config.py
4.47 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
"""WeChat Viewer — 微信消息管理 Web 服务"""
import subprocess
import json
import os
from pathlib import Path
BASE_DIR = Path(__file__).parent
# ── 账号自动发现 ──────────────────────────────────────
DEFAULT_KEY = "e03ce81"
def _su_exists(path):
try:
r = subprocess.run(["su","-c",f"test -f {path} && echo yes || echo no"],
capture_output=True, text=True, timeout=5)
return "yes" in r.stdout
except:
return False
def _get_key_for_account(account_id):
"""读取 .wechat_key_<id> 文件，不存在用默认密钥"""
key_file = BASE_DIR / f".wechat_key_{account_id}"
try:
with open(key_file) as f:
k = f.read().strip()
if k:
return k
except:
pass
return DEFAULT_KEY
def _get_nickname(db_path, key):
"""从账号自己的 DB 读取昵称"""
try:
inner = (
f"{SQLCIPHER_BIN} '{db_path}' "
f"-cmd \"PRAGMA key='{key}';\" "
f"-cmd \"PRAGMA cipher_compatibility=1;\" "
f"-cmd \"PRAGMA cipher_page_size=1024;\" "
f"-cmd \"PRAGMA query_only=true;\" "
f"-cmd \".mode json\" "
f"\"SELECT value FROM userinfo WHERE id=4 LIMIT 1;\""
)
r = subprocess.run(["su","-c",inner], capture_output=True, text=True, timeout=10)
# Parse JSON output
for line in r.stdout.strip().split("\n"):
line = line.strip()
if line.startswith("[{") and '"value"' in line:
import json as _json
rows = _json.loads(line)
if rows and isinstance(rows, list) and rows[0].get("value"):
return rows[0]["value"]
except:
pass
return None
def _get_custom_name(account_id):
"""读取 .wechat_name_<id> 文件（用户自定义昵称）"""
name_file = BASE_DIR / f".wechat_name_{account_id}"
try:
with open(name_file) as f:
n = f.read().strip()
if n:
return n
except:
pass
return None
def discover_accounts():
accounts = []
base = "/data/data/com.tencent.mm/MicroMsg"
try:
result = subprocess.run(
["su","-c",f"ls -d {base}/*/ 2>/dev/null"],
capture_output=True, text=True, timeout=5
)
for line in result.stdout.strip().split("\n"):
line = line.strip()
if not line:
continue
db_path = os.path.join(line, "EnMicroMsg.db")
if _su_exists(db_path):
folder = os.path.basename(line.rstrip("/"))
key = _get_key_for_account(folder)
# 名字优先级：自定义名 > 从DB读 > hash前缀
name = _get_custom_name(folder) or _get_nickname(db_path, key) or folder[:8]
accounts.append({
"id": folder,
"name": name,
"db_path": db_path,
"data_root": line.rstrip("/"),
"key": key,
})
except Exception:
pass
if not accounts:
accounts = [{"id": "default", "name": "默认", "db_path": "", "data_root": "", "key": DEFAULT_KEY}]
return accounts
SQLCIPHER_BIN = "/data/data/com.termux/files/usr/bin/sqlcipher"
ACCOUNTS = discover_accounts()
# ── 当前账号状态 ─────────────────────────────────────
ACCOUNT_FILE = str(BASE_DIR / ".current_account")
def get_current_account_id():
try:
with open(ACCOUNT_FILE) as f:
aid = f.read().strip()
if any(a["id"] == aid for a in ACCOUNTS):
return aid
except:
pass
return ACCOUNTS[0]["id"] if ACCOUNTS else "default"
def set_current_account_id(aid):
with open(ACCOUNT_FILE, "w") as f:
f.write(aid)
def get_current_account():
aid = get_current_account_id()
for a in ACCOUNTS:
if a["id"] == aid:
return a
return ACCOUNTS[0] if ACCOUNTS else {"id": "default", "name": "默认", "db_path": "", "data_root": "", "key": DEFAULT_KEY}
# ── 通用配置 ──────────────────────────────────────────
WECHAT_KEY = DEFAULT_KEY # 向后兼容，实际用 account["key"]
LOCAL_DB = str(BASE_DIR / "circle.db")
HOST = "0.0.0.0"
PORT = 5000
DEBUG = False