diff --git a/yunke-server/app/sync/scheduler.py b/yunke-server/app/sync/scheduler.py index d85d3fa..8a6c47d 100644 --- a/yunke-server/app/sync/scheduler.py +++ b/yunke-server/app/sync/scheduler.py @@ -701,25 +701,27 @@ class SyncScheduler: print(line, flush=True) async def _analytics_tick(self) -> None: - """4h 分析链(stat_daily_staff 滚动3天 + 敏感词增量)——异常只记日志不断链。 + """4h 分析链(staff 滚动3天 + 敏感词增量 + 管理域滚动)——异常只记日志不断链。 - 纯 DB 无浏览器子进程,不抢 ``_page_lock``(DB 写毫秒级; - ``max_instances=1`` 防自身重叠即可)。与夜间 full_sync 的 - staff-daily-stats 步骤重叠时,两侧 DELETE+INSERT 在重叠 PK 范围上 - 可能死锁/锁等待——非致命:tick 吞异常下轮重试,夜间脚本无 set -e - 次日自愈(UNIQUE/PK 幂等只保证串行提交下不重复,不消锁竞争)。 + 两段独立事务:段1 staff+敏感词(既有);段2 管理域(fact 近3天活跃会话 + + 宽表/通话 滚动3天,批次F 2026-09-10)。段2 失败只回滚自身,不牵连 + 段1 已提交结果(v2 Task12:指标失败不中断主分析链)。纯 DB 无浏览器 + 子进程,不抢 ``_page_lock``;``max_instances=1`` 防自身重叠即可。与 + 夜间 full_sync 的汇总/管理步重叠时,两侧 DELETE+INSERT 在重叠 PK + 范围可能锁等待——非致命:tick 吞异常下轮重试,夜间脚本幂等自愈。 时区:``day_key`` 内部钉死 Asia/Shanghai,喂 UTC epoch - (``time.time()*1000``)即无歧义——用 epoch 而非 naive datetime - 只为读法直接,数学上两者经本地时区往返等价。 + (``time.time()*1000``)即无歧义。 """ + from app.storage.summary import day_key + + today = day_key(int(time.time() * 1000)) + # ── 段1:staff 滚动 + 敏感词增量(既有,语义原样) ── 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() @@ -730,6 +732,32 @@ class SyncScheduler: except Exception: # noqa: BLE001 — scheduler MUST NOT crash logger.exception("analytics tick failed") + # ── 段2:管理域滚动(fact 活跃会话 + 宽表/通话,当日+滚动3天) ── + try: + async with self._session_maker() as session: + from app.storage.employee_metrics import ( + recompute_call_days, + recompute_metric_days, + ) + from app.storage.first_reply_facts import ( + pairs_active_in_days, + recompute_pairs, + ) + from app.storage.staff_summary import rolling_days + + rd = rolling_days(today, 3) + pairs = await pairs_active_in_days(session, rd) + f = await recompute_pairs(session, pairs) + w = await recompute_metric_days(session, rd) + c = await recompute_call_days(session, rd) + await session.commit() + logger.info( + "[analytics-tick] mgmt fact_pairs=%s wide_rows=%s call_rows=%s", + f, w, c, + ) + except Exception: # noqa: BLE001 + logger.exception("management metrics tick failed") + async def _page_subprocess(self, args: list[str], *, timeout: float | None = None) -> int: """启动页面子进程并等待结束;继承环境 + 补 DB_URL/PYTHONIOENCODING。 diff --git a/yunke-server/tests/test_scheduler_analytics_tick.py b/yunke-server/tests/test_scheduler_analytics_tick.py index 680be4e..3e8db97 100644 --- a/yunke-server/tests/test_scheduler_analytics_tick.py +++ b/yunke-server/tests/test_scheduler_analytics_tick.py @@ -1,4 +1,4 @@ -"""_analytics_tick:滚动刷新 + 增量扫描 + 异常吞掉(数据分析板块 v1, T11)。 +"""_analytics_tick:滚动刷新 + 增量扫描 + 管理域滚动段 + 异常吞掉(v1 T11;批次F F1)。 行为测试是底线——start() 的 job 注册断言在 test_sync_scheduler.py 的 注册测试里同步更新(那里 spy 断言 add_job 全集)。 @@ -15,6 +15,7 @@ from sqlalchemy.pool import StaticPool from app.db.base import Base from app.db.models import MsgDailyStats, SensitiveWord, StatDailyStaff +from app.storage.staff_summary import rolling_days from app.storage.summary import day_key from app.sync.scheduler import SyncScheduler @@ -74,3 +75,110 @@ async def test_analytics_tick_swallows_errors(sched_engine, monkeypatch, caplog) 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) + + +async def _seed_today(sched_engine) -> int: + """seed 当日一条 wechat friend 会话 + 敏感词,返回 seed 用的 day 键。""" + _, 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() + return today + + +async def test_tick_calls_management_chain_in_order(sched_engine, monkeypatch): + """两段链完整顺序:recompute_days → scan → pairs_active_in_days → + recompute_pairs → recompute_metric_days → recompute_call_days。 + + 记录桩替换全部六个上游——_analytics_tick 段内函数延迟 import, + monkeypatch 模块属性即命中(惯例见 test_analytics_tick_swallows_errors)。 + """ + engine, maker = sched_engine + calls: list[str] = [] + + def rec(name: str, ret): + async def _stub(*args, **kwargs): + calls.append(name) + return ret + return _stub + + monkeypatch.setattr("app.storage.staff_summary.recompute_days", + rec("recompute_days", 0)) + monkeypatch.setattr("app.analytics.sensitive.scanner.run_incremental", + rec("scan", {"scanned": 0, "hits": 0})) + monkeypatch.setattr("app.storage.first_reply_facts.pairs_active_in_days", + rec("pairs_active_in_days", [])) + monkeypatch.setattr("app.storage.first_reply_facts.recompute_pairs", + rec("recompute_pairs", 0)) + monkeypatch.setattr("app.storage.employee_metrics.recompute_metric_days", + rec("recompute_metric_days", 0)) + monkeypatch.setattr("app.storage.employee_metrics.recompute_call_days", + rec("recompute_call_days", 0)) + + sch = SyncScheduler(engine=engine, session_maker=maker) + await sch._analytics_tick() + assert calls == [ + "recompute_days", "scan", # 段1 + "pairs_active_in_days", "recompute_pairs", # 段2 + "recompute_metric_days", "recompute_call_days", + ] + + +async def test_tick_management_failure_isolated(sched_engine, monkeypatch, caplog): + """段2 失败隔离:recompute_metric_days 抛 RuntimeError → ①tick 不抛; + ②记 "management metrics tick failed";③段1 已提交的 StatDailyStaff + 不被段2 回滚(两段独立事务)。 + """ + engine, maker = sched_engine + await _seed_today(sched_engine) + + async def _boom(*args, **kwargs): + raise RuntimeError("boom") + + monkeypatch.setattr("app.storage.employee_metrics.recompute_metric_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( # ②段2 独立记日志 + "management metrics tick failed" in r.getMessage() for r in caplog.records + ) + async with maker() as s: # ③段1 结果已提交 + n = (await s.execute( + select(func.count()).select_from(StatDailyStaff) + )).scalar_one() + assert n >= 1 + + +async def test_tick_management_rolling_window(sched_engine, monkeypatch): + """管理域三函数收到的 days == rolling_days(today, 3)(长度 3,当日+前 + 两天),且与 staff 段捕获的窗口一致(两段共用函数顶提取的同一 today)。 + """ + engine, maker = sched_engine + got: dict[str, list[int]] = {} + + def cap(name: str, ret): + async def _stub(session, days, *args, **kwargs): + got[name] = list(days) + return ret + return _stub + + monkeypatch.setattr("app.storage.staff_summary.recompute_days", + cap("staff", 0)) + monkeypatch.setattr("app.storage.first_reply_facts.pairs_active_in_days", + cap("pairs", [])) + monkeypatch.setattr("app.storage.employee_metrics.recompute_metric_days", + cap("wide", 0)) + monkeypatch.setattr("app.storage.employee_metrics.recompute_call_days", + cap("call", 0)) + + sch = SyncScheduler(engine=engine, session_maker=maker) + await sch._analytics_tick() + + rd = rolling_days(day_key(int(time.time() * 1000)), 3) + assert len(rd) == 3 # 当日 + 前两天 + assert got["staff"] == got["pairs"] == got["wide"] == got["call"] == rd