如何写一个简易文件同步器

本文目录导读：

1. 基础文件同步器实现
2. 增强版：带命令行界面的同步器
3. 使用说明

我来帮你写一个简易文件同步器,这个工具可以在源目录和目标目录之间保持文件同步。

基础文件同步器实现

import os
import shutil
import hashlib
from pathlib import Path
import time
from datetime import datetime
class FileSync:
    def __init__(self, source_dir, target_dir):
        """
        初始化文件同步器
        Args:
            source_dir: 源目录路径
            target_dir: 目标目录路径
        """
        self.source_dir = Path(source_dir)
        self.target_dir = Path(target_dir)
    def get_file_hash(self, filepath):
        """
        计算文件的MD5哈希值，用于比较文件内容是否相同
        """
        hash_md5 = hashlib.md5()
        with open(filepath, "rb") as f:
            for chunk in iter(lambda: f.read(4096), b""):
                hash_md5.update(chunk)
        return hash_md5.hexdigest()
    def get_relative_paths(self, base_dir):
        """
        递归获取目录下所有文件的相对路径
        Returns:
            返回相对路径列表
        """
        paths = []
        for root, dirs, files in os.walk(base_dir):
            for file in files:
                full_path = Path(root) / file
                relative_path = full_path.relative_to(base_dir)
                paths.append(relative_path)
        return paths
    def sync_to_target(self):
        """
        将源目录同步到目标目录
        复制新增/更新的文件，删除目标目录中不存在于源目录的文件
        """
        # 确保源目录存在
        if not self.source_dir.exists():
            print(f"源目录不存在: {self.source_dir}")
            return
        # 创建目标目录
        self.target_dir.mkdir(parents=True, exist_ok=True)
        # 获取源目录和目标目录的文件列表
        source_files = set(self.get_relative_paths(self.source_dir))
        target_files = set(self.get_relative_paths(self.target_dir))
        # 记录统计信息
        copied = 0
        updated = 0
        deleted = 0
        # 1. 复制新增文件和更新已存在的文件
        for relative_path in source_files:
            source_file = self.source_dir / relative_path
            target_file = self.target_dir / relative_path
            # 创建必要的目录
            target_file.parent.mkdir(parents=True, exist_ok=True)
            # 如果文件不存在，直接复制
            if not target_file.exists():
                shutil.copy2(source_file, target_file)
                copied += 1
                print(f"新增: {relative_path}")
            else:
                # 文件存在，比较内容
                if self.get_file_hash(source_file) != self.get_file_hash(target_file):
                    shutil.copy2(source_file, target_file)
                    updated += 1
                    print(f"更新: {relative_path}")
                else:
                    print(f"相同: {relative_path}")
        # 2. 删除目标目录中不存在于源目录的文件
        files_to_delete = target_files - source_files
        for relative_path in files_to_delete:
            target_file = self.target_dir / relative_path
            if target_file.exists():
                target_file.unlink()  # 删除文件
                deleted += 1
                print(f"删除: {relative_path}")
        # 输出统计信息
        print(f"\n同步完成!")
        print(f"新增文件: {copied}")
        print(f"更新文件: {updated}")
        print(f"删除文件: {deleted}")
    def preview_changes(self):
        """
        预览将要执行的同步操作，不实际执行
        """
        if not self.source_dir.exists():
            print(f"源目录不存在: {self.source_dir}")
            return
        source_files = set(self.get_relative_paths(self.source_dir))
        target_files = set(self.get_relative_paths(self.target_dir)) if self.target_dir.exists() else set()
        print("=== 将要执行的操作 ===")
        # 预览新增/更新
        for relative_path in sorted(source_files):
            source_file = self.source_dir / relative_path
            target_file = self.target_dir / relative_path
            if not target_file.exists():
                print(f"[新增] {relative_path}")
            elif self.get_file_hash(source_file) != self.get_file_hash(target_file):
                print(f"[更新] {relative_path}")
        # 预览删除
        for relative_path in sorted(target_files - source_files):
            print(f"[删除] {relative_path}")
    def sync_from_target(self):
        """
        反向同步：从目标目录同步到源目录
        """
        # 交换源和目标
        temp = self.source_dir
        self.source_dir = self.target_dir
        self.target_dir = temp
        self.sync_to_target()
        # 恢复原目录
        self.source_dir, self.target_dir = self.target_dir, self.source_dir
# 使用示例
if __name__ == "__main__":
    # 创建同步器
    syncer = FileSync(
        source_dir="/path/to/source",  # 修改为你的源目录
        target_dir="/path/to/target"   # 修改为你的目标目录
    )
    # 预览将要执行的操作
    print("预览同步操作:")
    syncer.preview_changes()
    # 执行同步
    print("\n开始同步:")
    syncer.sync_to_target()

增强版：带命令行界面的同步器

import argparse
import logging
import json
from datetime import datetime
class EnhancedFileSync(FileSync):
    def __init__(self, source_dir, target_dir, log_file=None):
        super().__init__(source_dir, target_dir)
        # 设置日志
        self.setup_logging(log_file)
    def setup_logging(self, log_file):
        """设置日志记录"""
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler(log_file) if log_file else logging.NullHandler(),
                logging.StreamHandler()
            ]
        )
        self.logger = logging.getLogger(__name__)
    def sync_with_options(self, ignore_patterns=None, only_newer=True):
        """
        带更多选项的同步
        Args:
            ignore_patterns: 忽略的文件模式列表，如 ['*.tmp', '*.bak']
            only_newer: 只同步较新的文件
        """
        # 在get_relative_paths中添加过滤逻辑
        import fnmatch
        def get_filtered_paths(base_dir):
            paths = []
            for path in super().get_relative_paths():
                if ignore_patterns:
                    if any(fnmatch.fnmatch(str(path), pattern) for pattern in ignore_patterns):
                        continue
                paths.append(path)
            return paths
        self.source_files = set(get_filtered_paths(self.source_dir))
        self.target_files = set(get_filtered_paths(self.target_dir)) if self.target_dir.exists() else set()
        # 继续同步逻辑...
    def create_backup(self):
        """同步前创建备份"""
        backup_dir = self.target_dir.parent / f"{self.target_dir.name}_backup_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
        if self.target_dir.exists():
            shutil.copytree(self.target_dir, backup_dir)
            self.logger.info(f"已创建备份: {backup_dir}")
            return backup_dir
        return None
def main():
    parser = argparse.ArgumentParser(description='简易文件同步器')
    parser.add_argument('source', help='源目录路径')
    parser.add_argument('target', help='目标目录路径')
    parser.add_argument('--preview', action='store_true', help='预览模式，不执行实际同步')
    parser.add_argument('--reverse', action='store_true', help='反向同步')
    parser.add_argument('--backup', action='store_true', help='同步前备份')
    parser.add_argument('--ignore', nargs='+', help='忽略的文件模式')
    args = parser.parse_args()
    syncer = EnhancedFileSync(args.source, args.target)
    if args.preview:
        syncer.preview_changes()
    else:
        if args.backup:
            syncer.create_backup()
        if args.reverse:
            syncer.sync_from_target()
        else:
            syncer.sync_to_target()
if __name__ == "__main__":
    main()

使用说明

基本用法：

# 预览模式
python file_sync.py source_folder target_folder --preview
# 执行同步
python file_sync.py source_folder target_folder
# 反向同步（从目标到源）
python file_sync.py source_folder target_folder --reverse
# 带备份和忽略规则
python file_sync.py source_folder target_folder --backup --ignore "*.tmp" "*.bak"

主要功能：

1. 双向同步：源→目标或目标→源
2. 增量同步：只复制新增和修改的文件校验**：使用MD5哈希检查文件是否真的不同
3. 预览模式：执行前查看将要进行的操作
4. 备份功能：同步前自动备份目标目录

这个同步器适合个人使用，比如同步项目文件、备份重要数据等场景，需要更复杂的功能（如实时监控、加密等）可以在基础上扩展。