spb/company-atlas

1"""Events: `/events`, `/events/types`, `/events/summary`, `/events/{id}`."""2from __future__ import annotations34from collections import defaultdict5from typing import Any67from fastapi import APIRouter, HTTPException, Query, Request, Response89from companyatlas.api import queries as q10from companyatlas.api import serializers as ser11from companyatlas.api.common import PageDep, cached, cached_response, page_payload, public_cache_value12from companyatlas.db import connection, fetch_all, fetch_one13from companyatlas.taxonomy import EVENT_SUBTYPES, EventType1415ORDER = 2016router = APIRouter(prefix="/api/v1", tags=["events"])171819@router.get("/events", summary="Event feed with filters")20async def list_events(response: Response, p: PageDep, event_type: str | None = None, event_subtype: str | None = None,21                      country: str | None = None, industry: str | None = None, since: str | None = None, until: str | None = None,22                      min_importance: float | None = Query(None, ge=0, le=1), min_confidence: float | None = Query(None, ge=0, le=1),23                      q_: str | None = Query(None, alias="q", max_length=200), surface: str | None = None, origin: str | None = None,24                      company: str | None = None, status: str = Query("active", pattern="^(active|retracted|duplicate|review|all)$"),25                      sort: str = Query("recent", pattern="^(recent|importance)$")) -> dict[str, Any]:26    response.headers["cache-control"] = public_cache_value(30)27    async with connection() as conn:28        company_id = None29        if company:30            company_id = (await q.require_company(conn, company))["id"]31        where, params = q.event_filters(company_id=company_id, event_type=event_type, event_subtype=event_subtype, country=country, industry=industry,32                                        since=q.parse_iso(since), until=q.parse_iso(until, "until"), min_importance=min_importance,33                                        min_confidence=min_confidence, q=q_, surface=surface, origin=origin, status=None if status == "all" else status)34        rows = await q.fetch_events(conn, where, params, sort=sort, limit=p.per_page, offset=p.offset)35        total = await q.count_events(conn, where, params)36    return page_payload([ser.event(r) for r in rows], total, p)373839@router.get("/events/types", summary="Event taxonomy with 30-day counts")40async def event_types(request: Request) -> Any:41    async def produce() -> dict[str, Any]:42        async with connection() as conn:43            rows = await fetch_all(conn, "select event_type, event_subtype, count(*) as n from events where status = 'active' and detected_at >= :d "44                                         "group by event_type, event_subtype", d=q.days_ago(30))45        counts: dict[str, dict[str, int]] = defaultdict(dict)46        for r in rows:47            counts[r["event_type"]][r["event_subtype"]] = int(r["n"])48        subtypes_by_type: dict[str, list[str]] = defaultdict(list)49        for sub, (typ, _imp) in EVENT_SUBTYPES.items():50            subtypes_by_type[typ.value].append(sub)51        types = []52        for t in [x.value for x in EventType] + [k for k in counts if k not in {x.value for x in EventType}]:53            observed = counts.get(t, {})54            subs = list(dict.fromkeys(subtypes_by_type.get(t, []) + list(observed)))55            items = sorted(({"event_subtype": s, "count_30d": observed.get(s, 0)} for s in subs), key=lambda x: (-x["count_30d"], x["event_subtype"]))56            types.append({"event_type": t, "subtypes": items, "count_30d": sum(observed.values())})57        types.sort(key=lambda x: (-x["count_30d"], x["event_type"]))58        return {"types": types}59    return cached_response(request, await cached("events:types", 300, produce), 300)606162@router.get("/events/summary", summary="Event counts by type / industry / country with delta vs previous window")63async def event_summary(request: Request, days: int = Query(7, ge=1, le=365), group: str = Query("type", pattern="^(type|industry|country)$")) -> Any:64    async def produce() -> dict[str, Any]:65        col = {"type": "e.event_type", "industry": "c.industry_primary", "country": "c.country"}[group]66        d1, d2 = q.days_ago(days), q.days_ago(days * 2)67        async with connection() as conn:68            rows = await fetch_all(conn, f"select {col} as key, count(*) filter (where e.detected_at >= :d1) as cur, "69                                         f"count(*) filter (where e.detected_at < :d1) as prev from events e join companies c on c.id = e.company_id "70                                         f"where e.status = 'active' and e.detected_at >= :d2 and {col} is not null group by 1 order by cur desc limit 200",71                                   d1=d1, d2=d2)72        items = []73        for r in rows:74            cur, prev = int(r["cur"]), int(r["prev"])75            items.append({"key": r["key"], "count": cur, "previous": prev, "delta_pct": round((cur - prev) / prev * 100, 1) if prev else None})76        return {"days": days, "group": group, "items": items}77    return cached_response(request, await cached(f"events:summary:{days}:{group}", 300, produce), 300)787980@router.get("/events/{event_id}", summary="Event detail with sources, change and company")81async def event_detail(event_id: str, response: Response) -> dict[str, Any]:82    response.headers["cache-control"] = public_cache_value(60)83    async with connection() as conn:84        row = await q.fetch_event(conn, event_id)85        if row is None:86            raise HTTPException(status_code=404, detail="event not found")87        out = ser.event(row)88        sources = await fetch_all(conn, "select * from event_sources where event_id = :id order by detected_at asc limit 50", id=event_id)89        out["sources"] = [ser.event_source(s) for s in sources]90        if not out["sources"] and row.get("source_url"):91            out["sources"] = [{"source_url": row["source_url"], "surface": row.get("surface"), "detected_at": row["detected_at"], "kind": "primary",92                               "sensor_id": row.get("sensor_id"), "snapshot_id": row.get("snapshot_after")}]93        out["change"] = None94        if row.get("change_id"):95            ch = await fetch_one(conn, "select * from changes where id = :id", id=row["change_id"])96            out["change"] = ser.change(ch) if ch else None97        cards = await q.fetch_cards_by_ids(conn, [row["company_id"]])98        if cards:99            out["company"] = ser.company_card(cards[0])100        if row.get("cluster_id"):101            cl = await fetch_one(conn, "select id, cluster_key, source_count, surfaces, confidence, first_detected_at, last_detected_at, canonical_event_id "102                                       "from event_clusters where id = :id", id=row["cluster_id"])103            out["cluster"] = cl104    return out105