feat(analytics): 4h 调度分析链+CLI 回填+full_sync 夜间自愈步

- scheduler: yunke_analytics_tick job(与消息增量同节奏 IntervalTrigger),
  _analytics_tick 纯 DB——stat_daily_staff 滚动3天刷新 + 敏感词增量扫描,
  不抢 _page_lock 无子进程,异常只记日志不断链;today 用 time.time()
  UTC epoch 喂 day_key,规避服务器非上海时区的 naive now 误判
- cli: staff-daily-stats(全量重建 rollup,幂等)+ sensitive-scan
  (默认增量水位扫描,--full 首跑建水位/新词回填历史;description 明示
  加词不回填历史)+ 幂等补种起步词库
- full_sync_all.sh: daily-stats 后追加 staff-daily-stats(rollup 随
  msg_daily_stats 重建同步自愈)
- 测试: tick 行为测试(今日 seed→stat_daily_staff 出行)+ 异常吞噬
  (monkeypatch 抛错不外抛);同步更新注册断言(3→4 jobs)

Co-Authored-By: Claude Code <noreply@anthropic.com>

yunke-server/app/sync/cli.py

@@ -1,6 +1,6 @@
 """Command-line entry point for running syncs + the scheduler (M6, Part B).
 
-Three invocations::
+Three invocations (plus analytics backfills, see below)::
 
     # One-shot full sync (every entity) into the configured DB:
     python -m app.sync.cli sync
@@ -11,6 +11,11 @@ Three invocations::
     # Long-running scheduler (APScheduler; full sync daily + messages every 4h):
     python -m app.sync.cli scheduler
 
+    # One-shot analytics backfills / self-heal (all idempotent):
+    python -m app.sync.cli daily-stats
+    python -m app.sync.cli staff-daily-stats
+    python -m app.sync.cli sensitive-scan --full
+
 Auth + DB
 ---------
 Auth comes from env / ``.env`` via the same helpers the smoke script uses
@@ -101,6 +106,36 @@ def _build_parser() -> argparse.ArgumentParser:
         ),
     )
 
+    # ``staff-daily-stats`` ---------------------------------------------------
+    p_staff_stats = sub.add_parser(
+        "staff-daily-stats",
+        help="Rebuild stat_daily_staff (staff x day rollup).",
+        description=(
+            "Full rebuild of the staff daily rollup (stat_daily_staff) from "
+            "msg_daily_stats. Used as the nightly self-heal step in "
+            "scripts/full_sync_all.sh and as the one-off deploy backfill. "
+            "Idempotent."
+        ),
+    )
+
+    # ``sensitive-scan`` ------------------------------------------------------
+    p_scan = sub.add_parser(
+        "sensitive-scan",
+        help="Sensitive-word scan (incremental; --full for backfill).",
+        description=(
+            "Incremental sensitive-word scan by fetched_at watermark. "
+            "First run on a populated DB MUST use --full to build the "
+            "watermark (run_incremental refuses no-watermark on big tables). "
+            "Adding a new word does NOT backfill history (watermark only "
+            "sees new rows) — rerun --full to scan history for it."
+        ),
+    )
+    p_scan.add_argument(
+        "--full",
+        action="store_true",
+        help="Full backfill scan (first deploy / new-word history).",
+    )
+
     # ``scheduler`` -----------------------------------------------------------
     p_sched = sub.add_parser(
         "scheduler",
@@ -302,6 +337,43 @@ async def _cmd_daily_stats(args: argparse.Namespace) -> int:
         await engine.dispose()
 
 
+async def _cmd_staff_daily_stats(args: argparse.Namespace) -> int:
+    """全量重建 stat_daily_staff(夜间自愈 / 部署回填,幂等)。"""
+    settings = get_settings()
+    engine, session_maker = await _init_db(settings.db_url)
+    try:
+        from app.storage.staff_summary import recompute_all
+
+        async with session_maker() as session:
+            days = await recompute_all(session)
+            await session.commit()
+        print(f"[staff-daily-stats] rebuilt {days} days")
+        return 0
+    finally:
+        await engine.dispose()
+
+
+async def _cmd_sensitive_scan(args: argparse.Namespace) -> int:
+    """敏感词扫描:默认增量(水位后);--full 全量回填(首跑/新词历史)。"""
+    settings = get_settings()
+    engine, session_maker = await _init_db(settings.db_url)
+    try:
+        from app.analytics.sensitive import scanner
+        from app.analytics.sensitive import words as words_svc
+
+        async with session_maker() as session:
+            seeded = await words_svc.ensure_starter_words(session)  # 幂等补种起步词库
+            if args.full:
+                out = await scanner.run_full(session)
+            else:
+                out = await scanner.run_incremental(session)
+            await session.commit()
+        print(f"[sensitive-scan] seeded={seeded} {out}")
+        return 0
+    finally:
+        await engine.dispose()
+
+
 def main(argv: list[str] | None = None) -> int:
     """CLI entry point. Returns a process exit code."""
     # Configure logging so the scheduler's JSON line + the orchestrator's
@@ -322,6 +394,10 @@ def main(argv: list[str] | None = None) -> int:
             return asyncio.run(_cmd_scheduler(args))
         if args.command == "daily-stats":
             return asyncio.run(_cmd_daily_stats(args))
+        if args.command == "staff-daily-stats":
+            return asyncio.run(_cmd_staff_daily_stats(args))
+        if args.command == "sensitive-scan":
+            return asyncio.run(_cmd_sensitive_scan(args))
     except KeyboardInterrupt:
         # So `python -m app.sync.cli scheduler` exits cleanly on Ctrl-C.
         return 0

yunke-server/app/sync/scheduler.py

@@ -28,6 +28,7 @@ import logging
 import os
 import shutil
 import sys
+import time
 from collections.abc import Awaitable, Callable
 from pathlib import Path
 from typing import Any
@@ -535,6 +536,16 @@ class SyncScheduler:
             coalesce=True,
             max_instances=1,
         )
+        # 4h 分析链:stat_daily_staff 滚动3天刷新 + 敏感词增量扫描(纯 DB,
+        # 无浏览器子进程;与消息增量同节奏,读滞后语义与「数据截至」一致)
+        sched.add_job(
+            self._analytics_tick,
+            IntervalTrigger(hours=self.message_interval_hours),
+            id="yunke_analytics_tick",
+            replace_existing=True,
+            coalesce=True,
+            max_instances=1,
+        )
         sched.start()
         self._scheduler = sched
         logger.info(
@@ -662,6 +673,34 @@ class SyncScheduler:
             logger.info(line)
             print(line, flush=True)
 
+    async def _analytics_tick(self) -> None:
+        """4h 分析链(stat_daily_staff 滚动3天 + 敏感词增量)——异常只记日志不断链。
+
+        纯 DB 无浏览器子进程,不抢 ``_page_lock``(DB 写毫秒级;
+        ``max_instances=1`` 防自身重叠即可;与夜间 full_sync 的
+        staff-daily-stats 步骤重叠时 UNIQUE 兜底幂等)。
+
+        时区:today 用 ``time.time()``(UTC epoch,无时区歧义)喂
+        ``day_key``——naive ``now().timestamp()`` 按本地时区解释,服务器
+        时区未必是上海(2026-09-04 部署注意点)。
+        """
+        try:
+            async with self._session_maker() as session:
+                from app.analytics.sensitive import scanner
+                from app.storage.staff_summary import recompute_days, rolling_days
+                from app.storage.summary import day_key
+
+                today = day_key(int(time.time() * 1000))
+                n = await recompute_days(session, rolling_days(today, 3))
+                scan = await scanner.run_incremental(session)
+                await session.commit()
+            logger.info(
+                "[analytics-tick] staff_days=%s scanned=%s hits=%s",
+                n, scan.get("scanned"), scan.get("hits"),
+            )
+        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。"""
         env = {**os.environ, "PYTHONIOENCODING": "utf-8"}

yunke-server/scripts/full_sync_all.sh

@@ -49,6 +49,10 @@ $PY scripts/scrape_via_page.py --entity media
 # sessions/staff-chat-stats 读路径依赖此表,2026-09-03)
 echo "$(date) --- 11/11 Daily-stats rebuild ---"
 $PY -m app.sync.cli daily-stats
+# 12:员工×日汇总全量重建(rollup 层随 msg_daily_stats 重建同步自愈;
+# A1 员工指标读路径依赖此表,2026-09-04)
+echo "$(date) --- 12/12 Staff-daily-stats rebuild ---"
+$PY -m app.sync.cli staff-daily-stats
 
 echo "=============================================="
 echo "$(date) === FULL RE-SYNC COMPLETE ==="

yunke-server/tests/test_scheduler_analytics_tick.py

@@ -0,0 +1,76 @@
+"""_analytics_tick:滚动刷新 + 增量扫描 + 异常吞掉(数据分析板块 v1, T11)。
+
+行为测试是底线——start() 的 job 注册断言在 test_sync_scheduler.py 的
+注册测试里同步更新(那里 spy 断言 add_job 全集)。
+"""
+from __future__ import annotations
+
+import logging
+import time
+
+import pytest
+from sqlalchemy import func, select
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+from sqlalchemy.pool import StaticPool
+
+from app.db.base import Base
+from app.db.models import MsgDailyStats, SensitiveWord, StatDailyStaff
+from app.storage.summary import day_key
+from app.sync.scheduler import SyncScheduler
+
+
+@pytest.fixture
+async def sched_engine():
+    # StaticPool:内存库单连接共享,同 engine 的多 session 看到同一份数据
+    engine = create_async_engine(
+        "sqlite+aiosqlite:///:memory:", poolclass=StaticPool,
+        connect_args={"check_same_thread": False},
+    )
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+    maker = async_sessionmaker(engine, expire_on_commit=False)
+    yield engine, maker
+    await engine.dispose()
+
+
+async def test_analytics_tick_runs(sched_engine):
+    """滚动 3 天窗覆盖今日 seed → stat_daily_staff 出行;tick 不抛。
+
+    _analytics_tick 用真实「今天」算滚动窗(time.time() 喂 day_key,上海
+    自然日)——seed 的 day 必须相对当日,否则窗口错过。
+    """
+    engine, maker = sched_engine
+    today = day_key(int(time.time() * 1000))
+    async with maker() as s:
+        s.add(MsgDailyStats(channel="wechat", owner_account_id="wx1",
+                            peer_kind="friend", peer_id="t1", day=today,
+                            send_cnt=1, recv_cnt=0, last_ts=1756950000000,
+                            last_is_send=True))
+        s.add(SensitiveWord(word="私单"))
+        await s.commit()
+    sch = SyncScheduler(engine=engine, session_maker=maker)
+    await sch._analytics_tick()  # 不抛即过
+    async with maker() as s:
+        n = (await s.execute(
+            select(func.count()).select_from(StatDailyStaff)
+        )).scalar_one()
+        assert n >= 1
+
+
+async def test_analytics_tick_swallows_errors(sched_engine, monkeypatch, caplog):
+    """内部异常只记日志不抛(保住下一个调度周期)。
+
+    直接让 recompute_days 抛错——_analytics_tick 内部的
+    ``from app.storage.staff_summary import recompute_days`` 在调用时求值,
+    monkeypatch 模块属性即可命中。
+    """
+    engine, maker = sched_engine
+
+    async def _boom(*args, **kwargs):
+        raise RuntimeError("boom")
+
+    monkeypatch.setattr("app.storage.staff_summary.recompute_days", _boom)
+    sch = SyncScheduler(engine=engine, session_maker=maker)
+    with caplog.at_level(logging.ERROR, logger="app.sync.scheduler"):
+        await sch._analytics_tick()  # 不抛
+    assert any("analytics tick failed" in r.getMessage() for r in caplog.records)

yunke-server/tests/test_sync_scheduler.py

@@ -416,10 +416,13 @@ async def test_start_registers_two_jobs_with_expected_triggers(engine, session_m
     with patch.object(scheduler_module, "AsyncIOScheduler", return_value=spy):
         await sched.start()
 
-    # Three jobs registered (2026-08-21 加每周全量对齐).
+    # Four jobs registered (2026-08-21 加每周全量对齐;2026-09-04 加 4h 分析链).
-    assert spy.add_job.call_count == 3
+    assert spy.add_job.call_count == 4
     job_ids = {c.kwargs["id"] for c in spy.add_job.call_args_list}
-    assert job_ids == {"yunke_full_sync", "yunke_message_sync", "yunke_weekly_full_history"}
+    assert job_ids == {
+        "yunke_full_sync", "yunke_message_sync", "yunke_weekly_full_history",
+        "yunke_analytics_tick",
+    }
 
     # Full sync -> CronTrigger(hour=3); messages -> IntervalTrigger(hours=6).
     # add_job was called as (func, trigger, kwargs...) so trigger is args[1].
@@ -454,6 +457,16 @@ async def test_start_registers_two_jobs_with_expected_triggers(engine, session_m
     assert weekly_call.args[0].__name__ == "_safe_page_run"
     assert weekly_call.kwargs["args"] == ["full-history"]
 
+    # 2026-09-04 加 4h 分析链:纯 DB(非 _safe_page_run,无子进程),
+    # IntervalTrigger 与消息增量同节奏,不带 args。
+    tick_call = next(
+        c for c in spy.add_job.call_args_list if c.kwargs["id"] == "yunke_analytics_tick"
+    )
+    assert tick_call.args[0].__name__ == "_analytics_tick"
+    tick_interval = tick_call.args[1]
+    assert tick_interval.interval.total_seconds() == 6 * 3600
+    assert "args" not in tick_call.kwargs
+
     # coalesce=True + max_instances=1 protects against overlap.
     for c in spy.add_job.call_args_list:
         assert c.kwargs["coalesce"] is True