ROOT_DIR = r"D:\imgs"

python dedup_images_run.py

# -*- coding: utf-8 -*-
"""
dedup_images_run.py
Python 批量刪除重復(fù)圖片（MD5 + pHash + ORB 圖像識(shí)別）—— 單文件可直接運(yùn)行版
? 你只需要改 ROOT_DIR，然后 python 運(yùn)行即可
? 不用手動(dòng) pip install：腳本會(huì)自動(dòng)檢測缺少的庫并安裝
? 支持子文件夾遞歸
? 支持中文路徑（Windows）
? 默認(rèn)不直接刪除：先移動(dòng)到 _duplicates/（更安全）
? 導(dǎo)出 dedup_report.csv，方便回溯
去重策略（推薦）：
1) MD5：完全重復(fù)（內(nèi)容一模一樣）——最快最準(zhǔn)
2) pHash：近似重復(fù)（縮放/壓縮/輕微變化）——快速篩選候選
3) ORB 圖像識(shí)別：關(guān)鍵點(diǎn)特征匹配——二次確認(rèn)，減少誤判，更像“識(shí)圖”
依賴（自動(dòng)安裝）：
- opencv-python（cv2）
- numpy
- tqdm
注意：
- 第一次自動(dòng)安裝 opencv-python 可能較慢
- 如在國內(nèi)網(wǎng)絡(luò)環(huán)境，可設(shè)置 PIP_INDEX_URL 為鏡像
"""
 
import os
import sys
import csv
import time
import shutil
import hashlib
import subprocess
import importlib.util
from dataclasses import dataclass
from typing import Dict, List, Tuple, Optional
 
 
# =========================================================
# 1) 配置區(qū)：你只需要改 ROOT_DIR
# =========================================================
ROOT_DIR = r"C:\feeday\imgTest"          # <<< 改成你的圖片目錄
RECURSIVE = True               # 是否遞歸子文件夾
 
# 模式：
# - "md5"              只做完全重復(fù)
# - "phash"            只做近似重復(fù)（可能誤判多一點(diǎn)）
# - "md5+phash"        先md5后phash（推薦）
# - "md5+phash+orb"    先md5后phash，再用ORB圖像識(shí)別二次確認(rèn)（強(qiáng)烈推薦）
MODE = "md5+phash+orb"
 
PHASH_TH = 8                   # pHash 漢明距離閾值：常用 6~10，默認(rèn) 8
 
# ORB 圖像識(shí)別參數(shù)（用于“確認(rèn)相似”）
USE_ORB = True                 # MODE 包含 orb 時(shí)會(huì)自動(dòng)啟用，這里留著方便你手動(dòng)開關(guān)
ORB_NFEATURES = 1200           # ORB 特征點(diǎn)數(shù)量：800~2000 常用
ORB_RATIO_TEST = 0.75          # 0.70~0.80，越小越嚴(yán)格
ORB_MIN_GOOD_MATCHES = 25      # 至少多少個(gè)“好匹配”才算重復(fù)（建議 15~60）
ORB_SIMILARITY_TH = 0.12       # good_matches / min(kp1,kp2) 的比例閾值（建議 0.08~0.20）
ORB_MAX_IMAGE_EDGE = 1200      # 識(shí)別時(shí)把大圖等比縮小到最長邊（提速）
 
ACTION = "move"                # "move" / "delete" / "none"
MOVE_TO = "_duplicates"        # move 時(shí)重復(fù)圖片存放目錄（相對(duì) ROOT_DIR）
REPORT_NAME = "dedup_report.csv"  # 報(bào)告文件名（相對(duì) ROOT_DIR）
DRY_RUN = False                # True=只生成報(bào)告不移動(dòng)/不刪除
 
PREFER_KEEP = "area"           # 保留策略： "area"(分辨率優(yōu)先) / "size" / "newest" / "oldest"
 
# 自動(dòng)安裝依賴
AUTO_INSTALL = True
 
# 如在國內(nèi)網(wǎng)絡(luò)環(huán)境可設(shè)置鏡像（可選）
# 例如：PIP_INDEX_URL = "https://mirrors.tencent.com/pypi/simple/"
PIP_INDEX_URL = ""
 
 
# =========================================================
# 2) 自動(dòng)安裝依賴（寫在 py 里）
# =========================================================
def _module_exists(module_name: str) -> bool:
    return importlib.util.find_spec(module_name) is not None
 
def ensure_packages():
    """
    檢測并自動(dòng)安裝依賴庫：opencv-python / numpy / tqdm
    """
    if not AUTO_INSTALL:
        return
 
    need_install = []
 
    req = [
        ("cv2", "opencv-python"),
        ("numpy", "numpy"),
        ("tqdm", "tqdm"),
    ]
 
    for mod, pkg in req:
        if not _module_exists(mod):
            need_install.append(pkg)
 
    if not need_install:
        return
 
    print("?? 檢測到缺少依賴，將自動(dòng)安裝：", ", ".join(need_install))
    cmd = [sys.executable, "-m", "pip", "install", "--upgrade"] + need_install
 
    if PIP_INDEX_URL.strip():
        cmd += ["-i", PIP_INDEX_URL.strip()]
 
    print("?? 執(zhí)行：", " ".join(cmd))
    try:
        subprocess.check_call(cmd)
        print("? 依賴安裝完成")
    except subprocess.CalledProcessError as e:
        print("? 自動(dòng)安裝失?。?, e)
        print("你可以手動(dòng)執(zhí)行：")
        if PIP_INDEX_URL.strip():
            print(f'  {sys.executable} -m pip install -U {" ".join(need_install)} -i {PIP_INDEX_URL.strip()}')
        else:
            print(f'  {sys.executable} -m pip install -U {" ".join(need_install)}')
        sys.exit(1)
 
ensure_packages()
 
import numpy as np
import cv2
from tqdm import tqdm
 
 
# =========================================================
# 3) 主邏輯
# =========================================================
IMAGE_EXTS = {".jpg", ".jpeg", ".png", ".webp", ".bmp", ".tif", ".tiff"}
 
 
def imread_unicode(path: str) -> Optional[np.ndarray]:
    """
    OpenCV 兼容中文路徑讀取
    """
    try:
        data = np.fromfile(path, dtype=np.uint8)
        img = cv2.imdecode(data, cv2.IMREAD_COLOR)
        return img
    except Exception:
        return None
 
 
def get_image_hw(path: str) -> Tuple[int, int]:
    img = imread_unicode(path)
    if img is None:
        return 0, 0
    h, w = img.shape[:2]
    return int(h), int(w)
 
 
def file_md5(path: str, chunk_size: int = 1024 * 1024) -> str:
    md5 = hashlib.md5()
    with open(path, "rb") as f:
        while True:
            b = f.read(chunk_size)
            if not b:
                break
            md5.update(b)
    return md5.hexdigest()
 
 
def phash_64(path: str) -> Optional[int]:
    """
    pHash 64-bit：灰度->32x32->DCT->取8x8低頻->與中位數(shù)比較->64bit
    """
    img = imread_unicode(path)
    if img is None:
        return None
    try:
        gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
        gray = cv2.resize(gray, (32, 32), interpolation=cv2.INTER_AREA)
        gray = np.float32(gray)
 
        dct = cv2.dct(gray)
        dct_low = dct[:8, :8].copy()
 
        flat = dct_low.flatten()
        flat_no_dc = flat[1:]  # 去掉 DC
        median = np.median(flat_no_dc)
 
        bits = (flat > median).astype(np.uint8)
 
        h = 0
        for b in bits:
            h = (h << 1) | int(b)
        return int(h)
    except Exception:
        return None
 
 
def hamming_distance_64(a: int, b: int) -> int:
    return int((a ^ b).bit_count())
 
 
# -----------------------------
# ORB 圖像識(shí)別（關(guān)鍵點(diǎn)特征匹配）
# -----------------------------
_orb = None
_bf = None
_orb_cache: Dict[str, Tuple[int, Optional[np.ndarray]]] = {}
 
def _get_orb():
    global _orb
    if _orb is None:
        _orb = cv2.ORB_create(nfeatures=int(ORB_NFEATURES))
    return _orb
 
def _get_bf():
    global _bf
    if _bf is None:
        _bf = cv2.BFMatcher(cv2.NORM_HAMMING, crossCheck=False)
    return _bf
 
def _resize_for_orb(img: np.ndarray) -> np.ndarray:
    h, w = img.shape[:2]
    m = max(h, w)
    if m <= ORB_MAX_IMAGE_EDGE:
        return img
    scale = ORB_MAX_IMAGE_EDGE / float(m)
    new_w = max(1, int(w * scale))
    new_h = max(1, int(h * scale))
    return cv2.resize(img, (new_w, new_h), interpolation=cv2.INTER_AREA)
 
def orb_signature(path: str) -> Tuple[int, Optional[np.ndarray]]:
    """
    返回：(keypoints_count, descriptors)
    descriptors 可能為 None（例如純色圖/讀圖失?。?
    """
    if path in _orb_cache:
        return _orb_cache[path]
 
    img = imread_unicode(path)
    if img is None:
        _orb_cache[path] = (0, None)
        return (0, None)
 
    try:
        img = _resize_for_orb(img)
        gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
        orb = _get_orb()
        kps, des = orb.detectAndCompute(gray, None)
        kp_count = 0 if kps is None else len(kps)
        _orb_cache[path] = (kp_count, des)
        return _orb_cache[path]
    except Exception:
        _orb_cache[path] = (0, None)
        return (0, None)
 
def orb_similarity(path_a: str, path_b: str) -> Tuple[int, float]:
    """
    計(jì)算 ORB 相似度：
    - good_matches：通過 ratio test 的好匹配數(shù)量
    - similarity：good_matches / min(kp_count_a, kp_count_b)
    """
    kp_a, des_a = orb_signature(path_a)
    kp_b, des_b = orb_signature(path_b)
 
    if des_a is None or des_b is None:
        return 0, 0.0
    if kp_a <= 0 or kp_b <= 0:
        return 0, 0.0
 
    try:
        bf = _get_bf()
        matches = bf.knnMatch(des_a, des_b, k=2)
 
        good = 0
        ratio = float(ORB_RATIO_TEST)
        for m_n in matches:
            if len(m_n) < 2:
                continue
            m, n = m_n[0], m_n[1]
            if m.distance < ratio * n.distance:
                good += 1
 
        denom = max(1, min(kp_a, kp_b))
        sim = good / float(denom)
        return good, float(sim)
    except Exception:
        return 0, 0.0
 
 
@dataclass
class ImgInfo:
    path: str
    size: int
    w: int
    h: int
 
    @property
    def area(self) -> int:
        return self.w * self.h
 
 
def choose_keep(a: ImgInfo, b: ImgInfo, prefer: str) -> Tuple[ImgInfo, ImgInfo]:
    if prefer == "area":
        if a.area != b.area:
            return (a, b) if a.area > b.area else (b, a)
        if a.size != b.size:
            return (a, b) if a.size > b.size else (b, a)
        return (a, b)
 
    if prefer == "size":
        if a.size != b.size:
            return (a, b) if a.size > b.size else (b, a)
        if a.area != b.area:
            return (a, b) if a.area > b.area else (b, a)
        return (a, b)
 
    if prefer in {"newest", "oldest"}:
        ta = os.path.getmtime(a.path)
        tb = os.path.getmtime(b.path)
        if prefer == "newest":
            return (a, b) if ta >= tb else (b, a)
        else:
            return (a, b) if ta <= tb else (b, a)
 
    return choose_keep(a, b, "area")
 
 
def iter_images(root: str, recursive: bool) -> List[str]:
    root = os.path.abspath(root)
    out = []
    if recursive:
        for dp, _, fns in os.walk(root):
            for fn in fns:
                ext = os.path.splitext(fn)[1].lower()
                if ext in IMAGE_EXTS:
                    out.append(os.path.join(dp, fn))
    else:
        for fn in os.listdir(root):
            p = os.path.join(root, fn)
            if os.path.isfile(p):
                ext = os.path.splitext(fn)[1].lower()
                if ext in IMAGE_EXTS:
                    out.append(p)
    return out
 
 
def ensure_dir(p: str) -> None:
    os.makedirs(p, exist_ok=True)
 
 
def safe_move(src: str, dst: str) -> str:
    base, ext = os.path.splitext(dst)
    final = dst
    i = 1
    while os.path.exists(final):
        final = f"{base}({i}){ext}"
        i += 1
    ensure_dir(os.path.dirname(final))
    shutil.move(src, final)
    return final
 
 
def write_report_csv(report_path: str, rows: List[dict]) -> None:
    ensure_dir(os.path.dirname(report_path) if os.path.dirname(report_path) else ".")
    fieldnames = [
        "time", "mode",
        "keep_path", "dup_path",
        "reason", "value",
        "phash_distance",
        "orb_good_matches", "orb_similarity",
        "keep_w", "keep_h", "keep_size",
        "dup_w", "dup_h", "dup_size",
        "dup_new_path",
    ]
    with open(report_path, "w", newline="", encoding="utf-8-sig") as f:
        w = csv.DictWriter(f, fieldnames=fieldnames)
        w.writeheader()
        for r in rows:
            w.writerow(r)
 
 
def dedup_by_md5(paths: List[str], prefer_keep: str) -> List[Tuple[str, str, str]]:
    """
    返回重復(fù)對(duì)：(dup_path, keep_path, md5)
    """
    md5_map: Dict[str, ImgInfo] = {}
    dups: List[Tuple[str, str, str]] = []
 
    for p in tqdm(paths, desc="MD5 掃描", unit="img"):
        try:
            h = file_md5(p)
        except Exception:
            continue
 
        try:
            size = os.path.getsize(p)
        except Exception:
            size = 0
        hh, ww = get_image_hw(p)
        cur = ImgInfo(path=p, size=size, w=ww, h=hh)
 
        if h not in md5_map:
            md5_map[h] = cur
        else:
            old = md5_map[h]
            keep, drop = choose_keep(old, cur, prefer_keep)
            md5_map[h] = keep
            dups.append((drop.path, keep.path, h))
 
    return dups
 
 
def dedup_by_phash_with_orb(
    paths: List[str],
    prefer_keep: str,
    phash_th: int,
    use_orb: bool
) -> List[Tuple[str, str, str, int, int, float]]:
    """
    返回重復(fù)對(duì)：
      (dup_path, keep_path, phash_hex, phash_dist, orb_good, orb_sim)
    說明：
    - 先用 pHash 快速篩選候選（phash_dist <= phash_th）
    - 若啟用 ORB，則對(duì)候選進(jìn)行二次“圖像識(shí)別”確認(rèn)
    """
    hashes: Dict[str, int] = {}
    infos: Dict[str, ImgInfo] = {}
 
    for p in tqdm(paths, desc="pHash 計(jì)算", unit="img"):
        ph = phash_64(p)
        if ph is None:
            continue
        hashes[p] = ph
        try:
            size = os.path.getsize(p)
        except Exception:
            size = 0
        hh, ww = get_image_hw(p)
        infos[p] = ImgInfo(path=p, size=size, w=ww, h=hh)
 
    # 分桶加速：用高16位分桶
    buckets: Dict[int, List[str]] = {}
    for p, ph in hashes.items():
        key = (ph >> 48) & 0xFFFF
        buckets.setdefault(key, []).append(p)
 
    keep_map: Dict[str, str] = {p: p for p in hashes.keys()}
    dups: List[Tuple[str, str, str, int, int, float]] = []
 
    for key in tqdm(list(buckets.keys()), desc="pHash 分桶比對(duì)", unit="bucket"):
        group = buckets[key]
        if len(group) <= 1:
            continue
 
        for i in range(len(group)):
            a = group[i]
            a_keep = keep_map.get(a, a)
            if a_keep not in hashes:
                continue
 
            for j in range(i + 1, len(group)):
                b = group[j]
                b_keep = keep_map.get(b, b)
                if b_keep not in hashes:
                    continue
 
                ha = hashes[a_keep]
                hb = hashes[b_keep]
                dist = hamming_distance_64(ha, hb)
 
                # pHash 先篩候選
                if dist > phash_th:
                    continue
 
                orb_good = 0
                orb_sim = 0.0
                if use_orb:
                    orb_good, orb_sim = orb_similarity(a_keep, b_keep)
 
                    # ORB 二次確認(rèn)門檻：兩個(gè)條件都滿足才算重復(fù)
                    if orb_good < int(ORB_MIN_GOOD_MATCHES):
                        continue
                    if orb_sim < float(ORB_SIMILARITY_TH):
                        continue
 
                ia = infos.get(a_keep)
                ib = infos.get(b_keep)
                if ia is None or ib is None:
                    continue
 
                keep, drop = choose_keep(ia, ib, prefer_keep)
                keep_map[keep.path] = keep.path
                keep_map[drop.path] = keep.path
 
                ph_hex = f"{hashes[keep.path]:016x}"
                dups.append((drop.path, keep.path, ph_hex, dist, orb_good, orb_sim))
 
    return dups
 
 
def main():
    root = os.path.abspath(ROOT_DIR)
    if not os.path.isdir(root):
        print(f"? 目錄不存在：{root}")
        return
 
    t0 = time.time()
    paths = iter_images(root, RECURSIVE)
    if not paths:
        print("?? 未找到圖片文件。")
        return
 
    duplicates_dir = os.path.join(root, MOVE_TO)
    report_path = os.path.join(root, REPORT_NAME)
 
    mode = MODE.strip().lower()
    use_orb = (("orb" in mode) and USE_ORB)
 
    print("=================================================")
    print("?? 根目錄：", root)
    print("?? 圖片數(shù)量：", len(paths))
    print(f"?? 模式：{MODE} | pHash閾值：{PHASH_TH} | 保留策略：{PREFER_KEEP}")
    print(f"?? 動(dòng)作：{ACTION} | dry-run：{DRY_RUN}")
    if use_orb:
        print("?? ORB 圖像識(shí)別：啟用")
        print(f"   ORB_NFEATURES={ORB_NFEATURES}, RATIO={ORB_RATIO_TEST}, MIN_GOOD={ORB_MIN_GOOD_MATCHES}, SIM_TH={ORB_SIMILARITY_TH}")
    else:
        print("?? ORB 圖像識(shí)別：未啟用")
    print("=================================================")
 
    report_rows: List[dict] = []
    processed = 0
 
    # 1) MD5
    md5_dups: List[Tuple[str, str, str]] = []
    remain_after_md5 = paths
 
    if mode in {"md5", "md5+phash", "md5+phash+orb"} or ("md5" in mode):
        md5_dups = dedup_by_md5(paths, PREFER_KEEP)
        dup_set = set([d[0] for d in md5_dups])
        remain_after_md5 = [p for p in paths if p not in dup_set]
 
    # 2) pHash (+ ORB)
    phash_dups: List[Tuple[str, str, str, int, int, float]] = []
    if ("phash" in mode):
        phash_dups = dedup_by_phash_with_orb(
            remain_after_md5,
            prefer_keep=PREFER_KEEP,
            phash_th=PHASH_TH,
            use_orb=use_orb
        )
 
    def do_action(dup_path: str) -> str:
        nonlocal processed
        if DRY_RUN or ACTION == "none":
            return ""
        if not os.path.exists(dup_path):
            return ""
 
        if ACTION == "delete":
            try:
                os.remove(dup_path)
                processed += 1
                return ""
            except Exception:
                return ""
 
        if ACTION == "move":
            try:
                rel = os.path.relpath(dup_path, root)
                dst = os.path.join(duplicates_dir, rel)
                newp = safe_move(dup_path, dst)
                processed += 1
                return newp
            except Exception:
                return ""
 
        return ""
 
    # 記錄 MD5
    for dup_path, keep_path, md5v in md5_dups:
        if not os.path.exists(dup_path):
            continue
 
        keep_h, keep_w = get_image_hw(keep_path)
        dup_h, dup_w = get_image_hw(dup_path)
        keep_size = os.path.getsize(keep_path) if os.path.exists(keep_path) else 0
        dup_size = os.path.getsize(dup_path) if os.path.exists(dup_path) else 0
 
        new_path = do_action(dup_path)
 
        report_rows.append({
            "time": time.strftime("%Y-%m-%d %H:%M:%S"),
            "mode": "md5",
            "keep_path": keep_path,
            "dup_path": dup_path,
            "reason": "MD5 完全重復(fù)",
            "value": md5v,
            "phash_distance": "",
            "orb_good_matches": "",
            "orb_similarity": "",
            "keep_w": keep_w, "keep_h": keep_h, "keep_size": keep_size,
            "dup_w": dup_w, "dup_h": dup_h, "dup_size": dup_size,
            "dup_new_path": new_path,
        })
 
    # 記錄 pHash(+ORB)
    for dup_path, keep_path, ph_hex, ph_dist, orb_good, orb_sim in phash_dups:
        if not os.path.exists(dup_path):
            continue
 
        keep_h, keep_w = get_image_hw(keep_path)
        dup_h, dup_w = get_image_hw(dup_path)
        keep_size = os.path.getsize(keep_path) if os.path.exists(keep_path) else 0
        dup_size = os.path.getsize(dup_path) if os.path.exists(dup_path) else 0
 
        new_path = do_action(dup_path)
 
        reason = "pHash 近似重復(fù)"
        if use_orb:
            reason = "pHash + ORB 圖像識(shí)別確認(rèn)"
 
        report_rows.append({
            "time": time.strftime("%Y-%m-%d %H:%M:%S"),
            "mode": "phash+orb" if use_orb else "phash",
            "keep_path": keep_path,
            "dup_path": dup_path,
            "reason": reason,
            "value": ph_hex,
            "phash_distance": ph_dist,
            "orb_good_matches": orb_good if use_orb else "",
            "orb_similarity": f"{orb_sim:.4f}" if use_orb else "",
            "keep_w": keep_w, "keep_h": keep_h, "keep_size": keep_size,
            "dup_w": dup_w, "dup_h": dup_h, "dup_size": dup_size,
            "dup_new_path": new_path,
        })
 
    write_report_csv(report_path, report_rows)
 
    dt = time.time() - t0
    print("")
    print("? 完成")
    print(f"?? 報(bào)告：{report_path}")
    if ACTION == "move":
        print(f"?? 重復(fù)文件目錄：{duplicates_dir}（保持原目錄結(jié)構(gòu)）")
    print(f"?? 識(shí)別重復(fù)條目：{len(report_rows)}")
    if DRY_RUN or ACTION == "none":
        print("?? dry-run/none：未移動(dòng)/未刪除任何文件")
    else:
        print(f"?? 實(shí)際處理文件數(shù)（移動(dòng)/刪除）：{processed}")
    print(f"? 用時(shí)：{dt:.2f} 秒")
 
 
if __name__ == "__main__":
    main()