"""/methodology · /trending · POST /views · /sitemap · /api-keys/me · /licenses.""" from __future__ import annotations import hashlib from datetime import UTC, datetime from typing import Any from fastapi import APIRouter, Depends, Query, Request from pydantic import BaseModel, Field from aiatlas.api.common import ( ENTITY_COLS, ENTITY_FROM, EVENT_TYPE_LABELS, STATUS_VOCAB, ApiError, cached, entity_summary, event_type_importance, event_type_label, rate_limit, ) from aiatlas.db import connection, execute, fetch_all, fetch_one, transaction from aiatlas.ids import ENTITY_TYPES from aiatlas.ontology import anomalies as anomaly_checks from aiatlas.ontology.benchmarks import CONDITION_KEYS, IGNORED_KEYS, TASK_KEYS, TRUST_LABELS, TRUST_LEVELS from aiatlas.ontology.licenses import CATEGORIES, LICENSES, normalize_license from aiatlas.ontology.openness import OPENNESS_CATEGORIES, OPENNESS_DEFINITIONS, OPENNESS_DIMENSIONS, OPENNESS_LABELS from aiatlas.services import hardware_fit as hf from aiatlas.services.finder import RULES as FINDER_RULES from aiatlas.services.frontier import FRONTIER_METHODOLOGY from aiatlas.services.quality import EXPECTED_FIELDS, QUALITY_VERSION from aiatlas.services.stats import DEFINITIONS as COUNTER_DEFINITIONS router = APIRouter(prefix="/api/v1", tags=["misc"]) CONFIDENCE_LEVELS = [ {"key": "verified", "label": "Verified", "description": "Confirmed by at least two independent sources, one of them tier 1."}, {"key": "high", "label": "High", "description": "Stated explicitly by an official (tier 1) source and extracted deterministically."}, {"key": "medium", "label": "Medium", "description": "Quality secondary source, curated registry, or LLM extraction from an official document."}, {"key": "low", "label": "Low", "description": "Community or unverified source, or LLM extraction from a secondary document."}, {"key": "conflicted", "label": "Conflicted", "description": "Another source states a different value; both claims are kept and flagged, never averaged."}, ] TIERS = [ {"tier": 1, "label": "Official / primary", "description": "The organization's own site, docs, pricing pages, model cards, filings."}, {"tier": 2, "label": "Quality secondary", "description": "Peer-reviewed venues, arXiv, curated registries, major leaderboards."}, {"tier": 3, "label": "Community", "description": "Community-maintained hubs, forums, wikis."}, {"tier": 4, "label": "Unverified", "description": "Anything else; never overrides a better tier."}, ] EXTRACTORS = [ {"key": "deterministic", "description": "Rule-based parsers (tables, JSON-LD, meta tags, embedded JSON, regex) — always runs first."}, {"key": "curated", "description": "Hand-maintained registries shipped with the code (organizations, providers, benchmarks, hardware)."}, {"key": "llm", "description": "Local LLM extraction validated against a strict JSON schema; medium/low confidence, never overrides deterministic tier-1 claims."}, ] COMPARABILITY_RULES = { "comparable": "Same benchmark variant, same canonical metric, same task-defining configuration (variant, evaluator/harness, shots, pass regime, scaffold…) and same conditions.", "partially-comparable": "Same task, but conditions differ (reasoning effort, thinking budget, temperature, judge, tool use, max tokens…).", "not-comparable": "Different benchmark variant, different metric, or a task-defining configuration key differs.", "task_keys": list(TASK_KEYS), "condition_keys": list(CONDITION_KEYS), "ignored_keys": sorted(IGNORED_KEYS), "group": "A comparability group is (benchmark, canonical metric, config_key) where config_key = sha1 of the task keys + metric (12 hex chars).", "leaderboard": "One row per canonical model: its best current row inside the chosen group; effort variants folded into a model share its rows.", } ANOMALY_CHECKS = [ {"check": "params_too_large", "severity": "critical", "description": f"parameter_count above {anomaly_checks.MAX_PARAMS:.0e}"}, {"check": "params_too_small", "severity": "warning", "description": "parameter_count below 100K"}, {"check": "active_gt_total", "severity": "critical", "description": "active parameters exceed total parameters"}, {"check": "context_too_large", "severity": "critical", "description": "context_length above 100M tokens"}, {"check": "context_too_small", "severity": "warning", "description": "context_length below 256 tokens"}, {"check": "max_output_gt_context", "severity": "warning", "description": "max_output_tokens exceeds context_length"}, {"check": "release_in_future", "severity": "critical", "description": "release_date after today"}, {"check": "release_too_old", "severity": "warning", "description": "release_date before 2010"}, {"check": "deprecated_before_release", "severity": "critical", "description": "deprecation_date before release_date"}, {"check": "retired_before_release", "severity": "critical", "description": "retirement_date before release_date"}, {"check": "retired_before_deprecated", "severity": "warning", "description": "retirement_date before deprecation_date"}, {"check": "cutoff_after_release", "severity": "warning", "description": "knowledge_cutoff after release_date"}, {"check": "deprecated_future_release", "severity": "critical", "description": "deprecated/retired status with a future release date"}, {"check": "open_without_weights_url", "severity": "info", "description": "labelled open but no weights location recorded"}, {"check": "negative_price", "severity": "critical", "description": "a price field is negative"}, {"check": "price_too_high", "severity": "warning", "description": f"a price exceeds ${anomaly_checks.MAX_PRICE_PER_MTOK:g} per 1M tokens"}, {"check": "zero_output_price", "severity": "warning", "description": "output price 0 while input is positive and the offer is not free"}, {"check": "input_gt_output_price", "severity": "info", "description": "input price more than 4× the output price"}, {"check": "cached_gt_input_price", "severity": "warning", "description": "cached input price above the input price"}, {"check": "price_jump_100x", "severity": "critical", "description": "a price moved by more than 100× between two observations"}, {"check": "score_above_max", "severity": "critical", "description": "benchmark score above the metric maximum"}, {"check": "score_below_min", "severity": "critical", "description": "benchmark score below the metric minimum"}, {"check": "evaluated_before_release", "severity": "warning", "description": "result evaluated more than 45 days before the model's release"}, {"check": "memory_implausible", "severity": "critical", "description": "hardware memory_gb ≤ 0 or above 100 000"}, {"check": "bandwidth_implausible", "severity": "warning", "description": "memory bandwidth ≤ 0 or above 100 000 GB/s"}, {"check": "tdp_implausible", "severity": "warning", "description": "TDP ≤ 0 or above 200 000 W"}, ] @router.get("/methodology") @cached(600) async def methodology(request: Request) -> dict[str, Any]: async with connection() as conn: metrics = await fetch_all(conn, "select key, label, version, description, formula from metric_definitions order by key") event_types = await fetch_all(conn, "select event_type, category, count(*) as count, max(observed_at) as last_seen_at from change_events group by 1, 2 order by 3 desc") seen = {r["event_type"] for r in event_types} types = [{**r, "count": int(r["count"]), "label": event_type_label(r["event_type"]), "importance": event_type_importance(r["event_type"])} for r in event_types] types += [{"event_type": t, "category": None, "count": 0, "last_seen_at": None, "label": lbl, "importance": imp} for t, (lbl, imp) in EVENT_TYPE_LABELS.items() if t not in seen] return {"metrics": metrics, "quality_version": QUALITY_VERSION, "expected_fields": EXPECTED_FIELDS, "confidence_levels": CONFIDENCE_LEVELS, "tiers": TIERS, "event_types": types, "status_vocabulary": list(STATUS_VOCAB), "extractors": EXTRACTORS, "openness": {"categories": list(OPENNESS_CATEGORIES), "labels": OPENNESS_LABELS, "definitions": OPENNESS_DEFINITIONS, "dimensions": list(OPENNESS_DIMENSIONS), "note": "Categories are derived from measurable dimensions and the licence ontology; a custom community licence is never 'open-source'."}, "trust_levels": [{"key": k, "label": TRUST_LABELS[k]} for k in TRUST_LEVELS], "comparability": COMPARABILITY_RULES, "counters": COUNTER_DEFINITIONS, "anomaly_checks": ANOMALY_CHECKS, "event_semantics": {"occurred_at": "coalesce(effective_at, observed_at) — when the change happened (effective date when a source states it)", "observed_at": "when AI Atlas first saw the change", "recorded_at": "when the row was written", "is_backfill": "true for history imported when a source is first crawled (never shown as 'today' in feeds)", "group_key": "one release / announcement seen through several documents shares a group_key"}, "hardware_fit": {"assumptions": hf.ASSUMPTIONS, "bytes_per_param": hf.BYTES_PER_PARAM, "reserved_gb": hf.RESERVED_GB}, "frontier": FRONTIER_METHODOLOGY, "find_a_model": FINDER_RULES, "licence_categories": list(CATEGORIES), "principles": ["Never fabricate: missing data is reported as unavailable.", "Every fact carries provenance (source, snapshot, URL, tier, confidence, extractor).", "History is append-only: claims, prices and benchmark results are never overwritten.", "Conflicts between sources are stored side by side and flagged for review.", "Live counters and feeds are computed from the database.", "No composite 'best model' score: rankings are per benchmark comparability group; finders return the observed dimensions."]} @router.get("/trending") @cached(300) async def trending(request: Request, days: int = Query(7, ge=1, le=90), limit: int = Query(12, ge=1, le=60), type: str | None = Query(None, alias="type"), kind: str = Query("views", pattern="^(views|most_changed|new_listings|new_results)$")) -> dict[str, Any]: """`kind=views` (page views, v1) · `most_changed` (events per entity) · `new_listings` (PROVIDER_LISTED) · `new_results` (benchmark result events) — separate lists, never merged.""" async with connection() as conn: if kind == "views": rows = await fetch_all(conn, f""" with v as (select split_part(regexp_replace(path, '[?#].*$', ''), '/', 3) as slug, sum(views) as views from page_views where day >= ((now() at time zone 'UTC') - make_interval(days => :d))::date and path ~ '^/[a-z-]+/[^/?#]+' group by 1) select v.views as n, {ENTITY_COLS} from v join entities e on e.slug = v.slug left join entities eo on eo.id = e.organization_id where e.merged_into is null {"and e.entity_type = :t" if type else ""} order by v.views desc, e.updated_at desc limit :lim""", d=days, lim=limit, t=type) items = [{**(entity_summary(r) or {}), "views": int(r["n"] or 0)} for r in rows] definition = "Page views recorded by the site beacon in the window." else: cond = {"most_changed": "ev.event_type <> 'DOCUMENT_CHANGED'", "new_listings": "ev.event_type in ('PROVIDER_LISTED','PROVIDER_DELISTED')", "new_results": "ev.event_type in ('BENCHMARK_RESULT','BENCHMARK_UPDATED')"}[kind] rows = await fetch_all(conn, f""" with v as (select ev.entity_id, count(*) as n, max(ev.occurred_at) as last_at from change_events ev where ev.is_backfill = false and ev.occurred_at > now() - make_interval(days => :d) and {cond} group by 1) select v.n, v.last_at, {ENTITY_COLS} from v join entities e on e.id = v.entity_id left join entities eo on eo.id = e.organization_id where e.merged_into is null {"and e.entity_type = :t" if type else ""} order by v.n desc, v.last_at desc limit :lim""", d=days, lim=limit, t=type) items = [{**(entity_summary(r) or {}), "events": int(r["n"] or 0), "last_event_at": r["last_at"]} for r in rows] definition = {"most_changed": "Entities with the most non-backfill events (any type except source-document changes) that occurred in the window.", "new_listings": "Entities with the most provider listing / delisting events in the window.", "new_results": "Entities with the most new or updated benchmark result events in the window."}[kind] return {"days": days, "kind": kind, "items": items, "definition": definition} @router.get("/licenses") @cached(600) async def licenses(request: Request) -> dict[str, Any]: """Licence ontology + how many canonical models use each key (canonical `license_key` or a raw label the ontology maps to it).""" async with connection() as conn: rows = await fetch_all(conn, """select coalesce(e.attributes->>'license_key', e.attributes->>'license') as raw, count(*) as n from entities e where e.entity_type = 'model' and e.merged_into is null and (e.attributes ? 'license' or e.attributes ? 'license_key') group by 1""") counts: dict[str, int] = {} unclassified: dict[str, int] = {} for r in rows: key = r["raw"] if r["raw"] in LICENSES else normalize_license(r["raw"]) if key: counts[key] = counts.get(key, 0) + int(r["n"]) elif r["raw"]: unclassified[r["raw"]] = unclassified.get(r["raw"], 0) + int(r["n"]) items = [{**info.as_dict(), "aliases": list(info.aliases), "models": counts.get(key, 0)} for key, info in LICENSES.items()] items.sort(key=lambda x: (-x["models"], x["label"])) return {"items": items, "total": len(items), "categories": list(CATEGORIES), "unclassified": [{"raw": k, "models": v} for k, v in sorted(unclassified.items(), key=lambda kv: -kv[1])], "note": "Permissions are read from the licence text (null = the text is ambiguous). Counts cover canonical models only."} @router.get("/licenses/{key}") @cached(600) async def license_detail(request: Request, key: str, limit: int = Query(50, ge=1, le=200)) -> dict[str, Any]: canon = key if key in LICENSES else normalize_license(key) info = LICENSES.get(canon) if canon else None if not info: raise ApiError(404, f"unknown licence {key!r}") raw = sorted({info.key.lower(), *(a.lower() for a in info.aliases), *([info.spdx.lower()] if info.spdx else [])}) async with connection() as conn: rows = await fetch_all(conn, f"""select {ENTITY_COLS} from {ENTITY_FROM} where e.entity_type = 'model' and e.merged_into is null and (e.attributes->>'license_key' = :k or lower(e.attributes->>'license') = any(cast(:raw as text[]))) order by e.attributes->>'release_date' desc nulls last, e.canonical_name limit :lim""", k=info.key, raw=raw, lim=limit) total = await fetch_one(conn, """select count(*) as n from entities e where e.entity_type = 'model' and e.merged_into is null and (e.attributes->>'license_key' = :k or lower(e.attributes->>'license') = any(cast(:raw as text[])))""", k=info.key, raw=raw) return {**info.as_dict(), "aliases": list(info.aliases), "models": {"items": [entity_summary(r) for r in rows], "total": int((total or {}).get("n") or 0), "limit": limit, "offset": 0}} class ViewBeacon(BaseModel): path: str = Field(..., min_length=1, max_length=300) @router.post("/views", dependencies=[Depends(rate_limit("views"))]) async def record_view(body: ViewBeacon) -> dict[str, Any]: path = body.path.strip() if not path.startswith("/") or "\n" in path or "//" in path: raise ApiError(400, "path must be a site-relative path") path = path.split("?", 1)[0].split("#", 1)[0][:300] async with transaction() as conn: await execute(conn, "insert into page_views (path, day, views) values (:p, :d, 1) on conflict (path, day) do update set views = page_views.views + 1", p=path, d=datetime.now(UTC).date()) return {"ok": True} @router.get("/sitemap") @cached(600) async def sitemap(request: Request, type: str | None = Query(None, alias="type"), limit: int = Query(5000, ge=1, le=5000), offset: int = Query(0, ge=0)) -> dict[str, Any]: if type and type not in ENTITY_TYPES: raise ApiError(400, f"unknown entity type {type!r}") where = "e.merged_into is null" + (" and e.entity_type = :t" if type else "") async with connection() as conn: rows = await fetch_all(conn, f"select e.slug, e.entity_type, e.updated_at from entities e where {where} order by e.updated_at desc, e.id limit :lim offset :off", t=type, lim=limit, off=offset) total = await fetch_one(conn, f"select count(*) as n from entities e where {where}", t=type) return {"items": rows, "total": int(total["n"]) if total else 0, "limit": limit, "offset": offset} @router.get("/api-keys/me") async def api_key_me(request: Request) -> dict[str, Any]: key = request.headers.get("x-api-key") or "" if not key: raise ApiError(401, "x-api-key header required") digest = hashlib.sha256(key.encode()).hexdigest() async with transaction() as conn: row = await fetch_one(conn, "update api_keys set last_used_at = now(), usage_count = usage_count + 1 where key_hash = :h and enabled returning label, plan, rate_per_min, usage_count, created_at, last_used_at", h=digest) if not row: raise ApiError(401, "unknown or disabled API key") return row __all__ = ["ANOMALY_CHECKS", "COMPARABILITY_RULES", "router"]