1"""/diff?a=&b=&scope= — what changed between two dates: new / gone entities, property, price and benchmark changes."""2from __future__ import annotations34from datetime import UTC, datetime5from datetime import time as dtime6from typing import Any78from fastapi import APIRouter, Query, Request910from aiatlas.api.common import (11 ENTITY_COLS,12 ENTITY_FROM,13 EVENT_COLS,14 EVENT_FROM,15 ApiError,16 cached,17 change_event,18 entity_summary,19 parse_date,20)21from aiatlas.db import connection, fetch_all, fetch_one, fetch_val2223router = APIRouter(prefix="/api/v1/diff", tags=["diff"])24GONE_TYPES = ("RETIREMENT_ANNOUNCED", "DEPRECATION_ANNOUNCED", "STATUS_CHANGED", "ENTITY_MERGED", "MODEL_RETIRED", "MODEL_DEPRECATED")252627async def _scope(conn: Any, scope: str) -> tuple[str, str, dict[str, Any], dict[str, Any]]:28 """Returns (entity where-fragment on alias e, event where-fragment on alias e, params, scope description)."""29 s = (scope or "all").strip()30 if s == "all":31 return "true", "true", {}, {"kind": "all"}32 if s == "models":33 return "e.entity_type = 'model'", "e.entity_type = 'model'", {}, {"kind": "models"}34 if s.startswith("org:"):35 key = s[4:]36 org = await fetch_one(conn, "select id, slug, canonical_name from entities where slug = :k or id = :k limit 1", k=key)37 if not org:38 raise ApiError(404, f"organization {key!r} not found")39 frag = "(e.organization_id = :org or e.id = :org)"40 return frag, frag, {"org": org["id"]}, {"kind": "org", "organization": {"id": org["id"], "slug": org["slug"], "name": org["canonical_name"]}}41 if s.startswith("family:"):42 frag = "e.attributes->>'family' ilike :family"43 return frag, frag, {"family": s[7:]}, {"kind": "family", "family": s[7:]}44 raise ApiError(400, "scope must be all | models | org:<slug> | family:<name>")454647@router.get("")48@cached(300)49async def diff(request: Request, a: str = Query(..., description="YYYY-MM-DD"), b: str = Query(..., description="YYYY-MM-DD"), scope: str = "all",50 limit: int = Query(200, ge=1, le=1000)) -> dict[str, Any]:51 da, db = parse_date(a, "a"), parse_date(b, "b")52 assert da is not None and db is not None53 if da > db:54 da, db = db, da55 start, end = datetime.combine(da, dtime.max, UTC), datetime.combine(db, dtime.max, UTC)56 async with connection() as conn:57 ent_frag, ev_frag, params, scope_desc = await _scope(conn, scope)58 new_entities = await fetch_all(conn, f"select {ENTITY_COLS} from {ENTITY_FROM} where e.merged_into is null and {ent_frag} and e.first_seen_at > :s and e.first_seen_at <= :e "59 f"order by e.first_seen_at desc limit :lim", s=start, e=end, lim=limit, **params)60 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_id61 where {ev_frag} and ev.observed_at > :s and ev.observed_at <= :e and (ev.event_type = any(cast(:gt as text[])) or e.merged_into is not null)62 and (e.status in ('retired','deprecated','merged') or ev.event_type in ('RETIREMENT_ANNOUNCED','DEPRECATION_ANNOUNCED'))63 order by e.id limit :lim""", s=start, e=end, gt=list(GONE_TYPES), lim=limit, **params)6465 async def events(cond: str) -> list[dict[str, Any]]:66 rows = await fetch_all(conn, f"select {EVENT_COLS} from {EVENT_FROM} where {ev_frag} and ev.observed_at > :s and ev.observed_at <= :e and {cond} "67 f"order by ev.importance desc, ev.observed_at desc limit :lim", s=start, e=end, lim=limit, **params)68 return [change_event(r) for r in rows]6970 prop_changes = await events("ev.property is not null and ev.category not in ('price','benchmark') and ev.event_type <> 'DOCUMENT_CHANGED'")71 price_changes = await events("ev.category = 'price'")72 bench_changes = await events("ev.category = 'benchmark'")73 counts = await fetch_one(conn, f"""select (select count(*) from entities e where e.merged_into is null and {ent_frag} and e.first_seen_at > :s and e.first_seen_at <= :e) as new_entities,74 (select count(*) from change_events ev left join entities e on e.id = ev.entity_id where {ev_frag} and ev.observed_at > :s and ev.observed_at <= :e and ev.event_type <> 'DOCUMENT_CHANGED') as events,75 (select count(*) from change_events ev left join entities e on e.id = ev.entity_id where {ev_frag} and ev.observed_at > :s and ev.observed_at <= :e and ev.category = 'price') as price_changes,76 (select count(*) from change_events ev left join entities e on e.id = ev.entity_id where {ev_frag} and ev.observed_at > :s and ev.observed_at <= :e and ev.category = 'benchmark') as benchmark_changes,77 (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,78 (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,79 (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""",80 s=start, e=end, **params)81 entities_at_a = await fetch_val(conn, f"select count(*) from entities e where e.merged_into is null and {ent_frag} and e.first_seen_at <= :s", s=start, **params)82 entities_at_b = await fetch_val(conn, f"select count(*) from entities e where e.merged_into is null and {ent_frag} and e.first_seen_at <= :e", e=end, **params)83 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],84 "property_changes": prop_changes, "price_changes": price_changes, "benchmark_changes": bench_changes,85 "counts": {**{k: int(v or 0) for k, v in (counts or {}).items()}, "gone_entities": len(gone), "entities_at_a": int(entities_at_a or 0), "entities_at_b": int(entities_at_b or 0)}}86