wechat_db.py
10 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
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
"""WeChat DB 读取层 — 通过 sqlcipher CLI 解密查询（只读）"""
import subprocess
import json
import config
def query(sql: str) -> list[dict]:
"""执行只读 SQL 查询，返回 dict 列表"""
acct = config.get_current_account()
cmd = [
"su", "-c",
" ".join([
config.SQLCIPHER_BIN,
acct["db_path"],
f'-cmd "PRAGMA key=' + acct["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 count_sessions() -> int:
"""统计不同 talker 数量"""
sql = "SELECT COUNT(DISTINCT talker) AS cnt FROM message"
rows = query(sql)
return rows[0]["cnt"] if rows else 0
def get_session_list(page: int = 1, per_page: int = 10,
sort: str = 'time_desc') -> list[dict]:
"""获取会话列表：每个 talker 的最新消息 + 未读数。支持分页+排序。"""
# 边界保护
page = max(1, page)
per_page = max(1, min(500, per_page))
offset = (page - 1) * per_page
# sort clause
if sort == 'time_asc':
sort_clause = "m.createTime ASC"
elif sort == 'name':
# name sort needs subquery because display_name is computed
sql = f"""
SELECT * FROM (
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 display_name ASC
LIMIT {per_page} OFFSET {offset}
"""
return query(sql)
else: # time_desc (default)
sort_clause = "m.createTime DESC"
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 {sort_clause}
LIMIT {per_page} OFFSET {offset}
"""
return query(sql)
def get_messages(talker: str, limit: int = 100, offset: int = 0, date: str = "") -> list[dict]:
"""获取指定 talker 的消息列表
Args:
date: 可选，格式 YYYY-MM-DD。过滤当天 00:00:00 ~ 23:59:59 的消息。
createTime 是毫秒时间戳，需转为当天起止毫秒数做 BETWEEN。
"""
where_clause = f"m.talker = '{talker}'"
if date:
# date = "2026-07-20" → start_ms = 当天 00:00:00 的毫秒时间戳
import time as _time
from datetime import datetime as _dt
try:
d = _dt.strptime(date, "%Y-%m-%d")
start_s = _time.mktime(d.timetuple())
end_s = _time.mktime(d.replace(hour=23, minute=59, second=59).timetuple())
start_ms = int(start_s) * 1000
end_ms = int(end_s) * 1000
where_clause = f"m.talker = '{talker}' AND m.createTime BETWEEN {start_ms} AND {end_ms}"
except ValueError:
pass # date 格式不对，忽略 date 过滤
sql = f"""
SELECT
m.msgId,
m.msgSvrId,
m.type,
m.isSend,
m.talker,
m.content,
m.createTime,
m.imgPath
FROM message m
WHERE {where_clause}
ORDER BY m.createTime DESC
LIMIT {limit} OFFSET {offset}
"""
rows = query(sql)
# Extract sender_username from content for group messages
import re
is_group = talker.endswith("@chatroom")
for row in rows:
if is_group and row.get("isSend") != 1:
content = row.get("content") or ""
m = re.match(r"^([^:\n<]+):\n", content)
row["sender_username"] = m.group(1) if m else None
elif not is_group:
row["sender_username"] = talker
else:
row["sender_username"] = None
return rows
def get_video_meta_batch(msg_ids: list) -> dict:
"""videoinfo2 批量查询（2026-09-06 新增）：msglocalid -> {duration, filename, human}
type=43 视频消息的 content 只有 'sender:0:0' 占位（无标题无XML），
视频时长/拍摄者只在 videoinfo2 表里。msglocalid 精确等于 message.msgId。
"""
if not msg_ids:
return {}
ids = ",".join(str(int(i)) for i in msg_ids if i)
if not ids:
return {}
sql = (
"SELECT msglocalid, videolength, filename, human FROM videoinfo2 "
f"WHERE msglocalid IN ({ids})"
)
out = {}
try:
for r in query(sql):
try:
dur = int(r.get("videolength") or 0)
except (TypeError, ValueError):
dur = 0
out[r.get("msglocalid")] = {
"duration": dur,
"filename": r.get("filename") or "",
"human": r.get("human") or "",
}
except Exception:
pass
return out
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
def get_talker_infos_batch(talkers: list[str]) -> dict[str, str]:
"""批量获取 talker 名称，最多 2 次 DB 查询（chatroom + contact）"""
if not talkers:
return {}
result = {}
chatrooms = [t for t in talkers if t.endswith("@chatroom")]
others = [t for t in talkers if not t.endswith("@chatroom")]
if chatrooms:
placeholders = ",".join(f"'{t}'" for t in chatrooms)
sql = f"""
SELECT c.chatroomname AS talker,
COALESCE(NULLIF(r.nickname, ''), c.displayname, c.chatroomname) AS name
FROM chatroom c
LEFT JOIN rcontact r ON c.chatroomname = r.username
WHERE c.chatroomname IN ({placeholders})
"""
for row in query(sql):
result[row.get("talker", "")] = row.get("name", "")
if others:
placeholders = ",".join(f"'{t}'" for t in others)
sql = f"""
SELECT username AS talker,
COALESCE(NULLIF(conRemark, ''), NULLIF(nickname, ''), username) AS name
FROM rcontact WHERE username IN ({placeholders})
"""
for row in query(sql):
result[row.get("talker", "")] = row.get("name", "")
return result