为何重复文件生成不同哈希？百万级文件查重优化求助

问题背景 #

需要检测包含数百万文件的文件夹中的重复文件与损坏文件：

• 初始方法读取完整文件内容计算哈希，但大文件耗时过长；
• 尝试改为对文件路径字符串计算哈希，却发现重复文件哈希值不同，无法识别重复。

错误原因分析 #

第二种方法的核心错误是对文件路径字符串而非文件内容计算哈希：

• 即使两个文件内容完全一致，只要文件路径不同（如不同目录下的同名文件），路径字符串的哈希值就会不同；
• 这种方式完全无法反映文件内容的一致性，自然不能识别重复文件。

高效正确的解决方案 #

要兼顾速度与正确性，需从以下几个方向优化：

1. 分块读取计算哈希 #

避免一次性读取大文件到内存，分块读取并更新哈希值，既节省内存，又不影响哈希结果的正确性。

2. 先按文件大小过滤 #

大小不同的文件必然不是重复文件，先按文件大小分组，仅对大小相同的文件计算哈希，大幅减少哈希计算的工作量。

3. 批量处理数据 #

不要逐行向DataFrame添加数据，先收集结果列表，最后一次性转换为DataFrame，提升数据写入效率。

4. 多线程并行处理 #

文件IO属于IO密集型操作，使用多线程并行处理多个文件，能显著提升整体处理速度。

优化后代码示例 #

import hashlib
import os
import pandas as pd
from concurrent.futures import ThreadPoolExecutor

def calculate_file_hash(file_path, chunk_size=4*1024*1024):
    """分块计算文件的blake2b哈希值，返回(路径, 哈希值, 错误信息)"""
    hash_obj = hashlib.blake2b()
    try:
        with open(file_path, "rb") as f:
            # 分块读取文件内容更新哈希
            while chunk := f.read(chunk_size):
                hash_obj.update(chunk)
        return file_path, hash_obj.hexdigest(), None
    except Exception as e:
        # 捕获文件读取错误，标记为损坏文件
        return file_path, None, str(e)

def main(root_dir):
    # 收集所有文件的完整路径
    all_files = []
    for root, _, files in os.walk(root_dir):
        for file in files:
            full_path = os.path.join(root, file)
            all_files.append(full_path)
    
    # 按文件大小分组，过滤不可能重复的文件
    size_groups = {}
    for file_path in all_files:
        try:
            file_size = os.path.getsize(file_path)
            size_groups.setdefault(file_size, []).append(file_path)
        except Exception as e:
            print(f"无法获取文件大小：{file_path} | 错误：{e}")
            continue
    
    hash_results = []
    error_files = []
    
    # 多线程并行处理大小相同的文件（这些才可能重复）
    with ThreadPoolExecutor(max_workers=os.cpu_count()*2) as executor:
        futures = []
        for size, files in size_groups.items():
            if len(files) < 2:
                continue
            for file in files:
                futures.append(executor.submit(calculate_file_hash, file))
        
        # 收集多线程处理结果
        for future in futures:
            path, hash_val, error = future.result()
            if error:
                error_files.append((path, error))
            else:
                hash_results.append({"Path": path, "Hash": hash_val})
    
    # 处理单个文件的哈希（用于检测损坏）
    for size, files in size_groups.items():
        if len(files) == 1:
            path = files[0]
            _, hash_val, error = calculate_file_hash(path)
            if error:
                error_files.append((path, error))
            else:
                hash_results.append({"Path": path, "Hash": hash_val})
    
    # 转换为DataFrame并识别重复文件
    hash_files = pd.DataFrame(hash_results)
    duplicate_files = hash_files[hash_files.duplicated(subset="Hash", keep=False)]
    
    # 输出结果
    print("=== 重复文件列表 ===")
    print(duplicate_files)
    print("\n=== 损坏/无法读取的文件 ===")
    for path, error in error_files:
        print(f"{path} | 错误：{error}")
    
    return hash_files, duplicate_files, error_files

if __name__ == "__main__":
    rootdir = "//?/Z:/"
    hash_files, duplicates, errors = main(rootdir)

额外说明 #

• 可根据硬件调整chunk_size（如8MB）和线程数max_workers；
• 如果需要检测文件损坏，可对单个文件的哈希进行记录，后续若文件哈希变化则说明损坏；
• 若文件数量极多，可考虑将结果分批写入磁盘，避免内存占用过高。

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