"""Org-wide API-Key analytics for the admin "报表 / Reports" tab.

The admin API-Key management page (``web/src/pages/asset/ApiKeyManagementPage.tsx``)
has an Assignments tab and a Providers tab. This module powers a third
**Reports** tab that answers the three questions an IT/finance admin
asks about the API-key fleet at a glance:

* **分布 / Distribution** — how many keys, by provider, by status, by
  cost-attribution binding, by department, by owner. All cheap GROUP-BY
  reductions over ``api_key_assignments`` (+ a person/department side
  lookup). Computed over the FULL population.
* **费用 / Cost** — the committed monthly-budget pool *and* the actual
  month-to-date cloud spend. The report uses a bounded-query bulk
  projection of the exact attribution rules used by the per-key panel
  and budget-alert cron, so the numbers reconcile to the cent without
  issuing a query sequence for every key. Spend is
  computed only for keys that could plausibly have facts in the window
  (see :data:`_SPEND_LIVE_STATUSES` + the revoked-in-window carve-out)
  so the loop is bounded to the granted-this-or-prior-month universe.
* **量 / Volume & lifecycle** — 6-month created/revoked trend, validity
  buckets (expiring 7d / 30d / expired), extension pressure, revoke
  breakdown (by actor role + reason code), and shared-roster stats.

Honesty about spend gaps is first-class: the same buckets that return
``data_available=False`` per-key (Bailian via Aliyun BSS account-only,
Foundry without an RG binding, Bedrock without an IAM binding) are
counted into ``spend.unresolvable`` so the UI can show "actual spend
covers N of M keys; $X is not per-key attributable" rather than a
silently-undercounted total.

The whole payload is read-only — no audit row, same as the per-key
``/spend`` endpoints.
"""
from __future__ import annotations

import calendar
from collections import defaultdict
from datetime import date, datetime
from decimal import Decimal
from typing import Any
from uuid import UUID
from zoneinfo import ZoneInfo

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

from app.domains.apikey.analytics_spend import compute_assignments_spend_bulk
from app.domains.apikey.expiry import effective_assignment_status
from app.domains.apikey.models import (
    ApiKeyAssignment,
    ApiKeyAssignmentMember,
    ApiKeyProvider,
)
from app.identity.person_models import IdentityOrgMembership, IdentityPerson

SHANGHAI = ZoneInfo("Asia/Shanghai")

# Statuses whose keys are currently "live" enough to plausibly accrue
# cost in the requested month. ``pending_approval`` / ``rejected`` were
# never granted (no IAM/RG binding minted) so they cannot have facts;
# ``revoked`` is handled separately (only included when the revoke
# happened inside the window — a key revoked in a prior month has no
# facts in this month).
_SPEND_LIVE_STATUSES = ("active", "approved", "expired", "pending_release")

# How many rows to surface in the "top-N" tables. Beyond this the chart
# / table stops being scannable; the FE can always drill into the
# Assignments tab for the long tail.
_TOP_N = 15


def _resolve_month(month: str | None) -> tuple[int, int, date, date, date]:
    """Parse ``YYYY-MM`` (or default to current Shanghai month).

    Returns ``(year, month_num, month_start, month_end, today)``.
    ``month_end`` is clamped to ``today`` (Asia/Shanghai) for the current
    month so the spend window matches the per-key panel / budget cron;
    for a past month it is the calendar last day. Raises ``ValueError``
    on a malformed string — the router validates the pattern first, so
    this is defence-in-depth.
    """
    today = datetime.now(SHANGHAI).date()
    if month:
        year_str, month_str = month.split("-", 1)
        year, month_num = int(year_str), int(month_str)
        if not 1 <= month_num <= 12:
            raise ValueError(f"month out of range: {month!r}")
    else:
        year, month_num = today.year, today.month
    month_start = date(year, month_num, 1)
    last_day = calendar.monthrange(year, month_num)[1]
    month_end = min(date(year, month_num, last_day), today)
    return year, month_num, month_start, month_end, today


def _month_key(d: date) -> str:
    return f"{d.year:04d}-{d.month:02d}"


def _trailing_months(end: date, count: int) -> list[str]:
    """``count`` ``YYYY-MM`` keys ending at ``end``'s month, oldest first."""
    year, month_num = end.year, end.month
    out: list[str] = []
    for _ in range(count):
        out.append(f"{year:04d}-{month_num:02d}")
        month_num -= 1
        if month_num == 0:
            month_num = 12
            year -= 1
    return list(reversed(out))


def _binding_mode(a: ApiKeyAssignment) -> str:
    """Cost-attribution binding bucket — mirrors the FE ``getBindingMode``
    in ApiKeyManagementPage so the report and the table agree."""
    if a.iam_user_id or a.iam_user_arn:
        return "iam"
    if a.cloud_scope_id:
        return "scope"
    return "unattributed"


async def compute_apikey_analytics(
    session: AsyncSession,
    *,
    month: str | None = None,
) -> dict[str, Any]:
    """Build the full analytics payload for the admin Reports tab.

    See the module docstring for the three reported axes. All reductions
    run against the full ``api_key_assignments`` population except the
    actual-spend loop, which is bounded to the live-or-revoked-in-window
    subset (:data:`_SPEND_LIVE_STATUSES`).
    """
    year, month_num, month_start, month_end, today = _resolve_month(month)
    month_str = f"{year:04d}-{month_num:02d}"

    # ── Load the full population once ────────────────────────────────
    assignments = list(
        (await session.execute(select(ApiKeyAssignment))).scalars().all()
    )

    provider_names: dict[UUID, str] = dict(
        (await session.execute(
            select(ApiKeyProvider.id, ApiKeyProvider.name)
        )).all()
    )

    owner_ids = {a.assigned_to for a in assignments if a.assigned_to}
    persons: dict[UUID, IdentityPerson] = {}
    dept_by_person: dict[UUID, str | None] = {}
    if owner_ids:
        person_rows = (await session.execute(
            select(IdentityPerson).where(IdentityPerson.id.in_(owner_ids))
        )).scalars().all()
        persons = {p.id: p for p in person_rows}
        # Primary org membership → department name. One query, no lazy
        # relationship load (the ``primary_department`` property would
        # MissingGreenlet under the async session).
        dept_rows = (await session.execute(
            select(
                IdentityOrgMembership.person_id,
                IdentityOrgMembership.org_node_name,
            ).where(
                IdentityOrgMembership.person_id.in_(owner_ids),
                IdentityOrgMembership.is_primary.is_(True),
                IdentityOrgMembership.deleted_at.is_(None),
            )
        )).all()
        for pid, name in dept_rows:
            # First primary membership wins; the unique constraint makes
            # duplicates rare, and a tie is cosmetic for a report bucket.
            dept_by_person.setdefault(pid, name)

    # Member roster counts per assignment (one grouped query).
    member_counts: dict[UUID, int] = dict(
        (await session.execute(
            select(
                ApiKeyAssignmentMember.assignment_id,
                func.count(),
            ).group_by(ApiKeyAssignmentMember.assignment_id)
        )).all()
    )

    # ── Distribution + lifecycle (Python reductions) ─────────────────
    total_keys = len(assignments)
    by_status: dict[str, int] = defaultdict(int)
    by_provider_count: dict[UUID, int] = defaultdict(int)
    by_provider_active: dict[UUID, int] = defaultdict(int)
    by_provider_budget: dict[UUID, Decimal] = defaultdict(Decimal)
    by_binding: dict[str, int] = defaultdict(int)
    by_dept_count: dict[str, int] = defaultdict(int)
    by_dept_budget: dict[str, Decimal] = defaultdict(Decimal)
    by_owner_count: dict[UUID, int] = defaultdict(int)
    by_owner_budget: dict[UUID, Decimal] = defaultdict(Decimal)

    created_trend: dict[str, int] = defaultdict(int)
    revoked_trend: dict[str, int] = defaultdict(int)
    revoke_by_role: dict[str, int] = defaultdict(int)
    revoke_by_reason: dict[str, int] = defaultdict(int)

    active_statuses = {"active", "approved"}
    total_budget = Decimal(0)
    active_with_budget = 0

    # Validity buckets (only meaningful for live keys).
    valid_expiring_7d = valid_expiring_30d = valid_expired = valid_no_expiry = 0
    extended_keys = 0
    total_extensions = 0

    # Sharing.
    keys_with_members = 0
    total_members = 0
    max_roster = 0

    trend_window = set(_trailing_months(month_end, 6))

    for a in assignments:
        st = effective_assignment_status(a.status, a.valid_until, today=today)
        by_status[st] += 1
        budget = Decimal(str(a.monthly_budget_usd or 0))
        is_live = st in active_statuses

        by_provider_count[a.provider_id] += 1
        if is_live:
            by_provider_active[a.provider_id] += 1
            by_provider_budget[a.provider_id] += budget
            total_budget += budget
            if budget > 0:
                active_with_budget += 1

        # Cost-attribution binding (over live keys — the rebind-worthy set).
        if is_live:
            by_binding[_binding_mode(a)] += 1

        # Department + owner (over live keys for budget; count over all).
        dept = dept_by_person.get(a.assigned_to) or "(unknown)"
        by_owner_count[a.assigned_to] += 1
        by_dept_count[dept] += 1
        if is_live:
            by_dept_budget[dept] += budget
            by_owner_budget[a.assigned_to] += budget

        # Created / revoked trend (6-month window).
        if a.created_at is not None:
            ck = _month_key(a.created_at.date() if isinstance(a.created_at, datetime) else a.created_at)
            if ck in trend_window:
                created_trend[ck] += 1
        if a.revoked_at is not None:
            rk = _month_key(a.revoked_at.date() if isinstance(a.revoked_at, datetime) else a.revoked_at)
            if rk in trend_window:
                revoked_trend[rk] += 1

        if st == "revoked":
            revoke_by_role[a.revoke_actor_role or "admin"] += 1
            revoke_by_reason[a.revoke_reason_code or "(unspecified)"] += 1

        # Validity (live keys only).
        if is_live:
            if a.valid_until is None:
                valid_no_expiry += 1
            else:
                days_left = (a.valid_until - today).days
                if days_left < 0:
                    valid_expired += 1
                elif days_left <= 7:
                    valid_expiring_7d += 1
                elif days_left <= 30:
                    valid_expiring_30d += 1
            if (a.extended_count or 0) > 0:
                extended_keys += 1
                total_extensions += a.extended_count

        n_members = member_counts.get(a.id, 0)
        if is_live and n_members > 0:
            keys_with_members += 1
            total_members += n_members
            max_roster = max(max_roster, n_members)

    def _person_label(
        pid: UUID,
    ) -> tuple[str | None, str | None, str | None, str | None]:
        p = persons.get(pid)
        if p is None:
            return None, None, None, None
        return p.display_name, p.work_email, p.employee_no, p.employment_status

    by_provider = sorted(
        (
            {
                "provider_id": str(pid),
                "provider_name": provider_names.get(pid, "(unknown)"),
                "count": by_provider_count[pid],
                "active_count": by_provider_active.get(pid, 0),
                "budget_usd": float(by_provider_budget.get(pid, Decimal(0))),
            }
            for pid in by_provider_count
        ),
        key=lambda r: (-r["count"], r["provider_name"]),
    )

    top_owners = []
    for pid, _cnt in sorted(
        by_owner_count.items(), key=lambda kv: (-kv[1], str(kv[0]))
    )[:_TOP_N]:
        name, email, emp_no, emp_status = _person_label(pid)
        top_owners.append({
            "person_id": str(pid),
            "display_name": name,
            "work_email": email,
            "display_name_employee_no": emp_no,
            "display_name_employment_status": emp_status,
            "count": by_owner_count[pid],
            "budget_usd": float(by_owner_budget.get(pid, Decimal(0))),
        })

    by_department = sorted(
        (
            {
                "department": dept,
                "count": cnt,
                "budget_usd": float(by_dept_budget.get(dept, Decimal(0))),
            }
            for dept, cnt in by_dept_count.items()
        ),
        key=lambda r: (-r["count"], r["department"]),
    )[:_TOP_N]

    # ── Actual MTD spend (bounded-query batch) ───────────────────────
    spend_universe = [
        a for a in assignments
        if effective_assignment_status(
            a.status, a.valid_until, today=today,
        ) in _SPEND_LIVE_STATUSES
        or (a.status == "revoked" and a.revoked_at is not None
            and (a.revoked_at.date() if isinstance(a.revoked_at, datetime)
                 else a.revoked_at) >= month_start)
    ]

    total_mtd = Decimal(0)
    resolvable = 0
    unresolvable = 0
    unresolvable_by_bucket: dict[str, int] = defaultdict(int)
    spend_by_bucket_usd: dict[str, Decimal] = defaultdict(Decimal)
    spend_by_bucket_keys: dict[str, int] = defaultdict(int)
    spend_by_provider_usd: dict[UUID, Decimal] = defaultdict(Decimal)
    health = {"ok": 0, "warning": 0, "over": 0, "unknown": 0}
    last_fact_dates: list[str] = []
    per_key_spend: list[dict[str, Any]] = []

    spend_by_assignment = await compute_assignments_spend_bulk(
        session,
        assignments=spend_universe,
        provider_names=provider_names,
        month_start=month_start,
        month_end=month_end,
    )
    for a in spend_universe:
        spend = spend_by_assignment[a.id]
        bucket = spend["bucket"]
        if not spend["data_available"]:
            unresolvable += 1
            unresolvable_by_bucket[bucket] += 1
            health["unknown"] += 1
            continue
        resolvable += 1
        mtd: Decimal = spend["mtd_usd"]
        total_mtd += mtd
        spend_by_bucket_usd[bucket] += mtd
        spend_by_bucket_keys[bucket] += 1
        spend_by_provider_usd[a.provider_id] += mtd
        if spend.get("last_fact_date"):
            last_fact_dates.append(spend["last_fact_date"])

        budget = Decimal(str(a.monthly_budget_usd or 0))
        if budget <= 0:
            pct = 0.0
            health["ok"] += 1
        else:
            pct = round(float(mtd / budget * 100), 2)
            if pct < 80:
                health["ok"] += 1
            elif pct < 100:
                health["warning"] += 1
            else:
                health["over"] += 1

        name, email, emp_no, emp_status = _person_label(a.assigned_to)
        per_key_spend.append({
            "assignment_id": str(a.id),
            "request_number": a.request_number,
            "project_name": a.project_name,
            "provider_name": provider_names.get(a.provider_id, "(unknown)"),
            "bucket": bucket,
            "owner_display_name": name,
            "owner_work_email": email,
            "owner_employee_no": emp_no,
            "owner_employment_status": emp_status,
            "mtd_usd": float(round(mtd, 2)),
            "budget_usd": float(budget),
            "percent": pct,
        })

    top_spenders = sorted(
        per_key_spend, key=lambda r: -r["mtd_usd"]
    )[:_TOP_N]

    spend_by_bucket = sorted(
        (
            {
                "bucket": b,
                "mtd_usd": float(round(spend_by_bucket_usd[b], 2)),
                "key_count": spend_by_bucket_keys[b],
            }
            for b in spend_by_bucket_usd
        ),
        key=lambda r: -r["mtd_usd"],
    )

    spend_by_provider = sorted(
        (
            {
                "provider_id": str(pid),
                "provider_name": provider_names.get(pid, "(unknown)"),
                "mtd_usd": float(round(usd, 2)),
                "budget_usd": float(by_provider_budget.get(pid, Decimal(0))),
            }
            for pid, usd in spend_by_provider_usd.items()
        ),
        key=lambda r: -r["mtd_usd"],
    )

    overall_pct = (
        round(float(total_mtd / total_budget * 100), 2)
        if total_budget > 0 else 0.0
    )

    return {
        "month": month_str,
        "generated_at": datetime.now(SHANGHAI).isoformat(),
        "totals": {
            "total_keys": total_keys,
            "active_keys": by_status.get("active", 0) + by_status.get("approved", 0),
            "total_monthly_budget_usd": float(total_budget),
            "active_with_budget": active_with_budget,
            "total_mtd_usd": float(round(total_mtd, 2)),
            "budget_utilization_percent": overall_pct,
        },
        "distribution": {
            "by_status": [
                {"status": s, "count": c}
                for s, c in sorted(by_status.items(), key=lambda kv: -kv[1])
            ],
            "by_provider": by_provider,
            "by_cost_attribution": [
                {"mode": m, "count": by_binding.get(m, 0)}
                for m in ("iam", "scope", "unattributed")
            ],
            "by_department": by_department,
            "top_owners": top_owners,
        },
        "validity": {
            "expiring_7d": valid_expiring_7d,
            "expiring_30d": valid_expiring_30d,
            "expired": valid_expired,
            "no_expiry": valid_no_expiry,
            "extended_keys": extended_keys,
            "total_extensions": total_extensions,
        },
        "lifecycle": {
            "created_trend": [
                {"month": m, "count": created_trend.get(m, 0)}
                for m in _trailing_months(month_end, 6)
            ],
            "revoked_trend": [
                {"month": m, "count": revoked_trend.get(m, 0)}
                for m in _trailing_months(month_end, 6)
            ],
            "revoke_by_actor_role": [
                {"role": r, "count": c}
                for r, c in sorted(revoke_by_role.items(), key=lambda kv: -kv[1])
            ],
            "revoke_by_reason_code": [
                {"reason_code": r, "count": c}
                for r, c in sorted(revoke_by_reason.items(), key=lambda kv: -kv[1])
            ],
        },
        "sharing": {
            "keys_with_members": keys_with_members,
            "total_members": total_members,
            "max_roster": max_roster,
        },
        "spend": {
            "month": month_str,
            "total_mtd_usd": float(round(total_mtd, 2)),
            "resolvable_keys": resolvable,
            "unresolvable_keys": unresolvable,
            "data_complete": unresolvable == 0,
            "unresolvable_by_bucket": [
                {"bucket": b, "count": c}
                for b, c in sorted(
                    unresolvable_by_bucket.items(), key=lambda kv: -kv[1]
                )
            ],
            "by_bucket": spend_by_bucket,
            "by_provider": spend_by_provider,
            "top_spenders": top_spenders,
            "budget_health": health,
            "last_fact_date": max(last_fact_dates) if last_fact_dates else None,
        },
    }


__all__ = ["compute_apikey_analytics"]