spb/company-atlas

1"""Provenance: sensors, snapshots (text/blocks from the object store), on-demand diffs and changes."""2from __future__ import annotations34import json5import logging6from typing import Any78from fastapi import APIRouter, HTTPException, Query, Response910from companyatlas import archive11from companyatlas.api import queries as q12from companyatlas.api import serializers as ser13from companyatlas.api.common import public_cache_value14from companyatlas.db import connection, fetch_all, fetch_one, fetch_val1516log = logging.getLogger("companyatlas.api.provenance")17ORDER = 2018router = APIRouter(prefix="/api/v1", tags=["provenance"])19MAX_TEXT_BYTES = 200 * 102420MAX_BLOCKS = 2000212223async def _sensor_or_404(conn: Any, sensor_id: str) -> dict[str, Any]:24    row = await fetch_one(conn, "select s.*, c.slug as company_slug, c.display_name as company_display_name, c.canonical_domain as company_domain, "25                                "c.country as company_country, c.logo_url as company_logo_url from sensors s join companies c on c.id = s.company_id "26                                "where s.id = :id", id=sensor_id)27    if row is None:28        raise HTTPException(status_code=404, detail="sensor not found")29    return row303132@router.get("/sensors/{sensor_id}", summary="Sensor with company and latest snapshot")33async def sensor_detail(sensor_id: str, response: Response) -> dict[str, Any]:34    response.headers["cache-control"] = public_cache_value(60)35    async with connection() as conn:36        row = await _sensor_or_404(conn, sensor_id)37        out = ser.sensor(row)38        out["company"] = ser.company_ref(row)39        latest = None40        if row.get("last_snapshot_id"):41            latest = await fetch_one(conn, "select * from snapshots where id = :id", id=row["last_snapshot_id"])42        if latest is None:43            latest = await fetch_one(conn, "select * from snapshots where sensor_id = :id order by fetched_at desc limit 1", id=sensor_id)44        out["latest_snapshot"] = ser.snapshot(latest) if latest else None45        out["last_meaningful_change_at"] = row.get("last_meaningful_change_at")46    return out474849@router.get("/sensors/{sensor_id}/snapshots", summary="Snapshot versions of a sensor")50async def sensor_snapshots(sensor_id: str, response: Response, limit: int = Query(50, ge=1, le=500), before: str | None = None) -> dict[str, Any]:51    response.headers["cache-control"] = public_cache_value(60)52    before_dt = q.parse_iso(before, "before")53    async with connection() as conn:54        await _sensor_or_404(conn, sensor_id)55        extra = " and fetched_at < :before" if before_dt else ""56        params: dict[str, Any] = {"id": sensor_id, "limit": limit}57        if before_dt:58            params["before"] = before_dt59        rows = await fetch_all(conn, f"select * from snapshots where sensor_id = :id{extra} order by fetched_at desc limit :limit", **params)60    return {"items": [ser.snapshot(r) for r in rows]}616263@router.get("/sensors/{sensor_id}/changes", summary="Changes detected by a sensor")64async def sensor_changes(sensor_id: str, response: Response, limit: int = Query(50, ge=1, le=500),65                         min_significance: float | None = Query(None, ge=0, le=1)) -> dict[str, Any]:66    response.headers["cache-control"] = public_cache_value(60)67    async with connection() as conn:68        await _sensor_or_404(conn, sensor_id)69        extra = " and significance >= :ms" if min_significance is not None else ""70        params: dict[str, Any] = {"id": sensor_id, "limit": limit}71        if min_significance is not None:72            params["ms"] = min_significance73        rows = await fetch_all(conn, f"select * from changes where sensor_id = :id{extra} order by detected_at desc limit :limit", **params)74    return {"items": [ser.change(r) for r in rows]}757677def _load_text(key: str | None, limit: int = MAX_TEXT_BYTES) -> tuple[str | None, bool]:78    if not key:79        return None, False80    try:81        data = archive.get_bytes(key)82    except (OSError, ValueError):83        return None, False84    truncated = len(data) > limit85    return data[:limit].decode("utf-8", errors="replace"), truncated868788def _load_blocks(key: str | None) -> list[dict[str, Any]] | None:89    if not key:90        return None91    try:92        parsed = json.loads(archive.get_text(key))93    except (OSError, ValueError):94        return None95    if isinstance(parsed, dict) and isinstance(parsed.get("blocks"), list):96        parsed = parsed["blocks"]97    return [b for b in parsed if isinstance(b, dict)][:MAX_BLOCKS] if isinstance(parsed, list) else None9899100@router.get("/snapshots/{snapshot_id}", summary="Snapshot with normalized text, blocks and extracted fields")101async def snapshot_detail(snapshot_id: str, response: Response, include: str = Query("text,blocks,extracted")) -> dict[str, Any]:102    response.headers["cache-control"] = public_cache_value(300)103    parts = set(q.csv_list(include))104    async with connection() as conn:105        row = await fetch_one(conn, "select * from snapshots where id = :id", id=snapshot_id)106        if row is None:107            raise HTTPException(status_code=404, detail="snapshot not found")108        sensor = await fetch_one(conn, "select id, url, surface, company_id from sensors where id = :id", id=row["sensor_id"])109    out = ser.snapshot(row)110    out["sensor"] = sensor111    out["object_keys"] = {"raw": row.get("object_key"), "text": row.get("text_key"), "blocks": row.get("blocks_key")}112    if "text" in parts:113        text, truncated = _load_text(row.get("text_key"))114        out["text"], out["text_truncated"] = text, truncated115    if "blocks" in parts:116        out["blocks"] = _load_blocks(row.get("blocks_key")) or []117    if "extracted" in parts:118        out["extracted"] = ser._dict(row.get("extracted"))119    return out120121122def _compute_diff(before_blocks: list[dict[str, Any]], after_blocks: list[dict[str, Any]], *, surface: str, before_text: str = "",123                  after_text: str = "") -> dict[str, Any] | None:124    try:125        from companyatlas.sdk.diff import compare  # crawl agent's module — optional at run time126    except ImportError:127        return None128    from companyatlas.sdk.models import Block129130    def to_blocks(items: list[dict[str, Any]]) -> list[Block]:131        out = []132        for b in items:133            try:134                out.append(Block(**{k: v for k, v in b.items() if k in Block.__dataclass_fields__}))135            except TypeError:136                continue137        return out138139    try:140        result = compare(to_blocks(before_blocks), to_blocks(after_blocks), surface=surface, before_text=before_text, after_text=after_text)141    except TypeError:   # older/newer signature: positional blocks only142        result = compare(to_blocks(before_blocks), to_blocks(after_blocks))143    if hasattr(result, "to_json"):144        payload = result.to_json()145        payload["significance"] = getattr(result, "significance", None)146        return payload147    return dict(result) if isinstance(result, dict) else None148149150@router.get("/snapshots/{snapshot_id}/diff/{other_id}", summary="Diff between two snapshots (stored change when available, else computed)")151async def snapshot_diff(snapshot_id: str, other_id: str, response: Response) -> dict[str, Any]:152    response.headers["cache-control"] = public_cache_value(300)153    async with connection() as conn:154        a = await fetch_one(conn, "select * from snapshots where id = :id", id=snapshot_id)155        b = await fetch_one(conn, "select * from snapshots where id = :id", id=other_id)156        if a is None or b is None:157            raise HTTPException(status_code=404, detail="snapshot not found")158        before, after = (a, b) if a["fetched_at"] <= b["fetched_at"] else (b, a)159        stored = await fetch_one(conn, "select diff, significance, id from changes where snapshot_before = :b and snapshot_after = :a limit 1",160                                 b=before["id"], a=after["id"])161    out = {"before": ser.snapshot(before), "after": ser.snapshot(after), "diff": None, "source": None}162    if stored and ser._dict(stored["diff"]):163        diff = ser._dict(stored["diff"])164        diff.setdefault("significance", ser._float(stored["significance"], 3))165        out.update(diff=diff, source="change", change_id=stored["id"])166        return out167    before_blocks, after_blocks = _load_blocks(before.get("blocks_key")), _load_blocks(after.get("blocks_key"))168    if before_blocks is None or after_blocks is None:169        raise HTTPException(status_code=404, detail="block objects unavailable for one of the snapshots")170    async with connection() as conn:171        surface = await fetch_val(conn, "select surface from sensors where id = :id", id=after["sensor_id"]) or "other"172    before_text = _load_text(before.get("text_key"))[0] or ""173    after_text = _load_text(after.get("text_key"))[0] or ""174    try:175        diff = _compute_diff(before_blocks, after_blocks, surface=surface, before_text=before_text, after_text=after_text)176    except Exception:177        log.exception("on-demand diff failed", extra={"before": before["id"], "after": after["id"]})178        raise HTTPException(status_code=500, detail="diff computation failed") from None179    if diff is None:180        raise HTTPException(status_code=501, detail="diff engine not available (companyatlas.sdk.diff.compare)")181    out.update(diff=diff, source="computed")182    return out183184185@router.get("/changes/{change_id}", summary="Change with diff, structured delta and derived events")186async def change_detail(change_id: str, response: Response) -> dict[str, Any]:187    response.headers["cache-control"] = public_cache_value(300)188    async with connection() as conn:189        row = await fetch_one(conn, "select * from changes where id = :id", id=change_id)190        if row is None:191            raise HTTPException(status_code=404, detail="change not found")192        out = ser.change(row, with_diff=True)193        where, params = q.event_filters(status=None)194        where.append("e.change_id = :chg")195        params["chg"] = change_id196        out["events"] = [ser.event(e) for e in await q.fetch_events(conn, where, params, limit=100)]197        sensor = await fetch_one(conn, "select id, url, surface, connector_id from sensors where id = :id", id=row["sensor_id"])198        out["sensor"] = sensor199        cards = await q.fetch_cards_by_ids(conn, [row["company_id"]])200        out["company"] = ser.company_card(cards[0]) if cards else None201    return out202