diff --git a/yunke-server/app/sync/cli.py b/yunke-server/app/sync/cli.py index e856184..e5a9770 100644 --- a/yunke-server/app/sync/cli.py +++ b/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 diff --git a/yunke-server/app/sync/scheduler.py b/yunke-server/app/sync/scheduler.py index 8980c2b..1949c76 100644 --- a/yunke-server/app/sync/scheduler.py +++ b/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"} diff --git a/yunke-server/scripts/full_sync_all.sh b/yunke-server/scripts/full_sync_all.sh index 3c4079b..b8ceaa2 100644 --- a/yunke-server/scripts/full_sync_all.sh +++ b/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 ===" diff --git a/yunke-server/tests/test_scheduler_analytics_tick.py b/yunke-server/tests/test_scheduler_analytics_tick.py new file mode 100644 index 0000000..680be4e --- /dev/null +++ b/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) diff --git a/yunke-server/tests/test_sync_scheduler.py b/yunke-server/tests/test_sync_scheduler.py index 3002153..3095927 100644 --- a/yunke-server/tests/test_sync_scheduler.py +++ b/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 加每周全量对齐). - assert spy.add_job.call_count == 3 + # Four jobs registered (2026-08-21 加每周全量对齐;2026-09-04 加 4h 分析链). + 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