"""FactWriter — turns `Facts` into versioned database state. Temporal rules (per entity × property): * no current claim → insert current claim, set attribute, NEW_* event for new entities * same value → confirm (observed_at bumped), nothing else * different, source ≥ tier → supersede (valid_to = observed), insert new current, update attribute, CHANGE event for material props * different, worse source → store as `conflicting`, flag confidence, review-queue item — never overwrite Prices and benchmark results have their own append-only tables with the same close/open semantics. """ from __future__ import annotations import hashlib import json import logging from datetime import UTC, datetime from typing import Any from sqlalchemy.ext.asyncio import AsyncConnection from aiatlas.db import execute, fetch_one, jsonb from aiatlas.ids import new_id, normalize_alias from aiatlas.sdk.facts import EVENT_CATEGORY_BY_TYPE, MATERIAL_PROPERTIES, NOISY_PREFIXES, EntityRef, Facts, PriceObs, ResultObs from aiatlas.sdk.resolution import Resolver log = logging.getLogger(__name__) TIER_CONFIDENCE = {1: "high", 2: "medium", 3: "low", 4: "low"} SOFT_PROPERTIES = {"description", "summary", "tagline", "availability_note", "abstract", "notes", "training_data_notes", "safety_notes", "hardware_requirements"} NEW_IMPORTANCE = {"model": 3, "company": 2, "provider": 2, "paper": 1, "dataset": 1, "benchmark": 2, "framework": 1, "hardware": 2, "tool": 1, "repository": 0, "release": 2, "regulation": 2, "incident": 2, "organization": 1, "researcher": 0} def _norm_value(v: Any) -> Any: if isinstance(v, datetime): return v.astimezone(UTC).isoformat(timespec="seconds") if isinstance(v, (set, tuple)): return sorted(v) if all(isinstance(x, str) for x in v) else list(v) if isinstance(v, float) and v.is_integer() and abs(v) < 1e15: return int(v) return v def _same(a: Any, b: Any) -> bool: return json.dumps(_norm_value(a), sort_keys=True, default=str) == json.dumps(_norm_value(b), sort_keys=True, default=str) class WriteStats: def __init__(self) -> None: self.entities_created = 0 self.entities_updated = 0 self.claims = 0 self.relations = 0 self.events = 0 self.prices = 0 self.results = 0 self.conflicts = 0 def as_dict(self) -> dict[str, int]: return dict(self.__dict__) class FactWriter: def __init__(self, conn: AsyncConnection, *, source_id: str | None, snapshot_id: str | None, source_url: str | None, tier: int = 2, connector_name: str | None = None, extractor: str = "deterministic", extractor_version: str = "1", observed_at: datetime | None = None): self.conn = conn self.source_id = source_id self.snapshot_id = snapshot_id self.source_url = source_url self.tier = tier self.connector_name = connector_name self.extractor = extractor self.extractor_version = extractor_version self.observed_at = observed_at or datetime.now(UTC) self.resolver = Resolver(conn, snapshot_id=snapshot_id, source_tier=tier) self.stats = WriteStats() # ---------------------------------------------------------------------------------------------- entry point async def write(self, facts: Facts) -> WriteStats: new_before = len(self.resolver.created) for ref in facts.entities: await self.resolver.resolve(ref) for prop, value in ref.attributes.items(): await self.write_claim(ref, prop, value) for c in facts.claims: await self.write_claim(c.entity, c.property, c.value, unit=c.unit, confidence=c.confidence, observed_at=c.observed_at, effective_at=c.effective_at, source_url=c.source_url) for r in facts.relations: await self.write_relation(r.subject, r.predicate, r.object, r.attributes, confidence=r.confidence, source_url=r.source_url) for p in facts.prices: await self.write_price(p) for res in facts.results: await self.write_result(res) for e in facts.events: eid = await self.resolver.resolve(e.entity) if e.entity else None await self.emit_event(e.event_type, e.category, e.summary, entity_id=eid, old_value=e.old_value, new_value=e.new_value, importance=e.importance, effective_at=e.effective_at, dedupe_key=e.dedupe_key, source_url=e.source_url, meta=e.meta) # NEW_* events for every entity created in this write for eid in self.resolver.created[new_before:]: await self._new_entity_event(eid) self.stats.entities_created = len(self.resolver.created) self.stats.entities_updated = len(self.resolver.updated - set(self.resolver.created)) return self.stats # ---------------------------------------------------------------------------------------------- claims async def write_claim(self, ref: EntityRef, prop: str, value: Any, *, unit: str | None = None, confidence: str | None = None, observed_at: datetime | None = None, effective_at: datetime | None = None, source_url: str | None = None) -> None: if value is None or value == "" or value == [] or value == {}: return eid = await self.resolver.resolve(ref) assert eid value = _norm_value(value) observed = observed_at or self.observed_at conf = confidence or TIER_CONFIDENCE.get(self.tier, "medium") url = source_url or self.source_url noisy = prop.startswith(NOISY_PREFIXES) current = await fetch_one(self.conn, """select id, value, tier, source_url, observed_at from claims where entity_id = :e and property = :p and status = 'current' order by valid_from desc limit 1""", e=eid, p=prop) is_new_entity = eid in self.resolver.created if current is None: await self._insert_claim(eid, prop, value, unit, conf, "current", observed, effective_at, url) await self._set_attribute(eid, prop, value, unit, conf, url, observed) return if _same(current["value"], value): await execute(self.conn, "update claims set observed_at = greatest(observed_at, :o) where id = :id", o=observed, id=current["id"]) if noisy: await self._set_attribute(eid, prop, value, unit, conf, url, observed) return same_source = bool(url and current["source_url"] == url) if prop in SOFT_PROPERTIES and not same_source: # soft text (descriptions, notes): first statement wins until *its own* source changes; never an event, never a conflict return if self.tier <= (current["tier"] or 2) or same_source: # supersede await execute(self.conn, "update claims set status = 'superseded', valid_to = :o where id = :id", o=observed, id=current["id"]) await self._insert_claim(eid, prop, value, unit, conf, "current", observed, effective_at, url) await self._set_attribute(eid, prop, value, unit, conf, url, observed) if not noisy and not is_new_entity and prop not in SOFT_PROPERTIES: await self._property_change_event(eid, prop, current["value"], value, observed, effective_at, url) else: await self._insert_claim(eid, prop, value, unit, conf, "conflicting", observed, effective_at, url) await execute(self.conn, "update claims set confidence = 'conflicted' where id = :id", id=current["id"]) await execute(self.conn, """update entities set quality = quality || jsonb_build_object('conflicts', coalesce((quality->>'conflicts')::int, 0) + 1) where id = :id""", id=eid) self.stats.conflicts += 1 dedupe = f"conflict:{eid}:{prop}:{hashlib.sha1(json.dumps(value, sort_keys=True, default=str).encode()).hexdigest()[:10]}" await execute(self.conn, """insert into review_queue (id, kind, entity_ids, payload, reason, dedupe_key) values (:id, 'conflict', :ids, cast(:p as jsonb), :r, :d) on conflict (dedupe_key) do nothing""", id=new_id("review"), ids=[eid], p=jsonb({"property": prop, "current": current["value"], "current_source": current["source_url"], "claimed": value, "claimed_source": url, "tier": self.tier}), r=f"source disagrees on {prop}", d=dedupe) async def _insert_claim(self, eid: str, prop: str, value: Any, unit: str | None, conf: str, status: str, observed: datetime, effective_at: datetime | None, url: str | None) -> None: text_val = value if isinstance(value, str) else None num_val = float(value) if isinstance(value, (int, float)) and not isinstance(value, bool) else None await execute(self.conn, """insert into claims (id, entity_id, property, value, value_text, value_num, unit, source_id, snapshot_id, source_url, tier, confidence, status, extractor, extractor_version, observed_at, effective_at, valid_from) values (:id, :e, :p, cast(:v as jsonb), :vt, :vn, :u, :src, :snap, :url, :tier, :conf, :status, :ex, :exv, :o, :eff, :vf)""", id=new_id("claim"), e=eid, p=prop, v=jsonb(value), vt=text_val[:2000] if text_val else None, vn=num_val, u=unit, src=self.source_id, snap=self.snapshot_id, url=url, tier=self.tier, conf=conf, status=status, ex=self.extractor, exv=self.extractor_version, o=observed, eff=effective_at, vf=effective_at or observed) self.stats.claims += 1 async def _set_attribute(self, eid: str, prop: str, value: Any, unit: str | None, conf: str, url: str | None, observed: datetime) -> None: prov = {"source_id": self.source_id, "snapshot_id": self.snapshot_id, "url": url, "observed_at": observed.isoformat(timespec="seconds"), "tier": self.tier, "confidence": conf, "extractor": self.extractor} if unit: prov["unit"] = unit await execute(self.conn, """update entities set attributes = attributes || jsonb_build_object(cast(:p as text), cast(:v as jsonb)), provenance = provenance || jsonb_build_object(cast(:p as text), cast(:prov as jsonb)), last_seen_at = greatest(last_seen_at, :o) where id = :id""", p=prop, v=jsonb(value), prov=jsonb(prov), o=observed, id=eid) if prop == "status" and isinstance(value, str): await execute(self.conn, "update entities set status = :s where id = :id", s=value[:40], id=eid) if prop == "description" and isinstance(value, str): await execute(self.conn, "update entities set description = :d where id = :id", d=value[:4000], id=eid) async def _property_change_event(self, eid: str, prop: str, old: Any, new: Any, observed: datetime, effective_at: datetime | None, url: str | None) -> None: event_type, importance = MATERIAL_PROPERTIES.get(prop, ("PROPERTY_CHANGED", 0)) row = await fetch_one(self.conn, "select canonical_name, entity_type from entities where id = :id", id=eid) name = row["canonical_name"] if row else eid category = EVENT_CATEGORY_BY_TYPE.get(row["entity_type"] if row else "", "update") if prop == "status" and str(new).lower() in ("deprecated", "retired", "discontinued"): importance = 3 summary = f"{name}: {prop.replace('_', ' ')} changed from {_short(old)} to {_short(new)}" dedupe = f"{event_type}:{eid}:{prop}:{hashlib.sha1(json.dumps([old, new], sort_keys=True, default=str).encode()).hexdigest()[:12]}" await self.emit_event(event_type, category, summary, entity_id=eid, old_value=old, new_value=new, importance=importance, effective_at=effective_at, dedupe_key=dedupe, source_url=url, meta={"property": prop}, observed_at=observed) # ---------------------------------------------------------------------------------------------- relations async def write_relation(self, subject: EntityRef, predicate: str, obj: EntityRef, attributes: dict[str, Any] | None = None, *, confidence: str | None = None, source_url: str | None = None) -> None: sid = await self.resolver.resolve(subject) oid = await self.resolver.resolve(obj) if not sid or not oid or sid == oid: return conf = confidence or TIER_CONFIDENCE.get(self.tier, "medium") existing = await fetch_one(self.conn, "select id, attributes from relations where subject_id = :s and predicate = :p and object_id = :o and valid_to is null", s=sid, p=predicate, o=oid) if existing: if attributes and not _same(existing["attributes"], {**existing["attributes"], **attributes}): await execute(self.conn, "update relations set attributes = attributes || cast(:a as jsonb), observed_at = :o where id = :id", a=jsonb(attributes), o=self.observed_at, id=existing["id"]) else: await execute(self.conn, "update relations set observed_at = :o where id = :id", o=self.observed_at, id=existing["id"]) return await execute(self.conn, """insert into relations (id, subject_id, predicate, object_id, attributes, source_id, snapshot_id, source_url, tier, confidence, observed_at, valid_from) values (:id, :s, :p, :o, cast(:a as jsonb), :src, :snap, :url, :tier, :conf, :obs, :obs)""", id=new_id("relation"), s=sid, p=predicate, o=oid, a=jsonb(attributes or {}), src=self.source_id, snap=self.snapshot_id, url=source_url or self.source_url, tier=self.tier, conf=conf, obs=self.observed_at) self.stats.relations += 1 # ---------------------------------------------------------------------------------------------- prices async def write_price(self, p: PriceObs) -> None: mid = await self.resolver.resolve(p.model) pid = await self.resolver.resolve(p.provider) if not mid or not pid: return if all(v is None for v in p.price_tuple()[:-1]): return url = p.source_url or self.source_url current = await fetch_one(self.conn, """select * from prices where model_id = :m and provider_id = :p and coalesce(provider_model_id,'') = :pm and valid_to is null""", m=mid, p=pid, pm=p.provider_model_id or "") new_tuple = p.price_tuple() if current: cur_tuple = (current["input_per_mtok"], current["output_per_mtok"], current["cached_input_per_mtok"], current["cache_write_per_mtok"], current["batch_input_per_mtok"], current["batch_output_per_mtok"], current["per_image"], current["per_request"], current["currency"]) if _same(cur_tuple, new_tuple) and (p.context_length in (None, current["context_length"])): await execute(self.conn, "update prices set observed_at = :o where id = :id", o=self.observed_at, id=current["id"]) return await execute(self.conn, "update prices set valid_to = :o where id = :id", o=self.observed_at, id=current["id"]) await execute(self.conn, """insert into prices (id, model_id, provider_id, provider_model_id, input_per_mtok, output_per_mtok, cached_input_per_mtok, cache_write_per_mtok, batch_input_per_mtok, batch_output_per_mtok, per_image, per_request, currency, context_length, max_output_tokens, features, observed_at, valid_from, source_id, snapshot_id, source_url, tier, meta) values (:id, :m, :p, :pm, :i, :o, :ci, :cw, :bi, :bo, :img, :req, :cur, :ctx, :mo, cast(:f as jsonb), :obs, :obs, :src, :snap, :url, :tier, cast(:meta as jsonb))""", id=new_id("price"), m=mid, p=pid, pm=p.provider_model_id, i=p.input_per_mtok, o=p.output_per_mtok, ci=p.cached_input_per_mtok, cw=p.cache_write_per_mtok, bi=p.batch_input_per_mtok, bo=p.batch_output_per_mtok, img=p.per_image, req=p.per_request, cur=p.currency, ctx=p.context_length, mo=p.max_output_tokens, f=jsonb(p.features), obs=self.observed_at, src=self.source_id, snap=self.snapshot_id, url=url, tier=self.tier, meta=jsonb(p.meta)) self.stats.prices += 1 model = await fetch_one(self.conn, "select canonical_name from entities where id = :id", id=mid) provider = await fetch_one(self.conn, "select canonical_name from entities where id = :id", id=pid) mname = model["canonical_name"] if model else mid pname = provider["canonical_name"] if provider else pid if current: old = {"input_per_mtok": current["input_per_mtok"], "output_per_mtok": current["output_per_mtok"]} new = {"input_per_mtok": p.input_per_mtok, "output_per_mtok": p.output_per_mtok} summary = f"{pname} changed pricing for {mname}: {_fmt_price(old)} → {_fmt_price(new)}" dedupe = f"PRICE_CHANGED:{mid}:{pid}:{p.provider_model_id or ''}:{hashlib.sha1(json.dumps([old, new], sort_keys=True, default=str).encode()).hexdigest()[:12]}" await self.emit_event("PRICE_CHANGED", "price", summary, entity_id=mid, old_value=old, new_value=new, importance=2, dedupe_key=dedupe, source_url=url, meta={"provider_id": pid, "provider": pname}) else: new = {"input_per_mtok": p.input_per_mtok, "output_per_mtok": p.output_per_mtok} summary = f"{pname} lists {mname} at {_fmt_price(new)}" dedupe = f"PROVIDER_LISTED:{mid}:{pid}:{p.provider_model_id or ''}" await self.emit_event("PROVIDER_LISTED", "provider", summary, entity_id=mid, new_value=new, importance=1, dedupe_key=dedupe, source_url=url, meta={"provider_id": pid, "provider": pname}) await self.write_relation(p.model, "available_through", p.provider, {"provider_model_id": p.provider_model_id}, source_url=url) # ---------------------------------------------------------------------------------------------- benchmark results async def write_result(self, r: ResultObs) -> None: mid = await self.resolver.resolve(r.model) bid = await self.resolver.resolve(r.benchmark) if not mid or not bid: return url = r.source_url or self.source_url cfg_hash = hashlib.sha1(json.dumps(r.config, sort_keys=True, default=str).encode()).hexdigest()[:12] dedupe = f"{mid}:{bid}:{cfg_hash}:{r.metric or ''}" existing = await fetch_one(self.conn, "select id, score from benchmark_results where dedupe_key = :d", d=dedupe) conf = r.confidence or TIER_CONFIDENCE.get(self.tier, "medium") if existing: if abs((existing["score"] or 0) - r.score) > 1e-9: await execute(self.conn, "update benchmark_results set valid_to = :o, dedupe_key = dedupe_key || ':' || :suffix where id = :id", o=self.observed_at, suffix=self.observed_at.strftime("%Y%m%d%H%M%S"), id=existing["id"]) model = await fetch_one(self.conn, "select canonical_name from entities where id = :id", id=mid) bench = await fetch_one(self.conn, "select canonical_name from entities where id = :id", id=bid) await self.emit_event("BENCHMARK_UPDATED", "benchmark", f"{model['canonical_name'] if model else mid} on {bench['canonical_name'] if bench else bid}: {existing['score']:g} → {r.score:g}", entity_id=mid, old_value=existing["score"], new_value=r.score, importance=1, dedupe_key=f"BENCHMARK_UPDATED:{dedupe}:{r.score:g}", source_url=url, meta={"benchmark_id": bid}) else: await execute(self.conn, "update benchmark_results set observed_at = :o where id = :id", o=self.observed_at, id=existing["id"]) return await execute(self.conn, """insert into benchmark_results (id, model_id, benchmark_id, score, metric, unit, higher_is_better, config, evaluated_at, observed_at, source_id, snapshot_id, source_url, tier, confidence, dedupe_key) values (:id, :m, :b, :s, :metric, :unit, :hib, cast(:cfg as jsonb), :ev, :obs, :src, :snap, :url, :tier, :conf, :d)""", id=new_id("result"), m=mid, b=bid, s=r.score, metric=r.metric, unit=r.unit, hib=r.higher_is_better, cfg=jsonb(r.config), ev=r.evaluated_at, obs=self.observed_at, src=self.source_id, snap=self.snapshot_id, url=url, tier=self.tier, conf=conf, d=dedupe) self.stats.results += 1 if not existing: model = await fetch_one(self.conn, "select canonical_name from entities where id = :id", id=mid) bench = await fetch_one(self.conn, "select canonical_name from entities where id = :id", id=bid) await self.emit_event("BENCHMARK_RESULT", "benchmark", f"{model['canonical_name'] if model else mid} scores {r.score:g}{r.unit or ''} on {bench['canonical_name'] if bench else bid}", entity_id=mid, new_value=r.score, importance=1, dedupe_key=f"BENCHMARK_RESULT:{dedupe}", source_url=url, meta={"benchmark_id": bid, "metric": r.metric}) await self.write_relation(r.model, "evaluated_on", r.benchmark, source_url=url) # ---------------------------------------------------------------------------------------------- events async def emit_event(self, event_type: str, category: str, summary: str, *, entity_id: str | None = None, old_value: Any = None, new_value: Any = None, importance: int = 2, effective_at: datetime | None = None, dedupe_key: str | None = None, source_url: str | None = None, meta: dict[str, Any] | None = None, observed_at: datetime | None = None) -> None: dedupe = dedupe_key or f"{event_type}:{entity_id or ''}:{normalize_alias(summary)[:120]}" await execute(self.conn, """insert into change_events (id, entity_id, event_type, category, property, old_value, new_value, summary, importance, observed_at, effective_at, source_id, snapshot_id, source_url, connector_name, dedupe_key, meta) values (:id, :e, :t, :c, :p, cast(:o as jsonb), cast(:n as jsonb), :s, :imp, :obs, :eff, :src, :snap, :url, :conn, :d, cast(:meta as jsonb)) on conflict (dedupe_key) do nothing""", id=new_id("change_event"), e=entity_id, t=event_type, c=category, p=(meta or {}).get("property"), o=jsonb(old_value) if old_value is not None else None, n=jsonb(new_value) if new_value is not None else None, s=summary[:500], imp=max(0, min(3, importance)), obs=observed_at or self.observed_at, eff=effective_at, src=self.source_id, snap=self.snapshot_id, url=source_url or self.source_url, conn=self.connector_name, d=dedupe[:400], meta=jsonb(meta or {})) self.stats.events += 1 async def _new_entity_event(self, eid: str) -> None: row = await fetch_one(self.conn, "select canonical_name, entity_type, attributes, organization_id from entities where id = :id", id=eid) if not row: return etype = row["entity_type"] if etype in ("researcher", "country", "license", "quantization"): return importance = NEW_IMPORTANCE.get(etype, 1) if self.tier > 2: importance = max(0, importance - 1) org = None if row["organization_id"]: o = await fetch_one(self.conn, "select canonical_name from entities where id = :id", id=row["organization_id"]) org = o["canonical_name"] if o else None label = etype.replace("_", " ") summary = f"New {label}: {row['canonical_name']}" + (f" ({org})" if org else "") effective = None rel = row["attributes"].get("release_date") or row["attributes"].get("published_at") if isinstance(rel, str): from aiatlas.sdk.extract.dates import parse_datetime effective = parse_datetime(rel) await self.emit_event(f"NEW_{etype.upper()}", EVENT_CATEGORY_BY_TYPE.get(etype, "update"), summary, entity_id=eid, importance=importance, dedupe_key=f"NEW_{etype.upper()}:{eid}", effective_at=effective) def _short(v: Any) -> str: s = json.dumps(v, default=str, ensure_ascii=False) if not isinstance(v, str) else v return s if len(s) <= 60 else s[:57] + "…" def _fmt_price(p: dict[str, Any]) -> str: i, o = p.get("input_per_mtok"), p.get("output_per_mtok") parts = [] if i is not None: parts.append(f"${i:g} in") if o is not None: parts.append(f"${o:g} out") return " / ".join(parts) + " per 1M tokens" if parts else "n/a" __all__ = ["FactWriter", "WriteStats"]