Python使用wxPython写一个MD5文件查重清理工具

摘要

电脑用得越久，重复文件就越猖獗。下载过好多遍的安装包、反复导出的照片、备份目录里散落的 Office 文档和压缩包副本——手动去翻它，又累又容易翻车。

这篇文章记录的是一个完整的 Python 桌面工具项目。技术选型很直接：用 wxPython 做界面，用 MD5 判断文件内容是否一致，用 SQLite 保存扫描记录，用 JSON 管理用户配置。最关键的一点是，删除文件时默认放入回收站，而不是直接永久干掉。

项目的目标很清晰——不是要做功能爆炸的商业软件，而是要做一个可运行、可维护、操作安全的本地小工具。先说几个判断：它不是什么惊天动地的大项目，但麻雀虽小，五脏俱全，涵盖了桌面工具的一套完整闭环。

功能效果

这个工具能干什么？下面这些功能，基本覆盖了日常查重的所有场景：

• 选择任意文件夹进行查重。
• 支持递归扫描子文件夹。
• 用 MD5 判断文件内容是否完全相同。
• 重复文件按大小从大到小排列显示。
• 每组重复文件默认保留修改时间最新那份。
• 支持批量勾选要删除的重复文件。
• 实时统计已选文件数量和预计释放空间。
• 删除前会弹框确认。
• 删除时走回收站，不走不归路。
• 支持图片预览、ZIP 文件列表预览。
• PDF、Word、Excel、视频等文件显示基本信息，并可调用系统默认程序打开。
• 使用 SQLite 保存扫描结果。
• 用配置文件保存上次选的目录和窗口设置。

项目结构

项目拆成了多个小模块，每个文件各司其职，互不干扰：

文件查重/

├── app.py # wxPython 主界面

├── config.py # JSON 配置读写

├── database.py # SQLite 数据库操作

├── duplicate_finder.py # MD5 扫描和重复文件分组

├── models.py # 数据模型

├── preview.py # 文件预览信息

├── recycle.py # 回收站删除

├── start.bat # Windows 启动脚本

├── README.md # 项目说明

└── tests/ # 单元测试

这么拆分的好处是：MD5 扫描、数据库、配置、删除逻辑都能独立测试，不会全部挤在 GUI 代码里变得一团糟。

一、数据模型设计

文件信息用 FileRecord 表示，重复文件组用 DuplicateGroup 表示：

@dataclass(frozen=True)
class FileRecord:
    path: str
    md5: str
    size: int
    modified_at: float


@dataclass(frozen=True)
class DuplicateGroup:
    md5: str
    size: int
    files: tuple[FileRecord, ...]

    @property
    def keep_file(self) -> FileRecord:
        return self.files[0]

    @property
    def delete_candidates(self) -> tuple[FileRecord, ...]:
        return self.files[1:]

这里有个约定：每组重复文件内部，files[0] 是要保留的。程序默认按修改时间倒序排序，所以最新文件放在最前面，剩下的都是可删除的候选者。

二、为什么用 MD5 查重

判断重复文件，只看文件名显然不靠谱。相同内容的文件可能名字五花八门，比如：

report.docx
report - 副本.docx
final-report.docx

只靠文件大小来判断也不行。两个文件大小相同，内容却可能完全不同。

所以本项目采用了两步判断：

1. 先按文件大小分组，只有大小相同的文件才有可能是重复的。
2. 再对同大小的文件计算 MD5，MD5 一致才能判定内容重复。

核心逻辑在 duplicate_finder.py 里：

by_size: dict[int, list[Path]] = defaultdict(list)
for path in paths:
    by_size[path.stat().st_size].append(path)

by_hash: dict[str, list[FileRecord]] = defaultdict(list)
for size, same_size_paths in by_size.items():
    if len(same_size_paths) < 2:
        continue
    for path in same_size_paths:
        record = self._record_for(path, size)
        if record is not None:
            by_hash[record.md5].append(record)

这个设计的好处在于——它能帮我们减少大量不必要的 MD5 计算。因为只有同大小的文件才需要进一步计算哈希，对于大量不重复的文件来说，速度会快不少。

MD5 计算采用分块读取的方式，避免大文件一次性读入内存：

def md5_for_file(self, path: Path) -> str:
    hasher = hashlib.md5()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(self.chunk_size), b""):
            hasher.update(chunk)
    return hasher.hexdigest()

三、重复组排序策略

用户最关心的通常是“能释放多少空间”，所以扫描结果按文件大小从大到小排序：

return sorted(groups, key=lambda group: group.size, reverse=True)

每组内部则按修改时间从新到旧排序：

files=tuple(sorted(records, key=lambda item: item.modified_at, reverse=True))

因此界面中每组第一条就是默认保留文件，其余文件可以批量勾选后一起删掉。

四、配置文件：保存用户习惯

配置使用 JSON 保存，比如上次选中的文件夹、是否递归扫描、窗口大小等。

@dataclass(frozen=True)
class AppConfig:
    last_folder: str = ""
    recursive: bool = True
    keep_strategy: str = "newest"
    window_width: int = 1180
    window_height: int = 760

加载配置时，如果文件不存在或 JSON 损坏，就返回默认配置：

@classmethod
def load(cls, path: Path) -> "AppConfig":
    if not path.exists():
        return cls()
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        return cls()
    valid = {field: data[field] for field in cls.__dataclass_fields__ if field in data}
    return cls(**valid)

这个处理方式比较稳妥，不会因为配置文件异常就导致程序启动不了。

五、SQLite：保存扫描记录

项目用 SQLite 保存扫描会话和重复文件记录。数据库包含两张表：

• scan_sessions：记录扫描目录和扫描时间。
• duplicate_files：记录每个重复文件的路径、大小、MD5、修改时间、是否保留、是否删除。

建表逻辑如下：

CREATE TABLE IF NOT EXISTS scan_sessions (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    folder TEXT NOT NULL,
    scanned_at REAL NOT NULL
)

CREATE TABLE IF NOT EXISTS duplicate_files (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    session_id INTEGER NOT NULL,
    group_index INTEGER NOT NULL,
    md5 TEXT NOT NULL,
    path TEXT NOT NULL,
    size INTEGER NOT NULL,
    modified_at REAL NOT NULL,
    keep_file INTEGER NOT NULL,
    deleted INTEGER NOT NULL DEFAULT 0,
    FOREIGN KEY(session_id) REFERENCES scan_sessions(id)
)

保存扫描结果时，先插入一次扫描会话，再插入每个文件记录：

session_id = int(cursor.lastrowid)
for group_index, group in enumerate(groups, start=1):
    keep_path = group.keep_file.path
    for record in group.files:
        conn.execute(
            """
            INSERT INTO duplicate_files(
                session_id, group_index, md5, path, size, modified_at, keep_file
            )
            VALUES (?, ?, ?, ?, ?, ?, ?)
            """,
            (
                session_id,
                group_index,
                group.md5,
                record.path,
                record.size,
                record.modified_at,
                1 if record.path == keep_path else 0,
            ),
        )

这里有一个细节：数据库连接使用 contextlib.closing 明确关闭。在 Windows 下，如果 SQLite 连接没有及时关闭，很可能导致数据库文件被占用，测试或删除临时目录时就会失败。

六、wxPython 界面设计

主界面由四部分组成：

1. 顶部工具栏：选择文件夹、递归选项、开始扫描、停止、批量勾选、清空勾选、删除所选。
2. 左侧列表：显示重复文件。
3. 右侧预览：显示图片、ZIP 内容或文件基本信息。
4. 底部状态栏：显示已选数量、预计释放空间和扫描进度。

列表使用 wx.ListCtrl：

self.result_list = wx.ListCtrl(list_panel, style=wx.LC_REPORT | wx.LC_SINGLE_SEL)
if hasattr(self.result_list, "EnableCheckBoxes"):
    self.result_list.EnableCheckBoxes(True)

列设计如下：

columns = [
    ("组", 56),
    ("状态", 80),
    ("大小", 90),
    ("修改时间", 150),
    ("文件路径", 520),
    ("MD5", 220),
]

这种布局很适合工具类软件：信息密度高，用户可以快速比对路径、大小和修改时间。

七、扫描线程：避免界面卡死

文件扫描和 MD5 计算可能会比较耗时，如果直接在主线程执行，GUI 必然卡死。因此程序使用了后台线程扫描：

self.scan_thread = threading.Thread(
    target=self._scan_worker,
    args=(folder, self.recursive_checkbox.GetValue()),
    daemon=True,
)
self.scan_thread.start()

后台线程不能直接操作 wxPython 控件，需要借助 wx.CallAfter 回到主线程更新界面：

progress=lambda path: wx.CallAfter(self.progress_label.SetLabel, f"正在计算: {path}")

扫描完成后同样通过 wx.CallAfter 填充列表：

wx.CallAfter(self._scan_finished, session_id, rows)

八、批量勾选和总大小统计

用户可以点击“勾选重复项”，程序会自动勾选每组中除了保留文件以外的候选文件：

def on_select_duplicates(self, _event) -> None:
    for index, row in enumerate(self.rows):
        if hasattr(self.result_list, "CheckItem"):
            self.result_list.CheckItem(index, not row["keep"] and not row["deleted"])
    self._update_selected_summary()

统计已选文件路径：

def _selected_paths(self) -> list[str]:
    paths = []
    if not hasattr(self.result_list, "IsItemChecked"):
        return paths
    for index, row in enumerate(self.rows):
        if self.result_list.IsItemChecked(index) and not row["keep"] and not row["deleted"]:
            paths.append(row["path"])
    return paths

统计预计释放空间：

def _selected_size(self) -> int:
    selected = set(self._selected_paths())
    return sum(row["size"] for row in self.rows if row["path"] in selected)

底部实时显示：

self.summary_label.SetLabel(f"已选 {len(paths)} 个文件，预计释放 {format_size(total)}")

九、文件预览设计

预览模块没有强行解析所有格式，而是分层处理：

• 图片：内置缩略图预览。
• ZIP：读取压缩包文件列表。
• PDF、Word、Excel、视频：显示文件基本信息，并提供一个“打开文件”按钮。
• 其他文件：显示路径、大小、MIME 类型等基础信息。

判断逻辑如下：

def build_preview(path: Path) -> Preview:
    suffix = path.suffix.lower()
    if suffix in IMAGE_EXTENSIONS:
        return Preview("image", _metadata_text(path), str(path))
    if suffix == ".zip":
        return Preview("zip", _zip_text(path))
    if suffix in DOCUMENT_EXTENSIONS:
        return Preview("document", _metadata_text(path) + "nn可使用“打开文件”调用系统默认程序查看。")
    if suffix in VIDEO_EXTENSIONS:
        return Preview("video", _metadata_text(path) + "nn可使用“打开文件”调用系统默认播放器查看。")
    return Preview("generic", _metadata_text(path))

这个设计相当务实。如果硬要对 Office、PDF、视频做内嵌预览，会引入一大堆依赖和兼容性问题；调用系统默认程序反而更稳定、更省心。

十、删除安全：放入回收站

删除重复文件是高风险的敏感操作，所以程序做了三层保护：

1. 默认保留每组最新文件。
2. 删除前弹窗确认。
3. 删除时放入回收站，而不是直接永久删除。

确认框会显示删除数量和预计释放空间：

message = f"确认将 {len(paths)} 个文件放入回收站？n预计释放空间：{format_size(total)}"
if wx.MessageBox(message, "确认删除", wx.YES_NO | wx.NO_DEFAULT | wx.ICON_WARNING) != wx.YES:
    return

回收站删除优先使用 send2trash：

try:
    from send2trash import send2trash
except ImportError:
    _windows_recycle(path)
else:
    send2trash(path)

如果没有安装 send2trash，则使用 Windows Shell API 作为备用方案：

operation.wFunc = 3
operation.pFrom = path + ""
operation.fFlags = 0x0040 | 0x0010 | 0x0400
result = shell32.SHFileOperationW(ctypes.byref(operation))

删除完成后，数据库会标记文件状态：

def mark_deleted(self, paths: list[str]) -> None:
    if not paths:
        return
    with closing(self._connect()) as conn:
        conn.executemany(
            "UPDATE duplicate_files SET deleted = 1 WHERE path = ?",
            [(path,) for path in paths],
        )
        conn.commit()

十一、启动脚本

为了双击就能启动，项目提供了一个 start.bat：

@echo off
cd /d "%~dp0"

python app.py

if errorlevel 1 (
    echo.
    echo Failed to start the application.
    echo Please check that Python and wxPython are installed.
    pause
)

cd /d "%~dp0" 可以切换到 bat 文件所在目录，避免双击运行时工作目录不对头。

十二、测试验证

项目用 Python 自带的 unittest 覆盖核心逻辑，内容包括：

• 配置文件默认值和读写。
• MD5 重复文件分组。
• 重复组按大小降序排序。
• 每组默认保留最新文件。
• SQLite 保存和读取扫描结果。
• ZIP 和普通文件预览。
• 回收站删除抽象。

运行测试：

python -m unittest discover -s tests -v

语法检查：

python -m py_compile app.py config.py database.py duplicate_finder.py models.py preview.py recycle.py

检查 wxPython：

python -c "import wx; print('wxPython ok', wx.version())"

十三、运行方式

安装依赖：

pip install wxPython send2trash

启动程序：

python app.py

或者直接双击：

start.bat

总结

这个项目虽然不算大，但覆盖了一个桌面工具从 GUI 交互、后台耗时任务、文件哈希计算、SQLite 持久化、配置保存、文件预览到安全删除、单元测试的完整闭环。

其中三个设计点最值得留意：

1. 先按大小分组，再计算 MD5，可以大幅减少不必要的哈希计算。
2. 扫描线程和 GUI 线程分离，确保界面永远不卡死。
3. 删除走回收站并二次确认，把误删风险降到最低。