怎样实现相似图片清理脚本

wen 实用脚本 2026-06-28 231

本文目录导读：

怎样实现相似图片清理脚本

完整实现方案
使用示例
安装依赖
功能特点
注意事项

我来帮你创建一个相似图片清理脚本,这个脚本可以找出并清理目录中相似度高的重复图片。

完整实现方案

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
相似图片清理脚本
功能：找出目录中相似度高于阈值的图片，并提供清理选项
"""
import os
import sys
import shutil
import hashlib
from pathlib import Path
from PIL import Image
import imagehash
from collections import defaultdict
class SimilarImageCleaner:
    def __init__(self, similarity_threshold=0.9):
        """
        初始化清理器
        :param similarity_threshold: 相似度阈值 (0.0-1.0)，默认0.9表示90%相似
        """
        self.similarity_threshold = similarity_threshold
        self.image_hashes = defaultdict(list)
        self.duplicate_groups = []
        self.processed_files = 0
        self.total_size_saved = 0
    def get_image_hash(self, image_path, hash_size=8):
        """
        计算图片的感知哈希值
        :param image_path: 图片路径
        :param hash_size: 哈希大小，越大越精确
        :return: 哈希值或None
        """
        try:
            with Image.open(image_path) as img:
                # 转换为RGB模式以确保一致性
                if img.mode != 'RGB':
                    img = img.convert('RGB')
                # 计算感知哈希
                phash = imagehash.phash(img, hash_size=hash_size)
                return phash
        except Exception as e:
            print(f"❌ 无法处理图片 {image_path}: {e}")
            return None
    def calculate_similarity(self, hash1, hash2, hash_size=8):
        """
        计算两个哈希值的相似度
        :param hash1: 哈希值1
        :param hash2: 哈希值2
        :param hash_size: 哈希大小
        :return: 相似度 (0.0-1.0)
        """
        # 汉明距离
        hamming_distance = hash1 - hash2
        # 最大可能的汉明距离
        max_distance = hash_size * hash_size
        # 转换为相似度
        similarity = 1 - (hamming_distance / max_distance)
        return similarity
    def scan_directory(self, directory, extensions=None):
        """
        扫描目录中的所有图片
        :param directory: 要扫描的目录
        :param extensions: 图片扩展名列表，默认常见格式
        """
        if extensions is None:
            extensions = {'.jpg', '.jpeg', '.png', '.gif', '.bmp', '.tiff', '.webp'}
        directory = Path(directory)
        if not directory.exists():
            raise ValueError(f"目录不存在: {directory}")
        print(f"🔍 正在扫描目录: {directory}")
        print(f"📁 支持的格式: {', '.join(extensions)}")
        # 递归遍历所有图片文件
        image_files = []
        for file_path in directory.rglob('*'):
            if file_path.suffix.lower() in extensions and file_path.is_file():
                image_files.append(file_path)
        print(f"📊 共找到 {len(image_files)} 个图片文件")
        # 计算每个图片的哈希值
        print("⚙️ 正在计算图片哈希值...")
        for i, img_path in enumerate(image_files, 1):
            if i % 10 == 0 or i == len(image_files):
                print(f"  进度: {i}/{len(image_files)}")
            img_hash = self.get_image_hash(img_path)
            if img_hash is not None:
                self.image_hashes[img_hash].append(img_path)
                self.processed_files += 1
        print(f"✅ 成功处理 {self.processed_files} 个图片文件")
    def find_similar_groups(self):
        """
        查找相似图片组
        :return: 相似图片组的列表
        """
        print("🔎 正在查找相似图片...")
        hash_list = list(self.image_hashes.keys())
        processed = set()
        similar_groups = []
        for i in range(len(hash_list)):
            if i in processed:
                continue
            current_group = {
                'hashes': [hash_list[i]],
                'files': self.image_hashes[hash_list[i]][:]
            }
            for j in range(i + 1, len(hash_list)):
                if j in processed:
                    continue
                similarity = self.calculate_similarity(hash_list[i], hash_list[j])
                if similarity >= self.similarity_threshold:
                    current_group['hashes'].append(hash_list[j])
                    current_group['files'].extend(self.image_hashes[hash_list[j]])
                    processed.add(j)
            if len(current_group['files']) > 1:
                similar_groups.append(current_group)
                processed.add(i)
        self.duplicate_groups = similar_groups
        # 计算可节省的空间
        for group in similar_groups:
            # 保留第一张图片，计算其他图片的总大小
            for file_path in group['files'][1:]:
                try:
                    self.total_size_saved += file_path.stat().st_size
                except:
                    pass
        return similar_groups
    def print_results(self):
        """
        打印相似图片分析结果
        """
        if not self.duplicate_groups:
            print("🎉 没有找到相似的图片！")
            return
        print(f"\n{'='*60}")
        print(f"📋 分析结果")
        print(f"{'='*60}")
        print(f"总处理图片: {self.processed_files}")
        print(f"相似图片组数: {len(self.duplicate_groups)}")
        print(f"可节省空间: {self._format_size(self.total_size_saved)}")
        print()
        for i, group in enumerate(self.duplicate_groups, 1):
            print(f"📌 组 {i}: {len(group['files'])} 张相似图片")
            for j, file_path in enumerate(group['files']):
                size = file_path.stat().st_size if file_path.exists() else 0
                print(f"   {j+1}. {file_path} ({self._format_size(size)})")
            # 显示相似度
            if len(group['hashes']) > 1:
                sim = self.calculate_similarity(group['hashes'][0], group['hashes'][1])
                print(f"   相似度: {sim:.2%}")
            print()
    def clean_duplicates(self, mode='move', backup_dir='backup'):
        """
        清理相似图片
        :param mode: 清理模式 ('move' 移动到备份目录, 'delete' 直接删除, 'list' 仅列出)
        :param backup_dir: 备份目录名
        """
        if not self.duplicate_groups:
            print("没有需要清理的图片")
            return
        print(f"\n{'='*60}")
        print(f"🗑️ 开始清理 (模式: {mode})")
        print(f"{'='*60}")
        cleaned_count = 0
        cleaned_size = 0
        for group in self.duplicate_groups:
            # 保留第一张图片，处理其余图片
            original = group['files'][0]
            for duplicate in group['files'][1:]:
                if not duplicate.exists():
                    continue
                try:
                    size = duplicate.stat().st_size
                    if mode == 'move':
                        # 创建备份目录（保持相对路径结构）
                        rel_path = duplicate.relative_to(duplicate.anchor)
                        backup_path = Path(backup_dir) / rel_path
                        backup_path.parent.mkdir(parents=True, exist_ok=True)
                        # 如果备份文件已存在，添加时间戳
                        if backup_path.exists():
                            timestamp = int(time.time())
                            backup_path = backup_path.with_stem(
                                f"{backup_path.stem}_{timestamp}"
                            )
                        shutil.move(str(duplicate), str(backup_path))
                        print(f"📦 已移动: {duplicate}")
                    elif mode == 'delete':
                        duplicate.unlink()
                        print(f"🗑️ 已删除: {duplicate}")
                    else:  # list mode
                        print(f"📋 将清理: {duplicate}")
                    cleaned_count += 1
                    cleaned_size += size
                except Exception as e:
                    print(f"❌ 清理失败 {duplicate}: {e}")
        print(f"\n✅ 清理完成!")
        print(f"清理文件数: {cleaned_count}")
        print(f"释放空间: {self._format_size(cleaned_size)}")
    def _format_size(self, bytes_size):
        """格式化文件大小"""
        for unit in ['B', 'KB', 'MB', 'GB']:
            if bytes_size < 1024:
                return f"{bytes_size:.2f} {unit}"
            bytes_size /= 1024
        return f"{bytes_size:.2f} TB"
def main():
    import time
    import argparse
    parser = argparse.ArgumentParser(description='相似图片清理工具')
    parser.add_argument('directory', help='要扫描的目录路径')
    parser.add_argument('-t', '--threshold', type=float, default=0.9,
                       help='相似度阈值 (0.0-1.0)，默认0.9')
    parser.add_argument('--hash-size', type=int, default=8,
                       help='哈希大小，越大越精确但越慢，默认8')
    parser.add_argument('--mode', choices=['move', 'delete', 'list'], 
                       default='list',
                       help='清理模式: move(移动到备份目录), delete(直接删除), list(仅列出)')
    parser.add_argument('--backup-dir', default='similar_image_backup',
                       help='备份目录名')
    parser.add_argument('--extensions', nargs='+',
                       default=['.jpg', '.jpeg', '.png', '.gif', '.bmp', '.tiff', '.webp'],
                       help='支持的图片扩展名')
    args = parser.parse_args()
    # 创建清理器并执行
    cleaner = SimilarImageCleaner(similarity_threshold=args.threshold)
    try:
        # 扫描目录
        cleaner.scan_directory(args.directory, set(args.extensions))
        # 查找相似图片
        similar_groups = cleaner.find_similar_groups()
        # 显示结果
        cleaner.print_results()
        # 如果有相似图片且不是list模式，询问确认
        if similar_groups and args.mode != 'list':
            response = input(f"\n确认以 '{args.mode}' 模式清理？(y/n): ")
            if response.lower() == 'y':
                cleaner.clean_duplicates(args.mode, args.backup_dir)
            else:
                print("已取消清理")
        elif similar_groups and args.mode == 'list':
            # list模式下直接列出清理内容
            cleaner.clean_duplicates('list')
    except Exception as e:
        print(f"❌ 错误: {e}")
        sys.exit(1)
if __name__ == '__main__':
    main()

使用示例

基本使用

# 扫描当前目录，查找相似图片（仅列出）
python similar_image_cleaner.py /path/to/images
# 扫描并设置相似度阈值为85%
python similar_image_cleaner.py /path/to/images -t 0.85
# 移动到备份目录
python similar_image_cleaner.py /path/to/images --mode move
# 直接删除（小心使用）
python similar_image_cleaner.py /path/to/images --mode delete
# 自定义支持的格式
python similar_image_cleaner.py /path/to/images --extensions .jpg .png .webp

高级选项

# 设置更高的哈希精度（更精确但更慢）
python similar_image_cleaner.py /path/to/images --hash-size 16
# 指定备份目录
python similar_image_cleaner.py /path/to/images --mode move --backup-dir my_backup