"""Weekly KB-gap digest (服务台收编 PR-4, IT-only).

``support.kb_gap_digest`` (seeded by migration 490, Mondays 09:00
Asia/Shanghai): deflection sessions from the past window where the
knowledge base had ZERO hits and the employee ended up filing a ticket
(escalated) or walking away (expired) are exactly the questions the KB
cannot answer. Cluster them by text similarity and digest the top
clusters to IT admins — the "which doc to write next" worklist. Silent
when the window has no gaps (no empty-digest spam), and strictly
IT-only per the notification红线 (no new employee-facing pushes).
"""

from __future__ import annotations

from datetime import UTC, datetime, timedelta
from typing import Any

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.domains.support.models import (
    DEFLECT_ESCALATED,
    DEFLECT_EXPIRED,
    SupportDeflectionEvent,
)
from app.domains.support.service import _jaccard, _shingles
from app.notifications.service import NotificationEnvelope
from app.scheduled_jobs.registry import register_scheduled_job

_SETTING_LAST_SENT = "support.kb_gap_digest.last_sent_at"

# Chinese paraphrases of the same question land around 0.25–0.5 bigram
# Jaccard (measured: 「怎么申请 GitLab 权限」 vs 「如何申请 GitLab 的权限」
# = 0.30); unrelated questions sit at ~0. 0.28 collapses paraphrases
# without merging genuinely different topics.
_CLUSTER_SIM = 0.28


def _cluster(queries: list[str]) -> list[tuple[str, int]]:
    """Greedy shingle-Jaccard clustering — near-duplicate questions collapse
    into one line with a count. Representative = the first (longest) query."""
    clusters: list[tuple[str, set[str], int]] = []
    for q in sorted({q.strip() for q in queries if q.strip()}, key=lambda x: (-len(x), x)):
        sh = _shingles(q)
        for i, (rep, rep_sh, n) in enumerate(clusters):
            if _jaccard(sh, rep_sh) >= _CLUSTER_SIM:
                clusters[i] = (rep, rep_sh, n + 1)
                break
        else:
            clusters.append((q, sh, 1))
    ranked = sorted(clusters, key=lambda c: c[2], reverse=True)
    return [(rep, n) for rep, _sh, n in ranked]


@register_scheduled_job(
    key="support.kb_gap_digest",
    display_name="求助 · 知识库缺口周报(IT-only)",
    description=(
        "聚合窗口期内「检索零命中且员工最终提单/放弃」的自助求助问题,"
        "聚类为「该补的文档清单」推送 IT 管理员;无缺口静默。"
    ),
)
async def run_support_kb_gap_digest(
    db: AsyncSession,
    config: dict[str, Any],
) -> dict[str, Any]:
    try:
        window_days = int((config or {}).get("window_days") or 7)
    except (TypeError, ValueError):
        window_days = 7
    try:
        top_n = int((config or {}).get("top_n") or 10)
    except (TypeError, ValueError):
        top_n = 10
    # Clamp: per-line text is truncated, but the LINE COUNT is the other
    # axis of the card/email size limit.
    top_n = max(1, min(top_n, 50))

    # Window dedup: a re-run inside the same window (scheduler refire after
    # a crash between enqueue and drain, or a manual run) must not send a
    # second digest to every IT admin. The marker is committed BEFORE the
    # fan-out (receipt-before-deliver, repo invariant): a crash after the
    # marker means one skipped week — never a duplicate blast.
    from app.integrations.models import PlatformSetting
    from app.integrations.service import get_platform_setting

    last_sent_raw = await get_platform_setting(_SETTING_LAST_SENT, None, session=db)
    if last_sent_raw:
        try:
            last_sent = datetime.fromisoformat(str(last_sent_raw))
            min_gap = timedelta(days=max(1, window_days - 1))
            if datetime.now(UTC) - last_sent < min_gap:
                return {
                    "window_days": window_days,
                    "skipped": "already_sent_this_window",
                    "last_sent_at": str(last_sent_raw),
                }
        except ValueError:
            pass  # unreadable marker — proceed rather than silently stall

    cutoff = datetime.now(UTC) - timedelta(days=window_days)
    rows = (
        await db.execute(
            select(
                SupportDeflectionEvent.query,
                SupportDeflectionEvent.kb_article_ids,
            ).where(
                SupportDeflectionEvent.outcome.in_(
                    (DEFLECT_ESCALATED, DEFLECT_EXPIRED),
                ),
                SupportDeflectionEvent.created_at >= cutoff,
            )
        )
    ).all()
    # Zero-hit = the KB (native + mirrored Feishu docs) had nothing to show.
    gaps = [q for q, kb_ids in rows if not kb_ids]
    if not gaps:
        return {"window_days": window_days, "gaps": 0, "skipped": "no_gaps"}

    clusters = _cluster(gaps)[:top_n]
    # Representatives are title+body strings up to 8000 chars — untruncated
    # they overflow every Feishu-card/email limit and the whole digest
    # silently fails per-recipient while the job still records ok.
    gap_lines = "\n".join(
        f"{i + 1}. {rep[:120]}{'…' if len(rep) > 120 else ''}（×{n}）"
        for i, (rep, n) in enumerate(clusters)
    )

    from app.notifications import fanout
    from app.scheduled_jobs.failure_notifier import list_admin_recipients

    await db.commit()  # nothing pending, but keep the primitive's contract
    admins = await list_admin_recipients(db)
    envelopes = [
        NotificationEnvelope(
            recipient=admin,
            title=f"知识库缺口周报 · {len(gaps)} 个未命中问题",
            body_text=gap_lines,
            channels=["feishu", "email"],
            template_key="support.kb_gap_digest",
            context={
                "window_days": str(window_days),
                "gap_count": str(len(gaps)),
                "cluster_count": str(len(clusters)),
                "gap_lines": gap_lines,
            },
            # 知识库 moved into 知识与制度 (/admin/knowledge) 2026-08-17 —
            # /admin/support/kb still redirects, but a digest card should
            # deep-link the live path, not a bounce.
            url="/admin/knowledge",
        )
        for admin in admins
    ]
    if envelopes:
        # Receipt BEFORE deliver (repo invariant): the marker must be durable
        # before any outbox row exists. enqueue_then_drain commits rows first
        # and delivers after — a deploy killing the drain, plus the scheduler
        # re-firing the "missed" cron, would otherwise double-send the digest
        # to every IT admin. Worst case is now one SKIPPED week, not spam.
        await db.merge(
            PlatformSetting(
                key=_SETTING_LAST_SENT,
                value=datetime.now(UTC).isoformat(),
            )
        )
        await db.commit()
        await fanout.enqueue_then_drain(
            db,
            envelopes,
            trace_id="",
            label="support.kb_gap_digest",
        )
    return {
        "window_days": window_days,
        "gaps": len(gaps),
        "clusters": len(clusters),
        "notified": len(envelopes),
    }