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"