"""/diff?a=&b=&scope= — what changed between two dates: new / gone entities, property, price and benchmark changes, new benchmark leaders,
provider listings, hardware, context changes, retired models. API 1.1: events use `occurred_at` and `is_backfill = false`; `new_entities`
excludes artifacts unless `include=artifacts`."""
from __future__ import annotations

from datetime import UTC, datetime
from datetime import time as dtime
from typing import Any

from fastapi import APIRouter, Query, Request

from aiatlas.api.common import (
    ENTITY_COLS,
    ENTITY_FROM,
    EVENT_COLS,
    EVENT_FROM,
    ApiError,
    cached,
    change_event,
    entity_summary,
    parse_date,
)
from aiatlas.db import connection, fetch_all, fetch_one, fetch_val
from aiatlas.services.frontier import benchmark_meta, leader_at, load_results

router = APIRouter(prefix="/api/v1/diff", tags=["diff"])
GONE_TYPES = ("RETIREMENT_ANNOUNCED", "DEPRECATION_ANNOUNCED", "STATUS_CHANGED", "ENTITY_MERGED", "MODEL_RETIRED", "MODEL_DEPRECATED")
RETIRED_PRED = "(ev.event_type in ('RETIREMENT_ANNOUNCED','DEPRECATION_ANNOUNCED','MODEL_RETIRED','MODEL_DEPRECATED') or (ev.event_type = 'STATUS_CHANGED' and ev.new_value::text ~* 'retired|deprecated'))"
_leaders_at = leader_at


async def _scope(conn: Any, scope: str) -> tuple[str, str, dict[str, Any], dict[str, Any]]:
    """Returns (entity where-fragment on alias e, event where-fragment on alias e, params, scope description)."""
    s = (scope or "all").strip()
    if s == "all":
        return "true", "true", {}, {"kind": "all"}
    if s == "models":
        return "e.entity_type = 'model'", "e.entity_type = 'model'", {}, {"kind": "models"}
    if s.startswith("org:"):
        key = s[4:]
        org = await fetch_one(conn, "select id, slug, canonical_name from entities where slug = :k or id = :k limit 1", k=key)
        if not org:
            raise ApiError(404, f"organization {key!r} not found")
        frag = "(e.organization_id = :org or e.id = :org)"
        return frag, frag, {"org": org["id"]}, {"kind": "org", "organization": {"id": org["id"], "slug": org["slug"], "name": org["canonical_name"]}}
    if s.startswith("family:"):
        frag = "(e.attributes->>'family' ilike :family or exists (select 1 from entities f where f.id = e.family_id and (f.slug = :family or f.canonical_name ilike :family)))"
        return frag, frag, {"family": s[7:]}, {"kind": "family", "family": s[7:]}
    raise ApiError(400, "scope must be all | models | org:<slug> | family:<name>")


@router.get("")
@cached(300)
async def diff(request: Request, a: str = Query(..., description="YYYY-MM-DD"), b: str = Query(..., description="YYYY-MM-DD"), scope: str = "all",
               limit: int = Query(200, ge=1, le=1000), include: str | None = None, include_backfill: int = Query(0, ge=0, le=1)) -> dict[str, Any]:
    da, db = parse_date(a, "a"), parse_date(b, "b")
    assert da is not None and db is not None
    if da > db:
        da, db = db, da
    start, end = datetime.combine(da, dtime.max, UTC), datetime.combine(db, dtime.max, UTC)
    include_artifacts = "artifacts" in {x.strip() for x in (include or "").split(",")}
    bf = "" if include_backfill else " and ev.is_backfill = false"
    win = f"ev.occurred_at > :s and ev.occurred_at <= :e{bf}"
    async with connection() as conn:
        ent_frag, ev_frag, params, scope_desc = await _scope(conn, scope)
        art = "" if include_artifacts else " and e.entity_type <> 'artifact'"
        new_entities = await fetch_all(conn, f"select {ENTITY_COLS} from {ENTITY_FROM} where e.merged_into is null and {ent_frag}{art} and e.first_seen_at > :s and e.first_seen_at <= :e "
                                             f"order by e.first_seen_at desc limit :lim", s=start, e=end, lim=limit, **params)
        gone = await fetch_all(conn, f"""select distinct on (e.id) {ENTITY_COLS} from change_events ev join entities e on e.id = ev.entity_id left join entities eo on eo.id = e.organization_id
                                         where {ev_frag} and {win} and (ev.event_type = any(cast(:gt as text[])) or e.merged_into is not null)
                                         and (e.status in ('retired','deprecated','merged') or ev.event_type in ('RETIREMENT_ANNOUNCED','DEPRECATION_ANNOUNCED'))
                                         order by e.id limit :lim""", s=start, e=end, gt=list(GONE_TYPES), lim=limit, **params)

        async def events(cond: str) -> list[dict[str, Any]]:
            rows = await fetch_all(conn, f"select {EVENT_COLS}, ev.occurred_at, ev.is_backfill from {EVENT_FROM} where {ev_frag} and {win} and {cond} "
                                         f"order by ev.importance desc, ev.occurred_at desc limit :lim", s=start, e=end, lim=limit, **params)
            out = []
            for r in rows:
                ev = change_event(r)
                ev["occurred_at"] = r.get("occurred_at")
                out.append(ev)
            return out

        prop_changes = await events("ev.property is not null and ev.category not in ('price','benchmark') and ev.event_type <> 'DOCUMENT_CHANGED'")
        price_changes = await events("ev.category = 'price'")
        bench_changes = await events("ev.category = 'benchmark'")
        provider_changes = await events("ev.event_type in ('PROVIDER_LISTED','PROVIDER_DELISTED','NEW_PROVIDER')")
        hardware_changes = await events("(ev.category = 'hardware' or ev.event_type = 'NEW_HARDWARE')")
        context_changes = await events("ev.event_type in ('CONTEXT_CHANGED','MAX_OUTPUT_CHANGED')")
        retired = await events(RETIRED_PRED)
        counts = await fetch_one(conn, f"""select (select count(*) from entities e where e.merged_into is null and {ent_frag}{art} and e.first_seen_at > :s and e.first_seen_at <= :e) as new_entities,
                                           (select count(*) from change_events ev left join entities e on e.id = ev.entity_id where {ev_frag} and {win} and ev.event_type <> 'DOCUMENT_CHANGED') as events,
                                           (select count(*) from change_events ev left join entities e on e.id = ev.entity_id where {ev_frag} and {win} and ev.category = 'price') as price_changes,
                                           (select count(*) from change_events ev left join entities e on e.id = ev.entity_id where {ev_frag} and {win} and ev.category = 'benchmark') as benchmark_changes,
                                           (select count(*) from change_events ev left join entities e on e.id = ev.entity_id where {ev_frag} and {win} and ev.event_type in ('PROVIDER_LISTED','PROVIDER_DELISTED','NEW_PROVIDER')) as provider_changes,
                                           (select count(*) from change_events ev left join entities e on e.id = ev.entity_id where {ev_frag} and {win} and ev.event_type in ('CONTEXT_CHANGED','MAX_OUTPUT_CHANGED')) as context_changes,
                                           (select count(*) from change_events ev left join entities e on e.id = ev.entity_id where {ev_frag} and {win} and {RETIRED_PRED}) as retired_models,
                                           (select count(*) from prices p join entities e on e.id = p.model_id where {ent_frag} and p.valid_from > :s and p.valid_from <= :e) as price_rows_opened,
                                           (select count(*) from prices p join entities e on e.id = p.model_id where {ent_frag} and p.valid_to > :s and p.valid_to <= :e) as price_rows_closed,
                                           (select count(*) from claims c join entities e on e.id = c.entity_id where {ent_frag} and c.valid_to > :s and c.valid_to <= :e) as claims_superseded""",
                                 s=start, e=end, **params)
        entities_at_a = await fetch_val(conn, f"select count(*) from entities e where e.merged_into is null and {ent_frag}{art} and e.first_seen_at <= :s", s=start, **params)
        entities_at_b = await fetch_val(conn, f"select count(*) from entities e where e.merged_into is null and {ent_frag}{art} and e.first_seen_at <= :e", e=end, **params)
        # new benchmark leaders: leader of each benchmark's primary group at a vs at b (from results observed by each date)
        meta = await benchmark_meta(conn)
        all_rows = await load_results(conn, current_only=False)
    by_bench: dict[str, list[dict[str, Any]]] = {}
    for r in all_rows:
        by_bench.setdefault(r["benchmark_id"], []).append(r)
    new_leaders = []
    for bid, rows in by_bench.items():
        m = meta.get(bid, {})
        la, lb = _leaders_at(rows, start, m.get("attributes")), _leaders_at(rows, end, m.get("attributes"))
        if lb and (la is None or la["model"]["id"] != lb["model"]["id"]):
            new_leaders.append({"benchmark": {"id": bid, "slug": m.get("slug"), "name": m.get("name"), "category": m.get("category")}, "at_a": la, "at_b": lb})
    new_leaders.sort(key=lambda x: x["benchmark"]["name"] or "")
    return {"a": da.isoformat(), "b": db.isoformat(), "scope": scope_desc, "new_entities": [entity_summary(r) for r in new_entities], "gone_entities": [entity_summary(r) for r in gone],
            "property_changes": prop_changes, "price_changes": price_changes, "benchmark_changes": bench_changes, "new_benchmark_leaders": new_leaders,
            "provider_changes": provider_changes, "hardware_changes": hardware_changes, "context_changes": context_changes, "retired_models": retired,
            "counts": {**{k: int(v or 0) for k, v in (counts or {}).items()}, "gone_entities": len(gone), "new_benchmark_leaders": len(new_leaders),
                       "entities_at_a": int(entities_at_a or 0), "entities_at_b": int(entities_at_b or 0)},
            "include_artifacts": include_artifacts, "include_backfill": bool(include_backfill),
            "note": "Events are keyed on occurred_at (effective date when known) and exclude back-filled history unless include_backfill=1; new_entities uses first_seen_at. "
                    "new_benchmark_leaders compares the primary-group leader computed from results observed by each date."}