From 602bee8fdfd7a8a8bde9e0e56cd71cb226e5042a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Thu, 10 Sep 2026 11:24:59 +0800 Subject: [PATCH] =?UTF-8?q?fix(sync):=20=E7=A9=BA=E5=AF=B9=E8=AF=9D?= =?UTF-8?q?=E3=80=8C=E5=B7=B2=E6=89=AB=E3=80=8D=E8=AE=B0=E5=BF=86=E2=80=94?= =?UTF-8?q?=E2=80=94=E6=A0=B9=E9=99=A4=202,468=20=E4=B8=AA=E6=97=A0?= =?UTF-8?q?=E6=B6=88=E6=81=AF=E5=AF=B9=E8=AF=9D=E7=9A=84=E6=AF=8F=E8=BD=AE?= =?UTF-8?q?=E5=85=A8=E9=87=8F=E7=BF=BB=E5=8F=B2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 前项审计的深挖:带窗验证发现活跃仅 27 对话,但访问集仍有 2,493 个—— 2,468 个「从未扫过」的对话(纯语音/从未聊过客户)水位恒 0,拉完 0 条 仍是 unscanned,下一轮又被全量翻史(~2.6s/个,每 4h×每晚都翻)—— 这才是消息任务时长的真正黑洞(昨夜 53,682 条绝大部分源于此)。 修:复用 sync_state 表记扫描标记(entity_type={channel}_conv_empty)—— - 拉到 0 条有效消息即打标(微信/WA 1:1) - 发现层排除「已扫空且不活跃」的;客户一旦发消息(last_chat_time 进 活跃窗)自动回到访问集——不会漏新消息 - 周日 full_history 强制全访问兜底(云客补历史场景) - 空群(115 个,每轮 ~5min)本轮不动,量小 722 测试绿。预期效果:4h 序列访问集 2,493→~30,时长分钟级。 Co-Authored-By: Claude Code --- yunke-server/scripts/scrape_via_page.py | 66 ++++++++++++++++++++++++- 1 file changed, 64 insertions(+), 2 deletions(-) diff --git a/yunke-server/scripts/scrape_via_page.py b/yunke-server/scripts/scrape_via_page.py index 42b45ed..0be29da 100644 --- a/yunke-server/scripts/scrape_via_page.py +++ b/yunke-server/scripts/scrape_via_page.py @@ -514,8 +514,49 @@ async def paginate_chats(fetch_page, end_ms: int, max_pages: int = 200, sleep=No await sleep(1) +# --------------------------------------------------------------------------- +# 空对话/空群「已扫」记忆(2026-09-10 同步审计):从未有消息的对话(纯语音/ +# 未聊过客户,WA 实测 2,468 个)水位恒 0,每轮都被当 unscanned 全量翻史 +# (~2.6s/个)——时长黑洞本体。拉到 0 条有效消息即记 sync_state 扫描标记; +# 发现层排除「已扫空且不活跃」的——客户一旦开始发消息(last_chat_time 进 +# 活跃窗)自动回到访问集;周日 full_history 强制全访问兜底(云客补历史)。 +# --------------------------------------------------------------------------- + + +async def _mark_scan_empty(kind: str, scope: str) -> None: + """记录「该对话/群已扫且无消息」。UPDATE 0 行则 INSERT(sync_state 无唯一键)。""" + import time as _t + + from app.db.base import async_engine + from sqlalchemy import text as sa_text + + now_ms = int(_t.time() * 1000) + async with async_engine.begin() as conn: + res = await conn.execute(sa_text( + "UPDATE sync_state SET last_timestamp = :ts " + "WHERE entity_type = :et AND scope_id = :sid" + ), {"et": kind, "sid": scope, "ts": now_ms}) + if not res.rowcount: + await conn.execute(sa_text( + "INSERT INTO sync_state (entity_type, channel, scope_id, last_timestamp) " + "VALUES (:et, NULL, :sid, :ts)" + ), {"et": kind, "sid": scope, "ts": now_ms}) + + +async def _load_scan_empty(kind: str) -> set[str]: + from app.db.base import async_engine + from sqlalchemy import text as sa_text + + async with async_engine.connect() as conn: + rows = (await conn.execute(sa_text( + "SELECT scope_id FROM sync_state WHERE entity_type = :et" + ), {"et": kind})).fetchall() + return {r[0] for r in rows if r[0]} + + async def _discover_conversations( - channel: str, voice_channel: str, active_since_ms: int | None = None + channel: str, voice_channel: str, active_since_ms: int | None = None, + *, exclude_scanned_empty: bool = True, ) -> list[tuple[str, str]]: """Conversation-pair discovery for the message scrapers (wechat + WA). @@ -561,6 +602,9 @@ async def _discover_conversations( ).fetchall() have = {(r[0], r[1]) for r in with_msgs} + scanned_empty = ( + await _load_scan_empty(f"{channel}_conv_empty") if exclude_scanned_empty else set() + ) seen: set[tuple[str, str]] = set() fresh: list[tuple[str, str]] = [] known: list[tuple[str, str]] = [] @@ -576,7 +620,18 @@ async def _discover_conversations( return seen.add(key) if key not in have: - fresh.append(key) # 从未扫过:新覆盖优先,不受窗口限制 + # 从未扫过:新覆盖优先,不受窗口限制。但「已扫空且不活跃」的除外 + # (2026-09-10:WA 2,468 个无消息对话每轮全翻史的黑洞)——一旦 + # 活跃(last_chat_time 进窗)自动回到访问集;full_history 强制全访。 + scope = f"{key[0]}:{key[1]}" + if ( + exclude_scanned_empty + and scope in scanned_empty + and active_since_ms is not None + and not _active(last_ts) + ): + return + fresh.append(key) elif _active(last_ts): known.append(key) # else: 已扫过且陈旧 → 本轮跳过 @@ -674,6 +729,8 @@ async def scrape_messages(cookie_str: str, max_conversations: int = 100, full_hi int((__import__("time").time() - active_window_hours * 3600) * 1000) if active_window_hours else None ), + # full_history(周日对齐)强制全访问:云客可能补出历史消息 + exclude_scanned_empty=not full_history, ) watermarks = await _conversation_watermarks("wechat") cust_activity = await _customer_activity("wechat") @@ -815,6 +872,8 @@ async def scrape_messages(cookie_str: str, max_conversations: int = 100, full_hi continue total_fetched += len(records) + if not records: + await _mark_scan_empty("wechat_conv_empty", f"{owner_id}:{friend_id}") total_pages += pages if records: @@ -2201,6 +2260,7 @@ async def scrape_wa_messages(cookie_str: str, max_conversations: int = 100, full int((__import__("time").time() - active_window_hours * 3600) * 1000) if active_window_hours else None ), + exclude_scanned_empty=not full_history, ) watermarks = await _conversation_watermarks("whatsapp") cust_activity = await _customer_activity("whatsapp") @@ -2285,6 +2345,8 @@ async def scrape_wa_messages(cookie_str: str, max_conversations: int = 100, full continue total_fetched += len(records) + if not records: + await _mark_scan_empty("whatsapp_conv_empty", f"{uid}:{friend_uid}") if records: # upsert_messages 收整列表(逐条版 _upsert_all 会把 dict 迭代成 # key 字符串 → TypeError, 2026-08-19 WA 阶段首条即崩的根因)