HTML 77.2%
TypeScript 10.5%
Python 9.6%
JavaScript 2.5%
1"""/claims/{id} — claim lifecycle (API 1.1): the claim, its entity, the chain of superseded / superseding / conflicting siblings, source, extractor,2run id and evidence pointer (snapshot id + document URL + archived flag; never raw content)."""3from __future__ import annotations45from typing import Any67from fastapi import APIRouter, Request89from aiatlas.api.common import (10 CLAIM_COLS,11 CLAIM_FROM,12 ENTITY_COLS,13 ENTITY_FROM,14 ApiError,15 cached,16 claim_row,17 entity_summary,18)19from aiatlas.db import connection, fetch_all, fetch_one2021router = APIRouter(prefix="/api/v1/claims", tags=["claims"])222324@router.get("/{claim_id}")25@cached(120)26async def claim_detail(request: Request, claim_id: str) -> dict[str, Any]:27 async with connection() as conn:28 c = await fetch_one(conn, f"select {CLAIM_COLS}, c.entity_id, c.snapshot_id, c.run_id, c.extractor_version, c.value_raw, c.value_text, c.value_num, c.source_id, "29 f"s.domain as source_domain, s.tier as source_tier from {CLAIM_FROM} where c.id = :id", id=claim_id)30 if not c:31 raise ApiError(404, "claim not found")32 ent = await fetch_one(conn, f"select {ENTITY_COLS} from {ENTITY_FROM} where e.id = :id", id=c["entity_id"])33 siblings = await fetch_all(conn, f"select {CLAIM_COLS}, c.snapshot_id, c.run_id from {CLAIM_FROM} where c.entity_id = :e and c.property = :p and c.id <> :id order by c.valid_from asc, c.observed_at asc limit 200",34 e=c["entity_id"], p=c["property"], id=claim_id)35 snap = None36 if c.get("snapshot_id"):37 snap = await fetch_one(conn, """select s.id, s.observed_at, s.http_status, s.content_type, s.parser_version, s.raw_path is not null as archived, s.text_path is not null as has_text,38 d.url as document_url, d.title as document_title, d.doc_type from snapshots s join documents d on d.id = s.document_id where s.id = :id""", id=c["snapshot_id"])39 vf = c["valid_from"]40 previous = [claim_row(s) for s in siblings if s["status"] == "superseded" and (s["valid_to"] is None or s["valid_to"] <= vf)]41 superseding = [claim_row(s) for s in siblings if s["valid_from"] >= vf and s["status"] in ("current", "superseded") and c["status"] != "current"]42 conflicting = [claim_row(s) for s in siblings if s["status"] == "conflicting"]43 return {"claim": {**claim_row(c), "value_raw": c.get("value_raw"), "run_id": c.get("run_id"), "extractor_version": c.get("extractor_version")},44 "entity": entity_summary(ent) if ent else {"id": c["entity_id"]}, "property": c["property"],45 "chain": {"previous": previous[-5:], "superseding": superseding[:5], "conflicting": conflicting, "history_count": len(siblings) + 1},46 "source": {"id": c.get("source_id"), "name": c.get("source_name"), "domain": c.get("source_domain"), "tier": c.get("tier"), "url": c.get("source_url"),47 "snapshot_id": c.get("snapshot_id"), "observed_at": c.get("observed_at")},48 "extractor": {"name": c.get("extractor"), "version": c.get("extractor_version"), "confidence": c.get("confidence")}, "run_id": c.get("run_id"),49 "evidence": {"snapshot_id": c.get("snapshot_id"), "document_url": (snap or {}).get("document_url") or c.get("source_url"), "archived": bool((snap or {}).get("archived")),50 "snapshot_observed_at": (snap or {}).get("observed_at"), "document_title": (snap or {}).get("document_title"), "doc_type": (snap or {}).get("doc_type")},51 "note": "Evidence is a pointer to the archived snapshot; raw content is available to administrators only (/admin/extractions/{snapshot_id})."}52