spb/company-atlas

1"""Admin API (`X-CA-Admin-Token`). Nothing is ever deleted: actions update `status` columns and append audit rows in payloads."""2from __future__ import annotations34from datetime import UTC, datetime5from typing import Any67from fastapi import APIRouter, Depends, HTTPException, Query, Response8from pydantic import BaseModel, Field910from companyatlas import archive11from companyatlas.api import aggregates as agg12from companyatlas.api import queries as q13from companyatlas.api import serializers as ser14from companyatlas.api.common import NO_STORE, PageDep, cache, page_payload, require_admin15from companyatlas.config import settings16from companyatlas.db import connection, execute, fetch_all, fetch_one, fetch_val, jsonb, transaction17from companyatlas.ids import new_id, slugify18from companyatlas.taxonomy import FailureClass, SensorStatus, tier_for_interval19from companyatlas.urls import canonicalize_url, registrable_domain2021ORDER = 1022router = APIRouter(prefix="/api/v1/admin", tags=["admin"], dependencies=[Depends(require_admin)])23LOW_QUALITY = 30.024STALE_DAYS = 725WORKER_WINDOW_MIN = 15262728def _ns(response: Response) -> None:29    response.headers["cache-control"] = NO_STORE303132# ------------------------------------------------------------------------------------------------ overview333435@router.get("/overview")36async def overview(response: Response) -> dict[str, Any]:37    _ns(response)38    today = datetime.now(UTC).replace(hour=0, minute=0, second=0, microsecond=0)39    async with connection() as conn:40        cs = await fetch_all(conn, "select status, count(*) as n from companies group by status")41        cos = await fetch_all(conn, "select onboarding_status, count(*) as n from companies group by onboarding_status")42        ss = await fetch_all(conn, "select status, count(*) as n from sensors group by status")43        st = await fetch_all(conn, "select tier, count(*) as n from sensors where status <> 'retired' group by tier")44        qs = await fetch_one(conn, "select count(*) filter (where status = 'pending') as pending, count(*) filter (where status = 'running') as running, "45                                   "count(*) filter (where status = 'dead') as dead, count(*) filter (where status = 'failed') as failed, "46                                   "extract(epoch from (now() - min(run_at) filter (where status = 'pending' and run_at <= now()))) as oldest_pending_s from queue_jobs") or {}47        llm = await fetch_one(conn, "select count(*) filter (where status = 'pending') as pending, count(*) filter (where status = 'done' and finished_at >= :t) as done_today, "48                                    "count(*) filter (where status = 'failed' and finished_at >= :t) as failed_today from llm_jobs", t=today) or {}49        fails = await fetch_all(conn, "select failure_class, count(*) as n from failures where at >= now() - interval '24 hours' group by failure_class order by n desc")50        rates = await fetch_one(conn, "select (select count(*) from observations where fetched_at >= now() - interval '1 hour') as fetch_rate_1h, "51                                      "(select count(*) from changes where detected_at >= now() - interval '1 hour') as change_rate_1h, "52                                      "(select count(*) from changes where detected_at >= now() - interval '1 hour' and kind in ('meaningful','major','critical')) as meaningful_rate_1h") or {}53        workers = await fetch_all(conn, "select worker as name, max(fetched_at) as last_seen_at, count(*) as fetches_15m from observations "54                                        "where fetched_at >= now() - make_interval(mins => :m) and worker is not null group by worker order by last_seen_at desc limit 50",55                                  m=WORKER_WINDOW_MIN)56        inflight = {r["claimed_by"]: int(r["n"]) for r in await fetch_all(conn, "select claimed_by, count(*) as n from sensors where claimed_by is not null group by claimed_by")}57        runs = await fetch_all(conn, "select worker as name, max(started_at) as last_seen_at from crawl_runs where started_at >= now() - make_interval(mins => :m) "58                                     "and worker is not null group by worker", m=WORKER_WINDOW_MIN)59        cost = await fetch_all(conn, "select dimension, sum(cost_estimate) as cost, sum(units) as units from cost_ledger where day = current_date group by dimension")60        periodic = await q.settings_value(conn, "scheduler:heartbeat")61    names = {w["name"] for w in workers} | {r["name"] for r in runs} | set(inflight)62    seen = {w["name"]: w["last_seen_at"] for w in workers}63    for r in runs:64        if r["name"] not in seen or (r["last_seen_at"] and seen[r["name"]] and r["last_seen_at"] > seen[r["name"]]):65            seen[r["name"]] = r["last_seen_at"]66    done_today, failed_today = int(llm.get("done_today") or 0), int(llm.get("failed_today") or 0)67    costs = {r["dimension"]: round(float(r["cost"] or 0), 4) for r in cost}68    return {"companies_by_status": {r["status"]: int(r["n"]) for r in cs}, "companies_by_onboarding": {r["onboarding_status"]: int(r["n"]) for r in cos},69            "sensors_by_status": {r["status"]: int(r["n"]) for r in ss}, "sensors_by_tier": {(r["tier"] or "").strip(): int(r["n"]) for r in st},70            "queue": {"pending": int(qs.get("pending") or 0), "running": int(qs.get("running") or 0), "dead": int(qs.get("dead") or 0),71                      "failed": int(qs.get("failed") or 0), "oldest_pending_s": round(float(qs["oldest_pending_s"]), 1) if qs.get("oldest_pending_s") is not None else 0.0},72            "llm": {"pending": int(llm.get("pending") or 0), "done_today": done_today, "failed_today": failed_today,73                    "budget_left": max(0, settings.llm_daily_budget - done_today - failed_today), "budget": settings.llm_daily_budget, "configured": settings.llm_configured},74            "failures_24h_by_class": {r["failure_class"]: int(r["n"]) for r in fails}, "fetch_rate_1h": int(rates.get("fetch_rate_1h") or 0),75            "change_rate_1h": int(rates.get("change_rate_1h") or 0), "meaningful_rate_1h": int(rates.get("meaningful_rate_1h") or 0),76            "storage": await agg.archive_stats(),77            "workers": [{"name": n, "last_seen_at": seen.get(n), "inflight": inflight.get(n, 0)} for n in sorted(names)],78            "cost_today": {"fetch": costs.get("fetch", 0.0), "browser": costs.get("browser", 0.0), "llm": costs.get("llm", 0.0), "total": round(sum(costs.values()), 4)},79            "scheduler_heartbeat": periodic, "time": datetime.now(UTC)}808182# ------------------------------------------------------------------------------------------------ connectors838485@router.get("/connectors")86async def connectors(response: Response) -> dict[str, Any]:87    _ns(response)88    async with connection() as conn:89        rows = await fetch_all(conn, "select * from connectors order by category, id")90        sensors = await fetch_all(conn, "select connector_id, count(*) filter (where status = 'active') as active, "91                                        "count(*) filter (where status in ('failing','stale','blocked')) as failing, count(*) filter (where status <> 'retired') as total, "92                                        "max(last_run_at) as last_run_at from sensors group by connector_id")93        obs = await fetch_all(conn, "select s.connector_id, count(*) as n, count(*) filter (where o.failure_class is null) as ok, avg(o.duration_ms) as latency, "94                                    "count(*) filter (where o.changed) as changed, count(*) filter (where o.failure_class is not null) as errors "95                                    "from observations o join sensors s on s.id = o.sensor_id where o.fetched_at >= now() - interval '24 hours' group by s.connector_id")96    sm = {r["connector_id"]: r for r in sensors}97    om = {r["connector_id"]: r for r in obs}98    items = []99    known = {r["id"] for r in rows}100    for cid in list(known) + [k for k in sm if k not in known]:101        row = next((r for r in rows if r["id"] == cid), None)102        base = ser.connector(row) if row else {"id": cid, "name": cid, "version": "?", "category": "?", "enabled": True}103        s, o = sm.get(cid, {}), om.get(cid, {})104        n = int(o.get("n") or 0)105        base.update({"sensors_active": int(s.get("active") or 0), "sensors_failing": int(s.get("failing") or 0), "sensors_total": int(s.get("total") or 0),106                     "success_rate_24h": round(int(o.get("ok") or 0) / n, 4) if n else None, "avg_latency_ms": round(float(o["latency"]), 1) if o.get("latency") is not None else None,107                     "change_rate_24h": round(int(o.get("changed") or 0) / n, 4) if n else None, "errors_24h": int(o.get("errors") or 0), "fetches_24h": n,108                     "last_run_at": s.get("last_run_at")})109        items.append(base)110    items.sort(key=lambda x: (x.get("category") or "", x["id"]))111    return {"items": items}112113114# ------------------------------------------------------------------------------------------------ sensors115116SENSOR_FILTERS: dict[str, str] = {117    "healthy": "s.status = 'active' and s.consecutive_failures = 0",118    "failing": "(s.status = 'failing' or s.consecutive_failures > 0)",119    "stale": "(s.status = 'stale' or (s.status in ('active','failing') and (s.last_success_at is null or s.last_success_at < now() - make_interval(days => :stale_days))))",120    "blocked": "(s.status = 'blocked' or s.last_failure_class in ('BOT_CHALLENGE','ROBOTS','BLOCKED_DESTINATION','RATE_LIMIT'))",121    "redirected": "(s.status = 'redirected' or s.last_failure_class = 'REDIRECT')",122    "low_quality": "s.quality_score < :low_quality",123    "high_activity": "s.last_change_at >= now() - interval '24 hours'",124}125SENSOR_SORTS = {"recent": "s.last_run_at desc nulls last, s.id", "next_run": "s.next_run_at asc, s.id", "failures": "s.consecutive_failures desc, s.last_run_at desc nulls last, s.id",126                "quality": "s.quality_score asc, s.id", "changes": "s.change_count desc, s.id", "created": "s.created_at desc, s.id"}127SENSOR_SELECT = ("select s.*, c.slug as company_slug, c.display_name as company_display_name, c.canonical_domain as company_domain, c.country as company_country, "128                 "c.logo_url as company_logo_url from sensors s join companies c on c.id = s.company_id")129130131@router.get("/sensors")132async def admin_sensors(response: Response, p: PageDep, status: str | None = None, domain: str | None = None, connector: str | None = None,133                        company: str | None = None, surface: str | None = None,134                        filter: str | None = Query(None, pattern="^(healthy|failing|stale|blocked|redirected|low_quality|high_activity)$"),135                        sort: str = Query("recent", pattern="^(recent|next_run|failures|quality|changes|created)$")) -> dict[str, Any]:136    _ns(response)137    where, params = ["true"], {"stale_days": STALE_DAYS, "low_quality": LOW_QUALITY}138    if status:139        where.append("s.status = cast(:status as text)")140        params["status"] = status.lower()[:20]141    if domain:142        where.append("s.domain = cast(:domain as text)")143        params["domain"] = domain.lower()[:253]144    if connector:145        where.append("s.connector_id = cast(:connector as text)")146        params["connector"] = connector[:80]147    if surface:148        where.append("s.surface = cast(:surface as text)")149        params["surface"] = surface.lower()[:40]150    if filter:151        where.append(SENSOR_FILTERS[filter])152        if filter == "high_activity":153            sort = "changes"154    async with connection() as conn:155        if company:156            params["company_id"] = (await q.require_company(conn, company))["id"]157            where.append("s.company_id = :company_id")158        wsql = " and ".join(where)159        rows = await fetch_all(conn, f"{SENSOR_SELECT} where {wsql} order by {SENSOR_SORTS[sort]} limit :limit offset :offset", **params, limit=p.per_page, offset=p.offset)160        total = await q.bounded_count(conn, f"from sensors s join companies c on c.id = s.company_id where {wsql}", params)161    items = []162    for r in rows:163        item = ser.sensor_admin(r)164        item["company"] = ser.company_ref(r)165        items.append(item)166    return page_payload(items, total, p)167168169class SensorActionBody(BaseModel):170    interval_s: int | None = Field(None, ge=60, le=90 * 86400)171    connector_id: str | None = Field(None, max_length=80)172    reason: str | None = Field(None, max_length=500)173174175SENSOR_ACTIONS = ("pause", "resume", "retry", "rediscover", "retire", "run_now", "set_interval", "set_connector")176177178@router.post("/sensors/{sensor_id}/{action}")179async def sensor_action(sensor_id: str, action: str, response: Response, body: SensorActionBody | None = None) -> dict[str, Any]:180    _ns(response)181    if action not in SENSOR_ACTIONS:182        raise HTTPException(status_code=404, detail=f"unknown action (one of {', '.join(SENSOR_ACTIONS)})")183    body = body or SensorActionBody()184    now = datetime.now(UTC)185    async with transaction() as conn:186        s = await fetch_one(conn, "select * from sensors where id = :id", id=sensor_id)187        if s is None:188            raise HTTPException(status_code=404, detail="sensor not found")189        extra: dict[str, Any] = {}190        if action == "pause":191            await execute(conn, "update sensors set status = :st, claimed_by = null, claimed_at = null, updated_at = :now where id = :id", st=SensorStatus.PAUSED.value, now=now, id=sensor_id)192        elif action == "resume":193            await execute(conn, "update sensors set status = :st, consecutive_failures = 0, next_run_at = :now, retired_at = null, updated_at = :now where id = :id",194                          st=SensorStatus.ACTIVE.value, now=now, id=sensor_id)195        elif action in ("retry", "run_now"):196            prio = ", priority = greatest(priority, 0.95)" if action == "run_now" else ""197            await execute(conn, f"update sensors set next_run_at = :now, claimed_by = null, claimed_at = null{prio}, updated_at = :now where id = :id", now=now, id=sensor_id)198        elif action == "retire":199            await execute(conn, "update sensors set status = :st, retired_at = :now, claimed_by = null, claimed_at = null, updated_at = :now where id = :id",200                          st=SensorStatus.RETIRED.value, now=now, id=sensor_id)201        elif action == "rediscover":202            jid = new_id("queue_job")203            key = f"discover:{s['company_id']}:{int(now.timestamp())}"204            await execute(conn, "insert into queue_jobs (id, kind, key, payload, priority) values (:id, 'discover', :key, cast(:p as jsonb), 0.9) on conflict (key) do nothing",205                          id=jid, key=key, p=jsonb({"company_id": s["company_id"], "sensor_id": sensor_id, "reason": body.reason or "admin:rediscover"}))206            extra["queued"] = {"id": jid, "key": key}207        elif action == "set_interval":208            if body.interval_s is None:209                raise HTTPException(status_code=422, detail="interval_s required")210            iv = max(settings.min_interval_s, min(settings.max_interval_s, body.interval_s))211            await execute(conn, "update sensors set base_interval_s = :iv, current_interval_s = :iv, tier = :tier, next_run_at = :now, updated_at = :now where id = :id",212                          iv=iv, tier=tier_for_interval(iv), now=now, id=sensor_id)213            extra["interval_s"] = iv214        elif action == "set_connector":215            if not body.connector_id:216                raise HTTPException(status_code=422, detail="connector_id required")217            ok = await fetch_val(conn, "select enabled from connectors where id = :c", c=body.connector_id)218            if ok is None:219                raise HTTPException(status_code=422, detail="unknown connector_id")220            await execute(conn, "update sensors set connector_id = :c, next_run_at = :now, updated_at = :now where id = :id", c=body.connector_id, now=now, id=sensor_id)221        row = await fetch_one(conn, f"{SENSOR_SELECT} where s.id = :id", id=sensor_id)222    out = ser.sensor_admin(row or {})223    out["company"] = ser.company_ref(row or {})224    return {"ok": True, "action": action, "sensor": out, **extra}225226227# ------------------------------------------------------------------------------------------------ companies228229230@router.get("/companies")231async def admin_companies(response: Response, p: PageDep, onboarding_status: str | None = None, status: str | None = None,232                          q_: str | None = Query(None, alias="q", max_length=200), country: str | None = None,233                          sort: str = Query("recent", pattern="^(activity|events|hiring|name|importance|recent)$")) -> dict[str, Any]:234    _ns(response)235    where, params = q.company_filters(q=q_, country=country, status=status, onboarding_status=onboarding_status)236    async with connection() as conn:237        ids, total = await q.company_page_ids(conn, where, params, sort=sort, limit=p.per_page, offset=p.offset)238        rows = await q.fetch_cards_by_ids(conn, ids)239    items = []240    for r in rows:241        card = ser.company_card(r)242        card.update({"onboarding_error": r.get("onboarding_error"), "indexed": bool(r.get("indexed")), "discovered_at": r.get("discovered_at"), "updated_at": r.get("updated_at")})243        items.append(card)244    return page_payload(items, total, p)245246247class CompanyCreate(BaseModel):248    website: str = Field(min_length=4, max_length=500)249    display_name: str | None = Field(None, max_length=200)250    country: str | None = Field(None, min_length=2, max_length=2)251    industries: list[str] | None = None252    importance: float | None = Field(None, ge=0, le=1)253    tier: int | None = Field(None, ge=1, le=4)254255256@router.post("/companies", status_code=201)257async def admin_create_company(body: CompanyCreate, response: Response) -> dict[str, Any]:258    _ns(response)259    website = body.website.strip()260    if "://" not in website:261        website = "https://" + website262    if not website.lower().startswith(("http://", "https://")):263        raise HTTPException(status_code=422, detail="website: only http(s) URLs are accepted")264    domain = registrable_domain(website)265    if not domain or "." not in domain:266        raise HTTPException(status_code=422, detail="website: could not derive a registrable domain")267    website = canonicalize_url(website)268    display_name = (body.display_name or "").strip() or domain.split(".")[0].capitalize()269    country = body.country.upper() if body.country else None270    industries = [slugify(i) for i in (body.industries or []) if i.strip()][:10]271    async with transaction() as conn:272        existing = await fetch_one(conn, "select id, slug from companies where canonical_domain = :d", d=domain)273        if existing:274            raise HTTPException(status_code=409, detail=f"company already exists: {existing['slug']}")275        if country and not await fetch_val(conn, "select 1 from countries where code = cast(:c as char(2))", c=country):276            raise HTTPException(status_code=422, detail="country: unknown ISO-2 code")277        if industries:278            known = {r["slug"] for r in await fetch_all(conn, "select slug from industries where slug = any(cast(:s as text[]))", s=industries)}279            missing = [i for i in industries if i not in known]280            if missing:281                raise HTTPException(status_code=422, detail=f"industries: unknown slugs {', '.join(missing)}")282        base = slugify(display_name)283        slug, n = base, 2284        while await fetch_val(conn, "select 1 from companies where slug = :s", s=slug):285            slug = f"{base}-{n}"286            n += 1287        cid = new_id("company")288        await execute(conn, "insert into companies (id, slug, display_name, canonical_domain, website, country, industries, industry_primary, importance, tier, "289                            "source_meta) values (:id, :slug, :name, :domain, :website, :country, cast(:inds as text[]), :ip, :imp, :tier, cast(:meta as jsonb))",290                      id=cid, slug=slug, name=display_name, domain=domain, website=website, country=country, inds=industries, ip=industries[0] if industries else None,291                      imp=body.importance if body.importance is not None else 0.2, tier=body.tier or 4, meta=jsonb({"source": "admin_api", "created_at": datetime.now(UTC)}))292        await execute(conn, "insert into domains (id, company_id, domain, kind) values (:id, :cid, :d, 'primary') on conflict do nothing", id=new_id("domain"), cid=cid, d=domain)293        jid = new_id("queue_job")294        await execute(conn, "insert into queue_jobs (id, kind, key, payload, priority) values (:id, 'discover', :key, cast(:p as jsonb), 0.8) on conflict (key) do nothing",295                      id=jid, key=f"discover:{cid}", p=jsonb({"company_id": cid, "reason": "admin:create"}))296        cards = await q.fetch_cards_by_ids(conn, [cid])297    return {"ok": True, "company": ser.company_card(cards[0]), "queued": {"id": jid, "kind": "discover"}}298299300@router.post("/companies/{key}/rediscover")301async def admin_rediscover(key: str, response: Response) -> dict[str, Any]:302    _ns(response)303    now = datetime.now(UTC)304    async with transaction() as conn:305        c = await q.require_company(conn, key)306        jid = new_id("queue_job")307        jkey = f"discover:{c['id']}:{int(now.timestamp())}"308        await execute(conn, "insert into queue_jobs (id, kind, key, payload, priority) values (:id, 'discover', :key, cast(:p as jsonb), 0.9)",309                      id=jid, key=jkey, p=jsonb({"company_id": c["id"], "reason": "admin:rediscover"}))310        if c["onboarding_status"] in ("failed", "no_website"):311            await execute(conn, "update companies set onboarding_status = 'pending', onboarding_error = null, updated_at = :now where id = :id", now=now, id=c["id"])312    return {"ok": True, "company": c["slug"], "queued": {"id": jid, "key": jkey}}313314315# ------------------------------------------------------------------------------------------------ failures / queue / llm / reviews316317318@router.get("/failures")319async def admin_failures(response: Response, p: PageDep, class_: str | None = Query(None, alias="class", max_length=40), since: str | None = None,320                         sensor: str | None = None, company: str | None = None) -> dict[str, Any]:321    _ns(response)322    where, params = ["true"], {}323    if class_:324        where.append("f.failure_class = cast(:fc as text)")325        params["fc"] = class_.upper()326    since_dt = q.parse_iso(since) or q.days_ago(7)327    where.append("f.at >= :since")328    params["since"] = since_dt329    if sensor:330        where.append("f.sensor_id = :sid")331        params["sid"] = sensor332    async with connection() as conn:333        if company:334            params["cid"] = (await q.require_company(conn, company))["id"]335            where.append("f.company_id = :cid")336        wsql = " and ".join(where)337        rows = await fetch_all(conn, f"select f.*, c.slug as company_slug, s.surface from failures f left join companies c on c.id = f.company_id "338                                     f"left join sensors s on s.id = f.sensor_id where {wsql} order by f.at desc limit :limit offset :offset", **params, limit=p.per_page, offset=p.offset)339        total = await q.bounded_count(conn, f"from failures f where {wsql}", params)340        by_class = await fetch_all(conn, f"select f.failure_class, count(*) as n from failures f where {wsql} group by 1 order by n desc", **params)341    out = page_payload([ser.failure(r) for r in rows], total, p)342    out["by_class"] = {r["failure_class"]: int(r["n"]) for r in by_class}343    out["classes"] = [c.value for c in FailureClass]344    return out345346347@router.get("/queue")348async def admin_queue(response: Response, kind: str | None = None, status: str | None = None, limit: int = Query(100, ge=1, le=500)) -> dict[str, Any]:349    _ns(response)350    where, params = ["true"], {"limit": limit}351    if kind:352        where.append("kind = cast(:kind as text)")353        params["kind"] = kind[:40]354    if status:355        where.append("status = cast(:status as text)")356        params["status"] = status[:20]357    async with connection() as conn:358        counts = await fetch_all(conn, "select kind, status, count(*) as n from queue_jobs group by kind, status order by kind, status")359        rows = await fetch_all(conn, f"select * from queue_jobs where {' and '.join(where)} order by case status when 'running' then 0 when 'pending' then 1 else 2 end, "360                                     "run_at desc limit :limit", **params)361    return {"counts": [{"kind": r["kind"], "status": r["status"], "n": int(r["n"])} for r in counts], "items": [ser.queue_job(r) for r in rows]}362363364class RequeueBody(BaseModel):365    kind: str | None = Field(None, max_length=40)366367368@router.post("/queue/requeue-dead")369async def admin_requeue_dead(response: Response, body: RequeueBody | None = None) -> dict[str, Any]:370    _ns(response)371    body = body or RequeueBody()372    extra, params = "", {}373    if body.kind:374        extra = " and kind = cast(:kind as text)"375        params["kind"] = body.kind376    async with transaction() as conn:377        n = await fetch_val(conn, "with u as (update queue_jobs set status = 'pending', attempts = 0, run_at = now(), locked_at = null, locked_by = null, "378                                  f"last_error = null, finished_at = null where status = 'dead'{extra} returning 1) select count(*) from u", **params)379    return {"ok": True, "requeued": int(n or 0)}380381382@router.get("/llm")383async def admin_llm(response: Response, p: PageDep, status: str | None = None, kind: str | None = None) -> dict[str, Any]:384    _ns(response)385    where, params = ["true"], {}386    if status:387        where.append("status = cast(:status as text)")388        params["status"] = status[:20]389    if kind:390        where.append("kind = cast(:kind as text)")391        params["kind"] = kind[:40]392    wsql = " and ".join(where)393    async with connection() as conn:394        rows = await fetch_all(conn, f"select * from llm_jobs where {wsql} order by created_at desc limit :limit offset :offset", **params, limit=p.per_page, offset=p.offset)395        total = await q.bounded_count(conn, f"from llm_jobs where {wsql}", params)396        stats = await fetch_one(conn, "select count(*) filter (where status = 'pending') as pending, count(*) filter (where status = 'running') as running, "397                                      "count(*) filter (where status = 'done' and finished_at >= current_date) as done_today, "398                                      "count(*) filter (where status = 'failed' and finished_at >= current_date) as failed_today, "399                                      "avg(latency_ms) filter (where status = 'done' and finished_at >= current_date) as avg_latency_ms, "400                                      "sum(request_tokens + coalesce(response_tokens, 0)) filter (where finished_at >= current_date) as tokens_today from llm_jobs") or {}401    out = page_payload([ser.llm_job(r) for r in rows], total, p)402    out["stats"] = {k: (round(float(v), 1) if k == "avg_latency_ms" and v is not None else (int(v) if v is not None else 0)) for k, v in stats.items()}403    out["stats"]["budget"] = settings.llm_daily_budget404    out["stats"]["configured"] = settings.llm_configured405    return out406407408@router.get("/reviews")409async def admin_reviews(response: Response, p: PageDep, kind: str | None = None, status: str = Query("open", max_length=20)) -> dict[str, Any]:410    _ns(response)411    where, params = ["true"], {}412    if kind:413        where.append("r.kind = cast(:kind as text)")414        params["kind"] = kind[:40]415    if status and status != "all":416        where.append("r.status = cast(:status as text)")417        params["status"] = status418    wsql = " and ".join(where)419    async with connection() as conn:420        rows = await fetch_all(conn, f"select r.*, c.slug as company_slug, c.display_name as company_display_name from review_queue r left join companies c on c.id = r.company_id "421                                     f"where {wsql} order by r.created_at asc limit :limit offset :offset", **params, limit=p.per_page, offset=p.offset)422        total = await q.bounded_count(conn, f"from review_queue r where {wsql}", params)423        by_kind = await fetch_all(conn, "select kind, count(*) as n from review_queue where status = 'open' group by kind order by n desc")424    out = page_payload([ser.review(r) for r in rows], total, p)425    out["open_by_kind"] = {r["kind"]: int(r["n"]) for r in by_kind}426    return out427428429class ReviewBody(BaseModel):430    resolution: str = Field(pattern="^(accepted|rejected)$")431    note: str | None = Field(None, max_length=1000)432    label: str | None = Field(None, pattern="^(correct|duplicate|noise|misclassified)$")433434435@router.post("/reviews/{review_id}")436async def admin_resolve_review(review_id: str, body: ReviewBody, response: Response) -> dict[str, Any]:437    _ns(response)438    async with transaction() as conn:439        r = await fetch_one(conn, "select * from review_queue where id = :id", id=review_id)440        if r is None:441            raise HTTPException(status_code=404, detail="review not found")442        if r["status"] != "open":443            raise HTTPException(status_code=409, detail=f"review already {r['status']}")444        payload = ser._dict(r["payload"])445        payload["resolution"] = {"status": body.resolution, "note": body.note, "label": body.label, "at": datetime.now(UTC)}446        await execute(conn, "update review_queue set status = :st, resolution = :res, resolved_at = now(), payload = cast(:p as jsonb) where id = :id",447                      st=body.resolution, res=body.label or body.note or body.resolution, p=jsonb(payload), id=review_id)448        row = await fetch_one(conn, "select * from review_queue where id = :id", id=review_id)449    return {"ok": True, "review": ser.review(row or {})}450451452# ------------------------------------------------------------------------------------------------ events (corrections)453454455class RetractBody(BaseModel):456    reason: str = Field(min_length=3, max_length=500)457458459async def _audit(conn: Any, event_id: str, action: str, reason: str | None) -> None:460    entry = jsonb([{"action": action, "reason": reason, "at": datetime.now(UTC)}])461    await execute(conn, "update events set payload = jsonb_set(payload, '{_audit}', coalesce(payload->'_audit', '[]'::jsonb) || cast(:e as jsonb), true) where id = :id",462                  e=entry, id=event_id)463464465@router.post("/events/{event_id}/retract")466async def admin_retract_event(event_id: str, body: RetractBody, response: Response) -> dict[str, Any]:467    _ns(response)468    async with transaction() as conn:469        ev = await fetch_one(conn, "select id, status from events where id = :id", id=event_id)470        if ev is None:471            raise HTTPException(status_code=404, detail="event not found")472        await execute(conn, "update events set status = 'retracted', retracted_reason = :r where id = :id", r=body.reason.strip(), id=event_id)473        await _audit(conn, event_id, "retract", body.reason.strip())474        row = await q.fetch_event(conn, event_id)475    cache.clear()476    return {"ok": True, "event": ser.event(row or {})}477478479@router.post("/events/{event_id}/restore")480async def admin_restore_event(event_id: str, response: Response) -> dict[str, Any]:481    _ns(response)482    async with transaction() as conn:483        ev = await fetch_one(conn, "select id, status, retracted_reason from events where id = :id", id=event_id)484        if ev is None:485            raise HTTPException(status_code=404, detail="event not found")486        await execute(conn, "update events set status = 'active', retracted_reason = null where id = :id", id=event_id)487        await _audit(conn, event_id, "restore", ev.get("retracted_reason"))488        row = await q.fetch_event(conn, event_id)489    cache.clear()490    return {"ok": True, "event": ser.event(row or {})}491492493# ------------------------------------------------------------------------------------------------ quality / costs / cache494495496@router.get("/quality")497async def admin_quality(response: Response) -> dict[str, Any]:498    _ns(response)499    async with connection() as conn:500        r = await fetch_one(conn, """501            select (select count(*) from companies) as companies,502                   (select count(*) from companies where onboarding_status = 'active' and status = 'ACTIVE') as companies_active,503                   (select count(*) from sensors where status <> 'retired') as sensors,504                   (select count(*) from sensors where status = 'active') as sensors_active,505                   (select count(*) from sensors where status <> 'retired' and last_run_at >= now() - interval '24 hours') as checked_24h,506                   (select count(*) from sensors where status = 'stale') as stale,507                   (select count(*) from sensors where status in ('failing', 'blocked')) as failed_sensors,508                   (select count(*) from sensors where surface = 'other' and status <> 'retired') as unknown_surfaces,509                   (select count(*) from events where detected_at >= now() - interval '30 days') as events_30d,510                   (select count(*) from events where detected_at >= now() - interval '30 days' and status = 'duplicate') as duplicates_30d,511                   (select avg(confidence) from events where detected_at >= now() - interval '30 days' and status = 'active') as confidence_avg512        """) or {}513        cal = await fetch_all(conn, "select coalesce(payload->'resolution'->>'label', resolution) as label, count(*) as n from review_queue "514                                    "where status in ('accepted','rejected','resolved') group by 1")515    companies, sensors = int(r.get("companies") or 0), int(r.get("sensors") or 0)516    ev30 = int(r.get("events_30d") or 0)517    labels = {c["label"]: int(c["n"]) for c in cal if c["label"]}518    return {"coverage": {"companies_active_pct": round(int(r.get("companies_active") or 0) / companies * 100, 1) if companies else None,519                         "sensors_active_pct": round(int(r.get("sensors_active") or 0) / sensors * 100, 1) if sensors else None,520                         "companies": companies, "sensors": sensors},521            "freshness": {"sensors_checked_24h_pct": round(int(r.get("checked_24h") or 0) / sensors * 100, 1) if sensors else None, "stale": int(r.get("stale") or 0)},522            "duplicate_rate": round(int(r.get("duplicates_30d") or 0) / ev30, 4) if ev30 else None,523            "event_confidence_avg": round(float(r["confidence_avg"]), 3) if r.get("confidence_avg") is not None else None,524            "unknown_surfaces": int(r.get("unknown_surfaces") or 0), "failed_sensors": int(r.get("failed_sensors") or 0),525            "calibration": {k: labels.get(k, 0) for k in ("correct", "duplicate", "noise", "misclassified")}, "events_30d": ev30}526527528@router.get("/costs")529async def admin_costs(response: Response, days: int = Query(30, ge=1, le=365)) -> dict[str, Any]:530    _ns(response)531    async with connection() as conn:532        items = await fetch_all(conn, "select day, dimension, key, units, cost_estimate from cost_ledger where day >= :d order by day desc, dimension, key limit 5000",533                                d=q.days_ago(days).date())534        denom = await fetch_one(conn, "select (select count(*) from companies where status = 'ACTIVE') as companies, "535                                      "(select count(*) from observations where fetched_at >= :d) as observations, "536                                      "(select count(*) from events where status = 'active' and detected_at >= :d and importance >= :imp) as meaningful_events",537                                d=q.days_ago(days), imp=settings.meaningful_threshold) or {}538    total = sum(float(i["cost_estimate"] or 0) for i in items)539    by_dim: dict[str, float] = {}540    for i in items:541        by_dim[i["dimension"]] = by_dim.get(i["dimension"], 0.0) + float(i["cost_estimate"] or 0)542    comp, obs, mev = int(denom.get("companies") or 0), int(denom.get("observations") or 0), int(denom.get("meaningful_events") or 0)543    return {"days": days, "items": [{"day": i["day"], "dimension": i["dimension"], "key": i["key"], "units": float(i["units"] or 0), "cost_estimate": float(i["cost_estimate"] or 0)}544                                    for i in items],545            "total": round(total, 4), "by_dimension": {k: round(v, 4) for k, v in by_dim.items()},546            "per_1000_companies": round(total / comp * 1000, 4) if comp else None, "per_million_observations": round(total / obs * 1_000_000, 4) if obs else None,547            "per_meaningful_event": round(total / mev, 4) if mev else None}548549550@router.post("/cache/clear")551async def admin_cache_clear(response: Response, prefix: str | None = Query(None, max_length=60)) -> dict[str, Any]:552    _ns(response)553    cache.clear(prefix)554    return {"ok": True, "cleared": prefix or "all"}555556557@router.get("/storage")558async def admin_storage(response: Response) -> dict[str, Any]:559    _ns(response)560    stats = await agg.archive_stats()561    return {"objects_dir": str(settings.objects_dir), "exists": archive.object_path("00" * 32).parent.parent.parent.exists(), **stats}562