类代码

import os
import pysftp
import fnmatch

class Synchronizer:
    def __init__(self, local_dir, remote_dir, host, username, password, ignore_list=None):
        self.local_dir = local_dir
        self.remote_dir = remote_dir
        self.host = host
        self.username = username
        self.password = password
        self.ignore_list = ignore_list or []

    # 初始化同步本地数据至远程
    def synchronize_to_remote(self):
        with pysftp.Connection(self.host, username=self.username, password=self.password) as sftp:
            self._sync_recursive_to_remote(sftp, self.local_dir, self.remote_dir)
    
    # 初始化同步远程数据至本地
    def synchronize_to_local(self):
        with pysftp.Connection(self.host, username=self.username, password=self.password) as sftp:
            self._sync_recursive_to_local(sftp, self.remote_dir, self.local_dir)

    # 同步本地数据至远程
    def _sync_recursive_to_remote(self, sftp, local_path, remote_path):
        local_files = os.listdir(local_path)

        for local_file in local_files:
            local_file_path = os.path.join(local_path, local_file)
            remote_file_path = os.path.join(remote_path, local_file)

            # 检查是否需要忽略该文件
            if self.should_ignore(local_file):
                continue

            # 判断远程目录是否存在,不存在则创建
            if os.path.isdir(local_file_path):
                try:
                    sftp.mkdir(remote_file_path)
                except:
                    pass

                # 递归同步子目录
                self._sync_recursive_to_remote(sftp, local_file_path, remote_file_path)
            else:
                # 如果文件不一致，上传
                if not self._is_same(sftp, local_file_path, remote_file_path):
                    self._upload_file(sftp, local_file_path, remote_file_path)


    # 同步远程数据至本地
    def _sync_recursive_to_local(self, sftp, remote_path, local_path):
        remote_files = sftp.listdir(remote_path)

        for remote_file in remote_files:
            remote_file_path = os.path.join(remote_path, remote_file)
            local_file_path = os.path.join(local_path, remote_file)

             # 检查是否需要忽略该文件
            if self.should_ignore(remote_file):
                continue

            # 判断本地目录是否存在,不存在则创建
            if sftp.isdir(remote_file_path):
                try:
                    os.makedirs(local_file_path)
                except:
                    pass

                self._sync_recursive_to_local(sftp, remote_file_path, local_file_path)
            else:
                if not self._is_same(sftp, local_file_path, remote_file_path):
                    self._download_file(sftp, remote_file_path, local_file_path)


    # 检查是否需要忽略文件
    def should_ignore(self, filename):
        for pattern in self.ignore_list:
            if fnmatch.fnmatch(filename, pattern):
                return True
        return False

    # 比较本地和远程文件是否一致
    def _is_same(self, sftp, local_file_path, remote_file_path):
        try:
            remote_stat = sftp.stat(remote_file_path)
            remote_mtime = int(remote_stat.st_mtime)
            remote_size = remote_stat.st_size
        except:
            return False

        if not os.path.exists(local_file_path):
            return False

        local_mtime = int(os.path.getmtime(local_file_path))
        local_size = os.path.getsize(local_file_path)
        
        # 判断修改时间和文件大小是否一致
        return remote_mtime == local_mtime and remote_size == local_size

    # 上传文件到远程服务器
    def _upload_file(self, sftp, local_file_path, remote_file_path):
        with sftp.cd(os.path.dirname(remote_file_path)):
            sftp.put(local_file_path)
            remote_stat = sftp.stat(remote_file_path)
            os.utime(local_file_path, (remote_stat.st_atime, remote_stat.st_mtime))
            self._print_sync_info(local_file_path, remote_file_path, "上传")

    # 下载远程服务器文件到本地
    def _download_file(self, sftp, remote_file_path, local_file_path):
        with sftp.cd(os.path.dirname(remote_file_path)):
            sftp.get(remote_file_path, local_file_path)
            remote_stat = sftp.stat(remote_file_path)
            os.utime(local_file_path, (remote_stat.st_atime, remote_stat.st_mtime))
            self._print_sync_info(remote_file_path, local_file_path, "下载")


    # 输出同步信息
    def _print_sync_info(self, source_path, destination_path, action):
        print(f"{action}: {source_path} -> {destination_path}")

使用方法

# 本地目录
local_dir = '/Users/test/'
# 远程目录
remote_dir = '/home/test/'
# 远程服务器ip
host = "10.0.0.1"
# 远程服务器用户名
username = "root"
# 远程服务器密码
password = "root"
# 需要排除的目录或文件,支持通配符*
ignore_list = [".*", "logs", "venv", "del_pycache*", "*cache*"]

# 初始化对象
synchronizer = Synchronizer(local_dir, remote_dir, host, username, password, ignore_list)

# 本地同步到远程
synchronizer.synchronize_to_remote()

# 远程同步到本地  
synchronizer.synchronize_to_local()

# 注意 synchronize_to_remote 和 synchronize_to_local 方法不要同时使用.