"""Day-1 readiness analytics — onboarding-parity PR5 (2026-06-12).

Answers the operational question the cockpit couldn't: **"what share of
our hires had every Day-1 dimension in place by their start date"**,
month over month — the IT team's onboarding scorecard.

Definition of "Day-1 ready": every BLOCKER-severity dimension was in
place (``hard_ready_at`` set, migration 280) no later than the end of
the hire_date (Asia/Shanghai). ``hard_ready_at`` is stamped exactly
once by the state machine when the pending hard-blocker count first
hits zero — advisory dims (须知确认 etc., which a hire can't complete
pre-start) deliberately don't gate the metric.

Exclusions (both would lie about the rate):
* ``backfilled`` processes — the platform never observed their Day-1
  readiness end-to-end (migration-229 seed rows).
* ``cancelled`` processes — rescinded offers never had a Day-1.

Read-only; zero notification surfaces.
"""
from __future__ import annotations

from datetime import date, datetime
from typing import Any
from zoneinfo import ZoneInfo

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

from app.domains.onboarding.models import (
    PROC_CANCELLED,
    PROC_OPEN_STATUSES,
    RESO_PENDING,
    SEVERITY_BLOCKER,
    OnboardingBlocker,
    OnboardingProcess,
)

_APP_TZ = ZoneInfo("Asia/Shanghai")

MAX_MONTHS = 24


def _today() -> date:
    # Intentionally app-tz (single-viewpoint IT analytics aggregate): one
    # "today" for the whole board, not a per-employee state guard — so
    # per-person local-today does NOT apply here (contrast
    # service._local_today_for_person).
    return datetime.now(_APP_TZ).date()


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


def _months_back(today: date, months: int) -> date:
    """First day of the month ``months-1`` months before today's month."""
    year, month = today.year, today.month
    steps = months - 1
    month -= steps
    while month < 1:
        month += 12
        year -= 1
    return date(year, month, 1)


def _ready_on_day1(proc: OnboardingProcess) -> bool:
    """``hard_ready_at`` (BLOCKER dims clear) by end of hire_date.

    Review-r1: using ``ready_at`` made one required 入职须知 item drag
    the whole metric to ~0% (hires can't ack before they have an
    account). ``hard_ready_at`` is the SLA/escalation hard-gap口径
    (migration 280; backfilled from ready_at for pre-existing rows).
    """
    if proc.hard_ready_at is None:
        return False
    ready_local_date = (
        proc.hard_ready_at.astimezone(_APP_TZ).date()
        if proc.hard_ready_at.tzinfo is not None
        # MySQL DATETIME round-trips naive; rows are written as UTC.
        else proc.hard_ready_at.replace(tzinfo=ZoneInfo("UTC"))
        .astimezone(_APP_TZ)
        .date()
    )
    return ready_local_date <= proc.hire_date


class OnboardingAnalyticsService:
    def __init__(self, db: AsyncSession) -> None:
        self.db = db

    async def day1_readiness(
        self, *, months: int = 6, as_of_date: date | None = None,
    ) -> dict[str, Any]:
        """Per-month Day-1 readiness rate + current open-debt summary."""
        months = max(1, min(MAX_MONTHS, months))
        today = as_of_date or _today()
        window_start = _months_back(today, months)

        procs = list(
            (
                await self.db.execute(
                    select(OnboardingProcess).where(
                        OnboardingProcess.hire_date >= window_start,
                        OnboardingProcess.hire_date <= today,
                        OnboardingProcess.status != PROC_CANCELLED,
                        OnboardingProcess.backfilled.is_(False),
                    )
                )
            )
            .scalars()
            .all()
        )

        # Month buckets, oldest → newest, zero-filled so the chart
        # doesn't silently skip empty months.
        buckets: dict[str, dict[str, int]] = {}
        cursor = window_start
        while cursor <= today:
            buckets[_month_key(cursor)] = {"total": 0, "day1_ready": 0}
            # advance one month
            cursor = (
                date(cursor.year + 1, 1, 1)
                if cursor.month == 12
                else date(cursor.year, cursor.month + 1, 1)
            )

        for proc in procs:
            key = _month_key(proc.hire_date)
            bucket = buckets.get(key)
            if bucket is None:
                continue
            bucket["total"] += 1
            if _ready_on_day1(proc):
                bucket["day1_ready"] += 1

        month_rows = [
            {
                "month": key,
                "total": b["total"],
                "day1_ready": b["day1_ready"],
                "rate_pct": (
                    round(b["day1_ready"] / b["total"] * 100)
                    if b["total"]
                    else None
                ),
            }
            for key, b in buckets.items()
        ]
        window_total = sum(b["total"] for b in buckets.values())
        window_ready = sum(b["day1_ready"] for b in buckets.values())

        # Current debt: open processes past hire_date still carrying
        # pending BLOCKER-severity dims (the escalation-worthy subset).
        open_debt = 0
        open_procs = list(
            (
                await self.db.execute(
                    select(OnboardingProcess.id).where(
                        OnboardingProcess.status.in_(PROC_OPEN_STATUSES),
                        OnboardingProcess.hire_date <= today,
                        OnboardingProcess.backfilled.is_(False),
                    )
                )
            ).scalars()
        )
        if open_procs:
            debt_rows = (
                await self.db.execute(
                    select(OnboardingBlocker.process_id)
                    .where(
                        OnboardingBlocker.process_id.in_(open_procs),
                        OnboardingBlocker.resolution == RESO_PENDING,
                        OnboardingBlocker.severity == SEVERITY_BLOCKER,
                    )
                    .distinct()
                )
            ).scalars()
            open_debt = len(set(debt_rows))

        return {
            "window_start": window_start.isoformat(),
            "months": month_rows,
            "window_total": window_total,
            "window_day1_ready": window_ready,
            "window_rate_pct": (
                round(window_ready / window_total * 100)
                if window_total
                else None
            ),
            "open_past_due_with_hard_gaps": open_debt,
        }