Python文件完整性监控工具问题：需备份修改前原文件至安全目录

核心思路 #

要实现修改前文件的备份，核心是提前维护文件的快照副本与哈希记录：

• 初始化时，为目标目录的监控文件创建初始快照（保存当前版本）并记录哈希值
• 当检测到文件哈希变化（内容已修改）时，先将快照中的原文件复制到安全目录，再更新快照为当前修改后的版本

代码实现示例 #

1. 基础配置与初始化 #

import os
import hashlib
import shutil
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

# 自定义配置
TARGET_DIR = "/your/target/directory"  # 监控的目标目录
SAFE_BACKUP_DIR = "/your/secure/backup/directory"  # 存储原文件的安全目录
SNAPSHOT_DIR = "/your/snapshot/directory"  # 实时保存当前文件快照的目录
MONITORED_EXTENSIONS = ('.txt', '.docx', '.doc')  # 需要监控的文件类型

# 确保目录存在
for dir_path in [SAFE_BACKUP_DIR, SNAPSHOT_DIR]:
    os.makedirs(dir_path, exist_ok=True)

# 存储文件路径到哈希值的映射
file_hash_map = {}

2. 哈希计算与快照初始化 #

def calculate_file_hash(file_path):
    """计算文件的MD5哈希值，用于检测内容变更"""
    md5_hasher = hashlib.md5()
    with open(file_path, 'rb') as f:
        while chunk := f.read(4096):
            md5_hasher.update(chunk)
    return md5_hasher.hexdigest()

def init_snapshots():
    """初始化目标目录的文件快照与哈希记录"""
    for root, _, files in os.walk(TARGET_DIR):
        for file in files:
            if file.lower().endswith(MONITORED_EXTENSIONS):
                file_path = os.path.join(root, file)
                # 计算哈希并记录
                file_hash_map[file_path] = calculate_file_hash(file_path)
                # 复制文件到快照目录，保持原目录结构
                rel_file_path = os.path.relpath(file_path, TARGET_DIR)
                snapshot_file_path = os.path.join(SNAPSHOT_DIR, rel_file_path)
                os.makedirs(os.path.dirname(snapshot_file_path), exist_ok=True)
                shutil.copy2(file_path, snapshot_file_path)

3. 文件变更事件处理 #

class FileMonitorHandler(FileSystemEventHandler):
    def on_modified(self, event):
        if not event.is_directory:
            file_path = event.src_path
            # 仅处理指定类型的文件
            if file_path.lower().endswith(MONITORED_EXTENSIONS):
                current_hash = calculate_file_hash(file_path)
                # 对比哈希，确认文件内容已变更
                if file_path in file_hash_map and current_hash != file_hash_map[file_path]:
                    rel_file_path = os.path.relpath(file_path, TARGET_DIR)
                    # 从快照目录获取修改前的原文件
                    snapshot_file = os.path.join(SNAPSHOT_DIR, rel_file_path)
                    # 生成带时间戳的备份文件名，避免覆盖
                    backup_filename = f"{os.path.splitext(os.path.basename(file_path))[0]}_{int(os.path.getmtime(snapshot_file))}{os.path.splitext(file_path)[1]}"
                    safe_backup_path = os.path.join(SAFE_BACKUP_DIR, backup_filename)
                    
                    # 备份原文件到安全目录
                    shutil.copy2(snapshot_file, safe_backup_path)
                    print(f"已备份修改前的文件: {safe_backup_path}")
                    
                    # 更新快照与哈希记录，为下一次监控做准备
                    shutil.copy2(file_path, snapshot_file)
                    file_hash_map[file_path] = current_hash

4. 启动监控 #

if __name__ == "__main__":
    # 初始化快照
    init_snapshots()
    # 启动监控
    event_handler = FileMonitorHandler()
    observer = Observer()
    observer.schedule(event_handler, path=TARGET_DIR, recursive=True)
    observer.start()
    
    try:
        while True:
            pass
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

关键说明 #

• 快照目录：作为临时存储区，实时保存每个文件的最新版本，确保文件修改时能快速获取原文件
• 安全目录备份：通过时间戳命名备份文件，避免同名文件覆盖，方便后续追溯修改历史
• 性能优化：仅监控指定类型的文件，减少不必要的哈希计算与IO操作

内容的提问来源于stack exchange，提问作者Shivam Bhorde