!/usr/bin/env python3

-- coding: utf-8 --

Windows UTF-8 输出兼容

import sys
import io
if sys.platform == 'win32':
try:
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace')
sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8', errors='replace')
except Exception:
pass

"""
文档重复清理工具 (Document Duplicate Cleaner)
基于 SHA-256 内容哈希检测并清理重复文档
支持 .pdf, .docx, .xlsx, .pptx, .txt, .md, .csv, .json, .xml 等格式
"""

import os
import sys
import hashlib
import argparse
import re
from pathlib import Path
from collections import defaultdict
from datetime import datetime

支持的文件扩展名

SUPPORTED_EXTENSIONS = {
'.pdf', '.docx', '.doc', '.xlsx', '.xls', '.pptx', '.ppt',
'.txt', '.md', '.csv', '.json', '.xml', '.html', '.htm',
'.rtf', '.odt', '.ods', '.odp', '.eml', '.msg', '.rtf'
}

排除的目录

EXCLUDED_DIRS = {'.git', '.svn', '.hg', 'pycache', 'node_modules',
'.DS_Store', 'Thumbs.db', '.cache'}

def calculate_file_hash(filepath: Path, chunk_size: int = 65536) -> str:
"""计算文件内容的 SHA-256 哈希值"""
sha256 = hashlib.sha256()
try:
with open(filepath, 'rb') as f:
while chunk := f.read(chunk_size):
sha256.update(chunk)
return sha256.hexdigest()
except (IOError, OSError) as e:
print(f" [警告] 无法读取文件 {filepath}: {e}")
return ""

def scan_directory(root_path: Path, recursive: bool = True,
exclude_patterns: list = None,
file_extensions: set = None) -> dict:
"""
扫描目录中的所有支持的文件
返回 {文件路径: (哈希值, 文件大小, 修改时间)}
"""
if file_extensions is None:
file_extensions = SUPPORTED_EXTENSIONS

exclude_patterns = exclude_patterns or []
files_info = {}

# 将根目录转换为字符串用于比较
root_str = str(root_path.resolve())

for dirpath, dirnames, filenames in os.walk(root_path):
    # 过滤排除的目录（原地修改，防止 os.walk 进入）
    dirnames[:] = [d for d in dirnames if d not in EXCLUDED_DIRS]

    # 应用用户指定的排除模式
    for pattern in exclude_patterns:
        dirnames[:] = [d for d in dirnames if not re.search(pattern, d)]

    for filename in filenames:
        filepath = Path(dirpath) / filename

        # 检查扩展名
        ext = filepath.suffix.lower()
        if ext not in file_extensions:
            continue

        # 应用文件排除模式
        skip = False
        for pattern in exclude_patterns:
            if re.search(pattern, filename):
                skip = True
                break
        if skip:
            continue

        # 计算哈希
        file_hash = calculate_file_hash(filepath)
        if not file_hash:
            continue

        # 获取文件信息
        stat = filepath.stat()
        files_info[str(filepath)] = {
            'hash': file_hash,
            'size': stat.st_size,
            'mtime': stat.st_mtime,
            'name': filepath.name,
            'relative_path': str(filepath.relative_to(root_path))
        }

return files_info

def find_duplicates(files_info: dict) -> dict:
"""基于哈希值找出重复文件分组"""
hash_groups = defaultdict(list)

for filepath, info in files_info.items():
    hash_groups[info['hash']].append({
        'path': filepath,
        'size': info['size'],
        'mtime': info['mtime'],
        'name': info['name'],
        'relative_path': info['relative_path']
    })

# 只返回有重复的组（至少2个文件）
duplicates = {h: group for h, group in hash_groups.items() if len(group) > 1}
return duplicates

def format_size(size_bytes: int) -> str:
"""格式化文件大小"""
for unit in ['B', 'KB', 'MB', 'GB']:
if size_bytes < 1024:
return f"{size_bytes:.1f} {unit}"
size_bytes /= 1024
return f"{size_bytes:.1f} TB"

def format_time(timestamp: float) -> str:
"""格式化时间戳"""
return datetime.fromtimestamp(timestamp).strftime('%Y-%m-%d %H:%M:%S')

def print_duplicate_report(duplicates: dict, files_info: dict, root_path: Path):
"""打印重复文件报告"""
if not duplicates:
print("\n✅ 未发现重复文件！")
return

total_groups = len(duplicates)
total_duplicates = sum(len(group) - 1 for group in duplicates.values())
total_wasted = sum(
    group[0]['size'] * (len(group) - 1)
    for group in duplicates.values()
)

print(f"\n{'='*80}")
print(f"  📋 重复文件扫描报告")
print(f"{'='*80}")
print(f"  发现 {total_groups} 组重复文件，共 {total_duplicates} 个重复文件")
print(f"  可节省空间: {format_size(total_wasted)}")
print(f"{'='*80}\n")

for i, (file_hash, group) in enumerate(duplicates.items(), 1):
    # 按修改时间排序，最早的是原始文件
    group.sort(key=lambda x: x['mtime'])
    original = group[0]
    dupes = group[1:]

    print(f"\n{'─'*80}")
    print(f"  📁 分组 {i} | 哈希前16位: {file_hash[:16]}... | 文档数: {len(group)}")
    print(f"{'─'*80}")
    print(f"  📄 文件大小: {format_size(original['size'])}")
    print()

    # 显示原始文件
    print(f"  ✅ [原始] {original['relative_path']}")
    print(f"      修改时间: {format_time(original['mtime'])}")

    # 尝试预览内容
    preview = get_file_preview(Path(original['path']))
    if preview:
        print(f"      预览: {preview[:200]}")

    # 显示重复文件
    for dup in dupes:
        print(f"\n  ❌ [重复] {dup['relative_path']}")
        print(f"          修改时间: {format_time(dup['mtime'])}")

def get_file_preview(filepath: Path, max_chars: int = 200) -> str:
"""获取文件文本预览（仅适用于文本格式）"""
text_extensions = {'.txt', '.md', '.csv', '.json', '.xml', '.html', '.htm', '.eml'}
ext = filepath.suffix.lower()

if ext not in text_extensions:
    return None

try:
    # 尝试多种编码
    for enc in ['utf-8', 'gbk', 'gb2312', 'latin-1']:
        try:
            with open(filepath, 'r', encoding=enc) as f:
                content = f.read(max_chars)
                # 清理不可见字符
                content = re.sub(r'[\x00-\x08\x0b-\x0c\x0e-\x1f]', '', content)
                return content.strip()
        except (UnicodeDecodeError, IOError):
            continue
except Exception:
    pass
return None

def interactive_clean(duplicates: dict, dry_run: bool = True):
"""交互式清理重复文件"""
if not duplicates:
return 0

deleted_count = 0
deleted_size = 0
to_delete = []

print(f"\n{'='*80}")
print(f"  🗑️  清理模式: {'DRY-RUN（仅报告，不删除）' if dry_run else '实际删除'}")
print(f"{'='*80}\n")

all_groups = list(duplicates.items())

for i, (file_hash, group) in enumerate(all_groups, 1):
    group.sort(key=lambda x: x['mtime'])
    original = group[0]
    dupes = group[1:]

    print(f"\n{'─'*80}")
    print(f"  分组 {i}/{len(all_groups)} | 哈希: {file_hash[:16]}...")
    print(f"{'─'*80}")
    print(f"  保留: {original['relative_path']}")
    print(f"  重复文件列表:")
    for j, dup in enumerate(dupes, 1):
        print(f"    {j}. {dup['relative_path']} ({format_size(dup['size'])})")

    if dry_run:
        print(f"\n  [DRY-RUN] 将删除 {len(dupes)} 个文件，节省 {format_size(original['size'] * len(dupes))}")
        for dup in dupes:
            to_delete.append((dup['path'], dup['size']))
    else:
        print(f"\n  选择操作:")
        print(f"    1. 保留所有（不删除）")
        print(f"    2. 删除所有重复文件")
        print(f"    3. 选择性删除")

        choice = input(f"  请输入选择 (1/2/3) [默认 1]: ").strip() or "1"

        if choice == "2":
            for dup in dupes:
                to_delete.append((dup['path'], dup['size']))
        elif choice == "3":
            for j, dup in enumerate(dupes, 1):
                print(f"    {j}. {dup['relative_path']}")
            sel = input(f"  输入要删除的编号，用逗号分隔 [如 1,3]: ").strip()
            if sel:
                try:
                    indices = [int(x.strip()) - 1 for x in sel.split(',')]
                    for idx in indices:
                        if 0 <= idx < len(dupes):
                            to_delete.append((dupes[idx]['path'], dupes[idx]['size']))
                except ValueError:
                    print("  [跳过] 无效的输入")

# 执行删除
if to_delete:
    to_delete = list(set(to_delete))  # 去重
    print(f"\n{'='*80}")
    print(f"  ⚠️  即将 {'模拟' if dry_run else '删除'} {len(to_delete)} 个文件")
    print(f"{'='*80}")

    for path, size in to_delete:
        print(f"  {'将删除' if dry_run else '删除'}: {path}")

    if not dry_run:
        confirm = input(f"\n  ⚠️  确认删除 {len(to_delete)} 个文件？(y/N): ").strip()
        if confirm.lower() != 'y':
            print("  [取消] 删除操作已取消")
            return 0

    if dry_run:
        print(f"\n  [DRY-RUN] 模拟删除完成，共将删除 {len(to_delete)} 个文件")
        print(f"            将节省空间: {format_size(sum(s for _, s in to_delete))}")
    else:
        for path, size in to_delete:
            try:
                os.remove(path)
                deleted_count += 1
                deleted_size += size
                print(f"  ✅ 已删除: {path}")
            except OSError as e:
                print(f"  ❌ 删除失败: {path} - {e}")

        print(f"\n  📊 清理完成！已删除 {deleted_count} 个文件，节省 {format_size(deleted_size)}")

return deleted_count

def main():
parser = argparse.ArgumentParser(
description='文档重复清理工具 - 基于 SHA-256 检测并清理重复文档',
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
使用示例:
python duplicate_cleaner.py C:\Documents
python duplicate_cleaner.py . -r # 递归扫描当前目录
python duplicate_cleaner.py . --dry-run # 仅报告，不删除
python duplicate_cleaner.py . -e "temp|backup" # 排除匹配的文件/目录
python duplicate_cleaner.py . --delete # 实际删除（默认 dry-run）
"""
)

parser.add_argument('path', nargs='?', default='.',
                    help='要扫描的目录路径（默认: 当前目录）')
parser.add_argument('-r', '--recursive', action='store_true', default=True,
                    help='递归扫描子目录（默认: 开启）')
parser.add_argument('--no-recursive', action='store_true',
                    help='不扫描子目录')
parser.add_argument('-e', '--exclude', action='append',
                    help='排除匹配的文件/目录（支持正则，可用多次）')
parser.add_argument('--dry-run', action='store_true', default=True,
                    dest='dry_run',
                    help='仅报告，不删除（默认行为）')
parser.add_argument('--delete', action='store_false', default=True,
                    dest='dry_run',
                    help='实际删除重复文件（需确认）')
parser.add_argument('-o', '--output', metavar='FILE',
                    help='将报告保存到指定文件')
parser.add_argument('-q', '--quiet', action='store_true',
                    help='静默模式，仅输出关键信息')
parser.add_argument('--ext', metavar='EXT', action='append',
                    help='指定额外的文件扩展名（如 .pdf）')

args = parser.parse_args()

# 处理路径
scan_path = Path(args.path).resolve()
if not scan_path.exists():
    print(f"❌ 错误: 路径不存在: {scan_path}", file=sys.stderr)
    sys.exit(1)
if not scan_path.is_dir():
    print(f"❌ 错误: 路径不是目录: {scan_path}", file=sys.stderr)
    sys.exit(1)

# 配置
recursive = not args.no_recursive
exclude_patterns = args.exclude or []

# 添加扩展名
exts = SUPPORTED_EXTENSIONS.copy()
if args.ext:
    for e in args.ext:
        if not e.startswith('.'):
            e = '.' + e
        exts.add(e.lower())

print(f"\n🔍 正在扫描: {scan_path}")
if recursive:
    print(f"   模式: 递归扫描")
else:
    print(f"   模式: 仅当前目录")
if exclude_patterns:
    print(f"   排除: {', '.join(exclude_patterns)}")
print(f"   支持格式: {', '.join(sorted(exts))}")

# 扫描
files_info = scan_directory(scan_path, recursive, exclude_patterns, exts)
print(f"\n📄 共发现 {len(files_info)} 个文件")

if not files_info:
    print("❌ 未找到任何支持的文件格式")
    sys.exit(0)

# 找重复
duplicates = find_duplicates(files_info)

# 生成报告
import io
from contextlib import redirect_stdout

if args.output:
    # 保存到文件
    buffer = io.StringIO()
    with redirect_stdout(buffer):
        print_duplicate_report(duplicates, files_info, scan_path)
        if duplicates:
            deleted = interactive_clean(duplicates, args.dry_run)
    content = buffer.getvalue()
    with open(args.output, 'w', encoding='utf-8') as f:
        f.write(content)
    print(f"\n✅ 报告已保存到: {args.output}")
else:
    # 打印到控制台
    if not args.quiet:
        print_duplicate_report(duplicates, files_info, scan_path)

    if duplicates:
        deleted = interactive_clean(duplicates, args.dry_run)
        if args.dry_run:
            print(f"\n💡 提示: 以上为 Dry-Run 结果，如需实际删除，请使用 --delete 参数")
    else:
        print("\n✅ 未发现重复文件")

return 0

if name == 'main':
try:
sys.exit(main())
except KeyboardInterrupt:
print("\n\n[中断] 操作已取消")
sys.exit(130)