From e3fadfe1a6f09d24e6058d04784469fa9371caeb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Thu, 10 Sep 2026 10:38:47 +0800 Subject: [PATCH] =?UTF-8?q?fix(scheduler):=20=E8=B0=83=E5=BA=A6=E6=A0=B9?= =?UTF-8?q?=E6=B2=BB=E2=80=94=E2=80=944h=20=E5=A2=9E=E9=87=8F=E5=8F=AA?= =?UTF-8?q?=E6=89=AB=E6=B4=BB=E8=B7=83=E5=AF=B9=E8=AF=9D+=E8=B4=A6?= =?UTF-8?q?=E5=8F=B7=E8=A1=A8=E8=BF=9B=204h+=E8=B6=85=E6=97=B6=E4=BF=9D?= =?UTF-8?q?=E9=99=A9=E4=B8=9D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 用户报「员工列表时间与云客差 2 天」。根因链:4h 增量序列全对话遍历 (WA 9,357 对×~3.5s≈9h,整序列 15h+)超过自身 4h 间隔→与 02:00 全量在 asyncio.Lock 上背靠背互相排队→近两晚全量被挤掉→账号表(仅全量刷)停更 两天,员工列表「数据更新/最新消息时间」显示 09-08 快照。 三层修复: - 活跃窗:_discover_conversations/群发现加 active_since_ms——已扫过的 对话仅保留 last_chat_time 在窗内的(云客 lastChatTime 滞后小时级, 24h 窗足够容错),从未扫过的(新覆盖优先)不受限;陈旧对话无新消息, 每日 full 兜底。CLI --active-window-hours,四消息实体透传 - 4h 序列头部加 accounts/wa-accounts(~2min)——员工列表时间最多滞后 4h,不再依赖每晚全量 - 超时保险丝:_page_subprocess 加 timeout(超时 kill 返回 124,上游 break 下轮补)——messages 各实体 15-80min 预算,单命令拖不死相位; full/full-history 不设限 722 测试绿(序列断言 7→9 命令+窗口参数)。干跑验证:近 24h 活跃私聊/ 群相对全量降一个数量级(本地旧库群 2011→537;私聊依赖 friends 先刷, 序列顺序已保证)。 Co-Authored-By: Claude Code --- yunke-server/app/sync/scheduler.py | 120 ++++++++++++++-------- yunke-server/scripts/scrape_via_page.py | 92 ++++++++++++----- yunke-server/tests/test_sync_scheduler.py | 22 ++-- 3 files changed, 162 insertions(+), 72 deletions(-) diff --git a/yunke-server/app/sync/scheduler.py b/yunke-server/app/sync/scheduler.py index 675725f..ef52474 100644 --- a/yunke-server/app/sync/scheduler.py +++ b/yunke-server/app/sync/scheduler.py @@ -602,50 +602,72 @@ class SyncScheduler: if not bash: logger.error("page sync 'full': bash not found on PATH") return - commands: list[list[str]] = [[bash, "scripts/full_sync_all.sh"]] + commands = [([bash, "scripts/full_sync_all.sh"], None)] # (args, timeout)——full 不设超时(全量本身 10-20h) elif kind == "messages": - # 先刷客户列表(最近沟通时间是云客列表的实时排序键,friends 快照 - # 滞后会让本地排序与云客第一页完全不同,2026-08-20),再抓全部四类 - # 消息增量(2026-08-21 全接入:水位增量后单对话 1 页,可承受); - # 末尾追加员工朋友圈浅扫(2026-08-25,max-pages 3 即每员工最多 - # 3 页,新帖通常落第一页;全量翻页由每日 full 走 full_sync_all.sh) + # 2026-09-10 调度根治(用户报「员工列表时间差 2 天」): + # ① 序列头部刷员工账号表(accounts ~1min)——员工列表「数据 + # 更新时间/最新消息时间」从此最多滞后 4h(此前只在每日 + # full 刷,长任务挤掉 full 就停更); + # ② 消息实体只扫**近 24h 活跃**对话(--active-window-hours; + # 从未扫过的不受限)——全对话遍历(WA 9,357 个×~3.5s≈9h) + # 曾把序列拖到 15h+,超过 4h 间隔→与 02:00 full 背靠背 + # 互相挤死;陈旧对话无新消息,每日 full 兜底; + # ③ 每命令超时保险丝(超时 kill 下轮补,增量幂等)。 + # friends 仍先刷(最近沟通时间是排序键+活跃窗口的判定源)。 + msg_entities = [ + ("accounts", None, 900), + ("wa-accounts", None, 900), + ("friends", None, 2400), + ("wa-friends", None, 2400), + ("messages", 24, 4800), + ("wa-messages", 24, 4800), + ("group-messages", 24, 4800), + ("wa-group-messages", 24, 2400), + ("moments", None, 2400), + ] commands = [ - [sys.executable, "scripts/scrape_via_page.py", "--entity", "friends"], - [sys.executable, "scripts/scrape_via_page.py", "--entity", "wa-friends"], - ] + [ - [ - sys.executable, - "scripts/scrape_via_page.py", - "--entity", - e, - "--max-conversations", - "99999", - ] - for e in ("messages", "wa-messages", "group-messages", "wa-group-messages") - ] + [ - [ - sys.executable, - "scripts/scrape_via_page.py", - "--entity", - "moments", - "--max-pages", - "3", - ] + ( + [ + sys.executable, + "scripts/scrape_via_page.py", + "--entity", + entity, + ] + + ( + [ + "--max-conversations", + "99999", + "--active-window-hours", + str(window), + ] + if window + else ( + ["--max-pages", "3"] + if entity == "moments" + else [] + ) + ), + timeout, + ) + for entity, window, timeout in msg_entities ] elif kind == "full-history": # 每周全量对齐(周日 05:00):强制全量翻历史,对齐增量模式抓不到的 # 撤回/编辑消息 + 彻底消除 lastChatTime 滞后窗口(2026-08-21) entities = ["messages", "group-messages", "wa-messages", "wa-group-messages"] commands = [ - [ - sys.executable, - "scripts/scrape_via_page.py", - "--entity", - e, - "--max-conversations", - "99999", - "--full-history", - ] + ( + [ + sys.executable, + "scripts/scrape_via_page.py", + "--entity", + e, + "--max-conversations", + "99999", + "--full-history", + ], + None, # 每周对齐不设超时 + ) for e in entities ] else: @@ -653,8 +675,8 @@ class SyncScheduler: return rc = 0 - for args in commands: - rc = await self._page_subprocess(args) or rc + for args, timeout in commands: + rc = await self._page_subprocess(args, timeout=timeout) or rc if rc != 0: logger.warning("page sync %s: command failed rc=%s, skipping rest", kind, rc) break @@ -703,8 +725,13 @@ class SyncScheduler: except Exception: # noqa: BLE001 — scheduler MUST NOT crash logger.exception("analytics tick failed") - async def _page_subprocess(self, args: list[str]) -> int: - """启动页面子进程并等待结束;继承环境 + 补 DB_URL/PYTHONIOENCODING。""" + async def _page_subprocess(self, args: list[str], *, timeout: float | None = None) -> int: + """启动页面子进程并等待结束;继承环境 + 补 DB_URL/PYTHONIOENCODING。 + + ``timeout``(秒,2026-09-10 调度根治保险丝):超时 kill 子进程返回 + 非零——上游 break 跳过本序列剩余命令,增量幂等下轮自动补,防止 + 单命令拖死整个调度相位(4h 序列曾因此 15h+ 与 full 互相挤死)。 + """ env = {**os.environ, "PYTHONIOENCODING": "utf-8"} settings = get_settings() if settings.db_url and "DB_URL" not in env: @@ -714,7 +741,18 @@ class SyncScheduler: cwd=str(REPO_ROOT), env=env, ) - return await proc.wait() + if timeout is None: + return await proc.wait() + try: + return await asyncio.wait_for(proc.wait(), timeout) + except asyncio.TimeoutError: + logger.error( + "page sync subprocess timeout after %.0fs (killed): %s", + timeout, " ".join(args[-6:]), + ) + proc.kill() + await proc.wait() + return 124 # 超时语义(同 shell timeout 惯例) async def _safe_run( self, entity: str, *, log: bool = True diff --git a/yunke-server/scripts/scrape_via_page.py b/yunke-server/scripts/scrape_via_page.py index 395ed1d..1fd8222 100644 --- a/yunke-server/scripts/scrape_via_page.py +++ b/yunke-server/scripts/scrape_via_page.py @@ -514,7 +514,9 @@ async def paginate_chats(fetch_page, end_ms: int, max_pages: int = 200, sleep=No await sleep(1) -async def _discover_conversations(channel: str, voice_channel: str) -> list[tuple[str, str]]: +async def _discover_conversations( + channel: str, voice_channel: str, active_since_ms: int | None = None +) -> list[tuple[str, str]]: """Conversation-pair discovery for the message scrapers (wechat + WA). Sources (union, dedup): @@ -527,6 +529,11 @@ async def _discover_conversations(channel: str, voice_channel: str) -> list[tupl Order: pairs with no messages yet FIRST (new coverage survives an interruption), then the rest by customer last_chat_time DESC. + + ``active_since_ms``(2026-09-10 调度根治):4h 增量模式传「now-24h」—— + **已扫过**的对话仅保留 last_chat_time 在窗内的(活跃的),陈旧对话本轮 + 跳过(没有新消息,每日全量兜底);**从未扫过**的(fresh)无论新旧全保 + (新覆盖优先原则不变)。None=不过滤(全量模式)。 """ import json as _json @@ -536,14 +543,14 @@ async def _discover_conversations(channel: str, voice_channel: str) -> list[tupl async with async_engine.connect() as conn: cust = ( await conn.execute(sa_text( - "SELECT owner_account_id, friend_id FROM customer " + "SELECT owner_account_id, friend_id, last_chat_time FROM customer " "WHERE channel=:ch AND owner_account_id IS NOT NULL AND friend_id IS NOT NULL " "ORDER BY last_chat_time DESC" ), {"ch": channel}) ).fetchall() voice = ( await conn.execute(sa_text( - "SELECT raw_json FROM phone_call WHERE channel=:vc" + "SELECT raw_json, start_call_time FROM phone_call WHERE channel=:vc" ), {"vc": voice_channel}) ).fetchall() with_msgs = ( @@ -558,16 +565,27 @@ async def _discover_conversations(channel: str, voice_channel: str) -> list[tupl fresh: list[tuple[str, str]] = [] known: list[tuple[str, str]] = [] - def _add(key: tuple[str, str]) -> None: + def _active(last_ts: int | None) -> bool: + """窗口判定:未开窗口恒真;有窗时 last_ts 缺失(从未聊过)也保(fresh)。""" + if active_since_ms is None: + return True + return last_ts is None or last_ts >= active_since_ms + + def _add(key: tuple[str, str], last_ts: int | None) -> None: if key in seen: return seen.add(key) - (fresh if key not in have else known).append(key) + if key not in have: + fresh.append(key) # 从未扫过:新覆盖优先,不受窗口限制 + elif _active(last_ts): + known.append(key) + # else: 已扫过且陈旧 → 本轮跳过 - for owner, friend in cust: - _add((owner, friend)) + for owner, friend, last_ts in cust: + _add((owner, friend), last_ts) voice_only = 0 - for (rj,) in voice: + cust_keys = {(r[0], r[1]) for r in cust} + for rj, call_ts in voice: try: d = _json.loads(rj) except (TypeError, ValueError): @@ -576,13 +594,15 @@ async def _discover_conversations(channel: str, voice_channel: str) -> list[tupl if not owner or not talker or str(talker).endswith("@chatroom"): continue before = len(seen) - _add((str(owner), str(talker))) - if len(seen) > before and (str(owner), str(talker)) not in cust: + _add((str(owner), str(talker)), call_ts) + if len(seen) > before and (str(owner), str(talker)) not in cust_keys: voice_only += 1 + skipped = len(cust) + voice_only - len(fresh) - len(known) logger.info( f"discovery[{channel}]: {len(cust)} customer pairs + {voice_only} voice-only pairs, " - f"{len(fresh)} unscanned (first) / {len(known)} already have messages" + f"{len(fresh)} unscanned (first) / {len(known)} active in window" + + (f" / {skipped} stale skipped" if active_since_ms is not None else f" / {len(known)} already have messages") ) return fresh + known @@ -635,7 +655,7 @@ def _page_min_ts(data: dict) -> int: return min(ts_list) if ts_list else 0 -async def scrape_messages(cookie_str: str, max_conversations: int = 100, full_history: bool = False): +async def scrape_messages(cookie_str: str, max_conversations: int = 100, full_history: bool = False, active_window_hours: int | None = None): """Scrape wechat messages via browser fetch POST /wechatweb/msg/chats. **增量同步(默认)**:每对话水位 = 本地 max(timestamp);请求带 @@ -648,7 +668,13 @@ async def scrape_messages(cookie_str: str, max_conversations: int = 100, full_hi from app.db.base import async_session_maker from app.storage.repos import upsert_messages - conversations = await _discover_conversations("wechat", "voice_call") + conversations = await _discover_conversations( + "wechat", "voice_call", + active_since_ms=( + int((__import__("time").time() - active_window_hours * 3600) * 1000) + if active_window_hours else None + ), + ) watermarks = await _conversation_watermarks("wechat") cust_activity = await _customer_activity("wechat") @@ -853,7 +879,7 @@ def _normalize_group_message(row: dict, owner_id: str, room_id: str) -> dict | N } -async def scrape_group_messages(cookie_str: str, max_groups: int = 500, full_history: bool = False): +async def scrape_group_messages(cookie_str: str, max_groups: int = 500, full_history: bool = False, active_window_hours: int | None = None): """Scrape WeChat GROUP chat messages via browser fetch POST /wechatweb/msg/chats. Same endpoint as 1:1 chats — pass the room id as ``friendId`` (confirmed @@ -868,12 +894,17 @@ async def scrape_group_messages(cookie_str: str, max_groups: int = 500, full_his from app.storage.repos import upsert_messages from sqlalchemy import text as sa_text + active_since_ms = ( + int((__import__("time").time() - active_window_hours * 3600) * 1000) + if active_window_hours else None + ) async with async_engine.connect() as conn: result = await conn.execute(sa_text( "SELECT owner_account_id, room_id FROM chat_group " "WHERE channel='wechat' AND owner_account_id IS NOT NULL " + "AND (:t IS NULL OR last_chat_time IS NULL OR last_chat_time >= :t) " "ORDER BY last_chat_time DESC" - )) + ), {"t": active_since_ms}) groups = [(row[0], row[1]) for row in result.fetchall()] room_wms = await _room_watermarks() @@ -2143,7 +2174,7 @@ async def scrape_voice_clips(cookie_str: str, max_employees: int = 200, page_siz # --------------------------------------------------------------------------- -async def scrape_wa_messages(cookie_str: str, max_conversations: int = 100, full_history: bool = False): +async def scrape_wa_messages(cookie_str: str, max_conversations: int = 100, full_history: bool = False, active_window_hours: int | None = None): """Scrape WhatsApp chat records via browser fetch POST /qqweb/WhatsAppUser/chatRecords (JSON body {uid, friendUid, start, end}). @@ -2158,7 +2189,13 @@ async def scrape_wa_messages(cookie_str: str, max_conversations: int = 100, full from app.scraper.whatsapp import _is_wa_divider, _normalize_whatsapp_message from app.storage.repos import upsert_messages - conversations = await _discover_conversations("whatsapp", "wa_voice_call") + conversations = await _discover_conversations( + "whatsapp", "wa_voice_call", + active_since_ms=( + int((__import__("time").time() - active_window_hours * 3600) * 1000) + if active_window_hours else None + ), + ) watermarks = await _conversation_watermarks("whatsapp") cust_activity = await _customer_activity("whatsapp") @@ -2281,7 +2318,7 @@ def _normalize_wa_group_message(record: dict) -> dict: return rec -async def scrape_wa_group_messages(cookie_str: str, max_groups: int = 500, full_history: bool = False): +async def scrape_wa_group_messages(cookie_str: str, max_groups: int = 500, full_history: bool = False, active_window_hours: int | None = None): """WA 群消息:浏览器 fetch chatRecords,friendUid 直接传 @g.us 群 id (真机确认 2026-08-20,与微信群的 msg/chats+roomId 同套路)。 @@ -2295,11 +2332,16 @@ async def scrape_wa_group_messages(cookie_str: str, max_groups: int = 500, full_ from app.storage.repos import upsert_messages from sqlalchemy import text as sa_text + active_since_ms = ( + int((__import__("time").time() - active_window_hours * 3600) * 1000) + if active_window_hours else None + ) async with async_engine.connect() as conn: result = await conn.execute(sa_text( "SELECT owner_account_id, room_id FROM chat_group " - "WHERE channel='whatsapp' AND owner_account_id IS NOT NULL" - )) + "WHERE channel='whatsapp' AND owner_account_id IS NOT NULL " + "AND (:t IS NULL OR last_chat_time IS NULL OR last_chat_time >= :t)" + ), {"t": active_since_ms}) groups = [(row[0], row[1]) for row in result.fetchall()] room_wms = await _room_watermarks() @@ -2473,6 +2515,8 @@ async def main(): parser.add_argument("--max-pages", type=int, default=200, help="列表翻页上限 (accounts/wa-accounts/wa-friends/voice-calls/moments)") parser.add_argument("--batch-size", type=int, default=50, help="媒体下载批次大小 (media)") parser.add_argument("--full-history", action="store_true", help="消息类实体强制全量翻历史(默认增量水位)") + parser.add_argument("--active-window-hours", type=int, default=None, + help="只扫近 N 小时活跃的对话(4h 增量模式传 24;未扫过的对话不受限;默认不过滤=全量发现)") parser.add_argument("--day", type=str, default=None, help="chatroom-stats:单日窗 YYYY-MM-DD(落按天进出表);缺省=全窗档案基线") args = parser.parse_args() @@ -2496,9 +2540,9 @@ async def _dispatch_entity(args, cookie: str) -> None: elif args.entity == "groups": await scrape_groups(cookie, args.max_employees) elif args.entity == "messages": - await scrape_messages(cookie, args.max_conversations, args.full_history) + await scrape_messages(cookie, args.max_conversations, args.full_history, args.active_window_hours) elif args.entity == "group-messages": - await scrape_group_messages(cookie, args.max_conversations, args.full_history) + await scrape_group_messages(cookie, args.max_conversations, args.full_history, args.active_window_hours) elif args.entity == "accounts": await scrape_accounts(cookie, args.max_pages) elif args.entity == "wa-accounts": @@ -2514,9 +2558,9 @@ async def _dispatch_entity(args, cookie: str) -> None: elif args.entity == "voice-clips": await scrape_voice_clips(cookie, args.max_employees) elif args.entity == "wa-messages": - await scrape_wa_messages(cookie, args.max_conversations, args.full_history) + await scrape_wa_messages(cookie, args.max_conversations, args.full_history, args.active_window_hours) elif args.entity == "wa-group-messages": - await scrape_wa_group_messages(cookie, args.max_conversations, args.full_history) + await scrape_wa_group_messages(cookie, args.max_conversations, args.full_history, args.active_window_hours) elif args.entity == "moments": await scrape_moments(cookie, args.max_pages) elif args.entity == "media": diff --git a/yunke-server/tests/test_sync_scheduler.py b/yunke-server/tests/test_sync_scheduler.py index dcf9c22..db4e9eb 100644 --- a/yunke-server/tests/test_sync_scheduler.py +++ b/yunke-server/tests/test_sync_scheduler.py @@ -572,7 +572,12 @@ async def test_safe_page_run_full_invokes_bash_full_sync(engine, monkeypatch): async def test_safe_page_run_messages_builds_scrape_command(engine, monkeypatch): - """messages job = friends→wa-friends→四类消息→moments 浅扫的命令序列。""" + """messages job = accounts→friends→四类消息(活跃窗)→moments 浅扫的命令序列。 + + 2026-09-10 调度根治:头部刷员工账号表(员工列表时间最多滞后 4h); + 消息实体只扫近 24h 活跃对话(--active-window-hours 24,防序列 15h+ + 挤死 full);陈旧对话每日 full 兜底。 + """ sched = SyncScheduler(engine=engine, max_conversations_per_run=123) commands: list[list[str]] = [] @@ -584,22 +589,25 @@ async def test_safe_page_run_messages_builds_scrape_command(engine, monkeypatch) await sched._safe_page_run("messages") - # 2026-08-21: messages job 是 friends→wa-friends→四类消息的命令序列, - # 每条都调 scrape_via_page.py;2026-08-25 末尾追加 moments 浅扫 - assert len(commands) == 7 + assert len(commands) == 9 for args in commands: assert args[1].endswith("scrape_via_page.py") def entity_of(args: list[str]) -> str: return args[args.index("--entity") + 1] - assert entity_of(commands[0]) == "friends" - assert entity_of(commands[1]) == "wa-friends" - for args in commands[2:6]: + # 头部:账号刷新(员工列表「数据更新/最新消息时间」来源) + assert entity_of(commands[0]) == "accounts" + assert entity_of(commands[1]) == "wa-accounts" + assert entity_of(commands[2]) == "friends" + assert entity_of(commands[3]) == "wa-friends" + # 四类消息: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" # moments 浅扫在序列末尾:每员工最多 3 页 moments_cmd = commands[-1] assert entity_of(moments_cmd) == "moments"