"""/providers listing (with pricing aggregates) and /providers/{slug} alias. API 1.1 adds price distributions, 30-day listing churn,
organizations covered and the union of priced feature keys."""
from __future__ import annotations

from typing import Any

from fastapi import APIRouter, Request

from aiatlas.api.common import ENTITY_COLS, ENTITY_FROM, cached, entity_summary
from aiatlas.api.routers.entities import detail_for_type
from aiatlas.db import connection, fetch_all

router = APIRouter(prefix="/api/v1/providers", tags=["providers"])
FEATURE_KEYS = {"batch": ("batch_input_per_mtok", "batch_output_per_mtok", "batch_enabled", "batch"), "cached": ("cached_input_per_mtok", "cache_write_per_mtok", "cache_write_1h_per_mtok", "input_cache_write_1h", "prompt_caching"),
                "fine_tuning": ("fine_tuning", "fine_tuning_input_per_mtok", "fine_tuning_training_per_mtok"), "flex": ("flex_input_per_mtok", "flex_output_per_mtok"),
                "priority": ("priority", "priority_input_per_mtok", "priority_output_per_mtok"), "long_context": ("long_context_input_per_mtok", "long_context_output_per_mtok"),
                "audio": ("audio", "audio_input_per_mtok", "audio_output_per_mtok"), "image": ("image_input_per_mtok", "image_output", "image_output_per_mtok", "per_image"),
                "web_search": ("web_search", "search_grounding_per_1k_requests"), "free_tier": ("free_tier", "free"), "reasoning": ("internal_reasoning",), "serverless": ("serverless",)}


@router.get("")
@cached(300)
async def list_providers(request: Request) -> dict[str, Any]:
    async with connection() as conn:
        rows = await fetch_all(conn, f"""
            select {ENTITY_COLS},
                   (select count(distinct p.model_id) from prices p where p.provider_id = e.id and p.valid_to is null) as priced_models,
                   (select count(*) from relations r where r.object_id = e.id and r.predicate = 'available_through' and r.valid_to is null) as listed_models,
                   (select count(*) from prices p where p.provider_id = e.id and p.valid_to is null) as price_count,
                   (select min(p.input_per_mtok) from prices p where p.provider_id = e.id and p.valid_to is null and p.input_per_mtok > 0) as min_input_per_mtok,
                   (select min(p.output_per_mtok) from prices p where p.provider_id = e.id and p.valid_to is null and p.output_per_mtok > 0) as min_output_per_mtok,
                   (select jsonb_build_object('min', min(x.v), 'p25', percentile_cont(0.25) within group (order by x.v), 'median', percentile_cont(0.5) within group (order by x.v),
                                              'p75', percentile_cont(0.75) within group (order by x.v), 'max', max(x.v), 'n', count(*))
                    from (select p.input_per_mtok as v from prices p where p.provider_id = e.id and p.valid_to is null and p.input_per_mtok > 0) x) as input_dist,
                   (select jsonb_build_object('min', min(x.v), 'p25', percentile_cont(0.25) within group (order by x.v), 'median', percentile_cont(0.5) within group (order by x.v),
                                              'p75', percentile_cont(0.75) within group (order by x.v), 'max', max(x.v), 'n', count(*))
                    from (select p.output_per_mtok as v from prices p where p.provider_id = e.id and p.valid_to is null and p.output_per_mtok > 0) x) as output_dist,
                   (select count(distinct p.model_id) from prices p where p.provider_id = e.id and p.valid_from > now() - interval '30 days'
                        and not exists (select 1 from prices q where q.provider_id = e.id and q.model_id = p.model_id and q.valid_from <= now() - interval '30 days')) as models_added_30d,
                   (select count(distinct p.model_id) from prices p where p.provider_id = e.id and p.valid_to > now() - interval '30 days'
                        and not exists (select 1 from prices q where q.provider_id = e.id and q.model_id = p.model_id and q.valid_to is null)) as models_removed_30d,
                   (select count(*) from change_events ev join prices p on p.model_id = ev.entity_id where ev.category = 'price' and ev.is_backfill = false
                        and ev.occurred_at > now() - interval '30 days' and p.provider_id = e.id and p.valid_to is null) as price_changes_30d,
                   (select count(distinct m.organization_id) from prices p join entities m on m.id = p.model_id where p.provider_id = e.id and p.valid_to is null and m.organization_id is not null) as organizations_covered,
                   (select coalesce(jsonb_agg(distinct k.key), '[]'::jsonb) from prices p, jsonb_object_keys(p.features) k(key) where p.provider_id = e.id and p.valid_to is null) as feature_keys,
                   (select count(*) from prices p where p.provider_id = e.id and p.valid_to is null and p.cached_input_per_mtok is not null) as with_cached,
                   (select count(*) from prices p where p.provider_id = e.id and p.valid_to is null and p.batch_input_per_mtok is not null) as with_batch
            from {ENTITY_FROM} where e.entity_type = 'provider' and e.merged_into is null
            order by price_count desc, listed_models desc, e.canonical_name""")
    items = []
    for r in rows:
        keys = set(r["feature_keys"] or [])
        supported = sorted(f for f, ks in FEATURE_KEYS.items() if keys & set(ks))
        if int(r["with_cached"] or 0) and "cached" not in supported:
            supported.append("cached")
        if int(r["with_batch"] or 0) and "batch" not in supported:
            supported.append("batch")
        items.append({**(entity_summary(r) or {}), "model_count": int(max(r["priced_models"] or 0, r["listed_models"] or 0)), "price_count": int(r["price_count"] or 0),
                      "min_input_per_mtok": r["min_input_per_mtok"], "min_output_per_mtok": r["min_output_per_mtok"],
                      "input_price_distribution": _dist(r["input_dist"]), "output_price_distribution": _dist(r["output_dist"]),
                      "models_added_30d": int(r["models_added_30d"] or 0), "models_removed_30d": int(r["models_removed_30d"] or 0), "price_changes_30d": int(r["price_changes_30d"] or 0),
                      "organizations_covered": int(r["organizations_covered"] or 0), "features_supported": sorted(set(supported)), "feature_keys": sorted(keys)})
    return {"items": items, "note": "Distributions are over live offers with a positive price (USD per 1M tokens). models_added/removed_30d count model listings first opened / last closed "
                                    "in the window; price_changes_30d counts price events (not back-filled) on models the provider currently lists."}


def _dist(v: Any) -> dict[str, Any] | None:
    if not isinstance(v, dict) or not v.get("n"):
        return None
    return {k: v.get(k) for k in ("min", "p25", "median", "p75", "max", "n")}


@router.get("/{slug}")
@cached(300)
async def get_provider(request: Request, slug: str) -> dict[str, Any]:
    return await detail_for_type(slug, ("provider",))