feat: cleanup.py + improved error messages
This commit is contained in:
Davy
32
Makefile
32
Makefile
@@ -1,6 +1,24 @@
|
||||
.PHONY: keys decrypt web build
|
||||
.PHONY: setup build keys decrypt web export all status clean help
|
||||
|
||||
PYTHON ?= .venv/bin/python3
|
||||
export SHELL := /bin/bash
|
||||
|
||||
help:
|
||||
@echo "WeChat Decrypt — Makefile"
|
||||
@echo ""
|
||||
@echo " make setup 一键安装所有依赖 + 编译 + 初始配置"
|
||||
@echo " make build 编译 macOS 密钥扫描器"
|
||||
@echo " make keys 提取密钥（需要 root）"
|
||||
@echo " make decrypt 提取密钥 + 解密全部数据库"
|
||||
@echo " make web 启动 Web UI（实时消息监听）"
|
||||
@echo " make export 解密 + 批量导出聊天记录"
|
||||
@echo " make all 从零到完成：setup → keys → decrypt → export"
|
||||
@echo " make status 显示当前数据状态和磁盘用量"
|
||||
@echo " make clean 交互式清理临时数据（解密库/导出/缓存）"
|
||||
@echo ""
|
||||
|
||||
setup:
|
||||
@bash setup.sh
|
||||
|
||||
build:
|
||||
cc -O2 -o find_all_keys_macos find_all_keys_macos.c -framework Foundation
|
||||
@@ -14,3 +32,15 @@ decrypt:
|
||||
|
||||
web:
|
||||
$(PYTHON) main.py
|
||||
|
||||
export:
|
||||
$(PYTHON) main.py export
|
||||
|
||||
all:
|
||||
$(PYTHON) main.py all
|
||||
|
||||
status:
|
||||
$(PYTHON) main.py status
|
||||
|
||||
clean:
|
||||
$(PYTHON) cleanup.py
|
||||
240
cleanup.py
Normal file
240
cleanup.py
Normal file
@@ -0,0 +1,240 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
WeChat Decrypt — 数据清理工具
|
||||
|
||||
安全地查看和解密导出数据占用的磁盘空间，交互式清理。
|
||||
|
||||
用法:
|
||||
python3 cleanup.py # 交互式清理
|
||||
python3 cleanup.py status # 仅显示磁盘用量
|
||||
python3 cleanup.py --dry-run # 显示将删除的内容但不实际操作
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import glob
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import sys
|
||||
|
||||
|
||||
def format_size(size_bytes):
|
||||
"""格式化文件大小"""
|
||||
if size_bytes > 1024 * 1024 * 1024:
|
||||
return f"{size_bytes / 1024 / 1024 / 1024:.1f} GB"
|
||||
elif size_bytes > 1024 * 1024:
|
||||
return f"{size_bytes / 1024 / 1024:.0f} MB"
|
||||
elif size_bytes > 1024:
|
||||
return f"{size_bytes / 1024:.0f} KB"
|
||||
else:
|
||||
return f"{size_bytes} B"
|
||||
|
||||
|
||||
class CleanupItem:
|
||||
def __init__(self, name, path, is_dir=True, pattern=None, description=""):
|
||||
self.name = name
|
||||
self.path = path
|
||||
self.is_dir = is_dir
|
||||
self.pattern = pattern
|
||||
self.description = description
|
||||
|
||||
def size(self):
|
||||
if not self.exists():
|
||||
return 0
|
||||
if self.is_dir:
|
||||
if self.pattern:
|
||||
files = glob.glob(os.path.join(self.path, self.pattern), recursive=True)
|
||||
files = [f for f in files if os.path.isfile(f)]
|
||||
else:
|
||||
files = []
|
||||
for root, dirs, fnames in os.walk(self.path):
|
||||
for fname in fnames:
|
||||
files.append(os.path.join(root, fname))
|
||||
return sum(os.path.getsize(f) for f in files)
|
||||
else:
|
||||
return os.path.getsize(self.path) if os.path.isfile(self.path) else 0
|
||||
|
||||
def exists(self):
|
||||
if self.is_dir:
|
||||
return os.path.isdir(self.path)
|
||||
return os.path.isfile(self.path)
|
||||
|
||||
def delete(self):
|
||||
if not self.exists():
|
||||
return
|
||||
if self.is_dir:
|
||||
shutil.rmtree(self.path)
|
||||
else:
|
||||
os.unlink(self.path)
|
||||
|
||||
|
||||
def get_items():
|
||||
"""返回所有可清理项的列表"""
|
||||
items = []
|
||||
|
||||
# 解密数据库
|
||||
cfg = {}
|
||||
if os.path.exists("config.json"):
|
||||
with open("config.json") as f:
|
||||
cfg = json.load(f)
|
||||
decrypted_dir = cfg.get("decrypted_dir", "decrypted")
|
||||
items.append(CleanupItem(
|
||||
"解密数据库", decrypted_dir,
|
||||
description="解密后的 SQLite 数据库文件（可重新解密恢复）"
|
||||
))
|
||||
|
||||
# WAV 解码缓存
|
||||
items.append(CleanupItem(
|
||||
"语音 WAV 缓存", "decoded_voices",
|
||||
description="SILK 解码后的临时 WAV 文件（可重新解码）"
|
||||
))
|
||||
|
||||
# 图片解码缓存
|
||||
items.append(CleanupItem(
|
||||
"图片解码缓存", "decoded_images",
|
||||
description="解密后的图片缓存"
|
||||
))
|
||||
|
||||
# 导出 JSON
|
||||
items.append(CleanupItem(
|
||||
"导出聊天记录", "exported_chats",
|
||||
description="export_all_chats.py 导出的 JSON 文件（可重新导出）"
|
||||
))
|
||||
|
||||
# 旧格式导出
|
||||
items.append(CleanupItem(
|
||||
"旧格式导出", "exports",
|
||||
description="旧版本导出的数据"
|
||||
))
|
||||
|
||||
# 密钥文件
|
||||
for kf in sorted(glob.glob("all_keys*.json")):
|
||||
items.append(CleanupItem(
|
||||
os.path.basename(kf), kf, is_dir=False,
|
||||
description="密钥缓存文件（可重新提取）"
|
||||
))
|
||||
|
||||
return items
|
||||
|
||||
|
||||
def show_status(items):
|
||||
"""显示各项目的磁盘用量"""
|
||||
total = 0
|
||||
rows = []
|
||||
for item in items:
|
||||
sz = item.size()
|
||||
if sz > 0:
|
||||
total += sz
|
||||
rows.append((item.name, sz, item.description))
|
||||
|
||||
if not rows:
|
||||
print("没有需要清理的数据。")
|
||||
return 0
|
||||
|
||||
# 找最长的名称
|
||||
name_width = max(len(r[0]) for r in rows) + 2
|
||||
print(f"{'项目':<{name_width}}{'大小':>10} 说明")
|
||||
print("-" * (name_width + 45))
|
||||
for name, sz, desc in rows:
|
||||
print(f"{name:<{name_width}}{format_size(sz):>10} {desc}")
|
||||
print("-" * (name_width + 45))
|
||||
print(f"{'总计':<{name_width}}{format_size(total):>10}")
|
||||
return total
|
||||
|
||||
|
||||
def cleanup(dry_run=False):
|
||||
"""交互式清理"""
|
||||
items = get_items()
|
||||
|
||||
print("=" * 60)
|
||||
print(" 磁盘用量分析")
|
||||
print("=" * 60)
|
||||
print()
|
||||
total = show_status(items)
|
||||
if total == 0:
|
||||
print()
|
||||
print("没有需要清理的数据。")
|
||||
return
|
||||
|
||||
print()
|
||||
print("选择要删除的项目（逗号分隔，如: 1,3,5）:")
|
||||
print(" 输入 a 选择全部")
|
||||
print(" 输入 n 取消")
|
||||
choice = input("> ").strip().lower()
|
||||
|
||||
if choice in ("", "n"):
|
||||
print("已取消。")
|
||||
return
|
||||
|
||||
# 解析选择
|
||||
indices = []
|
||||
if choice == "a":
|
||||
indices = list(range(len(items)))
|
||||
else:
|
||||
for part in choice.split(","):
|
||||
part = part.strip()
|
||||
try:
|
||||
idx = int(part) - 1
|
||||
if 0 <= idx < len(items):
|
||||
indices.append(idx)
|
||||
except ValueError:
|
||||
pass
|
||||
|
||||
if not indices:
|
||||
print("未选择任何项目。")
|
||||
return
|
||||
|
||||
# 确认
|
||||
total_saved = 0
|
||||
print()
|
||||
for idx in indices:
|
||||
item = items[idx]
|
||||
if item.exists():
|
||||
sz = item.size()
|
||||
total_saved += sz
|
||||
print(f" [{idx+1}] {item.name} ({format_size(sz)})")
|
||||
|
||||
print(f"\n将释放 {format_size(total_saved)} 磁盘空间")
|
||||
if dry_run:
|
||||
print("（dry-run 模式，未实际删除）")
|
||||
return
|
||||
|
||||
confirm = input("确认删除？(y/N): ").strip().lower()
|
||||
if confirm != "y":
|
||||
print("已取消。")
|
||||
return
|
||||
|
||||
# 执行删除
|
||||
for idx in indices:
|
||||
item = items[idx]
|
||||
if item.exists():
|
||||
sz = item.size()
|
||||
item.delete()
|
||||
print(f" 已删除: {item.name} ({format_size(sz)})")
|
||||
|
||||
print()
|
||||
# 显示剩余
|
||||
remaining = sum(item.size() for item in get_items())
|
||||
print(f"剩余: {format_size(remaining)}")
|
||||
print("清理完成。")
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(
|
||||
description="WeChat Decrypt — 数据清理工具",
|
||||
)
|
||||
parser.add_argument("mode", nargs="?", default="interactive",
|
||||
choices=["interactive", "status"],
|
||||
help="interactive（默认）或 status（仅显示）")
|
||||
parser.add_argument("--dry-run", action="store_true",
|
||||
help="预览模式，不实际删除")
|
||||
args = parser.parse_args()
|
||||
|
||||
if args.mode == "status":
|
||||
show_status(get_items())
|
||||
else:
|
||||
cleanup(dry_run=args.dry_run)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -112,8 +112,8 @@ def main():
|
||||
|
||||
# 加载密钥
|
||||
if not os.path.exists(KEYS_FILE):
|
||||
print(f"[ERROR] 密钥文件不存在: {KEYS_FILE}")
|
||||
print("请先运行 find_all_keys.py")
|
||||
print(f"[ERROR] 密钥文件不存在: {KEYS_FILE}")
|
||||
print("请先运行 python main.py decrypt 提取密钥并解密")
|
||||
sys.exit(1)
|
||||
|
||||
with open(KEYS_FILE, encoding="utf-8") as f:
|
||||
@@ -146,7 +146,7 @@ def main():
|
||||
for rel, path, sz in db_files:
|
||||
key_info = get_key_info(keys, rel)
|
||||
if not key_info:
|
||||
print(f"SKIP: {rel} (无密钥)")
|
||||
print(f"SKIP: {rel} (无密钥，如已安装微信补丁可能需要重新运行密钥提取)")
|
||||
skipped += 1
|
||||
continue
|
||||
|
||||
|
||||
Reference in New Issue
Block a user