在判断文档重复时，MD5 和 Datasketch（通常指 MinHash）是两种常用的方法，它们在原理和适用场景上存在显著差异。

MD5 哈希： MD5 是一种加密哈希函数，能将任意长度的输入映射为固定长度的哈希值。其特点是：

• 精确匹配：即使输入有微小差异，生成的哈希值也会完全不同。因此，MD5 适用于检测完全相同的文件或文本。
• 局限性：对于内容稍有改动的文档（例如，修改了格式、添加了标点等），MD5 无法识别其相似性，因为哈希值会完全不同。

Datasketch（MinHash）： MinHash 是一种局部敏感哈希（LSH）技术，专门用于估计集合之间的相似度，特别适用于文本去重和相似性检测。其特点是：

• 近似匹配：即使文档内容有细微差异，MinHash 仍能检测出它们的相似性，适用于内容相近但不完全相同的文本。
• 高效性：通过将文档转换为签名矩阵，MinHash 可以在大规模数据集中高效地进行相似性查询。

选择适用方法：

• 使用 MD5：当需要检测完全相同的文档时，MD5 是简单且高效的选择。
• 使用 MinHash（Datasketch）：当需要识别内容相似但不完全相同的文档时，MinHash 更为适用，特别是在处理大规模文本数据时。

综上，MD5 适用于精确匹配，而 MinHash（Datasketch）适用于近似匹配，选择哪种方法取决于具体的应用需求。

• 计算复杂度
  MD5 只需一次单一哈希运算，经过底层优化（甚至硬件加速），计算速度非常快。相比之下，MinHash 需要对每个文件计算多个哈希函数（例如 128 或更多），再结合集合操作来估计 Jaccard 相似度，这大大增加了计算量。
• 预处理开销
  使用 MinHash 通常需要先对文件内容进行分词、预处理等操作，这些步骤也会带来额外的时间消耗，而 MD5 则直接对文件的二进制数据进行运算。

MD5

import os
import hashlib
from typing import Dict, List

def calculate_md5(file_path: str) -> str:
    """
    计算文件的MD5哈希值
    
    Args:
        file_path (str): 文件路径
    
    Returns:
        str: 文件的MD5哈希值
    """
    hash_md5 = hashlib.md5()
    try:
        with open(file_path, "rb") as f:
            # 分块读取文件，避免大文件内存溢出
            for chunk in iter(lambda: f.read(4096), b""):
                print(chunk) 
                hash_md5.update(chunk)
        return hash_md5.hexdigest()
    except Exception as e:
        print(f"Error calculating MD5 for {file_path}: {e}")
        return ""

def find_duplicate_files(directory: str) -> Dict[str, List[str]]:
    """
    查找目录中的重复文件
    
    Args:
        directory (str): 要扫描的目录路径
    
    Returns:
        Dict[str, List[str]]: 重复文件的哈希值和对应的文件路径列表
    """
    # 存储文件哈希值和对应文件路径的字典
    hash_dict: Dict[str, List[str]] = {}
    
    # 遍历目录下的所有文件
    for root, _, files in os.walk(directory):
        for filename in files:
            file_path = os.path.join(root, filename)
            
            # 跳过目录和符号链接
            if not os.path.isfile(file_path):
                continue
            
            # 计算文件MD5
            file_hash = calculate_md5(file_path)
            
            # 如果哈希值已存在，添加到对应列表
            if file_hash:
                if file_hash in hash_dict:
                    hash_dict[file_hash].append(file_path)
                else:
                    hash_dict[file_hash] = [file_path]
    
    # 过滤出重复文件
    duplicate_files = {k: v for k, v in hash_dict.items() if len(v) > 1}
    
    return duplicate_files

def print_duplicate_files(duplicate_files: Dict[str, List[str]]):
    """
    打印重复文件信息
    
    Args:
        duplicate_files (Dict[str, List[str]]): 重复文件字典
    """
    if not duplicate_files:
        print("未找到重复文件")
        return
    
    print("找到以下重复文件组：")
    for hash_value, file_paths in duplicate_files.items():
        print(f"\nMD5: {hash_value}")
        for path in file_paths:
            print(f"  - {path}")
        print(f"  共 {len(file_paths)} 个重复文件")

# 使用示例
if __name__ == "__main__":
    # 指定要扫描的目录
    scan_directory = "D:\文档测试\并发测试"
    
    # 查找重复文件
    duplicate_files = find_duplicate_files(scan_directory)
    
    # 打印重复文件
    print_duplicate_files(duplicate_files)

Datasketch

• MinHash 用于生成用于比较相似度的紧凑签名。
• LSH（例如 MinHashLSH、MinHashLSHForest 等） 则利用这些签名构建索引，实现快速的相似性匹配。

from datasketch import MinHash, MinHashLSH
import requests
import json

def get_minhash(text, num_perm=128):
    """
    计算文本的 MinHash 签名
    """
    # 简单分词，可以根据实际情况使用更高级的分词工具（如 jieba）
    tokens = text.lower().split()
    m = MinHash(num_perm=num_perm)
    for token in tokens:
        m.update(token.encode('utf8'))
    return m

def get_content(path):
# 1. 通过接口获取文档内容
    url = "http://XXX.XXX.XXX.XXX:15554/zlzspt/dataprocess/document/txt_analysis"
    payload = {
        "doc_id": "None",
        "doc_title": "XXXX.pdf",
        "file_type": "pdf",
        "path": "path",
        "ocr_open": 1,
        "ocr_type": 0,
        "associate_filename": 1,
        "processes": [
            {"name": "traditional_simplified"},
            {"name": "sensitive_filter"},
            {"name": "remove_url"}
        ]
    }
    headers = {
        "accept": "application/json",
        "Content-Type": "application/json"
    }

    response = requests.post(url, headers=headers, json=payload)
# 假设返回的 JSON 中有一个字段 'text' 存储了处理后的文档文本
    doc_text = response.json().get("data")
    print(doc_text['content'])
    return doc_text['content']

# 创建 LSH 索引，阈值设为 0.8
lsh = MinHashLSH(threshold=0.9, num_perm=128)
path1 = "/root/TEST-01.pdf"
path2 = "/root/TEST-02.pdf"
path3 = "/root/TEST-03.pdf"

# 示例文档
doc1 = get_content(path1)
doc2 = get_content(path2)
doc3 = get_content(path3)

# 计算 MinHash 签名
m1 = get_minhash(doc1)
m2 = get_minhash(doc2)
m3 = get_minhash(doc3)

# 将文档插入 LSH 索引
lsh.insert("doc1", m1)
lsh.insert("doc3", m3)

# 查询 doc2 是否存在重复文档
result = lsh.query(m2)
if result:
    print(f"检测到重复或近似重复文档，相关文档ID：{result}")
else:
    print("未检测到重复文档")

持久化

import pickle
from datasketch import MinHash, MinHashLSH

# 构建 LSH 索引并插入数据
lsh = MinHashLSH(threshold=0.5, num_perm=128)

# 假设你已经创建了一些 MinHash 对象并插入
# lsh.insert("doc1", minhash1)
# lsh.insert("doc2", minhash2)
# ...

# 将 LSH 对象持久化到磁盘
with open('minhash_lsh.pkl', 'wb') as f:
    pickle.dump(lsh, f)

# 需要时再加载
with open('minhash_lsh.pkl', 'rb') as f:
    lsh = pickle.load(f)

# 加载后可以直接进行查询
result = lsh.query(minhash_query)
print(result)