fix(scheduler): 调度根治——4h 增量只扫活跃对话+账号表进 4h+超时保险丝

用户报「员工列表时间与云客差 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 <noreply@anthropic.com>
2026-09-10 10:38:47 +08:00
parent 3cf77ec6d5
commit e3fadfe1a6
3 changed files with 162 additions and 72 deletions
@@ -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
@@ -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":
@@ -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"