很多团队的文件同步仍停留在 rsync 或 scp -r 一把梭的阶段，一旦目标机器只开放了 SFTP 且没有 shell 权限，或者需要在同步前做业务侧的白名单、过滤、二次处理，命令行工具就不够用了。更麻烦的是全量覆盖：源端有几千个文件，每次同步都重传一遍，带宽和时间都被浪费。本文给出一个纯 Python 的方案——用 paramiko 建立 SFTP 会话，用 os.walk 递归枚举本地目录，把两侧的路径、大小、mtime 拉平成字典做差集运算，只传输真正变化的文件，并支持删除远端多余文件、dry-run 预演、并发上传和断线重试。
为什么不用 rsync：SFTP-only 场景的现实约束
rsync 的效率来自 delta 算法和远端 rsync 进程，但它的前提是两端都能执行 rsync 二进制。在以下场景中它直接失效：
- 跳板机或托管环境只给 SFTP subsystem，
ssh登录后exec被禁用，报错通常是Administratively prohibited: open failed或subsystem request failed on channel 0。 - 目标端是对象存储网关、文件交换平台提供的 SFTP 接口，只实现了 RFC 兼容的最小子集，不支持
rsync、tar之类的外部命令。 - 需要在同步过程中插入业务逻辑，例如只同步某个扩展名、文件名包含日期分区的、或者需要在上传前对文件做重命名与校验。
- 需要产出结构化的同步报告（哪些新增、哪些更新、哪些删除、哪些跳过），供 CI 或巡检系统消费。
Python 方案的另一层收益是可测试性：比对逻辑是纯函数，喂两个字典就能断言结果，不需要真的连服务器。
环境准备与 paramiko 的坑位
依赖只有 paramiko 一个，建议锁在 3.x：
python -m venv .venv
source .venv/bin/activate
pip install "paramiko>=3.4,<4"
python -c "import paramiko; print(paramiko.__version__)"
几个实际使用中高频出现的报错与处理方式：
| 报错信息 | 根因 | 解决办法 |
|---|---|---|
paramiko.ssh_exception.BadHostKeyException |
远端主机密钥与 known_hosts 记录不符 | 确认主机确实变更后删除对应行，或在首次连接时用 AutoAddPolicy 并记录指纹 |
paramiko.ssh_exception.AuthenticationException: Authentication failed. |
密码或密钥不被接受，常见于只允许 publickey 的服务器 | 显式传 key_filename，并确认私钥权限为 600 |
OSError: [Errno 36] File name too long |
paramiko 对超长路径的处理在部分版本有问题 | 使用 posixpath 拼接相对路径，避免把本地绝对路径带到远端 |
IOError: Failure |
远端目录不存在或权限不足，SFTP 只返回通用错误码 | 上传前逐级 sftp.stat 判断并 sftp.mkdir 补建 |
socket.timeout: timed out |
文件较大，默认无超时或网络抖动 | 设置 channel_timeout，并实现带退避的重试 |
关于主机密钥策略，生产脚本不要无脑 AutoAddPolicy，推荐加载本地 known_hosts，失败时打印指纹让人工确认。
建立连接：Transport、SFTPClient 与目录补建
先封装一个连接层，把认证、超时、重试都收进去。注意 Transport 手工管理比 SSHClient 更灵活，也更容易复用 channel。
import posixpath
import socket
import time
import paramiko
from paramiko.ssh_exception import SSHException
DEFAULT_PORT = 22
CONNECT_RETRIES = 3
BACKOFF_BASE = 2.0
def open_sftp(host, port, username, key_path=None, password=None,
known_hosts=None, timeout=20):
"""建立 SFTP 会话，返回 (transport, sftp)。调用方负责关闭。"""
last_err = None
for attempt in range(1, CONNECT_RETRIES + 1):
try:
transport = paramiko.Transport((host, port))
transport.banner_timeout = timeout
transport.auth_timeout = timeout
sock = transport.sock
sock.settimeout(timeout)
if key_path:
pkey = paramiko.RSAKey.from_private_key_file(key_path)
transport.connect(username=username, pkey=pkey)
else:
transport.connect(username=username, password=password)
if known_hosts:
key = transport.get_remote_server_key()
hostkey = known_hosts.lookup(host)
if hostkey is None or key != hostkey:
raise paramiko.BadHostKeyException(host, key, hostkey)
sftp = paramiko.SFTPClient.from_transport(transport)
sftp.get_channel().settimeout(timeout)
return transport, sftp
except (socket.timeout, SSHException, paramiko.SSHException) as exc:
last_err = exc
time.sleep(BACKOFF_BASE ** attempt)
raise RuntimeError(f"SFTP 连接失败: {host}:{port} -> {last_err}")
def ensure_remote_dir(sftp, remote_dir):
"""逐级创建远端目录，已存在则跳过。"""
parts = [p for p in remote_dir.strip("/").split("/") if p]
cur = "/" if remote_dir.startswith("/") else ""
for part in parts:
cur = posixpath.join(cur, part) if cur not in ("", "/") else "/" + part
try:
sftp.stat(cur)
except FileNotFoundError:
sftp.mkdir(cur)
ensure_remote_dir 里的 sftp.mkdir 不会自动建父目录，所以必须逐级判断。这一步是后续上传的前提，很多人只建一层然后在深层目录上撞到 IOError: Failure。
两侧清单采集：os.walk 与 sftp.listdir_attr 的对称处理
比对的核心是两侧数据结构必须一致。本地用 os.walk，远端用 sftp.listdir_attr 递归，统一产出 {相对路径: (size, mtime)} 的字典。
import os
import stat
EXCLUDE_DIRS = {".git", "__pycache__", ".mypy_cache", "node_modules"}
EXCLUDE_SUFFIX = {".pyc", ".swp", ".tmp"}
def should_skip(name, is_dir):
if is_dir and name in EXCLUDE_DIRS:
return True
return any(name.endswith(s) for s in EXCLUDE_SUFFIX)
def scan_local(root):
"""返回 {rel_path: (size, mtime)}，rel_path 使用 / 分隔。"""
result = {}
for dirpath, dirnames, filenames in os.walk(root):
dirnames[:] = [d for d in dirnames if not should_skip(d, True)]
for fname in filenames:
if should_skip(fname, False):
continue
abs_path = os.path.join(dirpath, fname)
if os.path.islink(abs_path):
continue
st = os.stat(abs_path)
rel = os.path.relpath(abs_path, root).replace(os.sep, "/")
result[rel] = (st.st_size, int(st.st_mtime))
return result
def scan_remote(sftp, root):
"""递归枚举远端目录。root 为远端绝对路径。"""
result = {}
stack = [""]
while stack:
rel_dir = stack.pop()
abs_dir = posixpath.join(root, rel_dir) if rel_dir else root
for attr in sftp.listdir_attr(abs_dir):
rel = posixpath.join(rel_dir, attr.filename) if rel_dir else attr.filename
is_dir = stat.S_ISDIR(attr.st_mode)
if should_skip(attr.filename, is_dir):
continue
if is_dir:
stack.append(rel)
elif stat.S_ISREG(attr.st_mode):
result[rel] = (attr.st_size, int(attr.st_mtime or 0))
return result
注意
attr.st_mtime可能是None（部分 SFTP 服务端不返回该字段），务必做or 0兜底，否则后续比较会抛TypeError。另外os.walk的dirnames[:]原地赋值才是真正的剪枝，写成dirnames = [...]只是丢弃了引用，子目录照样会被遍历。
差集比对：新增、更新、删除的判定规则
把两侧字典对照，得到四个集合。判定策略可配置——纯时间戳、大小优先、或大小加时间戳双条件。
MTIME_TOLERANCE = 2 # 秒，容忍文件系统精度差异
def diff_manifests(local, remote, mode="size_or_mtime"):
"""返回 (to_upload, to_delete, skipped, same)。"""
to_upload, skipped, same = [], [], []
for rel, (l_size, l_mtime) in local.items():
if rel not in remote:
to_upload.append(rel)
continue
r_size, r_mtime = remote[rel]
size_diff = l_size != r_size
mtime_diff = abs(l_mtime - r_mtime) > MTIME_TOLERANCE
if mode == "size_only":
changed = size_diff
elif mode == "mtime_only":
changed = mtime_diff
else:
changed = size_diff or mtime_diff
if changed:
to_upload.append(rel)
else:
same.append(rel)
to_delete = sorted(set(remote) - set(local))
return sorted(to_upload), to_delete, skipped, same
关于比较模式的选择，实践建议如下：
size_only：吞吐最大，适合追加写日志、压缩包这类内容变化必然带来大小变化的场景。缺点是同尺寸内容替换检测不到。mtime_only：适合文件系统 mtime 可靠的场景，例如本地产物由构建系统生成。size_or_mtime（默认）：最稳妥。绝大多数本地编辑器、构建工具在改动文件时都会刷新 mtime。
MTIME_TOLERANCE 设在 2 秒而不是 0，是因为 FAT/exFAT 的 mtime 精度为 2 秒，跨文件系统比较时若取 0 会导致大量文件被反复判定为”有变化”，形成每次全量重传的假象。
执行同步：上传、远端删除、dry-run 与并发
有了 to_upload 和 to_delete，剩下的就是执行。注意 SFTP 的 put 在目标目录已存在时直接覆盖，不需要先删。并发方面，paramiko 的单个 SFTPClient 不是线程安全的，要多线程就必须每个线程一个 Transport，或者干脆串行 + 大文件流式写入。
import shutil
def sync(sftp, local_root, remote_root, mode="size_or_mtime",
delete_extra=False, dry_run=False, exclude=None):
exclude = set(exclude or [])
local = {k: v for k, v in scan_local(local_root).items() if k not in exclude}
remote = scan_remote(sftp, remote_root)
to_upload, to_delete, _, same = diff_manifests(local, remote, mode)
report = {
"upload": to_upload,
"delete": to_delete if delete_extra else [],
"unchanged": len(same),
"dry_run": dry_run,
}
if dry_run:
return report
for rel in to_upload:
remote_path = posixpath.join(remote_root, rel)
ensure_remote_dir(sftp, posixpath.dirname(remote_path))
local_path = os.path.join(local_root, rel.replace("/", os.sep))
if os.path.getsize(local_path) > 64 * 1024 * 1024:
with open(local_path, "rb") as fh, sftp.open(remote_path, "wb") as rf:
rf.set_pipelined(True)
shutil.copyfileobj(fh, rf, 1024 * 1024)
else:
sftp.put(local_path, remote_path, confirm=True)
if delete_extra:
for rel in to_delete:
try:
sftp.remove(posixpath.join(remote_root, rel))
except FileNotFoundError:
pass
return report
几个容易踩的实现细节：
- 大文件用流式写入。默认
sftp.put会按块读取，8GB 以上文件建议手动copyfileobj并把set_pipelined(True)打开，能显著减少往返延迟。 - 删除远端多余文件前先做清理。远端空目录不会自动消失，如果要连目录一起清理，需要按深度倒序
rmdir。 - 写入过程中断会留下半截文件。稳妥做法是先传到
.tmp后缀，成功后sftp.rename原子替换。 - dry-run 必须真的不写。
scan_remote是只读操作，ensure_remote_dir必须在 dry-run 分支之后调用。
串起入口：CLI 参数、报告输出与退出码
把上面几块拼成可执行的脚本，参数用 argparse，报告输出 JSON 供流水线消费，退出码区分”有变更”与”无变更”。
import argparse
import json
import sys
def main(argv=None):
parser = argparse.ArgumentParser(description="基于时间戳比对的 SFTP 增量同步")
parser.add_argument("--host", required=True)
parser.add_argument("--port", type=int, default=DEFAULT_PORT)
parser.add_argument("--user", required=True)
parser.add_argument("--password")
parser.add_argument("--key", dest="key_path")
parser.add_argument("--local", required=True)
parser.add_argument("--remote", required=True)
parser.add_argument("--mode", choices=["size_only", "mtime_only", "size_or_mtime"],
default="size_or_mtime")
parser.add_argument("--delete-extra", action="store_true")
parser.add_argument("--dry-run", action="store_true")
args = parser.parse_args(argv)
transport = sftp = None
try:
transport, sftp = open_sftp(
args.host, args.port, args.user,
key_path=args.key_path, password=args.password,
)
report = sync(
sftp, args.local, args.remote,
mode=args.mode, delete_extra=args.delete_extra, dry_run=args.dry_run,
)
print(json.dumps(report, ensure_ascii=False, indent=2))
return 1 if report["upload"] or report["delete"] else 0
finally:
if sftp is not None:
sftp.close()
if transport is not None:
transport.close()
if __name__ == "__main__":
sys.exit(main())
典型调用：
python sftp_sync.py \
--host 10.20.31.17 --port 22 --user deploy \
--key ~/.ssh/id_ed25519 \
--local ./dist --remote /srv/www/static \
--mode size_or_mtime --delete-extra --dry-run
报告里 upload 与 delete 都是相对路径列表，直接可以贴进部署日志。退出码 1 表示有变更、0 表示无变更、连接失败或其它异常会抛到顶层由 shell 拿到非零码。
验证与压测：确认比对逻辑没有假阳性
比对逻辑是最该写单元测试的部分，因为它不依赖网络。给两个手工构造的 manifest 就能断言：
def test_size_or_mtime():
local = {"a.txt": (10, 1700000000), "b.txt": (20, 1700000005)}
remote = {"a.txt": (10, 1700000000), "b.txt": (20, 1700000000)}
up, dele, _, same = diff_manifests(local, remote, "size_or_mtime")
assert up == ["b.txt"]
assert same == ["a.txt"]
assert dele == []
def test_tolerance_absorbs_fat():
local = {"c.bin": (1024, 1700000000)}
remote = {"c.bin": (1024, 1700000001)}
up, _, _, same = diff_manifests(local, remote, "mtime_only")
assert up == [] and same == ["c.bin"]
第一次对生产目录运行务必加 --dry-run，观察 upload 列表规模是否符合预期。如果发现同一批文件每次都出现在 upload 里，八成是 mtime 精度或时区问题：检查远端 sftp.stat(path).st_mtime 的返回值是否与本机 os.stat 的量级一致，SFTP 协议本身返回的是 UNIX 时间戳，若服务端实现有偏差会体现在这里。
性能上，枚举 1 万个文件大约需要 10 到 30 秒（取决于 RTT 和目录深度），因为 listdir_attr 是逐目录一次往返。Transport 默认窗口 2MB，可按需调大：transport.default_window_size = 8 * 1024 * 1024 与 transport.packetizer.REKEY_BYTES = 1 << 30 对高延迟链路有帮助，但会放大内存占用，按需调整。
小结
整套方案的可迁移部分有三块：scan_local / scan_remote 这一对对称的清单采集器、diff_manifests 这个把两侧字典做差集的纯函数、以及一个能在 dry-run 与真实执行之间切换的 sync 调度层。SFTP 只是其中一种传输后端，如果目标端是 S3 或 WebDAV，只要替换 scan_remote 与上传删除的实现，比对逻辑和报告结构可以原样复用。
FAQ
为什么每次运行都把文件判定为”有变化”，反复重传？
按可能性排序：一是远端文件系统 mtime 精度较低（FAT/exFAT 为 2 秒，某些网络存储裁剪到秒），MTIME_TOLERANCE 设得太小；二是本地 os.stat 与远端 st_mtime 时区处理不一致；三是脚本每次上传都覆盖后触发了远端 mtime 刷新，而本地又没有对应变化。排查方法是打印三组值（本地 size/mtime、远端 size/mtime、差值），别只看结果。
上传时报 IOError: Failure 怎么定位？
SFTP 协议只返回通用错误码，客户端拿不到细节。按顺序检查：远端父目录是否真的存在（用 sftp.stat 而不是猜）、当前用户对父目录是否有写权限、磁盘配额是否已满、是否有路径长度限制。托管环境里最常见的是”目录存在但属主不是当前用户”，此时 mkdir 也会失败但报同样的错。
能不能多线程并发上传？
可以，但不能共享同一个 SFTPClient。正确做法是每个工作线程各自 Transport + SFTPClient，或者用独立进程池。并发数建议控制在 4 到 8，再多通常会被服务端的 MaxStartups 或 MaxSessions 限制，表现为 ChannelException: Administratively prohibited。
删除远端多余文件时把目录也删掉怎么办？
sync 里只对文件调用 sftp.remove，不碰目录。如果确实要清理空目录，需要额外收集目录集合，按路径深度从深到浅 rmdir，并且 rmdir 在目录非空时会静默失败，正好可以当作安全网。
如何在 CI 里安全地传私钥？
把私钥内容放进 CI 的 secret 变量，运行前写入临时文件并 chmod 600，脚本结束后删除。不要用 --password 明文参数，进程列表里能看到。若服务端支持，优先使用 short-lived 证书或 ssh-agent 转发。