From a9197f696a6c4532f3ea7aedb0780bc594dd78f1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Thu, 10 Sep 2026 11:08:15 +0800 Subject: [PATCH] =?UTF-8?q?fix(sync):=20=E5=90=8C=E6=AD=A5=E9=93=BE?= =?UTF-8?q?=E8=B7=AF=E5=AE=A1=E8=AE=A1=E4=BF=AE=E5=A4=8D=E2=80=94=E2=80=94?= =?UTF-8?q?=E6=B0=B4=E4=BD=8D=E8=BE=B9=E7=95=8C=E7=99=BD=E6=8B=89/?= =?UTF-8?q?=E5=85=A8=E9=87=8F=E6=97=A0=E8=B6=85=E6=97=B6/=E7=BE=A4?= =?UTF-8?q?=E6=B4=BB=E8=B7=83=E7=AA=97=E5=88=A4=E5=AE=9A=E6=BA=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 全项目数据同步审计发现四漏洞(前项为证据): ① 消息增量边界白拉(证据:昨夜 wa-messages fetched=53,682 而真实新消息 仅 321 条,99.4% 是白拉白写的已入库旧消息)——云客 chatRecords/msg-chats 的 start 参数是 inclusive(返回 ts>=start 含边界),四个消息实体(微信 1:1/WA 1:1/微信群/WA 群)统一在归一化后过滤 ts<=watermark 的边界行, 不计数不写库(省 99% 写放大,计数从此反映真实新增) ② full_sync_all.sh 无超时——极端情况(常态 10-20h)拖进次日相位把 02:00 全量又挤掉(前两天事故同型);加 22h 上限 ③ 周日 full-history 无超时——4 实体全历史可能 24h+,必挤周一;单实体 加 6h 上限 ④ 群消息活跃窗判定源是昨晚 full 的 chat_group.last_chat_time 快照—— 「昨天安静、今天突然活跃」的群会被 24h 窗口漏到明晚;4h 序列在 group-messages 前加 chatroom-stats 全窗(~1min)刷新群活跃判定源 722 测试绿(4h 序列断言 9→10 命令+chatroom-stats 位序)。 Co-Authored-By: Claude Code --- yunke-server/app/sync/scheduler.py | 9 +++++++-- yunke-server/scripts/scrape_via_page.py | 21 +++++++++++++++++++-- yunke-server/tests/test_sync_scheduler.py | 15 ++++++++------- 3 files changed, 34 insertions(+), 11 deletions(-) diff --git a/yunke-server/app/sync/scheduler.py b/yunke-server/app/sync/scheduler.py index ef52474..d85d3fa 100644 --- a/yunke-server/app/sync/scheduler.py +++ b/yunke-server/app/sync/scheduler.py @@ -602,7 +602,7 @@ class SyncScheduler: if not bash: logger.error("page sync 'full': bash not found on PATH") return - commands = [([bash, "scripts/full_sync_all.sh"], None)] # (args, timeout)——full 不设超时(全量本身 10-20h) + commands = [([bash, "scripts/full_sync_all.sh"], 22 * 3600)] # 22h 上限:防极端情况拖进次日相位(全量常态 10-20h) elif kind == "messages": # 2026-09-10 调度根治(用户报「员工列表时间差 2 天」): # ① 序列头部刷员工账号表(accounts ~1min)——员工列表「数据 @@ -621,6 +621,10 @@ class SyncScheduler: ("wa-friends", None, 2400), ("messages", 24, 4800), ("wa-messages", 24, 4800), + # chatroom-stats 全窗(~1min):刷 chat_group.last_chat_time—— + # 群消息活跃窗的判定源。不加则窗口用的是昨晚 full 的快照, + # 「昨天安静、今天突然活跃」的群会被窗口漏到明晚(2026-09-10) + ("chatroom-stats", None, 1800), ("group-messages", 24, 4800), ("wa-group-messages", 24, 2400), ("moments", None, 2400), @@ -666,7 +670,8 @@ class SyncScheduler: "99999", "--full-history", ], - None, # 每周对齐不设超时 + 6 * 3600, # 单实体 6h 上限(2026-09-10:防周日的全历史 + # 对齐拖进周一,把 02:00 full 又挤掉——同类事故根治) ) for e in entities ] diff --git a/yunke-server/scripts/scrape_via_page.py b/yunke-server/scripts/scrape_via_page.py index 1fd8222..42b45ed 100644 --- a/yunke-server/scripts/scrape_via_page.py +++ b/yunke-server/scripts/scrape_via_page.py @@ -763,6 +763,9 @@ async def scrape_messages(cookie_str: str, max_conversations: int = 100, full_hi ts = int(str(ts_raw)) if ts_raw else None except (ValueError, TypeError): ts = None + # 增量模式过滤边界旧消息(云客 start inclusive,2026-09-10) + if watermark and ts is not None and ts <= watermark: + continue records.append({ "msg_id": str(row.get("msgId") or row.get("msgSvrId") or row.get("id") or f"{owner_id}:{friend_id}:{i}:{len(records)}"), "channel": "wechat", @@ -965,6 +968,9 @@ async def scrape_group_messages(cookie_str: str, max_groups: int = 500, full_his continue rec = _normalize_group_message(row, owner_id, room_id) if rec is not None: + # 增量模式过滤边界旧消息(同 1:1,2026-09-10) + if watermark and (rec.get("timestamp") or 0) <= watermark: + continue records.append(rec) except Exception as e: @@ -2241,7 +2247,14 @@ async def scrape_wa_messages(cookie_str: str, max_conversations: int = 100, full msgs = data.get("messages") or [] for row in msgs: if isinstance(row, dict) and not _is_wa_divider(row): - records.append(_normalize_whatsapp_message(row)) + rec = _normalize_whatsapp_message(row) + # 增量模式过滤边界旧消息:云客 start 参数是 inclusive + # (返回 ts>=start 含边界),不过滤则每对话白拉数条已 + # 入库旧消息(2026-09-10 实测一夜 fetched 53,682 仅 + # 321 条真新,99% 白写)。 + if watermark and (rec.get("timestamp") or 0) <= watermark: + continue + records.append(rec) if not data.get("hasLast"): break # 增量触底停:翻到已同步区 @@ -2385,7 +2398,11 @@ async def scrape_wa_group_messages(cookie_str: str, max_groups: int = 500, full_ for row in rows: if not isinstance(row, dict) or _is_wa_divider(row): continue - records.append(_normalize_wa_group_message(row)) + rec = _normalize_wa_group_message(row) + # 增量模式过滤边界旧消息(同 1:1,2026-09-10) + if watermark and (rec.get("timestamp") or 0) <= watermark: + continue + records.append(rec) except Exception as e: logger.warning(f" [{i+1}] {str(owner_id)[:18]}:{str(room_id)[:18]} failed: {e}") diff --git a/yunke-server/tests/test_sync_scheduler.py b/yunke-server/tests/test_sync_scheduler.py index db4e9eb..901bf6a 100644 --- a/yunke-server/tests/test_sync_scheduler.py +++ b/yunke-server/tests/test_sync_scheduler.py @@ -589,7 +589,7 @@ async def test_safe_page_run_messages_builds_scrape_command(engine, monkeypatch) await sched._safe_page_run("messages") - assert len(commands) == 9 + assert len(commands) == 10 for args in commands: assert args[1].endswith("scrape_via_page.py") @@ -601,13 +601,14 @@ async def test_safe_page_run_messages_builds_scrape_command(engine, monkeypatch) assert entity_of(commands[1]) == "wa-accounts" assert entity_of(commands[2]) == "friends" assert entity_of(commands[3]) == "wa-friends" + # 群消息前刷群活跃(chatroom-stats 全窗——群活跃窗判定源,2026-09-10) + entities_seq = [entity_of(a) for a in commands] + assert entities_seq.index("chatroom-stats") < entities_seq.index("group-messages") # 四类消息:24h 活跃窗(2026-09-10 根治调度堵塞) - for args in commands[4:8]: - assert entity_of(args) in ( - "messages", "wa-messages", "group-messages", "wa-group-messages", - ) - assert args[args.index("--max-conversations") + 1] == "99999" - assert args[args.index("--active-window-hours") + 1] == "24" + for args in commands: + if entity_of(args) in ("messages", "wa-messages", "group-messages", "wa-group-messages"): + assert args[args.index("--max-conversations") + 1] == "99999" + assert args[args.index("--active-window-hours") + 1] == "24" # moments 浅扫在序列末尾:每员工最多 3 页 moments_cmd = commands[-1] assert entity_of(moments_cmd) == "moments"