Python重复文件批量整理工具的设计与实现

import os
import hashlib
import shutil
import json
import threading
import re
import io
import exifread
import tkinter as tk
from tkinter import ttk, filedialog, messagebox
from datetime import datetime
from PIL import Image
from PIL.ExifTags import TAGS
import pillow_heif
from pillow_heif import register_heif_opener

class FileOrganizerApp:
    def __init__(self, root):
        self.root = root
        self.root.title("文件整理工具")
         
        # 变量初始化
        self.base_dir = tk.StringVar()
        self.target_dir = tk.StringVar()
        self.duplicate_dir = tk.StringVar()
        self.unique_dir = tk.StringVar()
        self.enable_classify = tk.BooleanVar(value=True)
        self.enable_md5list = tk.BooleanVar(value=False)
        
        self.dictfile_name = "base_md5_list.txt"
        
        # 创建register_heif_opener模块里的一个类然后再Image.open打开HEIC文件
        register_heif_opener()
        
        # 创建界面组件
        self.create_widgets()

    def create_widgets(self):
        # 目录选择部分
        dir_frame = ttk.LabelFrame(self.root, text="目录配置")
        dir_frame.pack(padx=10, pady=5, fill=tk.X)
        
        self.create_dir_selector(dir_frame, "基准目录:", self.base_dir, 0)
        self.create_dir_selector(dir_frame, "目标目录:", self.target_dir, 1)
        self.create_dir_selector(dir_frame, "重复文件目录:", self.duplicate_dir, 2)
        self.create_dir_selector(dir_frame, "非重复文件目录:", self.unique_dir, 3)
        
        # 选项配置
        opt_frame = ttk.LabelFrame(self.root, text="选项配置")
        opt_frame.pack(padx=10, pady=5, fill=tk.X)
        
        ttk.Checkbutton(opt_frame, text="启用分类存储", variable=self.enable_classify).pack(anchor=tk.W)
        ttk.Checkbutton(opt_frame, text="读取MD5列表", variable=self.enable_md5list).pack(anchor=tk.W)
        
        # 操作按钮
        btn_frame = ttk.Frame(self.root)
        btn_frame.pack(padx=10, pady=5, fill=tk.X)
        ttk.Button(btn_frame, text="开始处理", command=self.start_processing).pack(side=tk.LEFT)
        
        # 日志显示
        self.log_text = tk.Text(self.root, height=15)
        self.scrollbar = tk.Scrollbar(root, command=self.log_text.yview)  
        self.scrollbar.pack(side=tk.RIGHT, fill=tk.Y)  
        self.log_text.config(yscrollcommand=self.scrollbar.set)
        self.log_text.pack(padx=10, pady=5, fill=tk.BOTH, expand=True)

    def create_dir_selector(self, parent, label, var, row):
        frame = ttk.Frame(parent)
        frame.grid(row=row, column=0, sticky="ew", padx=5, pady=2)
        
        ttk.Label(frame, text=label).pack(side=tk.LEFT)
        entry = ttk.Entry(frame, textvariable=var, width=40)
        entry.pack(side=tk.LEFT, fill=tk.X, expand=True, padx=5)
        ttk.Button(frame, text="浏览...", command=lambda: self.select_directory(var)).pack(side=tk.LEFT)
    
    def select_directory(self, var):
        dir_path = filedialog.askdirectory()
        if dir_path:
            var.set(dir_path)
    
    def log(self, message):
        # 批量更新日志，减少频繁刷新UI
        self.log_text.insert(tk.END, message + "\n")
        self.root.update_idletasks()
    
    def calculate_md5(self, filepath):
        hash_md5 = hashlib.md5()
        with open(filepath, "rb") as f:
            for chunk in iter(lambda: f.read(4096), b""):
                hash_md5.update(chunk)
        return hash_md5.hexdigest()
    
    def save_dict_to_file(self, dictionary, file_path):
        with open(file_path, 'w', encoding='utf-8') as file:
            json.dump(dictionary, file, ensure_ascii=False, indent=4)

    def load_dict_from_file(self, file_path):
        with open(file_path, 'r', encoding='utf-8') as file:
            return json.load(file)
    
    def get_unique_filename(self, dest_dir, filename):
        base, ext = os.path.splitext(filename)
        counter = 1
        new_name = filename
        while os.path.exists(os.path.join(dest_dir, new_name)):
            new_name = f"{base}_{counter}{ext}"
            counter += 1
        return new_name

    def is_image(self, file_path):
        ext = os.path.splitext(file_path)[1].lower()
        return ext in [".jpg", ".jpeg", ".png", ".bmp", ".heic"]

    def get_file_time(self, file_path):
        mtime = os.path.getmtime(file_path)
        ext = os.path.splitext(file_path)[1].lower()

        if self.is_image(file_path):
            if ext == ".heic":
                mtime = self.get_heic_original(file_path)
                mtime = self.replace_limited(mtime, ':', '-', 2)
                dt_object = datetime.strptime(mtime, "%Y-%m-%d %H:%M:%S")
                mtime = dt_object.timestamp()
            else:
                with open(file_path, 'rb') as f:
                    tags = exifread.process_file(f, details=False)
                    if 'EXIF DateTimeOriginal' in tags:
                        mtime = str(tags['EXIF DateTimeOriginal'])
                        mtime = self.replace_limited(mtime, ':', '-', 2)
                        dt_object = datetime.strptime(mtime, "%Y-%m-%d %H:%M:%S")
                        mtime = dt_object.timestamp()
        return mtime

    def process_files(self):
        base_files = {}
        
        if self.enable_md5list.get():
            base_md5_path = os.path.join(self.base_dir.get(), self.dictfile_name)
            if os.path.exists(base_md5_path):
                base_files = self.load_dict_from_file(base_md5_path)
                self.log(f"基准MD5已加载: {base_md5_path}")
        else:
            for root, _, files in os.walk(self.base_dir.get()):
                for file in files:
                    path = os.path.join(root, file)
                    md5 = self.calculate_md5(path)
                    base_files[md5] = file
                    self.log(f"基准文件已扫描: {path}")

        for root, _, files in os.walk(self.target_dir.get()):
            for file in files:
                src_path = os.path.join(root, file)
                md5 = self.calculate_md5(src_path)
                self.log(f"正在处理: {src_path}")
                
                if md5 in base_files:
                    dest_dir = self.duplicate_dir.get()
                    dest_path = os.path.join(dest_dir, self.get_unique_filename(dest_dir, file))
                    shutil.move(src_path, dest_path)
                    self.log(f"重复文件已移动: {dest_path}")
                else:
                    if self.enable_classify.get():
                        mtime = self.get_file_time(src_path)
                        date_dir_Y = datetime.fromtimestamp(mtime).strftime("%Y")
                        date_dir_m = datetime.fromtimestamp(mtime).strftime("%m")
                        dest_dir = os.path.join(self.unique_dir.get(), date_dir_Y, date_dir_m)
                    else:
                        dest_dir = self.unique_dir.get()
                    
                    os.makedirs(dest_dir, exist_ok=True)
                    dest_path = os.path.join(dest_dir, self.get_unique_filename(dest_dir, file))
                    shutil.move(src_path, dest_path)
                    self.log(f"非重复文件已移动: {dest_path}")

        dest_md5_path = os.path.join(self.unique_dir.get(), self.dictfile_name)
        self.save_dict_to_file(base_files, dest_md5_path)
        self.log(f"基准MD5已输出: {dest_md5_path}")

    def start_processing(self):
        try:
            dirs = [
                self.base_dir.get(),
                self.target_dir.get(),
                self.duplicate_dir.get(),
                self.unique_dir.get()
            ]
            
            for d in dirs:
                if not d:
                    raise ValueError("所有目录都必须设置")
                os.makedirs(d, exist_ok=True)
            
            # 启动新线程处理文件
            threading.Thread(target=self.process_files, daemon=True).start()
            messagebox.showinfo("完成", "文件处理已开始，请稍候！")
        except Exception as e:
            messagebox.showerror("错误", str(e))
            self.log(f"错误发生: {str(e)}")

if __name__ == "__main__":
    root = tk.Tk()
    app = FileOrganizerApp(root)
    root.mainloop()