Python 实现 图片批量筛选去重整理工具

flyfish

batch_image_filter_deduplicator.py

功能

1. 全目录递归扫描

自动递归遍历源文件夹下所有层级子目录，识别全部 .jpg / .jpeg 格式图片（大小写兼容），无需提前手动整理文件夹结构。

2. 指定分辨率精准筛选

仅保留设定的三种标准分辨率图片：1280×720、1920×1080、2560×1440，其余分辨率图片自动过滤，处理结束后会单独统计过滤数量。

3. 图片内容级去重

基于感知哈希（pHash）算法实现纯内容去重，不受文件名、修改时间、存储路径影响，完全根据画面内容判断重复；
重复图片会完整记录原始路径并自动导出为日志文件，相同内容的图片仅保留1份进入后续归档流程。

4. 多进程并行加速

采用多进程架构并行执行图片读取、分辨率校验、哈希计算操作，充分利用CPU多核性能，适配万级以上图片的批量处理场景。

5. 自动分文件夹归档

拷贝时自动按每1000张为一组创建子文件夹，文件夹命名格式为 sisyphus_00000001 顺序递增，避免单目录文件过多导致的管理与打开卡顿问题。

6. 统一规范重命名

所有最终保留的图片按全局顺序统一重命名为8位数字格式（从 00000001.jpg 开始连续递增），序号全局唯一无重复。

7. 完整统计与异常兼容

• 处理结束输出全量对账统计：总图片数、分辨率不符数、损坏图片数、有效唯一图片数、重复图片数，数据可一一核对
• 自动跳过损坏、无法读取的异常图片，不会导致程序中途中断
• 自动生成 重复图片列表.txt 日志文件，留存所有重复文件的原始完整路径，方便后续核查处理

多进程

内部任务分配与执行流程

进程池采用 全局任务队列 + 空闲 Worker 动态取任务 的调度模式，并非预先平均分配任务

1. 任务提交阶段

当执行 executor.submit(process_single_image, p) 时：
传入的函数和参数会被序列化（pickle 编码），放入进程池内部的全局任务队列中排队
此时任务仅完成提交，尚未开始执行
通过推导式一次性提交了全部图片的处理任务，所有任务会按提交顺序进入队列等待。

2. 并行执行阶段

进程池启动 max_workers 个子进程（比如你设置的 8 个）：
每个子进程是独立的 worker，循环从全局任务队列中取下一个待执行任务
哪个 worker 先跑完当前任务、进入空闲状态，就自动领取下一个任务
不会预先给每个进程分配固定数量的图片，而是动态负载均衡，保证 CPU 持续满负载运行
同一时刻，最多有 max_workers 个任务在并行运行，剩余任务在队列中排队等待

3. 结果返回阶段

子进程执行完处理函数后：
将返回值序列化，通过进程间通信（IPC 管道）传回主进程
主进程中的 as_completed(...) 会按任务完成的先后顺序逐个返回结果（不是按提交顺序）
在这个循环中串行执行去重判断、计数统计等操作

import os
import shutil
from PIL import Image
import imagehash
from concurrent.futures import ProcessPoolExecutor, as_completed

# ===================== 【配置】 =====================
SOURCE_DIR = r"D:\data\original"
TARGET_DIR = r"D:\data\SisyphusPremium"
ALLOWED_SIZES = {(1280, 720), (1920, 1080), (2560, 1440)}
BATCH_SIZE = 1000
MAX_WORKERS = 8
# ====================================================

def process_single_image(img_path):
    """单张图片处理：返回状态标记 + 路径 + hash"""
    try:
        with Image.open(img_path) as img:
            if img.size not in ALLOWED_SIZES:
                return ("size_error", img_path, None)
            img_hash = str(imagehash.phash(img))
            return ("success", img_path, img_hash)
    except:
        return ("broken", img_path, None)

def get_all_image_paths(root_dir):
    """递归获取所有jpg路径"""
    paths = []
    for root, _, files in os.walk(root_dir):
        for f in files:
            if f.lower().endswith((".jpg", ".jpeg")):
                paths.append(os.path.join(root, f))
    return paths

def process_images():
    print("开始扫描图片...")
    all_images = get_all_image_paths(SOURCE_DIR)
    total = len(all_images)
    print(f"扫描完成，总计图片数：{total}")

    print("开始校验分辨率与去重（多进程）...")
    seen_hashes = set()
    valid_images = []
    duplicate_images = []
    broken_count = 0
    size_error_count = 0

    with ProcessPoolExecutor(max_workers=MAX_WORKERS) as executor:
        future_to_path = {executor.submit(process_single_image, p): p for p in all_images}
        count = 0
        for future in as_completed(future_to_path):
            count += 1
            # 每1000张打印一次进度，显示更连贯
            if count % 1000 == 0:
                print(f"已校验：{count}/{total}")
                
            status, path, img_hash = future.result()
            if status == "broken":
                broken_count += 1
                continue
            if status == "size_error":
                size_error_count += 1
                continue
            
            # 分辨率符合，进入去重逻辑
            if img_hash not in seen_hashes:
                seen_hashes.add(img_hash)
                valid_images.append(path)
            else:
                duplicate_images.append(path)

    # 完整统计输出
    print(f"\n================ 校验结果统计 ================")
    print(f"总图片数量：{total}")
    print(f"分辨率不符合：{size_error_count}")
    print(f"图片损坏/无法读取：{broken_count}")
    print(f"符合分辨率的图片总数：{len(valid_images) + len(duplicate_images)}")
    print(f"其中：唯一有效图片：{len(valid_images)}")
    print(f"其中：重复图片：{len(duplicate_images)}")
    print(f"==============================================")

    # 保存重复图片列表
    if duplicate_images:
        os.makedirs(TARGET_DIR, exist_ok=True)
        dup_log = os.path.join(TARGET_DIR, "重复图片列表.txt")
        with open(dup_log, "w", encoding="utf-8") as f:
            for p in duplicate_images:
                f.write(p + "\n")
        print(f"重复列表已保存到文件：{dup_log}")

    if len(valid_images) == 0:
        print("没有符合条件的有效图片，程序退出")
        return

    print("\n开始复制文件...")
    os.makedirs(TARGET_DIR, exist_ok=True)
    global_count = 1

    for img_path in valid_images:
        folder_num = (global_count - 1) // BATCH_SIZE + 1
        batch_folder = f"sisyphus_{folder_num:08d}"
        batch_path = os.path.join(TARGET_DIR, batch_folder)
        os.makedirs(batch_path, exist_ok=True)

        new_name = f"{global_count:08d}.jpg"
        target_path = os.path.join(batch_path, new_name)
        shutil.copy2(img_path, target_path)

        if global_count % 1000 == 0:
            print(f"已复制：{global_count}")
        global_count += 1

    print(f"\n处理完成，总复制：{global_count - 1} 张")

if __name__ == "__main__":
    process_images()