如何写文件夹增量同步脚本

wen 实用脚本 2026-06-28 297

本文目录导读：

如何写文件夹增量同步脚本

基础Python增量同步脚本
高级版本（支持排除规则和日志）
定时同步脚本
使用建议

我来为你提供一个文件夹增量同步脚本的完整方案。

基础Python增量同步脚本

#!/usr/bin/env python3
"""
文件夹增量同步脚本
功能：将源文件夹中新增或修改的文件同步到目标文件夹
"""
import os
import hashlib
import shutil
import json
import time
from datetime import datetime
from pathlib import Path
class IncrementalSync:
    def __init__(self, source_dir, target_dir, state_file="sync_state.json"):
        """
        初始化同步器
        :param source_dir: 源文件夹路径
        :param target_dir: 目标文件夹路径
        :param state_file: 状态文件路径，用于记录同步状态
        """
        self.source_dir = Path(source_dir)
        self.target_dir = Path(target_dir)
        self.state_file = Path(state_file)
        self.sync_state = self._load_state()
    def _load_state(self):
        """加载同步状态"""
        if self.state_file.exists():
            with open(self.state_file, 'r') as f:
                return json.load(f)
        return {}
    def _save_state(self):
        """保存同步状态"""
        with open(self.state_file, 'w') as f:
            json.dump(self.sync_state, f, indent=2)
    def _get_file_hash(self, file_path):
        """计算文件的MD5哈希值"""
        hash_md5 = hashlib.md5()
        with open(file_path, "rb") as f:
            for chunk in iter(lambda: f.read(4096), b""):
                hash_md5.update(chunk)
        return hash_md5.hexdigest()
    def _get_file_info(self, file_path):
        """获取文件信息"""
        stat = file_path.stat()
        return {
            "size": stat.st_size,
            "mtime": stat.st_mtime,
            "hash": self._get_file_hash(file_path)
        }
    def _needs_sync(self, relative_path, file_info):
        """检查文件是否需要同步"""
        if relative_path not in self.sync_state:
            return True
        prev_state = self.sync_state[relative_path]
        # 检查文件大小或修改时间是否变化
        if (file_info["size"] != prev_state["size"] or 
            abs(file_info["mtime"] - prev_state["mtime"]) > 1):
            return True
        # 检查哈希值是否变化
        if file_info["hash"] != prev_state["hash"]:
            return True
        return False
    def sync(self):
        """执行同步操作"""
        print(f"开始同步: {self.source_dir} -> {self.target_dir}")
        print(f"时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        synced_count = 0
        skipped_count = 0
        error_count = 0
        # 确保目标目录存在
        self.target_dir.mkdir(parents=True, exist_ok=True)
        # 遍历源文件夹
        for source_path in self.source_dir.rglob("*"):
            if source_path.is_file():
                relative_path = str(source_path.relative_to(self.source_dir))
                target_path = self.target_dir / relative_path
                try:
                    # 获取文件信息
                    file_info = self._get_file_info(source_path)
                    # 检查是否需要同步
                    if self._needs_sync(relative_path, file_info):
                        # 创建目标文件夹
                        target_path.parent.mkdir(parents=True, exist_ok=True)
                        # 复制文件
                        shutil.copy2(source_path, target_path)
                        # 更新状态
                        self.sync_state[relative_path] = file_info
                        synced_count += 1
                        print(f"  ✓ 同步: {relative_path}")
                    else:
                        skipped_count += 1
                except Exception as e:
                    print(f"  ✗ 错误: {relative_path} - {str(e)}")
                    error_count += 1
        # 保存状态
        self._save_state()
        print(f"\n同步完成!")
        print(f"  新增/更新: {synced_count} 个文件")
        print(f"  跳过: {skipped_count} 个文件")
        print(f"  错误: {error_count} 个文件")
        return synced_count, skipped_count, error_count
    def get_summary(self):
        """获取同步摘要"""
        return {
            "source": str(self.source_dir),
            "target": str(self.target_dir),
            "last_sync": datetime.now().isoformat(),
            "synced_files": len(self.sync_state)
        }
# 使用示例
def main():
    # 配置同步参数
    source = "/path/to/source/folder"
    target = "/path/to/target/folder"
    # 创建同步器
    syncer = IncrementalSync(source, target, "sync_state.json")
    # 执行同步
    syncer.sync()
if __name__ == "__main__":
    main()

高级版本（支持排除规则和日志）

#!/usr/bin/env python3
"""
高级增量同步脚本 - 支持排除规则、日志和通知
"""
import os
import hashlib
import shutil
import json
import logging
from pathlib import Path
from datetime import datetime
import fnmatch
class AdvancedIncrementalSync:
    def __init__(self, source_dir, target_dir, config=None):
        """
        初始化高级同步器
        :param source_dir: 源文件夹
        :param target_dir: 目标文件夹
        :param config: 配置字典
            - exclude_patterns: 排除的文件模式列表
            - exclude_dirs: 排除的目录列表
            - max_file_size: 最大文件大小(字节)
            - backup: 是否备份被覆盖的文件
        """
        self.source_dir = Path(source_dir)
        self.target_dir = Path(target_dir)
        self.config = config or {}
        # 设置日志
        self._setup_logging()
        # 加载状态
        self.state_file = self.target_dir / ".sync_state.json"
        self.sync_state = self._load_state()
        # 默认排除规则
        self.exclude_patterns = self.config.get('exclude_patterns', [
            '*.tmp', '*.log', '.DS_Store', 'Thumbs.db'
        ])
        self.exclude_dirs = self.config.get('exclude_dirs', [
            '.git', '__pycache__', 'node_modules'
        ])
        self.max_file_size = self.config.get('max_file_size', 100 * 1024 * 1024)  # 100MB
        self.backup_enabled = self.config.get('backup', True)
    def _setup_logging(self):
        """设置日志"""
        log_file = self.target_dir / f"sync_{datetime.now().strftime('%Y%m%d')}.log"
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler(log_file),
                logging.StreamHandler()
            ]
        )
        self.logger = logging.getLogger(__name__)
    def _load_state(self):
        """加载同步状态"""
        if self.state_file.exists():
            try:
                with open(self.state_file, 'r') as f:
                    return json.load(f)
            except:
                return {}
        return {}
    def _save_state(self):
        """保存同步状态"""
        try:
            with open(self.state_file, 'w') as f:
                json.dump(self.sync_state, f, indent=2)
        except Exception as e:
            self.logger.error(f"保存状态文件失败: {e}")
    def _should_exclude(self, path):
        """检查路径是否应该被排除"""
        # 检查排除目录
        for part in path.parts:
            if part in self.exclude_dirs:
                return True
        # 检查排除模式
        for pattern in self.exclude_patterns:
            if fnmatch.fnmatch(path.name, pattern):
                return True
        return False
    def _get_file_info(self, file_path):
        """获取文件信息"""
        try:
            stat = file_path.stat()
            # 只读取小文件的哈希值，大文件使用修改时间比较
            if stat.st_size <= 10 * 1024 * 1024:  # 10MB以下计算哈希
                hash_value = self._calculate_hash(file_path)
            else:
                hash_value = None
            return {
                "size": stat.st_size,
                "mtime": stat.st_mtime,
                "hash": hash_value
            }
        except Exception as e:
            self.logger.error(f"读取文件信息失败 {file_path}: {e}")
            return None
    def _calculate_hash(self, file_path):
        """计算文件MD5哈希"""
        hash_md5 = hashlib.md5()
        try:
            with open(file_path, "rb") as f:
                for chunk in iter(lambda: f.read(8192), b""):
                    hash_md5.update(chunk)
            return hash_md5.hexdigest()
        except:
            return None
    def _backup_file(self, file_path):
        """备份即将被覆盖的文件"""
        if not self.backup_enabled:
            return
        backup_dir = self.target_dir / ".sync_backup"
        backup_dir.mkdir(exist_ok=True)
        relative_path = file_path.relative_to(self.target_dir)
        backup_path = backup_dir / f"{relative_path}_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
        backup_path.parent.mkdir(parents=True, exist_ok=True)
        try:
            shutil.copy2(file_path, backup_path)
            self.logger.info(f"已备份: {relative_path} -> {backup_path}")
        except Exception as e:
            self.logger.error(f"备份失败 {file_path}: {e}")
    def sync(self):
        """执行同步操作"""
        self.logger.info(f"开始同步: {self.source_dir} -> {self.target_dir}")
        stats = {"synced": 0, "deleted": 0, "errors": 0, "skipped": 0}
        # 确保目标目录存在
        self.target_dir.mkdir(parents=True, exist_ok=True)
        # 同步文件
        for source_path in self.source_dir.rglob("*"):
            if not source_path.is_file():
                continue
            # 检查排除规则
            if self._should_exclude(source_path):
                continue
            relative_path = str(source_path.relative_to(self.source_dir))
            target_path = self.target_dir / relative_path
            try:
                # 检查文件大小限制
                if source_path.stat().st_size > self.max_file_size:
                    self.logger.warning(f"跳过大文件: {relative_path}")
                    stats["skipped"] += 1
                    continue
                # 获取文件信息
                file_info = self._get_file_info(source_path)
                if not file_info:
                    stats["errors"] += 1
                    continue
                # 检查是否需要同步
                needs_sync = True
                if relative_path in self.sync_state:
                    prev_state = self.sync_state[relative_path]
                    if (prev_state.get("size") == file_info["size"] and
                        prev_state.get("mtime") == file_info["mtime"]):
                        needs_sync = False
                    # 如果有哈希值，进行精确比较
                    if (file_info["hash"] and prev_state.get("hash") and
                        file_info["hash"] == prev_state["hash"]):
                        needs_sync = False
                if needs_sync:
                    # 备份旧文件
                    if target_path.exists():
                        self._backup_file(target_path)
                    # 创建目标目录
                    target_path.parent.mkdir(parents=True, exist_ok=True)
                    # 复制文件
                    shutil.copy2(source_path, target_path)
                    # 更新状态
                    self.sync_state[relative_path] = file_info
                    stats["synced"] += 1
                    self.logger.info(f"同步: {relative_path}")
                else:
                    stats["skipped"] += 1
            except Exception as e:
                self.logger.error(f"同步失败 {relative_path}: {e}")
                stats["errors"] += 1
        # 清理目标文件夹中已删除的文件（可选）
        if self.config.get('delete_removed', False):
            for target_path in self.target_dir.rglob("*"):
                if not target_path.is_file():
                    continue
                relative_path = str(target_path.relative_to(self.target_dir))
                source_path = self.source_dir / relative_path
                if not source_path.exists():
                    try:
                        target_path.unlink()
                        self.sync_state.pop(relative_path, None)
                        stats["deleted"] += 1
                        self.logger.info(f"删除: {relative_path}")
                    except Exception as e:
                        self.logger.error(f"删除失败 {relative_path}: {e}")
        # 保存同步状态
        self._save_state()
        # 输出统计
        self.logger.info(f"同步完成!")
        self.logger.info(f"  同步: {stats['synced']} 个文件")
        self.logger.info(f"  跳过: {stats['skipped']} 个文件")
        self.logger.info(f"  删除: {stats['deleted']} 个文件")
        self.logger.info(f"  错误: {stats['errors']} 个文件")
        return stats
# 使用示例
def main():
    config = {
        'exclude_patterns': ['*.tmp', '*.log', '.DS_Store', 'Thumbs.db'],
        'exclude_dirs': ['.git', '__pycache__', 'node_modules'],
        'max_file_size': 500 * 1024 * 1024,  # 500MB
        'backup': True,
        'delete_removed': False  # 是否删除目标中已不存在的文件
    }
    syncer = AdvancedIncrementalSync(
        source_dir="/path/to/source",
        target_dir="/path/to/target",
        config=config
    )
    stats = syncer.sync()
    print(f"同步统计: {stats}")
if __name__ == "__main__":
    main()

定时同步脚本

#!/usr/bin/env python3
"""
定时增量同步脚本
"""
import schedule
import time
from datetime import datetime
def scheduled_sync():
    """定时执行同步"""
    print(f"\n=== 定时同步开始: {datetime.now()} ===")
    syncer = AdvancedIncrementalSync(
        source_dir="/path/to/source",
        target_dir="/path/to/target"
    )
    syncer.sync()
    print(f"=== 定时同步结束: {datetime.now()} ===\n")
def main():
    # 设置定时任务
    schedule.every().hour.do(scheduled_sync)  # 每小时同步一次
    # schedule.every().day.at("02:00").do(scheduled_sync)  # 每天凌晨2点同步
    # schedule.every(30).minutes.do(scheduled_sync)  # 每30分钟同步一次
    print("定时同步服务已启动...")
    # 执行一次立即同步
    scheduled_sync()
    # 持续运行
    while True:
        schedule.run_pending()
        time.sleep(60)  # 每分钟检查一次
if __name__ == "__main__":
    main()