Python批量文件查重：怎么高效遍历并比对文件夹内容

一、为什么直接计算哈希不是最优解：多级比对漏斗

很多初学者在编写查重脚本时，习惯遍历所有文件并直接调用 hashlib.md5() 或 hashlib.sha256()。在小规模测试时看似运转良好，但一旦面对含有几万张高清照片、安装包或视频的真实目录，整个系统的瓶颈几乎全在磁盘 I/O 上。

我们可以建立一个清晰的认知：两个内容完全相同的文件，其字节大小必须完全相同；但大小相同的文件，内容不一定相同。因此，文件大小是成本最低的第一道防线。

图 1：多级查重过滤漏斗架构图

二、高效遍历目录：为什么推荐 os.scandir

在 Python 中遍历文件夹通常有三种途径：传统递归 os.walk()、面向对象的 pathlib.Path.rglob()，以及 Python 3.5 引入的 os.scandir()。

在性能敏感的查重场景中，首选是 os.scandir()。原因在于操作系统的文件系统实现：

• 在 POSIX 系统与 Windows 系统中，读取目录项时，底层系统调用（如 Windows 的 FindNextFileW 或 Linux 的 readdir）其实已经携带了文件类型与部分属性。
• 传统的 os.walk() 在拿到文件名后，如果想要获取文件大小，还必须再次调用 os.stat(file_path)，这会导致二次系统调用。在遍历 10 万个文件时，就额外产生了 10 万次独立系统调用开销。
• os.scandir() 返回的是包含属性缓存的 DirEntry 对象。调用 entry.stat().st_size 在绝大多数平台上不需要额外的系统调用，遍历吞吐速度能提升 2 到 10 倍。

图 2：批量文件高效遍历与查重处理流转图

三、分级查重算法的完整 Python 实现

以下代码提供了一个完整、独立且生产可用的查重实现。代码完全基于 Python 标准库，无需安装任何额外依赖，能够优雅处理多级目录嵌套与潜在的权限异常。

import os
import hashlib
from collections import defaultdict
from typing import Dict, List

def scan_files_by_size(folder_path: str) -> Dict[int, List[str]]:
    """
    第一阶段：高效递归遍历目录，按文件大小聚类。
    利用 os.scandir 减少系统调用。
    """
    size_map = defaultdict(list)
    stack = [folder_path]

    while stack:
        current_dir = stack.pop()
        try:
            with os.scandir(current_dir) as entries:
                for entry in entries:
                    try:
                        # 忽略符号链接，避免环形遍历
                        if entry.is_symlink():
                            continue
                        if entry.is_dir(follow_symlinks=False):
                            stack.append(entry.path)
                        elif entry.is_file(follow_symlinks=False):
                            file_size = entry.stat().st_size
                            # 大小为 0 的空文件通常单独处理或直接忽略
                            if file_size > 0:
                                size_map[file_size].append(entry.path)
                    except (PermissionError, FileNotFoundError, OSError):
                        continue
        except (PermissionError, FileNotFoundError, OSError):
            continue

    # 仅保留存在潜在冲突的文件组（文件数 >= 2）
    return {size: paths for size, paths in size_map.items() if len(paths) > 1}

def get_chunk_hash(file_path: str, chunk_size: int = 4096) -> str:
    """
    第二阶段：仅读取文件头部数据，计算局部快速哈希。
    """
    hasher = hashlib.md5()
    try:
        with open(file_path, "rb") as f:
            chunk = f.read(chunk_size)
            hasher.update(chunk)
        return hasher.hexdigest()
    except (PermissionError, OSError):
        return ""

def get_full_hash(file_path: str, buffer_size: int = 65536) -> str:
    """
    第三阶段：流式读取全量文件，计算安全哈希摘要（SHA-256）。
    按 64KB 缓冲区分块读取，杜绝大文件爆内存。
    """
    hasher = hashlib.sha256()
    try:
        with open(file_path, "rb") as f:
            while True:
                data = f.read(buffer_size)
                if not data:
                    break
                hasher.update(data)
        return hasher.hexdigest()
    except (PermissionError, OSError):
        return ""

def find_duplicate_files(root_folder: str) -> Dict[str, List[str]]:
    """
    组合三级筛选机制的主函数：
    1. 大小聚合 -> 2. 首块哈希筛选 -> 3. 全量哈希确认
    返回格式：{全量哈希值: [重复文件路径1, 重复文件路径2, ...]}
    """
    print("[1/3] 正在遍历目录并按文件大小聚类...")
    size_candidates = scan_files_by_size(root_folder)
    print(f"      发现 {len(size_candidates)} 组相同大小的文件。")

    print("[2/3] 正在进行首块抽样哈希初筛...")
    sample_candidates = []
    for size, paths in size_candidates.items():
        chunk_map = defaultdict(list)
        for path in paths:
            c_hash = get_chunk_hash(path)
            if c_hash:
                chunk_map[c_hash].append(path)
        
        # 仅将头部哈希一致的组送入终验
        for c_hash, matched_paths in chunk_map.items():
            if len(matched_paths) > 1:
                sample_candidates.append(matched_paths)

    print(f"      通过首块初筛的文件组共有 {len(sample_candidates)} 组。")

    print("[3/3] 正在对候选组进行全量流式哈希终验...")
    final_duplicates = defaultdict(list)
    for paths in sample_candidates:
        full_map = defaultdict(list)
        for path in paths:
            f_hash = get_full_hash(path)
            if f_hash:
                full_map[f_hash].append(path)
        
        for f_hash, verified_paths in full_map.items():
            if len(verified_paths) > 1:
                final_duplicates[f_hash].extend(verified_paths)

    return dict(final_duplicates)

if __name__ == "__main__":
    target_directory = "/data/storage_pool"
    print(f"开始扫描目录: {target_directory}")
    duplicates = find_duplicate_files(target_directory)

    print("\n========= 查重比对结果 =========")
    total_wasted_groups = len(duplicates)
    print(f"共发现 {total_wasted_groups} 组完全重复的文件。")

    for idx, (digest, file_list) in enumerate(duplicates.items(), 1):
        file_size = os.path.getsize(file_list[0]) if os.path.exists(file_list[0]) else 0
        print(f"\n[组 {idx}] 哈希: {digest[:16]}... (单文件大小: {file_size / 1024:.2f} KB)")
        for file_path in file_list:
            print(f"  - {file_path}")

四、比对策略与性能收益对比

为了直观说明分级漏斗算法的优势，我们可以通过下表和性能对比图来观察在典型 100GB 混杂数据集（包含小文档、中等图片和数个大视频，共计约 50,000 个文件）下的资源占用差异。

图 3：传统全盘比对与多级哈希策略性能与 I/O 开销对比

五、安全处理重复文件：防御与去重策略

比对出重复文件往往只是第一步，真正的风险在于“如何处置重复文件”。在生产环境或个人数据归档中，直接执行 os.remove() 是极度危险的行为，容易因路径错误或依赖关联导致数据永久损坏。

建议遵循以下规范化的处理流转：

1. 审核第一，删除第二：不要在查重脚本中直接硬编码删除逻辑。应当先将查重结果导出为 JSON 或 CSV 清单，由人工复核或专门脚本处理。
2. 留存规则确定性：如果需要自动化清理，必须制定唯一的原文件保留规则（如：优先保留路径层级最短的文件、保留修改时间最早的文件，或保留特定主目录下的文件）。
3. 空间释放的高级替代方案——硬链接（Hard Link）：如果两个文件位于同一个磁盘分区，可以使用硬链接代替删除（os.link）。硬链接让两个文件路径指向磁盘底层的同一个 inode 数据块，既立竿见影释放了物理存储空间，又保证了原有路径的程序读取完全不受影响。

图 4：重复文件处置与安全防御决策流