"""BaseConnector — the lifecycle every connector shares: discover() → fetch() → parse() → extract() → diff/write → schedule_next() Subclasses usually implement `discover()` (seed URLs / feeds / sitemaps) and `extract()` (structured facts from a parsed document). `run()` does all bookkeeping: connector_runs, documents, snapshots (raw archive), content-hash change detection, structural diffs, DOCUMENT_CHANGED events, breakage detection (record count collapse ≠ deletion), circuit breaker and adaptive scheduling. """ from __future__ import annotations import asyncio import json import logging import statistics import time from dataclasses import dataclass, field from datetime import UTC, datetime, timedelta from typing import Any from aiatlas.db import connection, execute, fetch_one, jsonb, transaction from aiatlas.ids import new_id from aiatlas.sdk import archive from aiatlas.sdk.extract.feeds import FeedItem, parse_feed from aiatlas.sdk.extract.html import HtmlDoc, parse_html from aiatlas.sdk.extract.markdown import MarkdownDoc, parse_markdown from aiatlas.sdk.facts import EntityRef, Facts, Target, facts_to_json from aiatlas.sdk.fetch import BlockedError, Fetcher, FetchError, FetchResult, NotModified, canonicalize_url, file_result from aiatlas.sdk.resolution import Resolver from aiatlas.sdk.writer import FactWriter log = logging.getLogger(__name__) CIRCUIT_FAILURES = 3 CIRCUIT_COOLDOWN_S = 1800 BASELINE_RUNS = 5 # rolling window (median) of full-extraction runs used by the quarantine detector QUARANTINE_MIN_NEW_ENTITIES = 50 # a run may always create up to this many new entities without tripping the detector class ConnectorError(Exception): pass @dataclass class Parsed: kind: str # html|markdown|feed|json|pdf|text|xml html: HtmlDoc | None = None markdown: MarkdownDoc | None = None feed_meta: dict[str, Any] | None = None feed_items: list[FeedItem] = field(default_factory=list) json: Any = None text: str = "" def structured(self) -> dict[str, Any] | None: if self.html: return self.html.structured() if self.markdown: return {"front_matter": _jsonable(self.markdown.front_matter), "headings": self.markdown.headings[:200], "tables": self.markdown.tables[:40], "link_count": len(self.markdown.links), "text_length": len(self.markdown.text)} if self.kind == "feed": return {"feed": self.feed_meta, "items": [{"id": i.id, "url": i.url, "title": i.title, "published_at": i.published_at.isoformat() if i.published_at else None} for i in self.feed_items[:200]]} if self.kind == "json": return {"json_keys": list(self.json)[:100] if isinstance(self.json, dict) else None, "json_length": len(self.json) if isinstance(self.json, (list, dict)) else None} return {"text_length": len(self.text)} if self.text else None @dataclass class RunStats: docs_discovered: int = 0 docs_fetched: int = 0 docs_changed: int = 0 docs_unchanged: int = 0 docs_failed: int = 0 entities_created: int = 0 entities_updated: int = 0 claims_written: int = 0 relations_written: int = 0 events_emitted: int = 0 records: int = 0 # connector-defined "records" for breakage detection (models, papers, prices…) meta: dict[str, Any] = field(default_factory=dict) @dataclass class Buffered: """Facts extracted from one document, held until the end of the run when the connector is quarantine-capable.""" target: Target doc_id: str snapshot_id: str source_url: str fetched_at: datetime facts: Facts @dataclass class RunContext: run_id: str connector: BaseConnector fetcher: Fetcher started_at: datetime force: bool = False reprocess: bool = False # re-extract from stored snapshots without fetching file_overrides: dict[str, str] = field(default_factory=dict) # target.key or url → local path max_targets: int = 2000 stats: RunStats = field(default_factory=RunStats) seen_urls: set[str] = field(default_factory=set) is_first_run: bool = False # no earlier successful run → every event of this run is backfill source_key: str | None = None buffered: list[Buffered] = field(default_factory=list) # quarantine mode: facts held until the run-end anomaly check quarantined: bool = False quarantine_reason: str | None = None log: logging.LoggerAdapter[logging.Logger] = field(init=False) def __post_init__(self) -> None: self.log = logging.LoggerAdapter(logging.getLogger(f"connector.{self.connector.name}"), {"connector": self.connector.name, "run_id": self.run_id}) class BaseConnector: # ------------------------------------------------------------------------------------------ identity & policy name: str = "" label: str = "" description: str = "" source_key: str = "" # sources.key (seeded from registry/sources.yaml) version: str = "1" # connector version parser_version: str = "1" # bump when extraction improves → `aia reprocess ` replays snapshots interval_seconds: int = 3600 min_interval_seconds: int = 900 max_interval_seconds: int = 7 * 86400 rate_per_min: int = 30 respect_robots: bool = True expected_min_records: int = 0 priority: int = 2 tier: int = 1 concurrency: int = 3 default_doc_type: str = "page" needs_llm: bool = False # queue documents for LLM extraction after deterministic extraction # Hold every fact until the end of the run and compare the run's counts with the connector's rolling baseline before writing; # a collapse or an explosion parks the whole run in `quarantined_runs` for review. None = default (tier ≥ 2 hub/leaderboard/registry # connectors are quarantine-capable; tier-1 lab documentation is not — it is the primary source). quarantine: bool | None = None def __init__(self, config: dict[str, Any] | None = None): self.config = config or {} self.source_id: str | None = None @property def quarantine_enabled(self) -> bool: if self.quarantine is not None: return self.quarantine return self.tier >= 2 # ------------------------------------------------------------------------------------------ to implement async def discover(self, ctx: RunContext) -> list[Target]: """Return the targets for this run (seed pages, feeds, sitemaps, org pages…).""" seeds = self.config.get("seeds") or getattr(self, "seeds", []) return [Target(url=s) if isinstance(s, str) else Target(**s) for s in seeds] async def extract(self, ctx: RunContext, target: Target, res: FetchResult, parsed: Parsed) -> Facts: """Turn a parsed document into facts. Deterministic only; LLM extraction is queued separately.""" return Facts() def parse(self, target: Target, res: FetchResult) -> Parsed: """Default parser by content type; connectors may override for special formats.""" if res.is_pdf: return Parsed(kind="pdf", text=_pdf_text(res.content)) if target.doc_type == "feed" or (res.is_xml and b" bool: return True # ------------------------------------------------------------------------------------------ orchestration async def run(self, *, force: bool = False, reprocess: bool = False, file_overrides: dict[str, str] | None = None, max_targets: int | None = None, only_urls: list[str] | None = None) -> RunContext: started = datetime.now(UTC) run_id = new_id("connector_run") async with Fetcher(rate_per_min=self.rate_per_min, robots=self.respect_robots) as fetcher: ctx = RunContext(run_id=run_id, connector=self, fetcher=fetcher, started_at=started, force=force, reprocess=reprocess, file_overrides=file_overrides or {}, max_targets=max_targets or int(self.config.get("max_targets", 2000))) async with transaction() as conn: state = await fetch_one(conn, """select c.source_id, c.enabled, c.circuit_open_until, s.key as source_key, exists(select 1 from connector_runs r where r.connector_name = c.name and r.status in ('success','unchanged','suspect')) as has_success from connectors c left join sources s on s.id = c.source_id where c.name = :n""", n=self.name) if state is None: raise ConnectorError(f"connector {self.name!r} not registered (run `aia seed`)") self.source_id = state["source_id"] ctx.source_key = state["source_key"] or self.source_key or None ctx.is_first_run = not state["has_success"] if not state["enabled"] and not force: ctx.log.info("connector disabled, skipping") await self._record(conn, ctx, "skipped", error="disabled") return ctx if state["circuit_open_until"] and state["circuit_open_until"] > started and not force: ctx.log.warning("circuit open, skipping", extra={"until": state["circuit_open_until"].isoformat()}) await self._record(conn, ctx, "skipped", error="circuit open") return ctx await execute(conn, "insert into connector_runs (id, connector_name, started_at, status) values (:id, :c, :t, 'running')", id=run_id, c=self.name, t=started) await execute(conn, "update connectors set last_attempt_at = :t, health = case when health = 'unknown' then 'degraded' else health end where name = :n", t=started, n=self.name) t0 = time.perf_counter() try: targets = await self.discover(ctx) if only_urls: targets = await self._select_targets(targets, only_urls) ctx.stats.docs_discovered = len(targets) await self._process_all(ctx, targets) if ctx.buffered: await self._flush_or_quarantine(ctx) status = "quarantined" if ctx.quarantined else self._final_status(ctx) async with transaction() as conn: await self._record(conn, ctx, status) changed = ctx.stats.docs_changed > 0 await execute(conn, """update connectors set consecutive_failures = 0, circuit_open_until = null, last_success_at = case when cast(:q as boolean) then last_success_at else now() end, last_change_at = case when cast(:changed as boolean) and not cast(:q as boolean) then now() else last_change_at end, consecutive_unchanged = case when cast(:changed as boolean) then 0 else consecutive_unchanged + 1 end, interval_seconds = cast(:interval as integer), next_run_at = now() + make_interval(secs => cast(:interval as double precision)), last_duration_ms = :d, health = :health, updated_at = now() where name = :n""", changed=changed, q=ctx.quarantined, interval=await self._next_interval(conn, changed), d=int((time.perf_counter() - t0) * 1000), health="degraded" if status in ("suspect", "quarantined") else "ok", n=self.name) if status == "suspect": await execute(conn, """insert into review_queue (id, kind, entity_ids, payload, reason, dedupe_key) values (:id, 'parser_breakage', '{}', cast(:p as jsonb), :r, :d) on conflict (dedupe_key) do nothing""", id=new_id("review"), p=jsonb({"connector": self.name, "run_id": run_id, "records": ctx.stats.records, "expected_min": self.expected_min_records}), r=f"{self.name}: {ctx.stats.records} records < expected {self.expected_min_records} — connector failure suspected, nothing deleted", d=f"parser_breakage:{self.name}:{started.date()}") ctx.log.info("run finished", extra={"status": status, **{k: v for k, v in ctx.stats.__dict__.items() if k != "meta"}, "ms": int((time.perf_counter() - t0) * 1000)}) except Exception as exc: ctx.log.exception("run failed", extra={"error": str(exc)}) async with transaction() as conn: await self._record(conn, ctx, "failed", error=f"{exc.__class__.__name__}: {exc}"[:2000]) await execute(conn, "insert into connector_errors (connector_name, run_id, error_type, message) values (:n, :r, :t, :m)", n=self.name, r=run_id, t=exc.__class__.__name__, m=str(exc)[:4000]) row = await fetch_one(conn, """update connectors set consecutive_failures = consecutive_failures + 1, last_duration_ms = :d, health = 'failing', next_run_at = now() + make_interval(secs => least(interval_seconds, 3600)), updated_at = now() where name = :n returning consecutive_failures""", d=int((time.perf_counter() - t0) * 1000), n=self.name) if row and row["consecutive_failures"] >= CIRCUIT_FAILURES: until = datetime.now(UTC) + timedelta(seconds=CIRCUIT_COOLDOWN_S) await execute(conn, "update connectors set circuit_open_until = :u where name = :n", u=until, n=self.name) ctx.log.error("circuit opened", extra={"until": until.isoformat()}) raise return ctx async def _select_targets(self, targets: list[Target], only_urls: list[str]) -> list[Target]: """`--url` / single-document reprocess: keep the discovered targets for those URLs; for URLs `discover()` no longer lists, rebuild the Target from what the first fetch persisted in `documents.meta` (key, doc_type, meta, needs_llm, entity) so extractors that branch on `target.key`/`target.meta` behave exactly as they did originally.""" wanted = {canonicalize_url(u) for u in only_urls} selected = [t for t in targets if canonicalize_url(t.url) in wanted] missing = wanted - {canonicalize_url(t.url) for t in selected} if missing: async with transaction() as conn: for url in sorted(missing): doc = await fetch_one(conn, "select url, doc_type, meta, needs_llm, entity_id, priority from documents where url = :u", u=url) if not doc: selected.append(Target(url=url)) continue meta = dict(doc["meta"] or {}) stored = meta.pop("_target", {}) if isinstance(meta.get("_target"), dict) else {} entity = None if stored.get("entity"): try: from aiatlas.sdk.facts import _ref_from # noqa: PLC0415 entity = _ref_from(stored["entity"]) except Exception: # noqa: BLE001 entity = None elif doc["entity_id"]: row = await fetch_one(conn, "select entity_type, canonical_name from entities where id = :id", id=doc["entity_id"]) if row: entity = EntityRef(entity_type=row["entity_type"], name=row["canonical_name"], id=doc["entity_id"]) selected.append(Target(url=doc["url"], doc_type=stored.get("doc_type") or doc["doc_type"] or self.default_doc_type, entity=entity, meta=stored.get("meta") if isinstance(stored.get("meta"), dict) else meta, key=stored.get("key"), needs_llm=bool(stored.get("needs_llm", doc["needs_llm"])), priority=int(doc["priority"] or 2))) return selected @staticmethod def _target_meta(target: Target, doc_type: str) -> dict[str, Any]: """What `documents.meta` remembers about the target that produced the document (rebuilt by `_select_targets`).""" stored: dict[str, Any] = {"key": target.key, "doc_type": doc_type, "meta": _jsonable(target.meta), "needs_llm": bool(target.needs_llm)} if target.entity is not None: from aiatlas.sdk.facts import _encode # noqa: PLC0415 stored["entity"] = _encode(target.entity) return {**_jsonable(target.meta), "_target": stored} def _full_extraction(self, ctx: RunContext) -> bool: s = ctx.stats return s.docs_fetched > 0 and s.docs_changed >= s.docs_fetched - s.docs_failed and s.docs_unchanged == 0 def _final_status(self, ctx: RunContext) -> str: s = ctx.stats # Breakage detection only makes sense when every fetched document was (re)extracted: unchanged documents legitimately # produce zero records (304 / same hash), so an incremental run is never "suspect". full_extraction = self._full_extraction(ctx) if self.expected_min_records and full_extraction and s.records < self.expected_min_records and not ctx.reprocess: return "suspect" if s.docs_fetched and s.docs_failed == s.docs_fetched and s.docs_fetched > 0: return "failed" if s.docs_changed == 0 else "success" return "unchanged" if s.docs_changed == 0 else "success" async def _next_interval(self, conn: Any, changed: bool) -> int: row = await fetch_one(conn, "select interval_seconds, consecutive_unchanged, min_interval_seconds, max_interval_seconds from connectors where name = :n", n=self.name) if not row: return self.interval_seconds cur = row["interval_seconds"] or self.interval_seconds lo, hi = row["min_interval_seconds"] or self.min_interval_seconds, row["max_interval_seconds"] or self.max_interval_seconds if changed: return max(lo, int(cur * 0.7)) if row["consecutive_unchanged"] >= 3: return min(hi, int(cur * 1.5)) return cur async def _process_all(self, ctx: RunContext, targets: list[Target]) -> None: queue: asyncio.Queue[Target] = asyncio.Queue() for t in targets: queue.put_nowait(t) processed = 0 sem = asyncio.Semaphore(max(1, self.concurrency)) pending: set[asyncio.Task[None]] = set() async def worker(target: Target) -> None: nonlocal processed try: async with sem: followups = await self._process_target(ctx, target) except Exception as exc: ctx.stats.docs_failed += 1 ctx.log.exception("target failed", extra={"url": target.url}) try: async with transaction() as conn: await execute(conn, "insert into connector_errors (connector_name, run_id, url, error_type, message) values (:n, :r, :u, :t, :m)", n=self.name, r=ctx.run_id, u=target.url, t=exc.__class__.__name__, m=str(exc)[:2000]) except Exception: # noqa: BLE001 pass return for f in followups: cu = canonicalize_url(f.url) if cu not in ctx.seen_urls and processed + queue.qsize() < ctx.max_targets: queue.put_nowait(f) while not queue.empty() or pending: while not queue.empty() and processed < ctx.max_targets: target = queue.get_nowait() cu = canonicalize_url(target.url) if cu in ctx.seen_urls: continue ctx.seen_urls.add(cu) processed += 1 task = asyncio.create_task(worker(target)) pending.add(task) task.add_done_callback(pending.discard) if pending: await asyncio.wait(pending, return_when=asyncio.FIRST_COMPLETED) elif queue.empty(): break async def _process_target(self, ctx: RunContext, target: Target) -> list[Target]: url = canonicalize_url(target.url) doc_type = target.doc_type or self.default_doc_type async with transaction() as conn: doc = await fetch_one(conn, "select * from documents where url = :u", u=url) if doc is None: doc_id = new_id("document") await execute(conn, """insert into documents (id, source_id, connector_name, url, doc_type, priority, meta) values (:id, :s, :c, :u, :t, :p, cast(:m as jsonb)) on conflict (url) do nothing""", id=doc_id, s=self.source_id, c=self.name, u=url, t=doc_type, p=target.priority, m=jsonb(self._target_meta(target, doc_type))) doc = await fetch_one(conn, "select * from documents where url = :u", u=url) elif (target.key or target.meta) and not ctx.reprocess and (doc["meta"] or {}).get("_target", {}).get("key") != target.key: await execute(conn, "update documents set meta = meta || cast(:m as jsonb) where id = :id", m=jsonb(self._target_meta(target, doc_type)), id=doc["id"]) assert doc # ---- reprocess mode: replay latest snapshot without network if ctx.reprocess: async with transaction() as conn: snap = await fetch_one(conn, "select * from snapshots where document_id = :d and changed order by observed_at desc limit 1", d=doc["id"]) if not snap or not snap["raw_path"]: return [] res = FetchResult(url=url, final_url=snap["final_url"] or url, status=snap["http_status"] or 200, headers=snap["headers"] or {}, content=archive.load_raw(snap["raw_path"]), content_type=snap["content_type"] or "", fetched_at=snap["observed_at"], duration_ms=0, transport="file") return await self._extract_and_write(ctx, target, doc, snap["id"], res, first_time=False) # ---- fetch override = ctx.file_overrides.get(target.key or "") or ctx.file_overrides.get(url) or ctx.file_overrides.get(target.url) try: if override: res = file_result(override, url=url, content_type=target.meta.get("content_type", "text/html")) else: res = await ctx.fetcher.get(url, etag=doc["etag"] if not ctx.force else None, last_modified=doc["last_modified"] if not ctx.force else None, min_bytes=target.min_bytes, escalate=target.escalate, accept=target.accept, rate_per_min=target.rate_per_min) except NotModified: ctx.stats.docs_fetched += 1 ctx.stats.docs_unchanged += 1 async with transaction() as conn: await execute(conn, "update documents set last_fetched_at = now(), fetch_count = fetch_count + 1, last_status = 304, fail_count = 0 where id = :id", id=doc["id"]) return [] except BlockedError as exc: ctx.stats.docs_fetched += 1 ctx.stats.docs_failed += 1 ctx.log.warning("blocked", extra={"url": url, "error": str(exc)}) async with transaction() as conn: await execute(conn, "update documents set last_fetched_at = now(), fetch_count = fetch_count + 1, fail_count = fail_count + 1, last_status = :s, status = 'blocked' where id = :id", s=exc.status, id=doc["id"]) await execute(conn, "insert into connector_errors (connector_name, run_id, url, error_type, message) values (:n, :r, :u, 'blocked', :m)", n=self.name, r=ctx.run_id, u=url, m=str(exc)[:2000]) await execute(conn, """insert into review_queue (id, kind, entity_ids, payload, reason, dedupe_key) values (:id, 'blocked_source', '{}', cast(:p as jsonb), :r, :d) on conflict (dedupe_key) do nothing""", id=new_id("review"), p=jsonb({"url": url, "connector": self.name, "status": exc.status}), r=f"{self.name}: access denied for {url}", d=f"blocked:{url}") return [] except FetchError as exc: ctx.stats.docs_fetched += 1 ctx.stats.docs_failed += 1 ctx.log.warning("fetch failed", extra={"url": url, "error": str(exc)}) async with transaction() as conn: gone = exc.status in (404, 410) await execute(conn, """update documents set last_fetched_at = now(), fetch_count = fetch_count + 1, fail_count = fail_count + 1, last_status = :s, status = case when :gone and fail_count >= 2 then 'gone' else status end where id = :id""", s=exc.status, gone=gone, id=doc["id"]) await execute(conn, "insert into connector_errors (connector_name, run_id, url, error_type, message) values (:n, :r, :u, :t, :m)", n=self.name, r=ctx.run_id, u=url, t=exc.__class__.__name__, m=str(exc)[:2000]) return [] ctx.stats.docs_fetched += 1 changed = res.sha256 != doc["content_hash"] first_time = doc["content_hash"] is None if not changed and not ctx.force: ctx.stats.docs_unchanged += 1 async with transaction() as conn: await execute(conn, """update documents set last_fetched_at = now(), fetch_count = fetch_count + 1, last_status = :s, fail_count = 0, etag = coalesce(:etag, etag), last_modified = coalesce(:lm, last_modified) where id = :id""", s=res.status, etag=res.headers.get("etag"), lm=res.headers.get("last-modified"), id=doc["id"]) return [] # ---- changed (or forced): archive + snapshot + extract parsed = self.parse(target, res) raw_path = archive.store_raw(res.content, sha256=res.sha256) text_path, text_hash = archive.store_text(parsed.text) if parsed.text else (None, None) structured = parsed.structured() snapshot_id = new_id("snapshot") async with transaction() as conn: prev = await fetch_one(conn, "select structured, text_hash from snapshots where document_id = :d and changed order by observed_at desc limit 1", d=doc["id"]) diff = _structural_diff(prev["structured"] if prev else None, structured) if prev else None semantic_change = not prev or prev["text_hash"] != text_hash or bool(diff) await execute(conn, """insert into snapshots (id, document_id, run_id, url, final_url, observed_at, http_status, headers, content_type, content_hash, byte_size, raw_path, text_path, text_hash, structured, parser_version, connector_version, transport, changed, diff, processing_status) values (:id, :d, :run, :u, :fu, :o, :st, cast(:h as jsonb), :ct, :hash, :size, :rp, :tp, :th, cast(:s as jsonb), :pv, :cv, :tr, :changed, cast(:diff as jsonb), 'stored')""", id=snapshot_id, d=doc["id"], run=ctx.run_id, u=url, fu=res.final_url, o=res.fetched_at, st=res.status, h=jsonb({k: v for k, v in res.headers.items() if k in ("etag", "last-modified", "content-type", "content-length", "server", "cache-control", "date")}), ct=res.content_type[:200], hash=res.sha256, size=len(res.content), rp=raw_path, tp=text_path, th=text_hash, s=jsonb(structured) if structured else None, pv=self.parser_version, cv=self.version, tr=res.transport, changed=changed, diff=jsonb(diff) if diff else None) title = (parsed.html.title if parsed.html else None) or (parsed.feed_meta or {}).get("title") if parsed.kind == "feed" else (parsed.html.title if parsed.html else None) await execute(conn, """update documents set last_fetched_at = now(), last_changed_at = case when :changed then now() else last_changed_at end, fetch_count = fetch_count + 1, change_count = change_count + case when :changed and not :first then 1 else 0 end, last_status = :s, fail_count = 0, content_hash = :hash, etag = :etag, last_modified = :lm, canonical_url = coalesce(:canon, canonical_url), title = coalesce(:title, title), status = 'active', needs_llm = :llm, doc_type = :dt where id = :id""", changed=changed, first=first_time, s=res.status, hash=res.sha256, etag=res.headers.get("etag"), lm=res.headers.get("last-modified"), canon=(parsed.html.canonical if parsed.html else None), title=(title or None), llm=bool(target.needs_llm or self.needs_llm), dt=doc_type, id=doc["id"]) if changed and not first_time and semantic_change and doc_type not in ("feed", "sitemap", "listing"): await execute(conn, """insert into change_events (id, entity_id, event_type, category, summary, importance, observed_at, source_id, snapshot_id, source_url, connector_name, dedupe_key, meta) values (:id, :e, 'DOCUMENT_CHANGED', 'source', :s, 0, now(), :src, :snap, :u, :c, :d, cast(:m as jsonb)) on conflict (dedupe_key) do nothing""", id=new_id("change_event"), e=doc["entity_id"], s=f"{title or url} changed", src=self.source_id, snap=snapshot_id, u=url, c=self.name, d=f"DOCUMENT_CHANGED:{snapshot_id}", m=jsonb({"doc_type": doc_type, "diff_keys": list(diff)[:20] if diff else []})) if changed: ctx.stats.docs_changed += 1 return await self._extract_and_write(ctx, target, doc, snapshot_id, res, first_time=first_time, parsed=parsed) async def _extract_and_write(self, ctx: RunContext, target: Target, doc: dict[str, Any], snapshot_id: str, res: FetchResult, *, first_time: bool, parsed: Parsed | None = None) -> list[Target]: parsed = parsed or self.parse(target, res) try: facts = await self.extract(ctx, target, res, parsed) except Exception as exc: ctx.log.exception("extract failed", extra={"url": res.url}) async with transaction() as conn: await execute(conn, "insert into connector_errors (connector_name, run_id, url, error_type, message) values (:n, :r, :u, :t, :m)", n=self.name, r=ctx.run_id, u=res.url, t=f"extract:{exc.__class__.__name__}", m=str(exc)[:2000]) await execute(conn, "update snapshots set processing_status = 'failed' where id = :id", id=snapshot_id) return [] if facts is None: return [] item = Buffered(target=target, doc_id=doc["id"], snapshot_id=snapshot_id, source_url=res.final_url or res.url, fetched_at=res.fetched_at, facts=facts) ctx.stats.records += len(facts.entities) + len(facts.prices) + len(facts.results) if self.quarantine_enabled and not ctx.reprocess: ctx.buffered.append(item) # written (or quarantined) at the end of the run, see _flush_or_quarantine else: await self._write_buffered(ctx, item) return facts.targets async def _write_buffered(self, ctx: RunContext, item: Buffered) -> None: facts, target, snapshot_id = item.facts, item.target, item.snapshot_id async with transaction() as conn: writer = FactWriter(conn, source_id=self.source_id, snapshot_id=snapshot_id, source_url=item.source_url, tier=self.tier, connector_name=self.name, extractor="deterministic", extractor_version=self.parser_version, observed_at=item.fetched_at, run_id=ctx.run_id, source_key=ctx.source_key, is_first_run=ctx.is_first_run) ws = await writer.write(facts) main = facts.document_entity or target.entity if main and main.id is None: await writer.resolver.resolve(main) if main and main.id: await execute(conn, "update documents set entity_id = coalesce(entity_id, :e), title = coalesce(:t, title) where id = :id", e=main.id, t=facts.document_title, id=item.doc_id) await execute(conn, "update snapshots set processing_status = :st where id = :id", st="llm_pending" if (target.needs_llm or self.needs_llm or facts.llm_hint) else "extracted", id=snapshot_id) if target.needs_llm or self.needs_llm or facts.llm_hint: from aiatlas.services.jobs import enqueue await enqueue(conn, "llm_extract", {"snapshot_id": snapshot_id, "task": facts.llm_hint or target.meta.get("llm_task") or "auto", "entity_id": main.id if main else None, "connector": self.name}, priority=4, dedupe_key=f"llm_extract:{snapshot_id}") s = ctx.stats s.entities_created += ws.entities_created s.entities_updated += ws.entities_updated s.claims_written += ws.claims s.relations_written += ws.relations s.events_emitted += ws.events # ------------------------------------------------------------------------------------------ quarantine async def _observed_counts(self, ctx: RunContext) -> dict[str, int]: """Counts of what the buffered facts would write: entity references, prices, results and *new* entity names (refs that resolve to nothing today). Resolution runs in a transaction that is rolled back so the check leaves no trace.""" refs: dict[str, EntityRef] = {} prices = results = 0 for item in ctx.buffered: prices += len(item.facts.prices) results += len(item.facts.results) for ref in item.facts.entities: refs.setdefault(ref.key(), ref) new_entities = 0 async with connection() as conn: trans = await conn.begin() try: resolver = Resolver(conn, source_tier=self.tier) for ref in refs.values(): probe = EntityRef(entity_type=ref.entity_type, name=ref.name, identifiers=dict(ref.identifiers), aliases=list(ref.aliases), slug_hint=ref.slug_hint) if await resolver.resolve(probe, create=False) is None: new_entities += 1 finally: await trans.rollback() return {"entities": len(refs), "prices": prices, "results": results, "new_entities": new_entities} @staticmethod def _quarantine_reason(observed: dict[str, int], baseline: dict[str, Any] | None, *, full_extraction: bool) -> str | None: """Explosion rules always apply; collapse rules only when every document was re-extracted (an incremental run legitimately yields little).""" if not baseline or not baseline.get("history"): return None med = baseline.get("medians") or {} b_new, b_ent, b_pr, b_res = med.get("new_entities", 0), med.get("entities", 0), med.get("prices", 0), med.get("results", 0) if observed["new_entities"] > max(QUARANTINE_MIN_NEW_ENTITIES, 1.3 * b_new): return f"new entities {observed['new_entities']} > max({QUARANTINE_MIN_NEW_ENTITIES}, 1.3 × baseline {b_new:g})" if b_pr > 0 and observed["prices"] > 3 * b_pr: return f"price rows {observed['prices']} > 3 × baseline {b_pr:g}" if full_extraction: if b_ent > 0 and observed["entities"] < 0.8 * b_ent: return f"entity references {observed['entities']} < 0.8 × baseline {b_ent:g}" if b_res > 0 and observed["results"] < 0.5 * b_res: return f"benchmark results {observed['results']} < 0.5 × baseline {b_res:g}" return None @staticmethod def _next_baseline(baseline: dict[str, Any] | None, observed: dict[str, int]) -> dict[str, Any]: history = list((baseline or {}).get("history") or []) history.append({k: int(observed[k]) for k in ("entities", "prices", "results", "new_entities")}) history = history[-BASELINE_RUNS:] medians = {k: float(statistics.median(h[k] for h in history)) for k in ("entities", "prices", "results", "new_entities")} return {"history": history, "medians": medians, "runs": len(history), "updated_at": datetime.now(UTC).isoformat(timespec="seconds")} async def _flush_or_quarantine(self, ctx: RunContext) -> None: observed = await self._observed_counts(ctx) full = self._full_extraction(ctx) async with transaction() as conn: row = await fetch_one(conn, "select baseline from connectors where name = :n", n=self.name) baseline = (row or {}).get("baseline") reason = None if ctx.force and ctx.file_overrides else self._quarantine_reason(observed, baseline, full_extraction=full) ctx.stats.meta["quarantine_check"] = {"observed": observed, "baseline": (baseline or {}).get("medians"), "full_extraction": full} if reason: ctx.quarantined = True ctx.quarantine_reason = reason ctx.log.warning("run quarantined", extra={"reason": reason, **observed}) payload = [{"doc_id": b.doc_id, "snapshot_id": b.snapshot_id, "source_url": b.source_url, "fetched_at": b.fetched_at.isoformat(), "target": {"url": b.target.url, "doc_type": b.target.doc_type, "key": b.target.key, "needs_llm": b.target.needs_llm, "meta": _jsonable(b.target.meta)}, "facts": facts_to_json(b.facts)} for b in ctx.buffered] qid = f"quar_{new_id('connector_run').split('_', 1)[1]}" async with transaction() as conn: await execute(conn, """insert into quarantined_runs (id, run_id, connector_name, reason, stats, facts) values (:id, :r, :c, :why, cast(:s as jsonb), cast(:f as jsonb))""", id=qid, r=ctx.run_id, c=self.name, why=reason, s=jsonb({"observed": observed, "baseline": baseline, "full_extraction": full, "documents": len(ctx.buffered)}), f=jsonb(payload)) await execute(conn, "update connector_runs set quarantined = true, baseline = cast(:b as jsonb) where id = :id", b=jsonb(baseline), id=ctx.run_id) await execute(conn, "update snapshots set processing_status = 'quarantined' where id = any(cast(:ids as text[]))", ids=[b.snapshot_id for b in ctx.buffered]) await execute(conn, """insert into review_queue (id, kind, entity_ids, payload, reason, dedupe_key) values (:id, 'quarantine', '{}', cast(:p as jsonb), :r, :d) on conflict (dedupe_key) do nothing""", id=new_id("review"), p=jsonb({"connector": self.name, "run_id": ctx.run_id, "quarantine_id": qid, "observed": observed, "baseline": (baseline or {}).get("medians")}), r=f"{self.name}: run held in quarantine — {reason}; nothing written, nothing deleted", d=f"quarantine:{qid}") ctx.buffered.clear() return for item in ctx.buffered: await self._write_buffered(ctx, item) ctx.buffered.clear() if full and not ctx.reprocess and not ctx.file_overrides: async with transaction() as conn: nb = self._next_baseline(baseline, observed) await execute(conn, "update connectors set baseline = cast(:b as jsonb) where name = :n", b=jsonb(nb), n=self.name) await execute(conn, "update connector_runs set baseline = cast(:b as jsonb) where id = :id", b=jsonb(nb), id=ctx.run_id) async def _record(self, conn: Any, ctx: RunContext, status: str, *, error: str | None = None) -> None: s = ctx.stats await execute(conn, """insert into connector_runs (id, connector_name, started_at, finished_at, status, duration_ms, docs_discovered, docs_fetched, docs_changed, docs_unchanged, docs_failed, entities_created, entities_updated, claims_written, relations_written, events_emitted, error, meta) values (:id, :c, :started, now(), :status, :dur, :dd, :df, :dc, :du, :dfail, :ec, :eu, :cw, :rw, :ee, :err, cast(:meta as jsonb)) on conflict (id) do update set finished_at = now(), status = excluded.status, duration_ms = excluded.duration_ms, docs_discovered = excluded.docs_discovered, docs_fetched = excluded.docs_fetched, docs_changed = excluded.docs_changed, docs_unchanged = excluded.docs_unchanged, docs_failed = excluded.docs_failed, entities_created = excluded.entities_created, entities_updated = excluded.entities_updated, claims_written = excluded.claims_written, relations_written = excluded.relations_written, events_emitted = excluded.events_emitted, error = excluded.error, meta = excluded.meta""", id=ctx.run_id, c=self.name, started=ctx.started_at, status=status, dur=int((datetime.now(UTC) - ctx.started_at).total_seconds() * 1000), dd=s.docs_discovered, df=s.docs_fetched, dc=s.docs_changed, du=s.docs_unchanged, dfail=s.docs_failed, ec=s.entities_created, eu=s.entities_updated, cw=s.claims_written, rw=s.relations_written, ee=s.events_emitted, err=error, meta=jsonb({**s.meta, "records": s.records})) def _structural_diff(prev: dict[str, Any] | None, cur: dict[str, Any] | None) -> dict[str, Any] | None: if not prev or not cur: return None diff: dict[str, Any] = {} for key in ("title", "description", "headings", "tables", "json_ld", "front_matter", "published_at", "modified_at", "items"): a, b = prev.get(key), cur.get(key) if a != b: if isinstance(a, list) and isinstance(b, list): sa = {json.dumps(x, sort_keys=True, default=str) for x in a} sb = {json.dumps(x, sort_keys=True, default=str) for x in b} added = [json.loads(x) for x in list(sb - sa)[:50]] removed = [json.loads(x) for x in list(sa - sb)[:50]] if added or removed: diff[key] = {"added": added, "removed": removed} else: diff[key] = {"old": a, "new": b} return diff or None def _pdf_text(content: bytes) -> str: try: import io from pypdf import PdfReader reader = PdfReader(io.BytesIO(content)) parts = [] for page in reader.pages[:60]: try: parts.append(page.extract_text() or "") except Exception: # noqa: BLE001 continue return "\n\n".join(parts).strip() except Exception: # noqa: BLE001 return "" def _jsonable(obj: Any) -> Any: try: json.dumps(obj) return obj except (TypeError, ValueError): return json.loads(json.dumps(obj, default=str)) __all__ = ["BaseConnector", "ConnectorError", "Parsed", "RunContext", "RunStats", "Target"]