wechat_db.py
4.65 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
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
"""WeChat DB 读取层 — 通过 sqlcipher CLI 解密查询（只读）"""
import subprocess
import json
import config
def query(sql: str) -> list[dict]:
"""执行只读 SQL 查询，返回 dict 列表"""
cmd = [
"su", "-c",
" ".join([
config.SQLCIPHER_BIN,
config.WECHAT_DB,
f'-cmd "PRAGMA key={config.WECHAT_KEY};"',
'-cmd "PRAGMA cipher_compatibility=1;"',
'-cmd "PRAGMA cipher_page_size=1024;"',
'-cmd "PRAGMA query_only=true;"',
'-cmd ".mode json"',
f'"{sql}"',
])
]
result = subprocess.run(cmd, capture_output=True, text=True, timeout=15)
if result.returncode != 0:
stderr = result.stderr.strip()
if stderr:
raise RuntimeError(f"SQLCipher error: {stderr[:500]}")
return []
output = result.stdout.strip()
if not output:
return []
# sqlcipher sometimes prints "ok\n" before the JSON
if output.startswith("ok"):
output = output[2:].strip()
try:
return json.loads(output)
except json.JSONDecodeError:
# 可能返回多行 JSON，取第一行
lines = output.strip().split("\n")
rows = []
for line in lines:
line = line.strip()
if line:
try:
rows.append(json.loads(line))
except json.JSONDecodeError:
pass
return rows
def get_session_list(limit: int = 200) -> list[dict]:
"""获取会话列表：每个 talker 的最新消息 + 未读数"""
sql = f"""
SELECT
m.talker,
CASE
WHEN m.talker LIKE '%@chatroom' THEN
COALESCE(NULLIF(r.nickname, ''), c.displayname, m.talker)
ELSE
COALESCE(NULLIF(r.conRemark, ''), r.nickname, m.talker)
END AS display_name,
COALESCE(r.conRemark, '') AS remark,
m.content,
m.createTime,
m.type,
m.isSend,
m.imgPath,
CASE
WHEN m.talker LIKE '%@chatroom' THEN 'group'
WHEN m.talker LIKE 'gh_%' THEN 'official'
ELSE 'personal'
END AS talker_type
FROM message m
LEFT JOIN rcontact r ON m.talker = r.username
LEFT JOIN chatroom c ON m.talker = c.chatroomname
WHERE m.createTime = (
SELECT MAX(m2.createTime) FROM message m2 WHERE m2.talker = m.talker
)
ORDER BY m.createTime DESC
LIMIT {limit}
"""
return query(sql)
def get_messages(talker: str, limit: int = 100, offset: int = 0) -> list[dict]:
"""获取指定 talker 的消息列表"""
sql = f"""
SELECT
m.msgId,
m.msgSvrId,
m.type,
m.isSend,
m.talker,
m.content,
m.createTime,
m.imgPath
FROM message m
WHERE m.talker = '{talker}'
ORDER BY m.createTime DESC
LIMIT {limit} OFFSET {offset}
"""
return query(sql)
def search_talker(query_text: str, limit: int = 50) -> list[dict]:
"""搜索会话"""
sql = f"""
SELECT
r.username AS talker,
COALESCE(r.conRemark, r.nickname, r.username) AS display_name,
r.nickname,
r.conRemark,
CASE
WHEN r.username LIKE '%@chatroom' THEN 'group'
WHEN r.username LIKE 'gh_%' THEN 'official'
ELSE 'personal'
END AS talker_type
FROM rcontact r
WHERE r.nickname LIKE '%{query_text}%'
OR r.conRemark LIKE '%{query_text}%'
OR r.username LIKE '%{query_text}%'
LIMIT {limit}
"""
# Also search chatroom
sql2 = f"""
SELECT
c.chatroomname AS talker,
c.displayname AS display_name,
'' AS nickname,
'' AS conRemark,
'group' AS talker_type
FROM chatroom c
WHERE c.displayname LIKE '%{query_text}%'
LIMIT {limit}
"""
rows = query(sql) + query(sql2)
return rows[:limit]
def get_talker_info(talker: str) -> dict | None:
"""获取单个 talker 的详细信息"""
if talker.endswith("@chatroom"):
sql = f"""
SELECT
c.chatroomname AS talker,
COALESCE(NULLIF(r.nickname, ''), c.displayname, c.chatroomname) AS displayname,
c.memberCount
FROM chatroom c
LEFT JOIN rcontact r ON c.chatroomname = r.username
WHERE c.chatroomname = '{talker}'
"""
else:
sql = f"""
SELECT username AS talker, nickname, alias, conRemark,
CASE WHEN username LIKE 'gh_%' THEN 'official'
ELSE 'personal' END AS talker_type
FROM rcontact WHERE username = '{talker}'
"""
rows = query(sql)
return rows[0] if rows else None