🐍 第25课：文件同步

—— 增量备份+校验：安全高效的数据保护

📌 本课目标

1️⃣ 增量备份原理

"""增量备份的核心思想：
- 全量备份：复制所有文件
- 增量备份：只复制自上次备份以来修改过的文件
- 判断依据：mtime（修改时间）+ size（文件大小）+ hash（内容校验）

判断文件是否修改的方法（由快到慢）：
1. mtime + size → 最快，99%场景足够
2. MD5 哈希 → 较慢，但更可靠
3. SHA256 哈希 → 最慢，安全性最高
"""

2️⃣ 文件哈希校验

import hashlib
from pathlib import Path

def file_hash(filepath, algorithm="md5"):
    """计算文件哈希值"""
    h = hashlib.new(algorithm)
    with open(filepath, "rb") as f:
        while chunk := f.read(8192):
            h.update(chunk)
    return h.hexdigest()

def verify_file(filepath, expected_hash, algorithm="md5"):
    """校验文件完整性"""
    actual = file_hash(filepath, algorithm)
    if actual == expected_hash:
        print(f"✅ {filepath}: 校验通过")
        return True
    else:
        print(f"❌ {filepath}: 校验失败!")
        print(f"  期望: {expected_hash}")
        print(f"  实际: {actual}")
        return False

# 示例
test_file = "/etc/hostname"
md5 = file_hash(test_file, "md5")
sha256 = file_hash(test_file, "sha256")
print(f"MD5:    {md5}")
print(f"SHA256: {sha256}")

# 性能对比
import time
for algo in ["md5", "sha1", "sha256"]:
    start = time.time()
    h = file_hash("/etc/hostname", algo)
    elapsed = time.time() - start
    print(f"{algo}: {h[:16]}... ({elapsed*1000:.1f}ms)")

3️⃣ 文件变更检测

import os
import hashlib
from pathlib import Path
from dataclasses import dataclass, field
from typing import Optional

@dataclass
class FileInfo:
    """文件信息快照"""
    path: str
    size: int
    mtime: float
    hash: Optional[str] = None

def get_file_info(filepath, compute_hash=False):
    """获取文件信息"""
    stat = os.stat(filepath)
    info = FileInfo(
        path=filepath,
        size=stat.st_size,
        mtime=stat.st_mtime,
    )
    if compute_hash:
        info.hash = file_hash(filepath)
    return info

def detect_changes(source_dir, snapshot_file):
    """检测文件变更"""
    import json
    source = Path(source_dir)
    
    # 加载上次快照
    old_snapshot = {}
    if os.path.exists(snapshot_file):
        with open(snapshot_file) as f:
            old_snapshot = json.load(f)
    
    # 扫描当前状态
    current_snapshot = {}
    changes = {"added": [], "modified": [], "deleted": []}
    
    for f in source.rglob("*"):
        if f.is_file():
            rel_path = str(f.relative_to(source))
            current_snapshot[rel_path] = {
                "size": f.stat().st_size,
                "mtime": f.stat().st_mtime,
            }
            
            if rel_path not in old_snapshot:
                changes["added"].append(rel_path)
            elif (current_snapshot[rel_path]["size"] != old_snapshot[rel_path]["size"] or
                  current_snapshot[rel_path]["mtime"] != old_snapshot[rel_path]["mtime"]):
                changes["modified"].append(rel_path)
    
    # 检测删除
    for rel_path in old_snapshot:
        if rel_path not in current_snapshot:
            changes["deleted"].append(rel_path)
    
    # 保存新快照
    with open(snapshot_file, "w") as f:
        json.dump(current_snapshot, f, indent=2)
    
    return changes

4️⃣ 增量备份实现

import os
import shutil
import hashlib
import json
from pathlib import Path
from datetime import datetime

class IncrementalBackup:
    """增量备份工具"""
    
    def __init__(self, source_dir, backup_dir):
        self.source = Path(source_dir)
        self.backup = Path(backup_dir)
        self.backup.mkdir(parents=True, exist_ok=True)
        self.manifest = self.backup / "manifest.json"
    
    def _load_manifest(self):
        """加载备份清单"""
        if self.manifest.exists():
            with open(self.manifest) as f:
                return json.load(f)
        return {"full_backup": None, "increments": [], "files": {}}
    
    def _save_manifest(self, manifest):
        """保存备份清单"""
        with open(self.manifest, "w") as f:
            json.dump(manifest, f, indent=2, ensure_ascii=False)
    
    def full_backup(self):
        """全量备份"""
        manifest = self._load_manifest()
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        backup_path = self.backup / f"full_{timestamp}"
        backup_path.mkdir()
        
        file_index = {}
        for f in self.source.rglob("*"):
            if f.is_file():
                rel = str(f.relative_to(self.source))
                dest = backup_path / rel
                dest.parent.mkdir(parents=True, exist_ok=True)
                shutil.copy2(f, dest)
                file_index[rel] = {
                    "size": f.stat().st_size,
                    "mtime": f.stat().st_mtime,
                    "hash": file_hash(str(f)),
                }
        
        manifest["full_backup"] = timestamp
        manifest["files"] = file_index
        self._save_manifest(manifest)
        print(f"✅ 全量备份完成: {len(file_index)} 个文件 → {backup_path}")
    
    def incremental_backup(self):
        """增量备份"""
        manifest = self._load_manifest()
        if not manifest["full_backup"]:
            print("❌ 请先执行全量备份!")
            return
        
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        inc_path = self.backup / f"inc_{timestamp}"
        inc_path.mkdir()
        
        changed_files = {}
        for f in self.source.rglob("*"):
            if f.is_file():
                rel = str(f.relative_to(self.source))
                current_mtime = f.stat().st_mtime
                current_size = f.stat().st_size
                
                old_info = manifest["files"].get(rel)
                is_changed = (
                    not old_info or
                    current_size != old_info["size"] or
                    current_mtime != old_info["mtime"]
                )
                
                if is_changed:
                    dest = inc_path / rel
                    dest.parent.mkdir(parents=True, exist_ok=True)
                    shutil.copy2(f, dest)
                    changed_files[rel] = {
                        "size": current_size,
                        "mtime": current_mtime,
                        "hash": file_hash(str(f)),
                    }
        
        manifest["increments"].append({
            "timestamp": timestamp,
            "files": list(changed_files.keys()),
        })
        manifest["files"].update(changed_files)
        self._save_manifest(manifest)
        print(f"✅ 增量备份完成: {len(changed_files)} 个文件变更 → {inc_path}")

5️⃣ 验证脚本

#!/usr/bin/env python3
"""第25课 文件同步验证"""
import hashlib
import tempfile
import os
import shutil
from pathlib import Path

def file_hash(filepath, algorithm="md5"):
    h = hashlib.new(algorithm)
    with open(filepath, "rb") as f:
        while chunk := f.read(8192):
            h.update(chunk)
    return h.hexdigest()

def test_file_hash():
    """文件哈希测试"""
    with tempfile.TemporaryDirectory() as tmpdir:
        test_file = os.path.join(tmpdir, "test.txt")
        with open(test_file, "w") as f:
            f.write("Hello, World!")
        h1 = file_hash(test_file)
        h2 = file_hash(test_file)
        assert h1 == h2  # 同一文件哈希相同
        assert len(h1) == 32  # MD5 是 32 位十六进制
    print("✅ 文件哈希测试通过")

def test_incremental_detection():
    """增量检测测试"""
    with tempfile.TemporaryDirectory() as tmpdir:
        source = Path(tmpdir) / "source"
        source.mkdir()
        
        # 创建初始文件
        (source / "a.txt").write_text("file A")
        (source / "b.txt").write_text("file B")
        
        # 第一次快照
        snapshot = {}
        for f in source.rglob("*"):
            if f.is_file():
                rel = str(f.relative_to(source))
                snapshot[rel] = {"size": f.stat().st_size, "mtime": f.stat().st_mtime}
        
        # 修改文件
        import time
        time.sleep(0.1)
        (source / "a.txt").write_text("file A modified")
        
        # 检测变更
        changes = []
        for f in source.rglob("*"):
            if f.is_file():
                rel = str(f.relative_to(source))
                current = {"size": f.stat().st_size, "mtime": f.stat().st_mtime}
                if rel in snapshot:
                    if current["size"] != snapshot[rel]["size"] or current["mtime"] != snapshot[rel]["mtime"]:
                        changes.append(rel)
        
        assert "a.txt" in changes
        assert "b.txt" not in changes
    print("✅ 增量检测测试通过")

def test_copy_preserves_metadata():
    """元数据保留测试"""
    with tempfile.TemporaryDirectory() as tmpdir:
        src = os.path.join(tmpdir, "src.txt")
        dst = os.path.join(tmpdir, "dst.txt")
        with open(src, "w") as f:
            f.write("test content")
        shutil.copy2(src, dst)
        assert os.path.getsize(src) == os.path.getsize(dst)
    print("✅ 元数据保留测试通过")

if __name__ == "__main__":
    test_file_hash()
    test_incremental_detection()
    test_copy_preserves_metadata()
    print("\n🎉 第25课全部验证通过！")

6️⃣ 双向同步与冲突处理

import os
import shutil
from pathlib import Path

class BidirectionalSync:
    """双向同步工具"""
    
    def __init__(self, dir_a, dir_b):
        self.dir_a = Path(dir_a)
        self.dir_b = Path(dir_b)
    
    def sync(self):
        """双向同步"""
        changes = {"a_to_b": [], "b_to_a": [], "conflicts": []}
        
        files_a = {str(f.relative_to(self.dir_a)): f for f in self.dir_a.rglob("*") if f.is_file()}
        files_b = {str(f.relative_to(self.dir_b)): f for f in self.dir_b.rglob("*") if f.is_file()}
        
        all_files = set(files_a.keys()) | set(files_b.keys())
        
        for rel in all_files:
            a_exists = rel in files_a
            b_exists = rel in files_b
            
            if a_exists and not b_exists:
                dst = self.dir_b / rel
                dst.parent.mkdir(parents=True, exist_ok=True)
                shutil.copy2(files_a[rel], dst)
                changes["a_to_b"].append(rel)
            elif b_exists and not a_exists:
                dst = self.dir_a / rel
                dst.parent.mkdir(parents=True, exist_ok=True)
                shutil.copy2(files_b[rel], dst)
                changes["b_to_a"].append(rel)
            elif a_exists and b_exists:
                mtime_a = files_a[rel].stat().st_mtime
                mtime_b = files_b[rel].stat().st_mtime
                if abs(mtime_a - mtime_b) < 1:
                    continue
                if mtime_a > mtime_b:
                    shutil.copy2(files_a[rel], self.dir_b / rel)
                    changes["a_to_b"].append(rel)
                else:
                    shutil.copy2(files_b[rel], self.dir_a / rel)
                    changes["b_to_a"].append(rel)
        
        return changes

7️⃣ 自动清理策略

import shutil
from pathlib import Path
from datetime import datetime

class BackupCleaner:
    """备份自动清理"""
    
    def __init__(self, backup_dir, retention=None):
        self.backup_dir = Path(backup_dir)
        self.policy = retention or {"daily": 7, "weekly": 4, "monthly": 12}
    
    def clean(self):
        """按策略清理旧备份"""
        now = datetime.now()
        deleted = 0
        kept = 0
        
        for item in sorted(self.backup_dir.iterdir()):
            if not item.is_dir():
                continue
            try:
                date_str = item.name.split("_")[-1]
                backup_date = datetime.strptime(date_str[:8], "%Y%m%d")
            except ValueError:
                continue
            
            age = (now - backup_date).days
            should_keep = False
            
            if age <= self.policy["daily"]:
                should_keep = True
            elif age <= 28:
                should_keep = backup_date.weekday() == 6
            elif age <= 365:
                should_keep = backup_date.day == 1
            
            if should_keep:
                kept += 1
            else:
                shutil.rmtree(item)
                deleted += 1
        
        print(f"清理完成: 保留{kept}, 删除{deleted}")
        return {"kept": kept, "deleted": deleted}

🔑 本课要点