Add cleanup audit manifests

Author: LeonidasZhak

main.py

@@ -64,6 +64,7 @@ def excepthook(exc_type, exc_value, exc_traceback):
 import datetime, time, re, math, shutil, json, logging
 
 from utils.deleteThread import *
+from utils.cleanupManifest import combine_cleanup_results, write_cleanup_manifest
 from utils.multiDeleteThread import multiDeleteThread
 from utils.selectVersion import *
 from utils.selectVersion import check_dir, existing_user_config, find_all_wechat_paths, get_dir_name, \
@@ -104,6 +105,7 @@ def ensure_writable_dir(path):
 STATE_PATH = os.path.join(working_dir, "clean_state.json")
 WHITELIST_PATH = os.path.join(working_dir, "whitelist.txt")
 PREVIEW_PATH = os.path.join(working_dir, "last_scan_preview.txt")
+CLEANUP_MANIFEST_DIR = os.path.join(working_dir, "cleanup_manifests")
 APP_ICON_PATH = os.path.join(resource_dir, "images", "wechat.png")
 
 logging.basicConfig(
@@ -1052,8 +1054,11 @@ def execute_delete(self):
 
         share_thread_arr = [0]
         direct_delete = load_config_file().get("global", {}).get("direct_delete", False)
+        self.cleanup_results = []
+        self.cleanup_expected_threads = 1
         thread = multiDeleteThread(selected_files, selected_dirs, share_thread_arr, direct_delete=direct_delete)
         thread.delete_process_signal.connect(self.callback)
+        thread.delete_complete_signal.connect(self.on_delete_complete)
         self.thread_list.append(thread)
         thread.start()
 
@@ -1771,6 +1776,26 @@ def callback(self, v):
                 self.auto_clean_running = False
             return
 
+    def on_delete_complete(self, result):
+        if not hasattr(self, "cleanup_results"):
+            self.cleanup_results = []
+        self.cleanup_results.append(result)
+        expected = max(1, int(getattr(self, "cleanup_expected_threads", 1)))
+        if len(self.cleanup_results) < expected:
+            return
+
+        combined = combine_cleanup_results(self.cleanup_results)
+        manifest_path = write_cleanup_manifest(combined, CLEANUP_MANIFEST_DIR)
+        self.last_cleanup_manifest = manifest_path
+        logging.info("清理审计记录已保存：%s", manifest_path)
+
+        failed_count = combined.get("failed_count", 0)
+        skipped_count = combined.get("skipped_count", 0)
+        if failed_count:
+            self.setWarninginfo(f"清理完成，但有 {failed_count} 个失败。审计记录：{manifest_path}")
+        else:
+            self.setSuccessinfo(f"清理完成，审计记录已保存：{manifest_path}。跳过 {skipped_count} 个受保护或不存在的路径。")
+
     def should_run_auto_clean(self, config):
         global_config = config.get("global", {})
         if not global_config.get("auto_clean_enable", False):
@@ -1804,6 +1829,8 @@ def justdoit(self, auto_mode=False):
         detail_lines = []
         share_thread_arr = [0]
         system_cache_added = False
+        self.cleanup_results = []
+        self.cleanup_expected_threads = 0
 
         for i, value in enumerate(self.config.get("users", [])):
             file_list = []
@@ -1832,6 +1859,7 @@ def justdoit(self, auto_mode=False):
                 direct_delete = self.config.get("global", {}).get("direct_delete", False)
                 thread = multiDeleteThread(file_list, dir_list, share_thread_arr, direct_delete=direct_delete)
                 thread.delete_process_signal.connect(self.callback)
+                thread.delete_complete_signal.connect(self.on_delete_complete)
                 self.thread_list.append(thread)
 
         if not need_clean:
@@ -1842,6 +1870,7 @@ def justdoit(self, auto_mode=False):
             self.total_file = total_file
             self.total_dir = total_dir
             self.total_size = total_stats.get("total_size", 0)
+            self.cleanup_expected_threads = len(self.thread_list)
             self.bar_progress.setRange(0, 100)
             self.bar_progress.setValue(0)
             if not auto_mode or self.config.get("global", {}).get("auto_clean_confirm", True):

readme.md

@@ -33,6 +33,7 @@ https://wwbie.lanzoue.com/iQlBl3v2rk6f)
 3. 自由设置需要删除的文件的距离时间，默认 365 天；
 4. 删除后的文件放置在回收站中，检查后自行清空，防止删错需要的文件；
 5. 支持定期自动清理；
+6. 清理完成后生成本地 JSON 审计记录，便于复查已处理、跳过和失败的路径；
 
 ## 运行截图
 

tests/test_cleanup_manifest.py

@@ -0,0 +1,76 @@
+import json
+import tempfile
+import unittest
+from pathlib import Path
+
+from utils.cleanupManifest import cleanup_result_summary, delete_path_for_manifest, write_cleanup_manifest
+
+
+def write_file(path, content=b"x"):
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_bytes(content)
+    return path
+
+
+class CleanupManifestTest(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.root = Path(self.tmp.name)
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def test_delete_records_success_skips_and_failures(self):
+        ok_file = write_file(self.root / "old-video.mp4", b"video")
+        protected_file = write_file(self.root / "message.db", b"sqlite")
+        busy_file = write_file(self.root / "busy.tmp", b"busy")
+        trashed = []
+
+        def fake_trash(path):
+            if Path(path).name == "busy.tmp":
+                raise RuntimeError("file is busy")
+            trashed.append(path)
+
+        records = [
+            delete_path_for_manifest(str(ok_file), "file", direct_delete=False, trash_func=fake_trash),
+            delete_path_for_manifest(str(protected_file), "file", direct_delete=False, trash_func=fake_trash),
+            delete_path_for_manifest(str(busy_file), "file", direct_delete=False, trash_func=fake_trash),
+        ]
+        result = cleanup_result_summary(records, direct_delete=False)
+
+        by_name = {Path(row["path"]).name: row for row in result["records"]}
+        self.assertEqual(by_name["old-video.mp4"]["status"], "trashed")
+        self.assertEqual(by_name["message.db"]["status"], "skipped")
+        self.assertEqual(by_name["message.db"]["reason"], "protected_extension")
+        self.assertEqual(by_name["busy.tmp"]["status"], "failed")
+        self.assertEqual(by_name["busy.tmp"]["error"], "file is busy")
+        self.assertEqual(result["processed_count"], 1)
+        self.assertEqual(result["skipped_count"], 1)
+        self.assertEqual(result["failed_count"], 1)
+        self.assertEqual([Path(path).name for path in trashed], ["old-video.mp4"])
+
+    def test_write_cleanup_manifest_persists_json_summary(self):
+        records = [
+            {
+                "path": "/tmp/old-video.mp4",
+                "type": "file",
+                "action": "trash",
+                "status": "trashed",
+                "size_bytes": 5,
+                "size": "5 B",
+                "error": "",
+            }
+        ]
+        result = cleanup_result_summary(records, direct_delete=False)
+
+        manifest_path = write_cleanup_manifest(result, self.root / "cleanup_manifests")
+        payload = json.loads(Path(manifest_path).read_text(encoding="utf-8"))
+
+        self.assertEqual(payload["schema_version"], 1)
+        self.assertEqual(payload["action"], "trash")
+        self.assertEqual(payload["processed_count"], 1)
+        self.assertEqual(payload["records"][0]["path"], "/tmp/old-video.mp4")
+
+
+if __name__ == "__main__":
+    unittest.main()

utils/cleanupManifest.py

@@ -0,0 +1,127 @@
+import json
+import os
+import shutil
+from datetime import datetime
+from pathlib import Path
+
+
+PROTECTED_EXTS = {
+    ".db", ".sqlite", ".sqlite3", ".db-shm", ".db-wal", ".ldb", ".sst",
+    ".dll", ".exe", ".msi", ".sys", ".ocx", ".pyd", ".so", ".dylib",
+    ".bat", ".cmd", ".ps1", ".vbs", ".js", ".jar", ".pak",
+}
+
+
+def is_protected_file(file_path):
+    return os.path.splitext(str(file_path))[1].lower() in PROTECTED_EXTS
+
+
+def human_size(num_bytes):
+    value = float(num_bytes or 0)
+    for unit in ["B", "KB", "MB", "GB", "TB"]:
+        if value < 1024 or unit == "TB":
+            if unit == "B":
+                return f"{int(value)} {unit}"
+            return f"{value:.2f} {unit}"
+        value /= 1024
+    return f"{num_bytes} B"
+
+
+def path_size(path):
+    path = Path(path)
+    try:
+        if path.is_file() or path.is_symlink():
+            return path.stat().st_size
+        if path.is_dir():
+            total = 0
+            for root, dirs, files in os.walk(path):
+                root_path = Path(root)
+                dirs[:] = [name for name in dirs if not (root_path / name).is_symlink()]
+                for filename in files:
+                    try:
+                        total += (root_path / filename).stat().st_size
+                    except OSError:
+                        continue
+            return total
+    except OSError:
+        return 0
+    return 0
+
+
+def permanent_delete(path):
+    if os.path.isdir(path) and not os.path.islink(path):
+        shutil.rmtree(path)
+    else:
+        os.remove(path)
+
+
+def delete_path_for_manifest(file_path, item_type, direct_delete=False, trash_func=None, delete_func=None):
+    size_bytes = path_size(file_path)
+    record = {
+        "path": str(file_path),
+        "type": item_type,
+        "action": "delete" if direct_delete else "trash",
+        "status": "",
+        "size_bytes": size_bytes,
+        "size": human_size(size_bytes),
+        "error": "",
+    }
+    if not os.path.exists(file_path):
+        record["status"] = "skipped"
+        record["reason"] = "missing"
+        return record
+    if is_protected_file(file_path):
+        record["status"] = "skipped"
+        record["reason"] = "protected_extension"
+        return record
+    try:
+        if direct_delete:
+            (delete_func or permanent_delete)(file_path)
+            record["status"] = "deleted"
+        else:
+            if trash_func is None:
+                raise RuntimeError("trash_func is not configured")
+            trash_func(file_path)
+            record["status"] = "trashed"
+    except Exception as exc:  # noqa: BLE001 - persisted for the cleanup report.
+        record["status"] = "failed"
+        record["error"] = str(exc)
+    return record
+
+
+def cleanup_result_summary(records, direct_delete=False):
+    processed = [row for row in records if row.get("status") in {"deleted", "trashed"}]
+    skipped = [row for row in records if row.get("status") == "skipped"]
+    failed = [row for row in records if row.get("status") == "failed"]
+    processed_size = sum(row.get("size_bytes", 0) for row in processed)
+    return {
+        "schema_version": 1,
+        "generated_at": datetime.now().isoformat(timespec="seconds"),
+        "action": "delete" if direct_delete else "trash",
+        "direct_delete": direct_delete,
+        "total_count": len(records),
+        "processed_count": len(processed),
+        "skipped_count": len(skipped),
+        "failed_count": len(failed),
+        "processed_size_bytes": processed_size,
+        "processed_size": human_size(processed_size),
+        "records": records,
+    }
+
+
+def combine_cleanup_results(results):
+    records = []
+    direct_delete = False
+    for result in results:
+        records.extend(result.get("records", []))
+        direct_delete = direct_delete or bool(result.get("direct_delete"))
+    return cleanup_result_summary(records, direct_delete=direct_delete)
+
+
+def write_cleanup_manifest(result, output_dir):
+    output_dir = Path(output_dir)
+    output_dir.mkdir(parents=True, exist_ok=True)
+    stamp = datetime.now().strftime("%Y%m%d_%H%M%S_%f")
+    path = output_dir / f"cleanup_manifest_{stamp}.json"
+    path.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
+    return str(path)

utils/multiDeleteThread.py

@@ -1,50 +1,42 @@
 import logging
-import os
-import shutil
 
 from PyQt5.QtCore import QMutex, QThread, pyqtSignal
 from send2trash import send2trash
 
-
-PROTECTED_EXTS = {
-    '.db', '.sqlite', '.sqlite3', '.db-shm', '.db-wal', '.ldb', '.sst',
-    '.dll', '.exe', '.msi', '.sys', '.ocx', '.pyd', '.so', '.dylib',
-    '.bat', '.cmd', '.ps1', '.vbs', '.js', '.jar', '.pak'
-}
-
-
-def is_protected_file(file_path):
-    return os.path.splitext(str(file_path))[1].lower() in PROTECTED_EXTS
+from utils.cleanupManifest import cleanup_result_summary, delete_path_for_manifest
 
 
 qmut = QMutex()
 
 
 class multiDeleteThread(QThread):
     delete_process_signal = pyqtSignal(int)
-    delete_complete_signal = pyqtSignal()
+    delete_complete_signal = pyqtSignal(dict)
 
-    def __init__(self, fileList, dirList, share_thread_arr, direct_delete=False):
+    def __init__(self, fileList, dirList, share_thread_arr, direct_delete=False, trash_func=None, delete_func=None):
         super(multiDeleteThread, self).__init__()
         self.fileList = fileList
         self.dirList = dirList
         self.share_thread_arr = share_thread_arr
         self.direct_delete = direct_delete
-
-    def _delete_path(self, file_path):
-        if is_protected_file(file_path):
-            logging.info("Skip protected file: %s", file_path)
-            return
-        try:
-            if self.direct_delete:
-                if os.path.isdir(file_path):
-                    shutil.rmtree(file_path)
-                else:
-                    os.remove(file_path)
-            else:
-                send2trash(file_path)
-        except Exception:
-            logging.exception("Failed to delete path: %s", file_path)
+        self.trash_func = trash_func or send2trash
+        self.delete_func = delete_func
+        self.records = []
+        self.result = cleanup_result_summary([], direct_delete=direct_delete)
+
+    def _delete_path(self, file_path, item_type):
+        record = delete_path_for_manifest(
+            file_path,
+            item_type,
+            direct_delete=self.direct_delete,
+            trash_func=self.trash_func,
+            delete_func=self.delete_func,
+        )
+        if record["status"] == "skipped":
+            logging.info("Skip cleanup path: %s (%s)", file_path, record.get("reason", ""))
+        elif record["status"] == "failed":
+            logging.error("Failed to delete path: %s: %s", file_path, record.get("error", ""))
+        return record
 
     def _emit_progress(self):
         qmut.lock()
@@ -57,13 +49,14 @@ def _emit_progress(self):
     def run(self):
         try:
             for file_path in self.fileList:
-                self._delete_path(file_path)
+                self.records.append(self._delete_path(file_path, "file"))
                 self._emit_progress()
 
             for file_path in self.dirList:
-                self._delete_path(file_path)
+                self.records.append(self._delete_path(file_path, "dir"))
                 self._emit_progress()
 
             logging.info("Delete thread finished")
         finally:
-            self.delete_complete_signal.emit()
+            self.result = cleanup_result_summary(self.records, direct_delete=self.direct_delete)
+            self.delete_complete_signal.emit(self.result)