本文目录导读：
我来帮你写一个简易文件同步器,这个工具可以在源目录和目标目录之间保持文件同步。
基础文件同步器实现
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"
主要功能：
- 双向同步：源→目标或目标→源
- 增量同步：只复制新增和修改的文件校验**：使用MD5哈希检查文件是否真的不同
- 预览模式：执行前查看将要进行的操作
- 备份功能：同步前自动备份目标目录
这个同步器适合个人使用，比如同步项目文件、备份重要数据等场景，需要更复杂的功能（如实时监控、加密等）可以在基础上扩展。
