1"""/changes (cursor feed) · /changes/daily ("What changed in AI today") · /changes/categories."""2from __future__ import annotations34from datetime import UTC, datetime5from typing import Any67from fastapi import APIRouter, Query, Request89from aiatlas.api.common import (10 ENTITY_COLS,11 ENTITY_FROM,12 EVENT_COLS,13 EVENT_FROM,14 ApiError,15 cached,16 change_event,17 csv,18 day_bounds,19 entity_summary,20 event_type_label,21 parse_date,22 parse_ts,23 resolve_id,24)25from aiatlas.db import connection, fetch_all, fetch_val2627router = APIRouter(prefix="/api/v1/changes", tags=["changes"])2829CATEGORY_LABELS = {"model": "Models", "price": "Pricing", "benchmark": "Benchmarks", "paper": "Research", "release": "Releases", "company": "Companies & labs",30 "provider": "Providers", "hardware": "Hardware", "framework": "Frameworks", "dataset": "Datasets", "regulation": "Regulation",31 "incident": "Incidents", "repository": "Repositories", "tool": "Tools", "source": "Sources & curation", "update": "Updates"}32CATEGORY_ORDER = list(CATEGORY_LABELS)33TOTAL_CAP = 10_000343536def _filters(*, category: str | None, types: list[str], entity_type: str | None, importance_min: int | None, since: datetime | None, until: datetime | None,37 q: str | None, include_documents: bool, entity_id: str | None = None, before: datetime | None = None) -> tuple[list[str], dict[str, Any]]:38 where: list[str] = []39 p: dict[str, Any] = {}40 if category:41 where.append("ev.category = any(cast(:cats as text[]))")42 p["cats"] = csv(category)43 if types:44 where.append("ev.event_type = any(cast(:types as text[]))")45 p["types"] = [t.upper() for t in types]46 elif not include_documents:47 where.append("ev.event_type <> 'DOCUMENT_CHANGED'")48 if entity_type:49 where.append("e.entity_type = any(cast(:etypes as text[]))")50 p["etypes"] = csv(entity_type)51 if importance_min is not None:52 where.append("ev.importance >= :imp")53 p["imp"] = importance_min54 if since is not None:55 where.append("ev.observed_at >= :since")56 p["since"] = since57 if until is not None:58 where.append("ev.observed_at <= :until")59 p["until"] = until60 if before is not None:61 where.append("ev.observed_at < :before")62 p["before"] = before63 if q:64 where.append("(ev.summary ilike :qlike or e.canonical_name ilike :qlike)")65 p["qlike"] = f"%{q}%"66 if entity_id:67 where.append("ev.entity_id = :eid")68 p["eid"] = entity_id69 return where or ["true"], p707172@router.get("")73@cached(60)74async def list_changes(request: Request, category: str | None = None, type: str | None = Query(None, alias="type"), entity_type: str | None = None,75 importance_min: int | None = Query(None, ge=0, le=3), since: str | None = None, until: str | None = None, q: str | None = Query(None, max_length=200),76 entity: str | None = None, limit: int = Query(50, ge=1, le=200), before: str | None = None, offset: int = Query(0, ge=0, le=10000),77 include_documents: int = Query(0, ge=0, le=1)) -> dict[str, Any]:78 before_ts, since_ts, until_ts = parse_ts(before, "before"), parse_ts(since, "since"), parse_ts(until, "until")79 async with connection() as conn:80 eid = await resolve_id(conn, entity) if entity else None81 where, params = _filters(category=category, types=csv(type), entity_type=entity_type, importance_min=importance_min, since=since_ts, until=until_ts, q=q,82 include_documents=bool(include_documents), entity_id=eid, before=before_ts)83 where_sql = " and ".join(where)84 rows = await fetch_all(conn, f"select {EVENT_COLS} from {EVENT_FROM} where {where_sql} order by ev.observed_at desc, ev.id desc limit :lim offset :off",85 lim=limit, off=offset, **params)86 total = await fetch_val(conn, f"select count(*) from (select 1 from {EVENT_FROM} where {where_sql} limit {TOTAL_CAP}) t", **params)87 items = [change_event(r) for r in rows]88 return {"items": items, "total": int(total or 0), "limit": limit, "offset": offset, "next_before": items[-1]["observed_at"] if len(items) == limit else None}899091@router.get("/daily")92@cached(120)93async def changes_daily(request: Request, date: str | None = None, per_section: int = Query(30, ge=1, le=100)) -> dict[str, Any]:94 d = parse_date(date, "date") or datetime.now(UTC).date()95 start, end = day_bounds(d)96 async with connection() as conn:97 rows = await fetch_all(conn, f"""select {EVENT_COLS} from (98 select ev.*, row_number() over (partition by ev.category order by ev.importance desc, ev.observed_at desc) as rn99 from change_events ev where ev.observed_at >= :s and ev.observed_at <= :e and ev.event_type <> 'DOCUMENT_CHANGED') ev100 left join entities e on e.id = ev.entity_id left join entities eo on eo.id = e.organization_id101 where ev.rn <= :n order by ev.category, ev.rn""", s=start, e=end, n=per_section)102 counts = await fetch_all(conn, "select category, count(*) as n from change_events where observed_at >= :s and observed_at <= :e and event_type <> 'DOCUMENT_CHANGED' group by 1", s=start, e=end)103 new_models = await fetch_all(conn, f"select {ENTITY_COLS} from {ENTITY_FROM} where e.entity_type = 'model' and e.merged_into is null and e.first_seen_at >= :s and e.first_seen_at <= :e "104 f"order by e.first_seen_at desc limit 100", s=start, e=end)105 prev = await fetch_val(conn, "select max(observed_at)::date from change_events where observed_at < :s and event_type <> 'DOCUMENT_CHANGED'", s=start)106 nxt = await fetch_val(conn, "select min(observed_at)::date from change_events where observed_at > :e and event_type <> 'DOCUMENT_CHANGED'", e=end)107 by_cat: dict[str, list[dict[str, Any]]] = {}108 for r in rows:109 by_cat.setdefault(r["category"], []).append(change_event(r))110 order = {c: i for i, c in enumerate(CATEGORY_ORDER)}111 sections = [{"category": c, "label": CATEGORY_LABELS.get(c, c.title()), "items": items} for c, items in sorted(by_cat.items(), key=lambda kv: (order.get(kv[0], 99), kv[0]))]112 return {"date": d.isoformat(), "counts": {r["category"]: int(r["n"]) for r in counts}, "total": sum(int(r["n"]) for r in counts), "sections": sections,113 "new_models": [entity_summary(r) for r in new_models], "labels": CATEGORY_LABELS,114 "previous_day": prev.isoformat() if prev else None, "next_day": nxt.isoformat() if nxt else None}115116117@router.get("/categories")118@cached(300)119async def changes_categories(request: Request, days: int = Query(7, ge=1, le=365)) -> dict[str, Any]:120 async with connection() as conn:121 rows = await fetch_all(conn, """select category, event_type, count(*) as count from change_events where observed_at > now() - make_interval(days => :d)122 and event_type <> 'DOCUMENT_CHANGED' group by 1, 2 order by 3 desc, 1, 2""", d=days)123 return {"days": days, "items": [{"category": r["category"], "label": CATEGORY_LABELS.get(r["category"], r["category"].title()), "event_type": r["event_type"],124 "event_label": event_type_label(r["event_type"]), "count": int(r["count"])} for r in rows]}125126127__all__ = ["CATEGORY_LABELS", "ApiError", "router"]128