feat(management): 两条回填CLI——first-reply-facts/employee-metrics全量重建幂等

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

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