"""Phase-7 weekly monitoring digest (plan §7) — all from the observability tables.

    docker compose exec bot python tools/monitor.py [--days 7] [--check] [--include-calibration]

- Prints a Persian digest and writes the same content as markdown to
  `data/reports/monitoring_<YYYYMMDD_HHMM>.md`.
- No dashboard by design (owner decision): raw tables + SELECTs are the
  source of truth; this script is the weekly read of them.
- Default window covers REAL traffic only: traces with source='calibration'
  and the test traces (chat_id=0) are excluded from the stats (pass
  `--include-calibration` to see everything; the volume section always shows
  the full breakdown).
- `--check` exits 1 when a warning trips (space/volume growth per plan §4.5 —
  >5GB or >100k events/month → the «archive, do not delete» session is due —
  plus Jev failure ratio and model drift) so it can be wired to a scheduler.
- Nothing is ever deleted here (no-deletion policy).
"""
import argparse
import asyncio
import sys
from datetime import datetime, timezone
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

from sqlalchemy import text

from app.config import settings
from app.db import SessionLocal

GB = 1024 ** 3
TABLE_SIZE_WARN = 5 * GB          # plan §4.5: >5GB of observability storage
MONTHLY_EVENTS_WARN = 100_000     # plan §4.5: >100k events / month
JEV_FAIL_RATIO_WARN = 0.20        # of jev.result events in the window (n>=5)

# window predicate reused everywhere: --days 0 → all time
WIN = "(:days = 0 or created_at >= now() - (:days * interval '1 day'))"
WIN_E = "(:days = 0 or e.created_at >= now() - (:days * interval '1 day'))"
# real-traffic filter: keep only traces of genuine chats (no calibration, no chat_id=0 tests)
REAL = ("trace_id in (select id from bot_traces where coalesce(source,'bale') <> 'calibration' "
        "and coalesce(chat_id, 0) <> 0)")
REAL_E = ("e.trace_id in (select id from bot_traces where coalesce(source,'bale') <> 'calibration' "
          "and coalesce(chat_id, 0) <> 0)")


def _n(x) -> str:
    try:
        return f"{int(x):,}"
    except (TypeError, ValueError):
        return "—"


def _human(nbytes: int) -> str:
    if not nbytes:
        return "0 B"
    step = 1024.0
    v = float(nbytes)
    for unit in ("B", "KB", "MB", "GB", "TB"):
        if v < step or unit == "TB":
            return f"{v:,.1f} {unit}"
        v /= step
    return f"{v:,.1f} TB"


def _join(rows, fmt) -> str:
    return " · ".join(fmt(r) for r in rows) or "—"


async def build(days: int, include_cal: bool = False) -> tuple[list[str], list[str]]:
    """Return (digest lines, warnings)."""
    flt = "" if include_cal else f" and {REAL}"
    flt_e = "" if include_cal else f" and {REAL_E}"
    warnings: list[str] = []
    L: list[str] = []

    async with SessionLocal() as s:
        async def rows(sql, **params):
            return (await s.execute(text(sql), {"days": days, **params})).all()

        async def one(sql, **params):
            return (await s.execute(text(sql), {"days": days, **params})).first()

        # ---------------------------------------------------------- volume (always all traces)
        by_src = await rows("""
            select coalesce(source,'?') src, count(*) n from bot_traces
            where (:days = 0 or created_at >= now() - (:days * interval '1 day'))
            group by 1 order by n desc
        """)
        per_day = await rows(f"""
            select to_char(date_trunc('day', created_at), 'MM-DD') d, count(*) n
            from bot_trace_events where {WIN}{flt}
            group by 1 order by min(created_at)
        """)
        total_events = await one(f"select count(*) n from bot_trace_events where {WIN}{flt}")

        L += [f"پنجره: {'از ابتدا' if days == 0 else f'{days} روز اخیر'}"
              + ("" if include_cal else " · فقط ترافیک واقعی (بدون calibration/test)"), ""]
        L.append("- منابع روندها: " + _join(by_src, lambda r: f"{r.src}: {_n(r.n)}"))
        if per_day:
            L.append(f"- رخداد{'' if include_cal else ' واقعی'} در روز: "
                     + _join(per_day, lambda r: f"{r.d}={_n(r.n)}")
                     + f"  (کل پنجره: {_n(total_events.n)})")

        # ---------------------------------------------------------- stages / latency
        stages = await rows(f"""
            select stage, count(*) n,
                   count(*) filter (where not ok) fails,
                   round(percentile_cont(0.5) within group (order by latency_ms)) p50,
                   round(percentile_cont(0.95) within group (order by latency_ms)) p95
            from bot_trace_events where {WIN}{flt}
            group by 1 order by n desc
        """)
        L += ["", "### مرحله‌ها (stage · تعداد · خطا · p50/p95 ms)"]
        if not stages:
            L.append("- —")
        for r in stages:
            lat = f" · {r.p50}/{r.p95}" if r.p50 is not None else ""
            bad = f" · ⚠️ {_n(r.fails)} خطا" if r.fails else ""
            L.append(f"- `{r.stage}`: {_n(r.n)}{bad}{lat}")
        fails = [(r.stage, r.fails) for r in stages if r.fails]
        if fails:
            warnings.append("خطا در مراحل: " + ", ".join(f"{a}×{b}" for a, b in fails))

        # ---------------------------------------------------------- funnel
        nlu = await one(f"""
            select count(*) n, count(*) filter (where ok) ok from bot_trace_events
            where stage='nlu.result' and {WIN}{flt}
        """)
        mentions = await one(f"""
            select coalesce(sum(jsonb_array_length(payload->'entity_mentions')), 0) n
            from bot_trace_events
            where stage='entities.extracted' and jsonb_typeof(payload->'entity_mentions')='array'
              and {WIN}{flt}
        """)
        jev = await one(f"""
            select count(*) n, count(*) filter (where ok) ok, count(*) filter (where not ok) fail
            from bot_trace_events where stage='jev.result' and {WIN}{flt}
        """)
        route_src = await rows(f"""
            select coalesce(payload->>'source','?') src, count(*) n from bot_trace_events
            where stage='routing.decision' and {WIN}{flt} group by 1 order by n desc
        """)
        jev_status = await rows(f"""
            select coalesce(m->>'status','?') st, count(*) n
            from bot_trace_events e, jsonb_array_elements(e.payload->'mentions') m
            where e.stage='routing.decision' and e.payload->>'source'='jev' and {WIN_E}{flt_e}
            group by 1 order by n desc
        """)
        bands = await rows(f"""
            select case
                     when (m->>'confidence')::float >= :ex then 'exact'
                     when (m->>'confidence')::float >= :fl then 'ambiguity'
                     else 'low' end band, count(*) n
            from bot_trace_events e, jsonb_array_elements(e.payload->'mentions') m
            where e.stage='routing.decision' and e.payload->>'source'='jev' and {WIN_E}{flt_e}
            group by 1 order by n desc
        """, ex=settings.JEV_CONF_EXACT, fl=settings.JEV_CONF_FLOOR)
        difflib = await rows(f"""
            select coalesce(payload->>'status','?') st, count(*) n from bot_trace_events
            where stage='routing.decision' and payload->>'source'='difflib' and {WIN}{flt}
            group by 1 order by n desc
        """)
        commits = await one(f"""
            select count(*) filter (where stage='draft.commit') committed,
                   count(*) filter (where stage='draft.cancel') cancelled,
                   count(*) filter (where stage='draft.render') rendered
            from bot_trace_events where {WIN}{flt}
        """)

        L += ["", "### قیف پردازش"]
        L.append(f"- NLU (nlu.result): {_n(nlu.n)} — mentionهای استخراج‌شده: {_n(mentions.n)}")
        L.append(f"- Jev (jev.result): {_n(jev.n)} درخواست/فیلد — خطا: {_n(jev.fail)}")
        L.append(f"- مسیریابی: " + _join(route_src, lambda r: f"{r.src}: {_n(r.n)}"))
        L.append(f"- وضعیت mentionهای Jev: " + _join(jev_status, lambda r: f"{r.st}: {_n(r.n)}"))
        L.append(f"- باندهای confidence (EXACT={settings.JEV_CONF_EXACT} / FLOOR={settings.JEV_CONF_FLOOR}): "
                 + _join(bands, lambda r: f"{r.band}: {_n(r.n)}"))
        L.append(f"- difflib (fallback): " + _join(difflib, lambda r: f"{r.st}: {_n(r.n)}"))
        L.append(f"- پیش‌نویس: نمایش {_n(commits.rendered)} · ثبت نهایی {_n(commits.committed)} · لغو {_n(commits.cancelled)}")

        if jev.n and jev.n >= 5 and jev.fail / jev.n > JEV_FAIL_RATIO_WARN:
            warnings.append(f"نسبت خطای Jev بالا: {jev.fail}/{jev.n}")

        # ---------------------------------------------------------- user choices (quality proxy)
        user_amb = await rows(f"""
            select coalesce(payload->>'action','?') act, coalesce(payload->>'arg','-') arg, count(*) n
            from bot_trace_events where stage='ambiguity.user_choice' and {WIN}{flt}
            group by 1,2 order by n desc
        """)
        picks = await rows(f"""
            select case when (payload->>'idx')::int = 0 then 'top' else 'other' end k, count(*) n
            from bot_trace_events where stage='user_choice.pick' and {WIN}{flt}
            group by 1 order by n desc
        """)
        L += ["", "### انتخاب کاربر (پروکسی کیفیت — A/B واقعی Jev↔difflib↔کاربر)"]
        L.append("- ابهام‌ها: " + (_join(user_amb, lambda r: f"{r.act}({r.arg}): {_n(r.n)}")
                 if user_amb else "— هنوز کاربری در ابهام انتخاب نکرده"))
        top = sum(r.n for r in picks if r.k == "top")
        other = sum(r.n for r in picks if r.k == "other")
        if picks:
            L.append(f"- دکمه‌های گزینه: گزینه اول (پیشنهاد مدل) {_n(top)} · گزینه دیگر {_n(other)}")
        new_rej = sum(r.n for r in user_amb if r.act == "new")
        if new_rej:
            L.append(f"- «هیچ‌کدام/جدید» در ابهام: {_n(new_rej)} (پیشنهاد مدل رد شد)")

        # ---------------------------------------------------------- jev health
        tokens = await one(f"""
            select coalesce(sum((payload->'usage'->>'input_tokens')::bigint),0) tin,
                   coalesce(sum((payload->'usage'->>'output_tokens')::bigint),0) tout
            from bot_trace_events
            where stage='jev.result' and payload->'usage' is not null and {WIN}{flt}
        """)
        models = await rows(f"""
            select coalesce(model,'?') m, count(*) n from bot_trace_events
            where stage='jev.result' and {WIN}{flt} group by 1 order by n desc
        """)
        L += ["", "### سلامت Jev"]
        L.append(f"- توکن مصرفی (از usage): ورودی {_n(tokens.tin)} · خروجی {_n(tokens.tout)}")
        L.append(f"- مدل‌ها: " + _join(models, lambda r: f"`{r.m}`×{_n(r.n)}"))
        drift = [r.m for r in models if r.m not in ("?", settings.JEV_MODEL)]
        if drift:
            warnings.append(f"انحراف نسخه مدل Jev: {drift} ≠ پین {settings.JEV_MODEL}")

        # ---------------------------------------------------------- stt health
        stt = await rows(f"""
            select coalesce(payload->>'provider','?') p, count(*) n, count(*) filter (where ok) ok,
                   round(percentile_cont(0.5) within group (order by latency_ms)) p50
            from bot_trace_events where stage='stt.result' and {WIN}{flt} group by 1 order by n desc
        """)
        if stt:
            L += ["", "### STT"]
            for r in stt:
                L.append(f"- {r.p}: {_n(r.ok)}/{_n(r.n)} موفق · p50 {r.p50 if r.p50 is not None else '—'} ms")

        # ---------------------------------------------------------- space / archive guard (always all data)
        sizes = await one("""
            select pg_total_relation_size('bot_trace_events') ev_bytes,
                   pg_total_relation_size('bot_traces') tr_bytes,
                   pg_total_relation_size('bot_messages') msg_bytes
        """)
        month_events = await one("""
            select count(*) n from bot_trace_events
            where created_at >= date_trunc('month', now())
        """)
        voices_dir = Path(settings.OBS_VOICE_DIR)
        v_files = v_bytes = 0
        if voices_dir.exists():
            for p in voices_dir.rglob("*"):
                if p.is_file():
                    v_files += 1
                    v_bytes += p.stat().st_size
        obs_bytes = (sizes.ev_bytes or 0) + (sizes.tr_bytes or 0) + v_bytes

        L += ["", "### فضا و آرشیو (سیاست: آرشیو، نه حذف — پلن §4.5)"]
        L.append(f"- bot_trace_events: {_human(sizes.ev_bytes or 0)} · bot_traces: {_human(sizes.tr_bytes or 0)} "
                 f"· bot_messages: {_human(sizes.msg_bytes or 0)} · صدای خام: {_human(v_bytes)} ({_n(v_files)} فایل)")
        L.append(f"- رخدادهای این ماه تقویمی: {_n(month_events.n)}")
        if obs_bytes > TABLE_SIZE_WARN:
            warnings.append(f"حجم Observability از ۵GB گذشت: {_human(obs_bytes)} → جلسه آرشیو لازم است")
        if month_events.n > MONTHLY_EVENTS_WARN:
            warnings.append(f"رخداد این ماه از ۱۰۰هزار گذشت: {_n(month_events.n)} → جلسه آرشیو لازم است")

    return L, warnings


async def main() -> None:
    ap = argparse.ArgumentParser(description="Phase-7 monitoring digest")
    ap.add_argument("--days", type=int, default=7, help="0 = از ابتدا (default: 7)")
    ap.add_argument("--check", action="store_true", help="exit 1 when a warning trips")
    ap.add_argument("--include-calibration", action="store_true",
                    help="شامل ترافیک calibration/test در آمار (پیش‌فرض: فقط ترافیک واقعی)")
    args = ap.parse_args()

    ts = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M")
    lines, warnings = await build(args.days, args.include_calibration)

    header = [f"# پایش هفتگی بات — {ts} UTC", ""]
    if warnings:
        header += ["## ⚠️ هشدارها", *[f"- {w}" for w in warnings], ""]

    out_dir = Path(settings.OBS_VOICE_DIR).parent / "reports"
    out_dir.mkdir(parents=True, exist_ok=True)
    report = out_dir / f"monitoring_{ts}.md"
    report.write_text("\n".join(header + lines), encoding="utf-8")

    print("\n".join(header + lines))
    print(f"\nگزارش: {report}")
    if args.check and warnings:
        sys.exit(1)


if __name__ == "__main__":
    asyncio.run(main())