From 7827921976d1979029676c908ed191a32de6e993 Mon Sep 17 00:00:00 2001 From: Leonidas Zhak <70497898+LeonidasZhak@users.noreply.github.com> Date: Wed, 3 Jun 2026 23:30:04 +0800 Subject: [PATCH] Add cleanup audit manifests --- main.py | 29 ++++++++ readme.md | 1 + tests/test_cleanup_manifest.py | 76 ++++++++++++++++++++ utils/cleanupManifest.py | 127 +++++++++++++++++++++++++++++++++ utils/multiDeleteThread.py | 57 +++++++-------- 5 files changed, 258 insertions(+), 32 deletions(-) create mode 100644 tests/test_cleanup_manifest.py create mode 100644 utils/cleanupManifest.py diff --git a/main.py b/main.py index 2e9275f95..2349027b0 100644 --- a/main.py +++ b/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): diff --git a/readme.md b/readme.md index ddc9b3cac..41549350d 100644 --- a/readme.md +++ b/readme.md @@ -33,6 +33,7 @@ https://wwbie.lanzoue.com/iQlBl3v2rk6f) 3. 自由设置需要删除的文件的距离时间，默认 365 天； 4. 删除后的文件放置在回收站中，检查后自行清空，防止删错需要的文件； 5. 支持定期自动清理； +6. 清理完成后生成本地 JSON 审计记录，便于复查已处理、跳过和失败的路径； ## 运行截图 diff --git a/tests/test_cleanup_manifest.py b/tests/test_cleanup_manifest.py new file mode 100644 index 000000000..e1860d981 --- /dev/null +++ b/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() diff --git a/utils/cleanupManifest.py b/utils/cleanupManifest.py new file mode 100644 index 000000000..00776bc24 --- /dev/null +++ b/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) diff --git a/utils/multiDeleteThread.py b/utils/multiDeleteThread.py index 53f5d5017..aedbccb23 100644 --- a/utils/multiDeleteThread.py +++ b/utils/multiDeleteThread.py @@ -1,20 +1,9 @@ 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() @@ -22,29 +11,32 @@ def is_protected_file(file_path): 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)