1"""Live counters and aggregate statistics — always computed from the database, never hardcoded."""2from __future__ import annotations34from typing import Any56from sqlalchemy.ext.asyncio import AsyncConnection78from aiatlas.db import execute, fetch_all, fetch_one, jsonb, transaction9from aiatlas.sdk.archive import archive_size101112async def live_counts(conn: AsyncConnection) -> dict[str, Any]:13 by_type = await fetch_all(conn, "select entity_type, count(*) as n from entities where merged_into is null group by 1")14 counts = {r["entity_type"]: int(r["n"]) for r in by_type}15 infra = await fetch_one(conn, """select (select count(*) from sources where enabled) as sources,16 (select count(*) from connectors) as connectors,17 (select count(*) from connectors where enabled) as connectors_enabled,18 (select count(*) from documents) as documents,19 (select count(*) from snapshots) as snapshots,20 (select count(*) from claims) as claims,21 (select count(*) from claims where status = 'current') as claims_current,22 (select count(*) from relations where valid_to is null) as relations,23 (select count(*) from change_events) as change_events,24 (select count(*) from change_events where observed_at > now() - interval '24 hours') as change_events_24h,25 (select count(*) from change_events where observed_at > now() - interval '7 days') as change_events_7d,26 (select count(*) from benchmark_results where valid_to is null) as benchmark_results,27 (select count(*) from prices where valid_to is null) as prices_current,28 (select count(*) from prices) as prices_total,29 (select count(*) from jobs where status = 'queued') as jobs_queued,30 (select count(*) from jobs where status = 'dead') as jobs_dead,31 (select count(*) from review_queue where status = 'pending') as review_pending,32 (select count(*) from llm_jobs) as llm_jobs,33 (select coalesce(sum(input_tokens),0) + coalesce(sum(output_tokens),0) from llm_jobs) as llm_tokens,34 (select max(observed_at) from snapshots) as last_snapshot_at,35 (select max(observed_at) from change_events) as last_event_at,36 (select min(first_seen_at) from entities) as first_entity_at""")37 return {"entities": counts, "entities_total": sum(counts.values()), **{k: (int(v) if isinstance(v, int) else v) for k, v in (infra or {}).items()}}383940async def compute_stats() -> dict[str, Any]:41 async with transaction() as conn:42 counts = await live_counts(conn)43 counts["archive"] = archive_size()44 await execute(conn, "insert into stats_snapshots (counts) values (cast(:c as jsonb))", c=jsonb(counts))45 return counts464748async def history(conn: AsyncConnection, days: int = 90) -> list[dict[str, Any]]:49 return await fetch_all(conn, """select distinct on (date_trunc('day', computed_at)) date_trunc('day', computed_at) as day, counts50 from stats_snapshots where computed_at > now() - make_interval(days => :d) order by 1 desc, computed_at desc""", d=days)515253__all__ = ["live_counts", "compute_stats", "history"]54