特别致谢：echotrace（本项目大量功能参考其实现，提供了重要技术支持）
+ 
From 93ad7b7a1cd1d2c47f89d1e74185b52c7e43566e Mon Sep 17 00:00:00 2001
From: 2977094657 <2977094657@qq.com>
Date: Sat, 24 Jan 2026 18:47:06 +0800
Subject: [PATCH 002/103] =?UTF-8?q?improvement(chat):=20realtime=20?=
=?UTF-8?q?=E7=9B=B4=E8=AF=BB=20WCDB=20=E5=B9=B6=E5=AE=8C=E5=96=84?=
=?UTF-8?q?=E8=BF=BD=E8=B8=AA=E6=97=A5=E5=BF=97?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
- SSE 变更扫描改用 asyncio.to_thread，避免阻塞事件循环
- sessions/messages 支持 source=realtime；realtime 下会话预览改用 session 信息避免缓存陈旧
- realtime sync/sync_all 增加 trace_id 与关键步骤日志，便于定位卡顿/锁竞争
- 支持通过 WECHAT_TOOL_LOG_LEVEL 环境变量覆盖日志级别
---
src/wechat_decrypt_tool/logging_config.py | 23 +-
src/wechat_decrypt_tool/routers/chat.py | 313 ++++++++++++++++++++--
2 files changed, 306 insertions(+), 30 deletions(-)
diff --git a/src/wechat_decrypt_tool/logging_config.py b/src/wechat_decrypt_tool/logging_config.py
index 4635353..35836b6 100644
--- a/src/wechat_decrypt_tool/logging_config.py
+++ b/src/wechat_decrypt_tool/logging_config.py
@@ -3,6 +3,7 @@
"""
import logging
+import os
import sys
from datetime import datetime
from pathlib import Path
@@ -58,6 +59,11 @@ def __init__(self):
def setup_logging(self, log_level: str = "INFO"):
"""设置日志配置"""
+ # Allow overriding via env var for easier debugging (e.g. WECHAT_TOOL_LOG_LEVEL=DEBUG)
+ env_level = str(os.environ.get("WECHAT_TOOL_LOG_LEVEL", "") or "").strip()
+ if env_level:
+ log_level = env_level
+
# 创建日志目录
now = datetime.now()
log_dir = Path("output/logs") / str(now.year) / f"{now.month:02d}" / f"{now.day:02d}"
@@ -88,46 +94,47 @@ def setup_logging(self, log_level: str = "INFO"):
# 文件处理器
file_handler = logging.FileHandler(self.log_file, encoding='utf-8')
file_handler.setFormatter(file_formatter)
- file_handler.setLevel(getattr(logging, log_level.upper()))
+ level = getattr(logging, str(log_level or "INFO").upper(), logging.INFO)
+ file_handler.setLevel(level)
# 控制台处理器
console_handler = logging.StreamHandler(sys.stdout)
console_handler.setFormatter(console_formatter)
- console_handler.setLevel(getattr(logging, log_level.upper()))
+ console_handler.setLevel(level)
# 配置根日志器
- root_logger.setLevel(getattr(logging, log_level.upper()))
+ root_logger.setLevel(level)
root_logger.addHandler(file_handler)
root_logger.addHandler(console_handler)
# 只为uvicorn日志器添加文件处理器，保持其原有的控制台处理器（带颜色）
uvicorn_logger = logging.getLogger("uvicorn")
uvicorn_logger.addHandler(file_handler)
- uvicorn_logger.setLevel(getattr(logging, log_level.upper()))
+ uvicorn_logger.setLevel(level)
# 只为uvicorn.access日志器添加文件处理器
uvicorn_access_logger = logging.getLogger("uvicorn.access")
uvicorn_access_logger.addHandler(file_handler)
- uvicorn_access_logger.setLevel(getattr(logging, log_level.upper()))
+ uvicorn_access_logger.setLevel(level)
# 只为uvicorn.error日志器添加文件处理器
uvicorn_error_logger = logging.getLogger("uvicorn.error")
uvicorn_error_logger.addHandler(file_handler)
- uvicorn_error_logger.setLevel(getattr(logging, log_level.upper()))
+ uvicorn_error_logger.setLevel(level)
# 配置FastAPI日志器
fastapi_logger = logging.getLogger("fastapi")
fastapi_logger.handlers = []
fastapi_logger.addHandler(file_handler)
fastapi_logger.addHandler(console_handler)
- fastapi_logger.setLevel(getattr(logging, log_level.upper()))
+ fastapi_logger.setLevel(level)
# 记录初始化信息
logger = logging.getLogger(__name__)
logger.info("=" * 60)
logger.info("微信解密工具日志系统初始化完成")
logger.info(f"日志文件: {self.log_file}")
- logger.info(f"日志级别: {log_level}")
+ logger.info(f"日志级别: {logging.getLevelName(level)}")
logger.info("=" * 60)
return self.log_file
diff --git a/src/wechat_decrypt_tool/routers/chat.py b/src/wechat_decrypt_tool/routers/chat.py
index 4597bf1..9455ad0 100644
--- a/src/wechat_decrypt_tool/routers/chat.py
+++ b/src/wechat_decrypt_tool/routers/chat.py
@@ -213,6 +213,13 @@ async def stream_chat_realtime_events(
if not db_storage_dir.exists() or not db_storage_dir.is_dir():
raise HTTPException(status_code=400, detail="db_storage directory not found for this account.")
+ logger.info(
+ "[realtime] SSE stream open account=%s interval_ms=%s db_storage=%s",
+ account_dir.name,
+ int(interval_ms),
+ str(db_storage_dir),
+ )
+
async def gen():
last_mtime_ns = 0
last_heartbeat = 0.0
@@ -226,27 +233,40 @@ async def gen():
}
yield f"data: {json.dumps(initial, ensure_ascii=False)}\n\n"
- while True:
- if await request.is_disconnected():
- break
+ try:
+ while True:
+ if await request.is_disconnected():
+ break
- mtime_ns = _scan_db_storage_mtime_ns(db_storage_dir)
- if mtime_ns and mtime_ns != last_mtime_ns:
- last_mtime_ns = mtime_ns
- payload = {
- "type": "change",
- "account": account_dir.name,
- "mtimeNs": int(mtime_ns),
- "ts": int(time.time() * 1000),
- }
- yield f"data: {json.dumps(payload, ensure_ascii=False)}\n\n"
+ # Avoid blocking the event loop on a potentially large directory walk.
+ scan_t0 = time.perf_counter()
+ try:
+ mtime_ns = await asyncio.to_thread(_scan_db_storage_mtime_ns, db_storage_dir)
+ except Exception:
+ mtime_ns = 0
+ scan_ms = (time.perf_counter() - scan_t0) * 1000.0
+ if scan_ms > 1000:
+ logger.warning("[realtime] SSE scan slow account=%s ms=%.1f", account_dir.name, scan_ms)
+
+ if mtime_ns and mtime_ns != last_mtime_ns:
+ last_mtime_ns = mtime_ns
+ payload = {
+ "type": "change",
+ "account": account_dir.name,
+ "mtimeNs": int(mtime_ns),
+ "ts": int(time.time() * 1000),
+ }
+ logger.info("[realtime] SSE change account=%s mtime_ns=%s", account_dir.name, int(mtime_ns))
+ yield f"data: {json.dumps(payload, ensure_ascii=False)}\n\n"
- now = time.time()
- if now - last_heartbeat > 15:
- last_heartbeat = now
- yield ": ping\n\n"
+ now = time.time()
+ if now - last_heartbeat > 15:
+ last_heartbeat = now
+ yield ": ping\n\n"
- await asyncio.sleep(interval_ms / 1000.0)
+ await asyncio.sleep(interval_ms / 1000.0)
+ finally:
+ logger.info("[realtime] SSE stream closed account=%s", account_dir.name)
headers = {"Cache-Control": "no-cache", "Connection": "keep-alive", "X-Accel-Buffering": "no"}
return StreamingResponse(gen(), media_type="text/event-stream", headers=headers)
@@ -337,7 +357,7 @@ def _ensure_session_last_message_table(conn: sqlite3.Connection) -> None:
@router.post("/api/chat/realtime/sync", summary="实时消息同步到解密库（按会话增量）")
-async def sync_chat_realtime_messages(
+def sync_chat_realtime_messages(
request: Request,
username: str,
account: Optional[str] = None,
@@ -357,11 +377,23 @@ async def sync_chat_realtime_messages(
max_scan = 5000
account_dir = _resolve_account_dir(account)
+ trace_id = f"rt-sync-{int(time.time() * 1000)}-{threading.get_ident()}"
+ logger.info(
+ "[%s] realtime sync start account=%s username=%s max_scan=%s",
+ trace_id,
+ account_dir.name,
+ username,
+ int(max_scan),
+ )
# Lock per (account, username) to avoid concurrent writes to the same sqlite tables.
+ logger.info("[%s] acquiring per-session lock account=%s username=%s", trace_id, account_dir.name, username)
with _realtime_sync_lock(account_dir.name, username):
+ logger.info("[%s] per-session lock acquired account=%s username=%s", trace_id, account_dir.name, username)
try:
+ logger.info("[%s] ensure wcdb connected account=%s", trace_id, account_dir.name)
rt_conn = WCDB_REALTIME.ensure_connected(account_dir)
+ logger.info("[%s] wcdb connected account=%s handle=%s", trace_id, account_dir.name, int(rt_conn.handle))
except WCDBRealtimeError as e:
raise HTTPException(status_code=400, detail=str(e))
@@ -369,6 +401,14 @@ async def sync_chat_realtime_messages(
if not resolved:
raise HTTPException(status_code=404, detail="Conversation table not found in decrypted databases.")
msg_db_path, table_name = resolved
+ logger.info(
+ "[%s] resolved decrypted table account=%s username=%s db=%s table=%s",
+ trace_id,
+ account_dir.name,
+ username,
+ str(msg_db_path),
+ table_name,
+ )
msg_conn = sqlite3.connect(str(msg_db_path))
msg_conn.row_factory = sqlite3.Row
@@ -457,8 +497,34 @@ def normalize(item: dict[str, Any]) -> dict[str, Any]:
while scanned < int(max_scan):
take = min(batch_size, int(max_scan) - scanned)
+ logger.info(
+ "[%s] wcdb_get_messages account=%s username=%s take=%s offset=%s",
+ trace_id,
+ account_dir.name,
+ username,
+ int(take),
+ int(offset),
+ )
+ wcdb_t0 = time.perf_counter()
with rt_conn.lock:
raw_rows = _wcdb_get_messages(rt_conn.handle, username, limit=take, offset=offset)
+ wcdb_ms = (time.perf_counter() - wcdb_t0) * 1000.0
+ logger.info(
+ "[%s] wcdb_get_messages done account=%s username=%s rows=%s ms=%.1f",
+ trace_id,
+ account_dir.name,
+ username,
+ len(raw_rows or []),
+ wcdb_ms,
+ )
+ if wcdb_ms > 2000:
+ logger.warning(
+ "[%s] wcdb_get_messages slow account=%s username=%s ms=%.1f",
+ trace_id,
+ account_dir.name,
+ username,
+ wcdb_ms,
+ )
if not raw_rows:
break
@@ -526,9 +592,27 @@ def normalize(item: dict[str, Any]) -> dict[str, Any]:
# Insert older -> newer to keep sqlite btree locality similar to existing data.
values = [tuple(r.get(c) for c in insert_cols) for r in reversed(new_rows)]
+ insert_t0 = time.perf_counter()
msg_conn.executemany(insert_sql, values)
msg_conn.commit()
+ insert_ms = (time.perf_counter() - insert_t0) * 1000.0
inserted = len(new_rows)
+ logger.info(
+ "[%s] sqlite insert done account=%s username=%s inserted=%s ms=%.1f",
+ trace_id,
+ account_dir.name,
+ username,
+ int(inserted),
+ insert_ms,
+ )
+ if insert_ms > 1000:
+ logger.warning(
+ "[%s] sqlite insert slow account=%s username=%s ms=%.1f",
+ trace_id,
+ account_dir.name,
+ username,
+ insert_ms,
+ )
if ("packed_info_data" in insert_cols) and backfill_rows:
update_values = []
@@ -539,12 +623,30 @@ def normalize(item: dict[str, Any]) -> dict[str, Any]:
update_values.append((pdata, int(r.get("local_id") or 0)))
if update_values:
before_changes = msg_conn.total_changes
+ update_t0 = time.perf_counter()
msg_conn.executemany(
f"UPDATE {quoted_table} SET packed_info_data = ? WHERE local_id = ? AND (packed_info_data IS NULL OR length(packed_info_data) = 0)",
update_values,
)
msg_conn.commit()
+ update_ms = (time.perf_counter() - update_t0) * 1000.0
backfilled = int(msg_conn.total_changes - before_changes)
+ logger.info(
+ "[%s] sqlite backfill done account=%s username=%s rows=%s ms=%.1f",
+ trace_id,
+ account_dir.name,
+ username,
+ int(backfilled),
+ update_ms,
+ )
+ if update_ms > 1000:
+ logger.warning(
+ "[%s] sqlite backfill slow account=%s username=%s ms=%.1f",
+ trace_id,
+ account_dir.name,
+ username,
+ update_ms,
+ )
# Update session.db so left sidebar ordering/time can follow new messages.
newest = new_rows[0] if new_rows else None
@@ -636,6 +738,16 @@ def normalize(item: dict[str, Any]) -> dict[str, Any]:
finally:
sconn.close()
+ logger.info(
+ "[%s] realtime sync done account=%s username=%s scanned=%s inserted=%s backfilled=%s maxLocalIdBefore=%s",
+ trace_id,
+ account_dir.name,
+ username,
+ int(scanned),
+ int(inserted),
+ int(backfilled),
+ int(max_local_id),
+ )
return {
"status": "success",
"account": account_dir.name,
@@ -750,8 +862,31 @@ def normalize(item: dict[str, Any]) -> dict[str, Any]:
while scanned < int(max_scan):
take = min(batch_size, int(max_scan) - scanned)
+ logger.info(
+ "[realtime] wcdb_get_messages account=%s username=%s take=%s offset=%s",
+ account_dir.name,
+ username,
+ int(take),
+ int(offset),
+ )
+ wcdb_t0 = time.perf_counter()
with rt_conn.lock:
raw_rows = _wcdb_get_messages(rt_conn.handle, username, limit=take, offset=offset)
+ wcdb_ms = (time.perf_counter() - wcdb_t0) * 1000.0
+ logger.info(
+ "[realtime] wcdb_get_messages done account=%s username=%s rows=%s ms=%.1f",
+ account_dir.name,
+ username,
+ len(raw_rows or []),
+ wcdb_ms,
+ )
+ if wcdb_ms > 2000:
+ logger.warning(
+ "[realtime] wcdb_get_messages slow account=%s username=%s ms=%.1f",
+ account_dir.name,
+ username,
+ wcdb_ms,
+ )
if not raw_rows:
break
@@ -816,9 +951,25 @@ def normalize(item: dict[str, Any]) -> dict[str, Any]:
continue
values = [tuple(r.get(c) for c in insert_cols) for r in reversed(new_rows)]
+ insert_t0 = time.perf_counter()
msg_conn.executemany(insert_sql, values)
msg_conn.commit()
+ insert_ms = (time.perf_counter() - insert_t0) * 1000.0
inserted = len(new_rows)
+ logger.info(
+ "[realtime] sqlite insert done account=%s username=%s inserted=%s ms=%.1f",
+ account_dir.name,
+ username,
+ int(inserted),
+ insert_ms,
+ )
+ if insert_ms > 1000:
+ logger.warning(
+ "[realtime] sqlite insert slow account=%s username=%s ms=%.1f",
+ account_dir.name,
+ username,
+ insert_ms,
+ )
if ("packed_info_data" in insert_cols) and backfill_rows:
update_values = []
@@ -829,12 +980,28 @@ def normalize(item: dict[str, Any]) -> dict[str, Any]:
update_values.append((pdata, int(r.get("local_id") or 0)))
if update_values:
before_changes = msg_conn.total_changes
+ update_t0 = time.perf_counter()
msg_conn.executemany(
f"UPDATE {quoted_table} SET packed_info_data = ? WHERE local_id = ? AND (packed_info_data IS NULL OR length(packed_info_data) = 0)",
update_values,
)
msg_conn.commit()
+ update_ms = (time.perf_counter() - update_t0) * 1000.0
backfilled = int(msg_conn.total_changes - before_changes)
+ logger.info(
+ "[realtime] sqlite backfill done account=%s username=%s rows=%s ms=%.1f",
+ account_dir.name,
+ username,
+ int(backfilled),
+ update_ms,
+ )
+ if update_ms > 1000:
+ logger.warning(
+ "[realtime] sqlite backfill slow account=%s username=%s ms=%.1f",
+ account_dir.name,
+ username,
+ update_ms,
+ )
newest = new_rows[0] if new_rows else None
preview = ""
@@ -938,7 +1105,7 @@ def normalize(item: dict[str, Any]) -> dict[str, Any]:
@router.post("/api/chat/realtime/sync_all", summary="实时消息同步到解密库（全会话增量）")
-async def sync_chat_realtime_messages_all(
+def sync_chat_realtime_messages_all(
request: Request,
account: Optional[str] = None,
max_scan: int = 200,
@@ -953,6 +1120,16 @@ async def sync_chat_realtime_messages_all(
说明：这是增量同步，不会每次全表扫描；priority_username 会优先同步并可设置更大的 priority_max_scan。
"""
account_dir = _resolve_account_dir(account)
+ trace_id = f"rt-syncall-{int(time.time() * 1000)}-{threading.get_ident()}"
+ logger.info(
+ "[%s] realtime sync_all start account=%s max_scan=%s priority=%s include_hidden=%s include_official=%s",
+ trace_id,
+ account_dir.name,
+ int(max_scan),
+ str(priority_username or "").strip(),
+ bool(include_hidden),
+ bool(include_official),
+ )
if max_scan < 20:
max_scan = 20
@@ -966,15 +1143,29 @@ async def sync_chat_realtime_messages_all(
priority = str(priority_username or "").strip()
started = time.time()
+ logger.info("[%s] acquiring global sync lock account=%s", trace_id, account_dir.name)
with _realtime_sync_all_lock(account_dir.name):
+ logger.info("[%s] global sync lock acquired account=%s", trace_id, account_dir.name)
try:
+ logger.info("[%s] ensure wcdb connected account=%s", trace_id, account_dir.name)
rt_conn = WCDB_REALTIME.ensure_connected(account_dir)
+ logger.info("[%s] wcdb connected account=%s handle=%s", trace_id, account_dir.name, int(rt_conn.handle))
except WCDBRealtimeError as e:
raise HTTPException(status_code=400, detail=str(e))
try:
+ logger.info("[%s] wcdb_get_sessions account=%s", trace_id, account_dir.name)
+ wcdb_t0 = time.perf_counter()
with rt_conn.lock:
raw_sessions = _wcdb_get_sessions(rt_conn.handle)
+ wcdb_ms = (time.perf_counter() - wcdb_t0) * 1000.0
+ logger.info(
+ "[%s] wcdb_get_sessions done account=%s sessions=%s ms=%.1f",
+ trace_id,
+ account_dir.name,
+ len(raw_sessions or []),
+ wcdb_ms,
+ )
except Exception:
raw_sessions = []
@@ -1018,6 +1209,13 @@ def _dedupe(items: list[tuple[int, str]]) -> list[tuple[int, str]]:
sessions = _dedupe(sessions)
sessions.sort(key=lambda x: int(x[0] or 0), reverse=True)
all_usernames = [u for _, u in sessions if u]
+ logger.info(
+ "[%s] sessions prepared account=%s raw=%s filtered=%s",
+ trace_id,
+ account_dir.name,
+ len(raw_sessions or []),
+ len(all_usernames),
+ )
# Skip sessions whose decrypted session.db already has a newer/equal sort_timestamp.
decrypted_ts_by_user: dict[str, int] = {}
@@ -1080,10 +1278,25 @@ def _dedupe(items: list[tuple[int, str]]) -> list[tuple[int, str]]:
continue
sync_usernames.append(u)
+ logger.info(
+ "[%s] sessions need_sync account=%s need_sync=%s skipped_up_to_date=%s",
+ trace_id,
+ account_dir.name,
+ len(sync_usernames),
+ int(skipped_up_to_date),
+ )
+
if priority and priority in sync_usernames:
sync_usernames = [priority] + [u for u in sync_usernames if u != priority]
table_map = _resolve_decrypted_message_tables(account_dir, sync_usernames)
+ logger.info(
+ "[%s] resolved decrypted tables account=%s resolved=%s need_sync=%s",
+ trace_id,
+ account_dir.name,
+ len(table_map),
+ len(sync_usernames),
+ )
scanned_total = 0
inserted_total = 0
@@ -1116,17 +1329,50 @@ def _dedupe(items: list[tuple[int, str]]) -> list[tuple[int, str]]:
inserted_total += ins
if ins:
updated_sessions += 1
+ logger.info(
+ "[%s] synced session account=%s username=%s inserted=%s scanned=%s",
+ trace_id,
+ account_dir.name,
+ uname,
+ ins,
+ int(result.get("scanned") or 0),
+ )
except HTTPException as e:
errors.append(f"{uname}: {str(e.detail or '')}".strip())
+ logger.warning(
+ "[%s] sync session failed account=%s username=%s err=%s",
+ trace_id,
+ account_dir.name,
+ uname,
+ str(e.detail or "").strip(),
+ )
continue
except Exception as e:
errors.append(f"{uname}: {str(e)}".strip())
+ logger.exception(
+ "[%s] sync session crashed account=%s username=%s",
+ trace_id,
+ account_dir.name,
+ uname,
+ )
continue
elapsed_ms = int((time.time() - started) * 1000)
if len(errors) > 20:
errors = errors[:20] + [f"... and {len(errors) - 20} more"]
+ logger.info(
+ "[%s] realtime sync_all done account=%s sessions_total=%s need_sync=%s synced=%s updated=%s inserted_total=%s elapsed_ms=%s errors=%s",
+ trace_id,
+ account_dir.name,
+ len(all_usernames),
+ len(sync_usernames),
+ int(synced),
+ int(updated_sessions),
+ int(inserted_total),
+ int(elapsed_ms),
+ len(errors),
+ )
return {
"status": "success",
"account": account_dir.name,
@@ -2134,7 +2380,7 @@ async def list_chat_accounts():
@router.get("/api/chat/sessions", summary="获取会话列表（聊天左侧列表）")
-async def list_chat_sessions(
+def list_chat_sessions(
request: Request,
account: Optional[str] = None,
limit: int = 400,
@@ -2157,10 +2403,32 @@ async def list_chat_sessions(
rows: list[Any]
if source_norm == "realtime":
+ trace_id = f"rt-sessions-{int(time.time() * 1000)}-{threading.get_ident()}"
+ logger.info(
+ "[%s] list_sessions realtime start account=%s limit=%s include_hidden=%s include_official=%s preview=%s",
+ trace_id,
+ account_dir.name,
+ int(limit),
+ bool(include_hidden),
+ bool(include_official),
+ str(preview or ""),
+ )
try:
+ logger.info("[%s] ensure wcdb connected account=%s", trace_id, account_dir.name)
conn = WCDB_REALTIME.ensure_connected(account_dir)
+ logger.info("[%s] wcdb connected account=%s handle=%s", trace_id, account_dir.name, int(conn.handle))
+ logger.info("[%s] wcdb_get_sessions account=%s", trace_id, account_dir.name)
+ wcdb_t0 = time.perf_counter()
with conn.lock:
raw = _wcdb_get_sessions(conn.handle)
+ wcdb_ms = (time.perf_counter() - wcdb_t0) * 1000.0
+ logger.info(
+ "[%s] wcdb_get_sessions done account=%s sessions=%s ms=%.1f",
+ trace_id,
+ account_dir.name,
+ len(raw or []),
+ wcdb_ms,
+ )
except WCDBRealtimeError as e:
raise HTTPException(status_code=400, detail=str(e))
@@ -2193,6 +2461,7 @@ def _ts(v: Any) -> int:
norm.sort(key=lambda r: _ts(r.get("sort_timestamp")), reverse=True)
rows = norm
+ logger.info("[%s] list_sessions realtime normalized account=%s rows=%s", trace_id, account_dir.name, len(rows))
else:
session_db_path = account_dir / "session.db"
sconn = sqlite3.connect(str(session_db_path))
@@ -2873,7 +3142,7 @@ def _collect_chat_messages(
@router.get("/api/chat/messages", summary="获取会话消息列表")
-async def list_chat_messages(
+def list_chat_messages(
request: Request,
username: str,
account: Optional[str] = None,
From ae2d7f128d3b1927aeb456266f886f8d6c50f762 Mon Sep 17 00:00:00 2001
From: 2977094657 <2977094657@qq.com>
Date: Sat, 24 Jan 2026 18:47:29 +0800
Subject: [PATCH 003/103] =?UTF-8?q?improvement(chat):=20realtime=20?=
=?UTF-8?q?=E5=88=B7=E6=96=B0=E5=8E=BB=E6=8A=96=E5=B9=B6=E7=BB=95=E8=BF=87?=
=?UTF-8?q?=E5=90=8E=E5=8F=B0=E5=85=A8=E9=87=8F=E5=90=8C=E6=AD=A5?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
- realtime 模式拉取消息时传 source=realtime，直接从 WCDB 读取
- SSE change 事件增加 500ms debounce，减少频繁刷新/请求抖动
- 停止 realtime 时清理 debounce timer
---
frontend/pages/chat/[[username]].vue | 32 ++++++++++++++++++++++------
1 file changed, 25 insertions(+), 7 deletions(-)
diff --git a/frontend/pages/chat/[[username]].vue b/frontend/pages/chat/[[username]].vue
index 48ed479..b2348ab 100644
--- a/frontend/pages/chat/[[username]].vue
+++ b/frontend/pages/chat/[[username]].vue
@@ -1894,6 +1894,7 @@ let realtimeSessionsRefreshQueued = false
let realtimeFullSyncFuture = null
let realtimeFullSyncQueued = false
let realtimeFullSyncPriority = ''
+let realtimeChangeDebounceTimer = null
const allMessages = ref({})
@@ -4644,9 +4645,9 @@ const loadMessages = async ({ username, reset }) => {
if (messageTypeFilter.value && messageTypeFilter.value !== 'all') {
params.render_types = messageTypeFilter.value
}
-
- if (reset) {
- await queueRealtimeFullSync(username)
+ if (realtimeEnabled.value) {
+ // In realtime mode, read directly from WCDB to avoid blocking on background sync.
+ params.source = 'realtime'
}
const resp = await api.listChatMessages(params)
@@ -4747,6 +4748,12 @@ const stopRealtimeStream = () => {
} catch {}
realtimeEventSource = null
}
+ if (realtimeChangeDebounceTimer) {
+ try {
+ clearTimeout(realtimeChangeDebounceTimer)
+ } catch {}
+ realtimeChangeDebounceTimer = null
+ }
}
const refreshRealtimeIncremental = async () => {
@@ -4774,8 +4781,8 @@ const refreshRealtimeIncremental = async () => {
if (messageTypeFilter.value && messageTypeFilter.value !== 'all') {
params.render_types = messageTypeFilter.value
}
+ params.source = 'realtime'
- await queueRealtimeFullSync(username)
const resp = await api.listChatMessages(params)
if (selectedContact.value?.username !== username) return
@@ -4820,6 +4827,19 @@ const queueRealtimeRefresh = () => {
})
}
+const queueRealtimeChange = () => {
+ if (!process.client || typeof window === 'undefined') return
+ if (!realtimeEnabled.value) return
+ if (realtimeChangeDebounceTimer) return
+
+ // Debounce noisy db_storage change events to avoid hammering the backend.
+ realtimeChangeDebounceTimer = setTimeout(() => {
+ realtimeChangeDebounceTimer = null
+ queueRealtimeRefresh()
+ queueRealtimeSessionsRefresh()
+ }, 500)
+}
+
const startRealtimeStream = () => {
stopRealtimeStream()
if (!process.client || typeof window === 'undefined') return
@@ -4840,9 +4860,7 @@ const startRealtimeStream = () => {
try {
const data = JSON.parse(String(ev.data || '{}'))
if (String(data?.type || '') === 'change') {
- queueRealtimeFullSync(selectedContact.value?.username || '')
- queueRealtimeRefresh()
- queueRealtimeSessionsRefresh()
+ queueRealtimeChange()
}
} catch {}
}
From d0d518aed9ba62d071d6416b51a30ced3e133387 Mon Sep 17 00:00:00 2001
From: 2977094657 <2977094657@qq.com>
Date: Tue, 27 Jan 2026 16:26:53 +0800
Subject: [PATCH 004/103] =?UTF-8?q?fix(chat):=20proxy=5Fimage=20=E5=85=BC?=
=?UTF-8?q?=E5=AE=B9=20tc.qq.com=20=E5=B9=B6=E5=A2=9E=E5=BC=BA=E9=98=B2?=
=?UTF-8?q?=E7=9B=97=E9=93=BE=20Referer?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
- proxy_image 放开 .tc.qq.com 白名单，兼容朋友圈/CDN 图片
- 下载时按多组 Referer/Origin 轮询，提高成功率
- 保持 host 校验与 10MB 限制
---
src/wechat_decrypt_tool/routers/chat_media.py | 67 ++++++++++++-------
1 file changed, 43 insertions(+), 24 deletions(-)
diff --git a/src/wechat_decrypt_tool/routers/chat_media.py b/src/wechat_decrypt_tool/routers/chat_media.py
index 655f0bc..0943374 100644
--- a/src/wechat_decrypt_tool/routers/chat_media.py
+++ b/src/wechat_decrypt_tool/routers/chat_media.py
@@ -414,7 +414,7 @@ def _is_allowed_proxy_image_host(host: str) -> bool:
if not h:
return False
# WeChat public account/article thumbnails and avatars commonly live on these CDNs.
- return h.endswith(".qpic.cn") or h.endswith(".qlogo.cn")
+ return h.endswith(".qpic.cn") or h.endswith(".qlogo.cn") or h.endswith(".tc.qq.com")
@router.get("/api/chat/media/proxy_image", summary="代理获取远程图片（解决微信公众号图片防盗链）")
@@ -435,33 +435,52 @@ async def proxy_image(url: str):
raise HTTPException(status_code=400, detail="Unsupported url host for proxy_image.")
def _download_bytes() -> tuple[bytes, str]:
- headers = {
+ base_headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120 Safari/537.36",
"Accept": "image/avif,image/webp,image/apng,image/*,*/*;q=0.8",
- # qpic/qlogo often require a mp.weixin.qq.com referer (anti-hotlink)
- "Referer": "https://mp.weixin.qq.com/",
- "Origin": "https://mp.weixin.qq.com",
}
- r = requests.get(u, headers=headers, timeout=20, stream=True)
- try:
- r.raise_for_status()
- content_type = str(r.headers.get("Content-Type") or "").strip()
- max_bytes = 10 * 1024 * 1024
- chunks: list[bytes] = []
- total = 0
- for ch in r.iter_content(chunk_size=64 * 1024):
- if not ch:
- continue
- chunks.append(ch)
- total += len(ch)
- if total > max_bytes:
- raise HTTPException(status_code=400, detail="Proxy image too large (>10MB).")
- return b"".join(chunks), content_type
- finally:
+
+ # Different Tencent CDNs enforce different anti-hotlink rules.
+ # Try a couple of safe referers so Moments(qpic) and MP(qpic) both work.
+ header_variants = [
+ {"Referer": "https://wx.qq.com/", "Origin": "https://wx.qq.com"},
+ {"Referer": "https://mp.weixin.qq.com/", "Origin": "https://mp.weixin.qq.com"},
+ {"Referer": "https://www.baidu.com/", "Origin": "https://www.baidu.com"},
+ {},
+ ]
+
+ last_err: Exception | None = None
+ for extra in header_variants:
+ headers = dict(base_headers)
+ headers.update(extra)
+ r = requests.get(u, headers=headers, timeout=20, stream=True)
try:
- r.close()
- except Exception:
- pass
+ r.raise_for_status()
+ content_type = str(r.headers.get("Content-Type") or "").strip()
+ max_bytes = 10 * 1024 * 1024
+ chunks: list[bytes] = []
+ total = 0
+ for ch in r.iter_content(chunk_size=64 * 1024):
+ if not ch:
+ continue
+ chunks.append(ch)
+ total += len(ch)
+ if total > max_bytes:
+ raise HTTPException(status_code=400, detail="Proxy image too large (>10MB).")
+ return b"".join(chunks), content_type
+ except HTTPException:
+ # Hard failure, don't retry with another referer.
+ raise
+ except Exception as e:
+ last_err = e
+ finally:
+ try:
+ r.close()
+ except Exception:
+ pass
+
+ # All variants failed.
+ raise last_err or RuntimeError("proxy_image download failed")
try:
data, ct = await asyncio.to_thread(_download_bytes)
From ba9eb5e2676ab3738fd80fe7b608fe1af262bbc3 Mon Sep 17 00:00:00 2001
From: 2977094657 <2977094657@qq.com>
Date: Tue, 27 Jan 2026 16:27:19 +0800
Subject: [PATCH 005/103] =?UTF-8?q?feat(sns):=20=E5=A2=9E=E5=8A=A0?=
=?UTF-8?q?=E6=9C=8B=E5=8F=8B=E5=9C=88=E6=97=B6=E9=97=B4=E7=BA=BF=E4=B8=8E?=
=?UTF-8?q?=E5=9B=BE=E7=89=87=E6=9C=AC=E5=9C=B0=E7=BC=93=E5=AD=98=E6=8E=A5?=
=?UTF-8?q?=E5=8F=A3?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
- 新增 /api/sns/timeline：优先走 WCDB realtime 读取 sns.db，支持分页/用户过滤/关键字
- 新增 /api/sns/media：本地缓存（cache/.../Sns/Img）解密优先，支持手动 pick/避开重复
- 新增 /api/sns/media_candidates 与 /api/sns/media_picks：候选 key 列表与本机持久化匹配表
- wcdb_realtime 增加 exec_query/get_sns_timeline 封装，并在连接时 set_my_wxid 上下文
- 更新 wcdb_api.dll 并补齐 MSVC runtime 依赖
---
src/wechat_decrypt_tool/api.py | 2 +
src/wechat_decrypt_tool/native/msvcp140.dll | Bin 0 -> 553552 bytes
src/wechat_decrypt_tool/native/msvcp140_1.dll | Bin 0 -> 35488 bytes
.../native/vcruntime140.dll | Bin 0 -> 123472 bytes
.../native/vcruntime140_1.dll | Bin 0 -> 47264 bytes
src/wechat_decrypt_tool/native/wcdb_api.dll | Bin 251904 -> 689152 bytes
src/wechat_decrypt_tool/routers/sns.py | 1068 +++++++++++++++++
src/wechat_decrypt_tool/wcdb_realtime.py | 154 +++
8 files changed, 1224 insertions(+)
create mode 100644 src/wechat_decrypt_tool/native/msvcp140.dll
create mode 100644 src/wechat_decrypt_tool/native/msvcp140_1.dll
create mode 100644 src/wechat_decrypt_tool/native/vcruntime140.dll
create mode 100644 src/wechat_decrypt_tool/native/vcruntime140_1.dll
create mode 100644 src/wechat_decrypt_tool/routers/sns.py
diff --git a/src/wechat_decrypt_tool/api.py b/src/wechat_decrypt_tool/api.py
index df3773f..97ad48b 100644
--- a/src/wechat_decrypt_tool/api.py
+++ b/src/wechat_decrypt_tool/api.py
@@ -18,6 +18,7 @@
from .routers.health import router as _health_router
from .routers.keys import router as _keys_router
from .routers.media import router as _media_router
+from .routers.sns import router as _sns_router
from .routers.wechat_detection import router as _wechat_detection_router
from .wcdb_realtime import WCDB_REALTIME, shutdown as _wcdb_shutdown
@@ -51,6 +52,7 @@
app.include_router(_chat_router)
app.include_router(_chat_export_router)
app.include_router(_chat_media_router)
+app.include_router(_sns_router)
class _SPAStaticFiles(StaticFiles):
diff --git a/src/wechat_decrypt_tool/native/msvcp140.dll b/src/wechat_decrypt_tool/native/msvcp140.dll
new file mode 100644
index 0000000000000000000000000000000000000000..554d2ffc5f5aa84a0598b2f167258187f1fa5842
GIT binary patch
literal 553552
zcmeFadwdkt`3Jn4&5{KIvp`mYsH`PwFbdIlNet=^*@aozg&cyLVP4%_WN`c%G+;tN5&mi1K!B|APEI{x}&
z>upIbukwYZKi6DG*^_am!)Mc2j-KD*(h%Oo8Y*12$6QPb6^-Q&sEMM@Cxz)aGtSCh
z!siyMC(Peqo`UD8^gB{)aj)3YF+)=8xfn!*HmLLyl8h_t-UAc6%$MaVzT;tk4_DD0
zOZ+J%TkeeuA8{~^!21Pc@7Dctc-eG`G9`yXoaB^6g=(JOq^agU0Z_*i1idRM+X>hw
zEiYyrLtboPD%;A74njb+9uFSfV=WhL2_WI%5N1MKaf=^+mIRgyeOC2et}m7Q3HdeX
zSiN}@BwG2^xWhC(B}c-_1cDEeXD)hAPJWqDIr>dmerF_`Yqm9C%JN$x-Rs`9?cbW0
z+iMC#M}WUS?;xX;1JUud&g7RJ+%!1ZkEnYZTQYLA`8{egrf-!p4lCuOD{orx5p}}i
z4Q-SOAA)+hRz)w|X@35mj1^(-e!cGi(Jh$YT98CIP^KO2#{bU$$oT!=YW$feIR3pI
z$KTLl{OD^Ec9`&t=Rt-ifWH0QyH>*6eDZbz{lpLFU2@uat&9-vopKTrDEo*cIClJV
zKE