基于Python编写一个重复文件查找工具

目录

• 简介
• 功能介绍
• 应用场景
• 报错处理
• 代码实现
• 使用方法
• 基本使用
• 命令行参数说明
• 使用示例
• 总结

在长期使用计算机的过程中，我们经常会无意中积累大量重复的文件，这些重复文件不仅占用宝贵的存储空间，还会造成文件管理的混乱。手动查找和删除重复文件是一项繁琐且容易出错的任务。本文将介绍一个实用的Python脚本——重复文件查找工具，它可以通过计算文件的哈希值来准确识别重复文件，并提供多种处理选项。

这个重复文件查找工具具有以下核心功能：

• 精准识别：通过计算文件的MD5哈希值来准确识别重复文件
• 多目录扫描：支持同时扫描多个目录中的文件
• 多种匹配模式：支持按文件名、文件大小或内容哈希值查找重复文件
• 详细报告：生成详细的重复文件报告，包括文件路径、大小等信息
• 多种处理选项：提供删除、移动或符号链接等多种处理重复文件的方式
• 进度显示：实时显示扫描进度和统计信息
• 安全保护：提供预览模式，避免误删重要文件
• 日志记录：记录所有操作历史，便于追踪和审计

这个工具适用于以下场景：

• 存储空间清理：查找并删除重复文件以释放存储空间
• 文件整理：在合并多个文件夹时识别重复内容
• 备份管理：在备份文件中查找重复项以优化备份策略
• 照片管理：查找重复的照片文件，特别是从不同设备导入的照片
• 文档去重：在企业环境中查找重复的文档文件
• 下载文件夹清理：清理下载文件夹中的重复内容

脚本包含了完善的错误处理机制：

• 路径验证：检查指定目录是否存在且可访问
• 权限检测：检测文件读取权限，防止因权限不足导致的错误
• 文件锁定处理：处理被其他程序占用的文件
• 大文件处理：优化大文件的哈希计算过程，防止内存溢出
• 符号链接处理：正确处理符号链接，避免无限循环
• 异常捕获：捕获并处理运行过程中可能出现的各种异常

import os
import sys
import hashlib
import argparse
import json
from collections import defaultdict
from datetime import datetime
import shutil

class DuplicateFileFinder:
    def __init__(self, directories, mode='content'):
        self.directories = directories if isinstance(directories, list) else [directories]
        self.mode = mode  # 'name', 'size', or 'content'
        self.duplicates = defaultdict(list)
        self.scan_stats = {
            'total_files': 0,
            'total_size': 0,
            'scanned_dirs': 0,
            'errors': 0
        }
        
    def calculate_md5(self, filepath, chunk_size=8192):
        """计算文件的MD5哈希值"""
        md5_hash = hashlib.md5()
        try:
            with open(filepath, "rb") as f:
                # 分块读取文件以节省内存
                for chunk in iter(lambda: f.read(chunk_size), b""):
                    md5_hash.update(chunk)
            return md5_hash.hexdigest()
        except Exception as e:
            print(f"计算文件哈希值时出错 {filepath}: {e}")
            return None
            
    def get_file_info(self, filepath):
        """获取文件信息"""
        try:
            stat = os.stat(filepath)
            return {
                'path': filepath,
                'size': stat.st_size,
                'mtime': stat.st_mtime,
                'hash': None
            }
        except Exception as e:
            print(f"获取文件信息时出错 {filepath}: {e}")
            return None
            
    def scan_directory(self, directory):
        """扫描目录中的文件"""
        if not os.path.exists(directory):
            print(f"警告: 目录不存在 {directory}")
            self.scan_stats['errors'] += 1
            return
            
        if not os.path.isdir(directory):
            print(f"警告: 路径不是目录 {directory}")
            self.scan_stats['errors'] += 1
            return
            
        print(f"正在扫描目录: {directory}")
        
        try:
            for root, dirs, files in os.walk(directory):
                self.scan_stats['scanned_dirs'] += 1
                
                # 过滤掉隐藏目录
                dirs[:] = [d for d in dirs if not d.startswith('.')]
                
                for filename in files:
                    # 跳过隐藏文件
                    if filename.startswith('.'):
                        continue
                        
                    filepath = os.path.join(root, filename)
                    
                    # 获取文件信息
                    file_info = self.get_file_info(filepath)
                    if not file_info:
                        self.scan_stats['errors'] += 1
                        continue
                        
                    self.scan_stats['total_files'] += 1
                    self.scan_stats['total_size'] += file_info['size']
                    
                    # 根据模式确定键值
                    if self.mode == 'name':
                        key = filename.lower()
                    elif self.mode == 'size':
                        key = file_info['size']
                    else:  # content
                        file_hash = self.calculate_md5(filepath)
                        if file_hash is None:
                            self.scan_stats['errors'] += 1
                            continue
                        file_info['hash'] = file_hash
                        key = file_hash
                        
                    self.duplicates[key].append(file_info)
                    
        except Exception as e:
            print(f"扫描目录时出错 {directory}: {e}")
            self.scan_stats['errors'] += 1
            
    def find_duplicates(self):
        """查找重复文件"""
        print("开始查找重复文件...")
        
        for directory in self.directories:
            self.scan_directory(directory)
            
        # 过滤出真正的重复文件（至少2个相同的）
        duplicates = {k: v for k, v in self.duplicates.items() if len(v) > 1}
        self.duplicates = duplicates
        
        print(f"n扫描完成:")
        print(f"  扫描目录数: {self.scan_stats['scanned_dirs']}")
        print(f"  总文件数: {self.scan_stats['total_files']}")
        print(f"  总大小: {self.scan_stats['total_size'] / (1024*1024):.2f} MB")
        print(f"  发现重复组: {len(self.duplicates)}")
        print(f"  错误数: {self.scan_stats['errors']}")
        
        return self.duplicates
        
    def print_duplicates(self, max_files_per_group=10):
        """打印重复文件信息"""
        if not self.duplicates:
            print("未发现重复文件")
            return
            
        print(f"n发现 {len(self.duplicates)} 组重复文件:")
        print("=" * 80)
        
        for i, (key, files) in enumerate(self.duplicates.items(), 1):
            print(f"n[{i}] {len(files)} 个重复文件:")
            
            # 显示文件信息
            total_size = files[0]['size']
            print(f"  文件大小: {total_size / 1024:.2f} KB")
            
            if self.mode == 'content' and files[0]['hash']:
                print(f"  文件哈希: {files[0]['hash'][:16]}...")
                
            print("  文件列表:")
            for j, file_info in enumerate(files[:max_files_per_group]):
                print(f"    {j+1}. {file_info['path']}")
                
            if len(files) > max_files_per_group:
                print(f"    ... 还有 {len(files) - max_files_per_group} 个文件")
                
    def save_report(self, report_file="duplicate_report.json"):
        """保存重复文件报告"""
        try:
            report_data = {
                'scan_time': datetime.now().isoformat(),
                'directories': self.directories,
                'mode': self.mode,
                'stats': self.scan_stats,
                'duplicates': {}
            }
            
            # 转换文件信息为可序列化的格式
            for key, files in self.duplicates.items():
                report_data['duplicates'][str(key)] = [
                    {
                        'path': f['path'],
                        'size': f['size'],
                        'modified': datetime.fromtimestamp(f['mtime']).isoformat() if 'mtime' in f else None,
                        'hash': f.get('hash')
                    }
                    for f in files
                ]
                
            with open(report_file, 'w', encoding='utf-8') as f:
                json.dump(report_data, f, indent=2, ensure_ascii=False)
                
            print(f"重复文件报告已保存到: {report_file}")
            return True
        except Exception as e:
            print(f"保存报告时出错: {e}")
            return False
            
    def remove_duplicates(self, keep_strategy='first', dry_run=True):
        """删除重复文件"""
        if not self.duplicates:
            print("没有重复文件需要处理")
            return
            
        action = "预览" if dry_run else "删除"
        print(f"n{action}重复文件 (保留策略: {keep_strategy}):")
        print("=" * 60)
        
        removed_count = 0
        removed_size = 0
        
        for key, files in self.duplicates.items():
            # 确定要保留的文件
            if keep_strategy == 'first':
                keep_index = 0
            elif keep_strategy == 'last':
                keep_index = -1
            elif keep_strategy == 'largest':
                keep_index = max(range(len(files)), key=lambda i: files[i]['size'])
            elif keep_strategy == 'smallest':
                keep_index = min(range(len(files)), key=lambda i: files[i]['size'])
            else:
                keep_index = 0  # 默认保留第一个
                
            keep_file = files[keep_index]
            print(f"n保留: {keep_file['path']}")
            
            # 处理其他重复文件
            for i, file_info in enumerate(files):
                if i == keep_index:
                    continue
                    
                filepath = file_info['path']
                filesize = file_info['size']
                
                if dry_run:
                    print(f"  将删除: {filepath} ({filesize / 1024:.2f} KB)")
                else:
                    try:
                        os.remove(filepath)
                        print(f"  已删除: {filepath}")
                        removed_count += 1
                        removed_size += filesize
                    except Exception as e:
                        print(f"  删除失败 {filepath}: {e}")
                        
        if not dry_run:
            print(f"n删除完成: {removed_count} 个文件, 释放空间 {removed_size / (1024*1024):.2f} MB")
            
    def move_duplicates(self, target_dir, keep_strategy='first', dry_run=True):
        """移动重复文件到指定目录"""
        if not self.duplicates:
            print("没有重复文件需要处理")
            return
            
        # 创建目标目录
        if not os.path.exists(target_dir):
            try:
                os.makedirs(target_dir)
                print(f"创建目录: {target_dir}")
            except Exception as e:
                print(f"创建目录失败 {target_dir}: {e}")
                return
                
        action = "预览" if dry_run else "移动"
        print(f"n{action}重复文件到 {target_dir} (保留策略: {keep_strategy}):")
        print("=" * 60)
        
        moved_count = 0
        
        for key, files in self.duplicates.items():
            # 确定要保留的文件
            if keep_strategy == 'first':
                keep_index = 0
            elif keep_strategy == 'last':
                keep_index = -1
            else:
                keep_index = 0  # 默认保留第一个
                
            keep_file = files[keep_index]
            print(f"n保留: {keep_file['path']}")
            
            # 处理其他重复文件
            for i, file_info in enumerate(files):
                if i == keep_index:
                    continue
                    
                filepath = file_info['path']
                filename = os.path.basename(filepath)
                
                # 构造目标路径
                target_path = os.path.join(target_dir, filename)
                # 如果目标文件已存在，添加序号
                counter = 1
                base_name, ext = os.path.splitext(filename)
                while os.path.exists(target_path):
                    new_name = f"{base_name}_{counter}{ext}"
                    target_path = os.path.join(target_dir, new_name)
                    counter += 1
                    
                if dry_run:
                    print(f"  将移动: {filepath} -> {target_path}")
                else:
                    try:
                        shutil.move(filepath, target_path)
                        print(f"  已移动: {filepath} -> {target_path}")
                        moved_count += 1
                    except Exception as e:
                        print(f"  移动失败 {filepath}: {e}")
                        
        if not dry_run:
            print(f"n移动完成: {moved_count} 个文件")

def main():
    parser = argparse.ArgumentParser(description="重复文件查找工具")
    parser.add_argument("directories", nargs='+', help="要扫描的目录路径")
    parser.add_argument("-m", "--mode", choices=['name', 'size', 'content'], 
                       default='content', help="匹配模式 (默认: content)")
    parser.add_argument("-r", "--report", help="保存报告到指定文件")
    parser.add_argument("--remove", action="store_true", help="删除重复文件")
    parser.add_argument("--move", help="移动重复文件到指定目录")
    parser.add_argument("--keep", choices=['first', 'last', 'largest', 'smallest'], 
                       default='first', help="保留策略 (默认: first)")
    parser.add_argument("--dry-run", action="store_true", help="预览模式，不执行实际操作")
    
    args = parser.parse_args()
    
    try:
        finder = DuplicateFileFinder(args.directories, args.mode)
        duplicates = finder.find_duplicates()
        
        if duplicates:
            finder.print_duplicates()
            
            if args.report:
                finder.save_report(args.report)
                
            if args.remove:
                finder.remove_duplicates(args.keep, dry_run=args.dry_run)
                
            if args.move:
                finder.move_duplicates(args.move, args.keep, dry_run=args.dry_run)
        else:
            print("未发现重复文件")
            
    except KeyboardInterrupt:
        print("nn用户中断操作")
    except Exception as e:
        print(f"程序执行出错: {e}")
        sys.exit(1)

if __name__ == "__main__":
    main()

# 基本用法，扫描单个目录
python duplicate_finder.py /path/to/directory

# 扫描多个目录
python duplicate_finder.py /path/to/dir1 /path/to/dir2 /path/to/dir3

# 按文件名查找重复文件
python duplicate_finder.py /path/to/directory -m name

# 按文件大小查找重复文件
python duplicate_finder.py /path/to/directory -m size

# 保存报告到文件
python duplicate_finder.py /path/to/directory -r report.json

# 预览删除操作（不实际删除）
python duplicate_finder.py /path/to/directory --remove --dry-run

# 删除重复文件（保留第一个）
python duplicate_finder.py /path/to/directory --remove

# 删除重复文件（保留最大的）
python duplicate_finder.py /path/to/directory --remove --keep largest

# 移动重复文件到指定目录
python duplicate_finder.py /path/to/directory --move /path/to/duplicates

• directories: 必需参数，指定要扫描的一个或多个目录路径
• -m, --mode: 匹配模式，可选 name(按文件名)、size(按文件大小)、content(按文件内容，默认)
• -r, --report: 保存详细报告到指定的JSON文件
• --remove: 删除重复文件
• --move: 移动重复文件到指定目录
• --keep: 保留策略，可选 first(第一个，默认)、last(最后一个)、largest(最大的)、smallest(最小的)
• --dry-run: 预览模式，只显示将要执行的操作，不实际执行

假设有以下文件结构：

/photos/
  vacation1.jpg (重复)
  vacation2.jpg
  family.jpg (重复)
  work/
    vacation1.jpg (重复)
    presentation.ppt
/downloads/
  family.jpg (重复)
  document.pdf

执行命令：

python duplicate_finder.py /photos /downloads -r duplicates.json

输出结果：

发现 2 组重复文件:

[1] 2 个重复文件:
  文件大小: 2048.00 KB
  文件哈希: abc123def456…
  文件列表:
    1. /photos/vacation1.jpg
    2. /photos/work/vacation1.jpg

[2] 2 个重复文件:
  文件大小: 1024.00 KB
  文件哈希: def456ghi789…
  文件列表:
    1. /photos/family.jpg
    2. /downloads/family.jpg

这个重复文件查找工具通过计算文件的MD5哈希值来准确识别重复文件，提供了多种匹配模式和处理选项。它不仅能帮助用户找出占用存储空间的重复文件，还提供了安全的处理方式，包括预览模式、保留策略选择等。工具生成的详细报告可以帮助用户更好地了解重复文件的情况。无论是个人用户清理存储空间，还是企业用户管理大量文件，这个工具都能提供有效的帮助。

以上就是基于Python编写一个重复文件查找工具的详细内容，更多关于Python查找重复文件的资料请关注风君子博客其它相关文章！

您可能感兴趣的文章:

• 使用Python编写文件重复检查器的完整代码
• 详解如何使用Python实现删除重复文件
• Python实现清除文件夹中重复视频
• Python自动化办公之清理重复文件详解
• 教你用Python寻找重复文件并删除的脚本写法
• python实现自动清理重复文件
• Python实现检测文件的MD5值来查找重复文件案例
• python如何删除文件中重复的字段