"""Events: `/events`, `/events/types`, `/events/summary`, `/events/{id}`.""" from __future__ import annotations from collections import defaultdict from typing import Any from fastapi import APIRouter, HTTPException, Query, Request, Response from companyatlas.api import queries as q from companyatlas.api import serializers as ser from companyatlas.api.common import PageDep, cached, cached_response, page_payload, public_cache_value from companyatlas.db import connection, fetch_all, fetch_one from companyatlas.taxonomy import EVENT_SUBTYPES, EventType ORDER = 20 router = APIRouter(prefix="/api/v1", tags=["events"]) @router.get("/events", summary="Event feed with filters") async def list_events(response: Response, p: PageDep, event_type: str | None = None, event_subtype: str | None = None, country: str | None = None, industry: str | None = None, since: str | None = None, until: str | None = None, min_importance: float | None = Query(None, ge=0, le=1), min_confidence: float | None = Query(None, ge=0, le=1), q_: str | None = Query(None, alias="q", max_length=200), surface: str | None = None, origin: str | None = None, company: str | None = None, status: str = Query("active", pattern="^(active|retracted|duplicate|review|all)$"), sort: str = Query("recent", pattern="^(recent|importance)$")) -> dict[str, Any]: response.headers["cache-control"] = public_cache_value(30) async with connection() as conn: company_id = None if company: company_id = (await q.require_company(conn, company))["id"] where, params = q.event_filters(company_id=company_id, event_type=event_type, event_subtype=event_subtype, country=country, industry=industry, since=q.parse_iso(since), until=q.parse_iso(until, "until"), min_importance=min_importance, min_confidence=min_confidence, q=q_, surface=surface, origin=origin, status=None if status == "all" else status) rows = await q.fetch_events(conn, where, params, sort=sort, limit=p.per_page, offset=p.offset) total = await q.count_events(conn, where, params) return page_payload([ser.event(r) for r in rows], total, p) @router.get("/events/types", summary="Event taxonomy with 30-day counts") async def event_types(request: Request) -> Any: async def produce() -> dict[str, Any]: async with connection() as conn: rows = await fetch_all(conn, "select event_type, event_subtype, count(*) as n from events where status = 'active' and detected_at >= :d " "group by event_type, event_subtype", d=q.days_ago(30)) counts: dict[str, dict[str, int]] = defaultdict(dict) for r in rows: counts[r["event_type"]][r["event_subtype"]] = int(r["n"]) subtypes_by_type: dict[str, list[str]] = defaultdict(list) for sub, (typ, _imp) in EVENT_SUBTYPES.items(): subtypes_by_type[typ.value].append(sub) types = [] for t in [x.value for x in EventType] + [k for k in counts if k not in {x.value for x in EventType}]: observed = counts.get(t, {}) subs = list(dict.fromkeys(subtypes_by_type.get(t, []) + list(observed))) items = sorted(({"event_subtype": s, "count_30d": observed.get(s, 0)} for s in subs), key=lambda x: (-x["count_30d"], x["event_subtype"])) types.append({"event_type": t, "subtypes": items, "count_30d": sum(observed.values())}) types.sort(key=lambda x: (-x["count_30d"], x["event_type"])) return {"types": types} return cached_response(request, await cached("events:types", 300, produce), 300) @router.get("/events/summary", summary="Event counts by type / industry / country with delta vs previous window") async def event_summary(request: Request, days: int = Query(7, ge=1, le=365), group: str = Query("type", pattern="^(type|industry|country)$")) -> Any: async def produce() -> dict[str, Any]: col = {"type": "e.event_type", "industry": "c.industry_primary", "country": "c.country"}[group] d1, d2 = q.days_ago(days), q.days_ago(days * 2) async with connection() as conn: rows = await fetch_all(conn, f"select {col} as key, count(*) filter (where e.detected_at >= :d1) as cur, " f"count(*) filter (where e.detected_at < :d1) as prev from events e join companies c on c.id = e.company_id " f"where e.status = 'active' and e.detected_at >= :d2 and {col} is not null group by 1 order by cur desc limit 200", d1=d1, d2=d2) items = [] for r in rows: cur, prev = int(r["cur"]), int(r["prev"]) items.append({"key": r["key"], "count": cur, "previous": prev, "delta_pct": round((cur - prev) / prev * 100, 1) if prev else None}) return {"days": days, "group": group, "items": items} return cached_response(request, await cached(f"events:summary:{days}:{group}", 300, produce), 300) @router.get("/events/{event_id}", summary="Event detail with sources, change and company") async def event_detail(event_id: str, response: Response) -> dict[str, Any]: response.headers["cache-control"] = public_cache_value(60) async with connection() as conn: row = await q.fetch_event(conn, event_id) if row is None: raise HTTPException(status_code=404, detail="event not found") out = ser.event(row) sources = await fetch_all(conn, "select * from event_sources where event_id = :id order by detected_at asc limit 50", id=event_id) out["sources"] = [ser.event_source(s) for s in sources] if not out["sources"] and row.get("source_url"): out["sources"] = [{"source_url": row["source_url"], "surface": row.get("surface"), "detected_at": row["detected_at"], "kind": "primary", "sensor_id": row.get("sensor_id"), "snapshot_id": row.get("snapshot_after")}] out["change"] = None if row.get("change_id"): ch = await fetch_one(conn, "select * from changes where id = :id", id=row["change_id"]) out["change"] = ser.change(ch) if ch else None cards = await q.fetch_cards_by_ids(conn, [row["company_id"]]) if cards: out["company"] = ser.company_card(cards[0]) if row.get("cluster_id"): cl = await fetch_one(conn, "select id, cluster_key, source_count, surfaces, confidence, first_detected_at, last_detected_at, canonical_event_id " "from event_clusters where id = :id", id=row["cluster_id"]) out["cluster"] = cl return out