"""What a connector produces: entity references, temporal claims, relations, events, prices, benchmark results. Connectors never touch the database — they return `Facts`; the writer resolves, versions and records them.""" from __future__ import annotations from dataclasses import dataclass, field from datetime import datetime from typing import Any CONFIDENCE = ("verified", "high", "medium", "low", "conflicted") @dataclass class EntityRef: entity_type: str name: str identifiers: dict[str, str] = field(default_factory=dict) # scheme -> value (hf_repo, github_repo, arxiv, doi, domain, pypi, provider_model_id…) aliases: list[str] = field(default_factory=list) slug_hint: str | None = None organization: EntityRef | None = None # developer / publisher / owner description: str | None = None status: str | None = None attributes: dict[str, Any] = field(default_factory=dict) # convenience: each item becomes a Claim first_seen_hint: datetime | None = None # e.g. release date, to backdate `first_seen_at` for historical backfill id: str | None = field(default=None, compare=False) def key(self) -> str: if self.identifiers: scheme, value = sorted(self.identifiers.items())[0] return f"{self.entity_type}:{scheme}={value}" return f"{self.entity_type}:name={self.name.strip().lower()}" @dataclass class Claim: entity: EntityRef property: str value: Any unit: str | None = None confidence: str | None = None # default from source tier observed_at: datetime | None = None effective_at: datetime | None = None source_url: str | None = None @dataclass class Relation: subject: EntityRef predicate: str object: EntityRef attributes: dict[str, Any] = field(default_factory=dict) confidence: str | None = None source_url: str | None = None @dataclass class Event: event_type: str category: str summary: str entity: EntityRef | 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] = field(default_factory=dict) @dataclass class PriceObs: model: EntityRef provider: EntityRef provider_model_id: str | None = None input_per_mtok: float | None = None output_per_mtok: float | None = None cached_input_per_mtok: float | None = None cache_write_per_mtok: float | None = None batch_input_per_mtok: float | None = None batch_output_per_mtok: float | None = None per_image: float | None = None per_request: float | None = None currency: str = "USD" context_length: int | None = None max_output_tokens: int | None = None features: dict[str, Any] = field(default_factory=dict) source_url: str | None = None meta: dict[str, Any] = field(default_factory=dict) def price_tuple(self) -> tuple[Any, ...]: return (self.input_per_mtok, self.output_per_mtok, self.cached_input_per_mtok, self.cache_write_per_mtok, self.batch_input_per_mtok, self.batch_output_per_mtok, self.per_image, self.per_request, self.currency) @dataclass class ResultObs: model: EntityRef benchmark: EntityRef score: float metric: str | None = None unit: str | None = "%" higher_is_better: bool = True config: dict[str, Any] = field(default_factory=dict) evaluated_at: datetime | None = None source_url: str | None = None confidence: str | None = None @dataclass class Target: """Something to fetch. Connectors return targets from `discover()` and may add more from `extract()`.""" url: str doc_type: str = "page" entity: EntityRef | None = None meta: dict[str, Any] = field(default_factory=dict) accept: str | None = None min_bytes: int = 64 escalate: bool = False needs_llm: bool = False priority: int = 2 key: str | None = None # short handle for --file overrides / fixtures rate_per_min: int | None = None @dataclass class Facts: entities: list[EntityRef] = field(default_factory=list) claims: list[Claim] = field(default_factory=list) relations: list[Relation] = field(default_factory=list) events: list[Event] = field(default_factory=list) prices: list[PriceObs] = field(default_factory=list) results: list[ResultObs] = field(default_factory=list) targets: list[Target] = field(default_factory=list) # follow-up fetches discovered while extracting document_title: str | None = None document_entity: EntityRef | None = None # main entity described by the document llm_hint: str | None = None # ask the LLM factory for a specific extraction task # ---------------------------------------------------------------------------------------------- builders def entity(self, entity_type: str, name: str, **kw: Any) -> EntityRef: ref = EntityRef(entity_type=entity_type, name=name.strip(), **kw) self.entities.append(ref) return ref def claim(self, entity: EntityRef, property: str, value: Any, **kw: Any) -> Claim | None: if value is None or value == "" or value == [] or value == {}: return None c = Claim(entity=entity, property=property, value=value, **kw) self.claims.append(c) return c def relate(self, subject: EntityRef, predicate: str, obj: EntityRef, **kw: Any) -> Relation: r = Relation(subject=subject, predicate=predicate, object=obj, **kw) self.relations.append(r) return r def event(self, event_type: str, category: str, summary: str, **kw: Any) -> Event: e = Event(event_type=event_type, category=category, summary=summary[:500], **kw) self.events.append(e) return e def price(self, **kw: Any) -> PriceObs: p = PriceObs(**kw) self.prices.append(p) return p def result(self, **kw: Any) -> ResultObs: r = ResultObs(**kw) self.results.append(r) return r def follow(self, url: str, **kw: Any) -> Target: t = Target(url=url, **kw) self.targets.append(t) return t def extend(self, other: Facts) -> None: self.entities += other.entities self.claims += other.claims self.relations += other.relations self.events += other.events self.prices += other.prices self.results += other.results self.targets += other.targets def is_empty(self) -> bool: return not (self.entities or self.claims or self.relations or self.events or self.prices or self.results) # Property → change-event mapping (material properties emit events; noisy metrics never do). MATERIAL_PROPERTIES: dict[str, tuple[str, int]] = { "context_length": ("CONTEXT_CHANGED", 2), "max_output_tokens": ("MAX_OUTPUT_CHANGED", 1), "status": ("STATUS_CHANGED", 2), "license": ("LICENSE_CHANGED", 2), "parameter_count": ("PARAMETERS_CHANGED", 1), "active_parameter_count": ("PARAMETERS_CHANGED", 1), "weights_availability": ("OPENNESS_CHANGED", 3), "openness": ("OPENNESS_CHANGED", 3), "release_date": ("RELEASE_DATE_CHANGED", 1), "knowledge_cutoff": ("KNOWLEDGE_CUTOFF_CHANGED", 1), "latest_version": ("VERSION_RELEASED", 2), "deprecation_date": ("DEPRECATION_ANNOUNCED", 3), "retirement_date": ("RETIREMENT_ANNOUNCED", 3), "modalities": ("CAPABILITIES_CHANGED", 2), "capabilities": ("CAPABILITIES_CHANGED", 1), "rate_limit": ("RATE_LIMIT_CHANGED", 1), "memory_gb": ("SPEC_CHANGED", 1), "price_usd": ("PRICE_CHANGED", 2), } NOISY_PREFIXES = ("metric.", "stats.", "counts.") EVENT_CATEGORY_BY_TYPE: dict[str, str] = { "model": "model", "company": "company", "organization": "company", "lab": "company", "university": "company", "paper": "paper", "dataset": "dataset", "benchmark": "benchmark", "provider": "provider", "framework": "framework", "library": "framework", "repository": "repository", "tool": "tool", "agent": "tool", "hardware": "hardware", "runtime": "framework", "quantization": "model", "regulation": "regulation", "incident": "incident", "release": "release", "mcp_server": "tool", "researcher": "company", "license": "model", "product": "tool", "robot": "hardware", } __all__ = ["EntityRef", "Claim", "Relation", "Event", "PriceObs", "ResultObs", "Target", "Facts", "MATERIAL_PROPERTIES", "NOISY_PREFIXES", "EVENT_CATEGORY_BY_TYPE", "CONFIDENCE"]