薪源論壇

代碼: 選擇全部

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
跨多個磁碟區（NAS 掛載點 + 外接硬碟）依「檔案內容」（MD5 雜湊值）找出重複檔案，
並把每一份多餘的副本搬到該檔案所在磁碟區第一層目錄下的
《<磁碟區名稱>_duFiles》資料夾中，同時保留原本的相對路徑。

絕對不會刪除任何檔案——只會搬移。可以安全地重複執行：先前已經搬移過的檔案，
因為在原本路徑已經不存在了，會自動被跳過。

保留規則：
  - 預設保留「NAS」磁碟區（可用 --nas-root 指定多個）上的檔案。
  - 只有當「NAS 磁碟區內部」自己也有重複檔案時，才會在 NAS 內部保留最舊的一份、
    搬移其餘的重複檔案。
  - 非 NAS 磁碟區（例如外接硬碟）若和 NAS 上的檔案重複，其副本一律視為多餘、
    全部搬移到自己磁碟區底下的 duFiles 資料夾。
  - 若重複檔案只出現在非 NAS 磁碟區（NAS 上沒有這份檔案），則在該磁碟區內部
    保留最舊的一份，其餘搬移。

用法：
    python3 dedupe_nas.py \
        --nas-root "home=/Volumes/home" \
        --nas-root "S-Private=/Volumes/S-Private" \
        --nas-root "S-Public=/Volumes/S-Public" \
        --other-root "usbshare2=/Volumes/usbshare2" \
        [--dry-run] [--max-seconds N] [--max-file-size BYTES]

每個 root 參數格式為 "顯示名稱=實際路徑"，顯示名稱會用來當作
《顯示名稱_duFiles》資料夾的前綴。

--max-file-size 預設 5GB，超過的檔案直接跳過（不計算雜湊值也不會搬動）。
--max-seconds 用於分段執行：處理完一個「大小分組」才會停下來，之後重新執行同一行
指令即可接續處理。
"""
import argparse
import hashlib
import os
import shutil
import sys
import time

SLOW_IO_SLEEP_SECONDS = 5 * 60  # 遇到讀寫效能不佳時，暫停 5 分鐘再繼續
SLOW_IO_RETRIES = 2


def md5_of(path, block_size=1024 * 1024):
    h = hashlib.md5()
    with open(path, 'rb') as f:
        while True:
            chunk = f.read(block_size)
            if not chunk:
                break
            h.update(chunk)
    return h.hexdigest()


def is_dup_output_dir(name):
    return name.endswith('_duFiles') or name.endswith('_DuplicateFiles') or name == 'DuplicateFiles'


def with_slow_io_retry(fn, desc, log):
    """對可能因磁碟效能不佳而失敗的操作做重試：失敗時 sleep 5 分鐘再重試，
    最多重試 SLOW_IO_RETRIES 次，之後仍失敗才真正放棄（回傳 None 並記錄警告）。"""
    for attempt in range(SLOW_IO_RETRIES + 1):
        try:
            return fn()
        except OSError as e:
            if attempt >= SLOW_IO_RETRIES:
                log(f"  警告：{desc} 持續失敗，放棄此項目：{e}")
                return None
            log(f"  偵測到讀寫效能不佳（{desc}：{e}），暫停 5 分鐘後重試...")
            time.sleep(SLOW_IO_SLEEP_SECONDS)
    return None


def parse_roots(values):
    roots = []
    for v in values or []:
        if '=' not in v:
            print(f"格式錯誤，應為 名稱=路徑：{v}", file=sys.stderr)
            sys.exit(1)
        name, path = v.split('=', 1)
        path = os.path.abspath(path)
        roots.append((name, path))
    return roots


def walk_files(name, base, log):
    for root, dirs, files in os.walk(base, onerror=lambda e: log(f"  警告：無法列出 {getattr(e, 'filename', '')}：{e}")):
        dirs[:] = [d for d in dirs if not is_dup_output_dir(d)]
        for fname in files:
            if fname == '.DS_Store':
                continue
            abspath = os.path.join(root, fname)
            def _stat():
                return os.stat(abspath)
            st = with_slow_io_retry(_stat, f"讀取檔案資訊 {abspath}", log)
            if st is None:
                continue
            if not os.path.isfile(abspath):
                continue
            yield st.st_size, st.st_mtime, name, base, abspath


def main():
    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument('--nas-root', action='append', default=[],
                     help='NAS 磁碟區，格式 "名稱=路徑"，可重複指定多次。')
    ap.add_argument('--other-root', action='append', default=[],
                     help='非 NAS 磁碟區（例如外接硬碟），格式 "名稱=路徑"，可重複指定多次。')
    ap.add_argument('--dry-run', action='store_true', help='只顯示會搬動哪些檔案，不實際搬動任何東西')
    ap.add_argument('--max-seconds', type=float, default=None,
                     help='大約執行這麼多秒後就乾淨地停下來（處理完一個大小分組後才停）。')
    ap.add_argument('--max-file-size', type=int, default=5 * 1024**3,
                     help='大於這個位元組數的檔案會直接跳過，不列入比對（預設 5GB，設 0 表示不設上限）。')
    args = ap.parse_args()

    def log(msg):
        print(msg)
        sys.stdout.flush()

    nas_roots = parse_roots(args.nas_root)
    other_roots = parse_roots(args.other_root)
    all_roots = nas_roots + other_roots
    nas_names = {name for name, _ in nas_roots}

    if not all_roots:
        print("至少要指定一個 --nas-root 或 --other-root", file=sys.stderr)
        sys.exit(1)

    for name, path in all_roots:
        if not os.path.isdir(path):
            print(f"不是一個資料夾：{path}", file=sys.stderr)
            sys.exit(1)

    size_limit = args.max_file_size if args.max_file_size and args.max_file_size > 0 else None

    log("正在掃描各磁碟區...")
    t0 = time.time()
    by_size = {}
    total_files = 0
    skipped_large = 0
    skipped_large_bytes = 0
    for name, base in all_roots:
        log(f"  掃描 {name}（{base}）...")
        for size, mtime, rname, rbase, path in walk_files(name, base, log):
            total_files += 1
            if size_limit is not None and size > size_limit:
                skipped_large += 1
                skipped_large_bytes += size
                continue
            by_size.setdefault(size, []).append((mtime, rname, rbase, path))
    log(f"共找到 {total_files} 個檔案，耗時 {time.time()-t0:.1f} 秒")
    if skipped_large:
        log(f"  其中 {skipped_large} 個檔案超過 {size_limit/1024**3:.1f}GB 上限"
            f"（共 {skipped_large_bytes/1024**3:.1f}GB），已跳過不列入比對")

    candidates = [entries for size, entries in by_size.items() if size > 0 and len(entries) > 1]
    n_candidates = sum(len(e) for e in candidates)
    log(f"其中 {n_candidates} 個檔案與其他檔案大小相同，需要計算內容雜湊值來比對")

    # per-root 統計，最後用來寫 report.md
    stats = {name: {'moved': 0, 'groups': 0, 'moved_files': []} for name, _ in all_roots}

    start_time = time.time()
    hashed = 0
    dup_groups_found = 0
    moved = 0
    stopped_early = False

    for entries in candidates:
        if args.max_seconds is not None and (time.time() - start_time) > args.max_seconds:
            stopped_early = True
            break

        by_hash = {}
        for mtime, rname, rbase, path in entries:
            if not os.path.exists(path):
                continue
            def _hash():
                return md5_of(path)
            h = with_slow_io_retry(_hash, f"計算雜湊值 {path}", log)
            if h is None:
                continue
            by_hash.setdefault(h, []).append((mtime, rname, rbase, path))
            hashed += 1
            if hashed % 500 == 0:
                log(f"  已計算 {hashed}/{n_candidates} 個檔案的雜湊值（已耗時 {time.time()-start_time:.0f} 秒）")

        for h, group in by_hash.items():
            if len(group) < 2:
                continue
            dup_groups_found += 1

            nas_entries = [e for e in group if e[1] in nas_names]
            other_entries = [e for e in group if e[1] not in nas_names]

            if nas_entries:
                # 保留全部 NAS 副本；但若 NAS 內部本身就重複，只留最舊的一份
                nas_entries.sort(key=lambda e: e[0])
                if len(nas_entries) > 1:
                    keep = nas_entries[0]
                    to_move = nas_entries[1:] + other_entries
                else:
                    keep = nas_entries[0]
                    to_move = other_entries
            else:
                # 沒有 NAS 副本，只在非 NAS 磁碟區內部找最舊的一份保留
                other_entries.sort(key=lambda e: e[0])
                keep = other_entries[0]
                to_move = other_entries[1:]

            for mtime, rname, rbase, path in to_move:
                rel = os.path.relpath(path, rbase)
                dup_root = os.path.join(rbase, f"{rname}_duFiles")
                dest = os.path.join(dup_root, rel)
                if args.dry_run:
                    log(f"  將會搬移：[{rname}] {rel}（與 [{keep[1]}] {os.path.relpath(keep[3], keep[2])} 內容重複）")
                    moved += 1
                    stats[rname]['moved'] += 1
                    stats[rname]['moved_files'].append(rel)
                    continue

                def _move():
                    os.makedirs(os.path.dirname(dest), exist_ok=True)
                    d = dest
                    if os.path.exists(d):
                        root_, ext = os.path.splitext(d)
                        i = 1
                        candidate_dest = f"{root_}__dup{i}{ext}"
                        while os.path.exists(candidate_dest):
                            i += 1
                            candidate_dest = f"{root_}__dup{i}{ext}"
                        d = candidate_dest
                    shutil.move(path, d)
                    return d

                result = with_slow_io_retry(_move, f"搬移檔案 {path}", log)
                if result is None:
                    continue
                moved += 1
                stats[rname]['moved'] += 1
                stats[rname]['moved_files'].append(rel)

            for rname in {e[1] for e in group}:
                stats[rname]['groups'] += 1

    log(f"\n已計算 {hashed} 個檔案的雜湊值，目前為止找到 {dup_groups_found} 組重複檔案"
        f"（耗時 {time.time()-start_time:.1f} 秒）")
    verb = "將會搬移" if args.dry_run else "已搬移"
    log(f"{verb} {moved} 個重複檔案")
    log("沒有刪除任何檔案。")
    if stopped_early:
        log(f"\n已達到 --max-seconds 設定的時間上限（{args.max_seconds} 秒），提早停止。"
            f"請重新執行同一行指令以繼續處理剩下的檔案。")

    # 寫 report.md（dry-run 時不寫入，避免留下不準確的報告）
    if not args.dry_run:
        for name, base in all_roots:
            dup_root = os.path.join(base, f"{name}_duFiles")
            s = stats[name]
            if s['moved'] == 0 and not os.path.isdir(dup_root):
                continue
            os.makedirs(dup_root, exist_ok=True)
            report_path = os.path.join(dup_root, 'report.md')
            is_nas = name in nas_names
            with open(report_path, 'a', encoding='utf-8') as f:
                f.write(f"\n---\n\n## 執行時間：{time.strftime('%Y-%m-%d %H:%M:%S')}\n\n")
                f.write(f"- 磁碟區：{name}（{base}）\n")
                f.write(f"- 類型：{'NAS' if is_nas else '非 NAS（外接硬碟）'}\n")
                f.write(f"- 本次搬移重複檔案數：{s['moved']}\n")
                f.write(f"- 涉及重複檔案組數：{s['groups']}\n")
                f.write(f"- 保留規則：{'優先保留本磁碟區檔案；若磁碟區內部本身重複，保留最舊一份' if is_nas else '若與 NAS 上的檔案重複，本磁碟區副本一律視為多餘並搬移；若僅本磁碟區內部重複，保留最舊一份'}\n")
                f.write(f"- 大於 {size_limit/1024**3:.1f}GB 的檔案已跳過，不列入比對\n\n")
                if s['moved_files']:
                    f.write("### 本次搬移的檔案清單\n\n")
                    for rel in s['moved_files']:
                        f.write(f"- {rel}\n")
                    f.write("\n")
            log(f"已寫入報告：{report_path}")


if __name__ == '__main__':
    main()

代碼: 選擇全部

    python3 dedupe_nas.py \
        --nas-root "home=/path/to/nasDir_1" \
        --nas-root "S-Private=/path/to/nasDir_2" \
        --nas-root "S-Public=/path/to/nasDir_3" \
        --other-root "usbshare2=/path/to/oursizeDir" \
        [--dry-run] [--max-seconds N] [--max-file-size BYTES]