diff --git a/yunke-server/app/sync/cli.py b/yunke-server/app/sync/cli.py index e5a9770..11372c6 100644 --- a/yunke-server/app/sync/cli.py +++ b/yunke-server/app/sync/cli.py @@ -15,6 +15,8 @@ Three invocations (plus analytics backfills, see below):: python -m app.sync.cli daily-stats python -m app.sync.cli staff-daily-stats python -m app.sync.cli sensitive-scan --full + python -m app.sync.cli first-reply-facts + python -m app.sync.cli employee-metrics Auth + DB --------- @@ -136,6 +138,30 @@ def _build_parser() -> argparse.ArgumentParser: help="Full backfill scan (first deploy / new-word history).", ) + # ``first-reply-facts`` ---------------------------------------------- + p_fr = sub.add_parser( + "first-reply-facts", + help="Rebuild first_reply_fact (cross-day exact threads).", + description=( + "Full rebuild of the first-reply fact table from the message " + "table (cross-day exact thread algorithm — overnight samples " + "not lost, unlike msg_daily_stats day-cut). Deploy backfill + " + "nightly self-heal step. Idempotent." + ), + ) + + # ``employee-metrics`` ------------------------------------------------ + p_emp = sub.add_parser( + "employee-metrics", + help="Rebuild employee_metric_daily + employee_call_daily.", + description=( + "Full rebuild of the management wide table (per-day x channel " + "x staff, 1:1 caliber) and the call daily table. Must run " + "AFTER first-reply-facts (thread columns share the same " + "source). Deploy backfill + nightly self-heal. Idempotent." + ), + ) + # ``scheduler`` ----------------------------------------------------------- p_sched = sub.add_parser( "scheduler", @@ -374,6 +400,50 @@ async def _cmd_sensitive_scan(args: argparse.Namespace) -> int: await engine.dispose() +async def _cmd_first_reply_facts(args: argparse.Namespace) -> int: + """全量重建 first_reply_fact(跨日精确串;部署回填/夜间自愈,幂等)。""" + settings = get_settings() + engine, session_maker = await _init_db(settings.db_url) + try: + from app.storage.first_reply_facts import recompute_all + + async with session_maker() as session: + pairs = await recompute_all(session) + await session.commit() + print(f"[first-reply-facts] rebuilt {pairs} conversations") + return 0 + finally: + await engine.dispose() + + +async def _cmd_employee_metrics(args: argparse.Namespace) -> int: + """全量重建 employee_metric_daily + employee_call_daily(幂等)。""" + settings = get_settings() + engine, session_maker = await _init_db(settings.db_url) + try: + import sqlalchemy as sa + + from app.db.models import MsgDailyStats + from app.storage.employee_metrics import ( + recompute_call_days, + recompute_metric_days, + ) + + async with session_maker() as session: + days = ( + await session.execute(sa.select(MsgDailyStats.day).distinct()) + ).scalars().all() + rows = await recompute_metric_days(session, list(days)) + calls = await recompute_call_days(session, list(days)) + await session.commit() + print( + f"[employee-metrics] rebuilt {rows} wide rows, {calls} call rows" + ) + 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 @@ -398,6 +468,10 @@ def main(argv: list[str] | None = None) -> int: return asyncio.run(_cmd_staff_daily_stats(args)) if args.command == "sensitive-scan": return asyncio.run(_cmd_sensitive_scan(args)) + if args.command == "first-reply-facts": + return asyncio.run(_cmd_first_reply_facts(args)) + if args.command == "employee-metrics": + return asyncio.run(_cmd_employee_metrics(args)) except KeyboardInterrupt: # So `python -m app.sync.cli scheduler` exits cleanly on Ctrl-C. return 0