HTML 77.2%
TypeScript 10.5%
Python 9.6%
JavaScript 2.5%
1"""BaseConnector — the lifecycle every connector shares:23 discover() → fetch() → parse() → extract() → diff/write → schedule_next()45Subclasses usually implement `discover()` (seed URLs / feeds / sitemaps) and `extract()` (structured facts from a parsed document).6`run()` does all bookkeeping: connector_runs, documents, snapshots (raw archive), content-hash change detection, structural diffs,7DOCUMENT_CHANGED events, breakage detection (record count collapse ≠ deletion), circuit breaker and adaptive scheduling.8"""9from __future__ import annotations1011import asyncio12import json13import logging14import statistics15import time16from dataclasses import dataclass, field17from datetime import UTC, datetime, timedelta18from typing import Any1920from aiatlas.db import connection, execute, fetch_one, jsonb, transaction21from aiatlas.ids import new_id22from aiatlas.sdk import archive23from aiatlas.sdk.extract.feeds import FeedItem, parse_feed24from aiatlas.sdk.extract.html import HtmlDoc, parse_html25from aiatlas.sdk.extract.markdown import MarkdownDoc, parse_markdown26from aiatlas.sdk.facts import EntityRef, Facts, Target, facts_to_json27from aiatlas.sdk.fetch import BlockedError, Fetcher, FetchError, FetchResult, NotModified, canonicalize_url, file_result28from aiatlas.sdk.resolution import Resolver29from aiatlas.sdk.writer import FactWriter3031log = logging.getLogger(__name__)3233CIRCUIT_FAILURES = 334CIRCUIT_COOLDOWN_S = 180035BASELINE_RUNS = 5 # rolling window (median) of full-extraction runs used by the quarantine detector36QUARANTINE_MIN_NEW_ENTITIES = 50 # a run may always create up to this many new entities without tripping the detector373839class ConnectorError(Exception):40 pass414243@dataclass44class Parsed:45 kind: str # html|markdown|feed|json|pdf|text|xml46 html: HtmlDoc | None = None47 markdown: MarkdownDoc | None = None48 feed_meta: dict[str, Any] | None = None49 feed_items: list[FeedItem] = field(default_factory=list)50 json: Any = None51 text: str = ""5253 def structured(self) -> dict[str, Any] | None:54 if self.html:55 return self.html.structured()56 if self.markdown:57 return {"front_matter": _jsonable(self.markdown.front_matter), "headings": self.markdown.headings[:200], "tables": self.markdown.tables[:40],58 "link_count": len(self.markdown.links), "text_length": len(self.markdown.text)}59 if self.kind == "feed":60 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}61 for i in self.feed_items[:200]]}62 if self.kind == "json":63 return {"json_keys": list(self.json)[:100] if isinstance(self.json, dict) else None,64 "json_length": len(self.json) if isinstance(self.json, (list, dict)) else None}65 return {"text_length": len(self.text)} if self.text else None666768@dataclass69class RunStats:70 docs_discovered: int = 071 docs_fetched: int = 072 docs_changed: int = 073 docs_unchanged: int = 074 docs_failed: int = 075 entities_created: int = 076 entities_updated: int = 077 claims_written: int = 078 relations_written: int = 079 events_emitted: int = 080 records: int = 0 # connector-defined "records" for breakage detection (models, papers, prices…)81 meta: dict[str, Any] = field(default_factory=dict)828384@dataclass85class Buffered:86 """Facts extracted from one document, held until the end of the run when the connector is quarantine-capable."""87 target: Target88 doc_id: str89 snapshot_id: str90 source_url: str91 fetched_at: datetime92 facts: Facts939495@dataclass96class RunContext:97 run_id: str98 connector: BaseConnector99 fetcher: Fetcher100 started_at: datetime101 force: bool = False102 reprocess: bool = False # re-extract from stored snapshots without fetching103 file_overrides: dict[str, str] = field(default_factory=dict) # target.key or url → local path104 max_targets: int = 2000105 stats: RunStats = field(default_factory=RunStats)106 seen_urls: set[str] = field(default_factory=set)107 is_first_run: bool = False # no earlier successful run → every event of this run is backfill108 source_key: str | None = None109 buffered: list[Buffered] = field(default_factory=list) # quarantine mode: facts held until the run-end anomaly check110 quarantined: bool = False111 quarantine_reason: str | None = None112 log: logging.LoggerAdapter[logging.Logger] = field(init=False)113114 def __post_init__(self) -> None:115 self.log = logging.LoggerAdapter(logging.getLogger(f"connector.{self.connector.name}"), {"connector": self.connector.name, "run_id": self.run_id})116117118class BaseConnector:119 # ------------------------------------------------------------------------------------------ identity & policy120 name: str = ""121 label: str = ""122 description: str = ""123 source_key: str = "" # sources.key (seeded from registry/sources.yaml)124 version: str = "1" # connector version125 parser_version: str = "1" # bump when extraction improves → `aia reprocess <connector>` replays snapshots126 interval_seconds: int = 3600127 min_interval_seconds: int = 900128 max_interval_seconds: int = 7 * 86400129 rate_per_min: int = 30130 respect_robots: bool = True131 expected_min_records: int = 0132 priority: int = 2133 tier: int = 1134 concurrency: int = 3135 default_doc_type: str = "page"136 needs_llm: bool = False # queue documents for LLM extraction after deterministic extraction137 # Hold every fact until the end of the run and compare the run's counts with the connector's rolling baseline before writing;138 # a collapse or an explosion parks the whole run in `quarantined_runs` for review. None = default (tier ≥ 2 hub/leaderboard/registry139 # connectors are quarantine-capable; tier-1 lab documentation is not — it is the primary source).140 quarantine: bool | None = None141142 def __init__(self, config: dict[str, Any] | None = None):143 self.config = config or {}144 self.source_id: str | None = None145146 @property147 def quarantine_enabled(self) -> bool:148 if self.quarantine is not None:149 return self.quarantine150 return self.tier >= 2151152 # ------------------------------------------------------------------------------------------ to implement153 async def discover(self, ctx: RunContext) -> list[Target]:154 """Return the targets for this run (seed pages, feeds, sitemaps, org pages…)."""155 seeds = self.config.get("seeds") or getattr(self, "seeds", [])156 return [Target(url=s) if isinstance(s, str) else Target(**s) for s in seeds]157158 async def extract(self, ctx: RunContext, target: Target, res: FetchResult, parsed: Parsed) -> Facts:159 """Turn a parsed document into facts. Deterministic only; LLM extraction is queued separately."""160 return Facts()161162 def parse(self, target: Target, res: FetchResult) -> Parsed:163 """Default parser by content type; connectors may override for special formats."""164 if res.is_pdf:165 return Parsed(kind="pdf", text=_pdf_text(res.content))166 if target.doc_type == "feed" or (res.is_xml and b"<rss" in res.content[:2000] or b"<feed" in res.content[:2000]):167 meta, items = parse_feed(res.content)168 if items or meta.get("title"):169 return Parsed(kind="feed", feed_meta=meta, feed_items=items)170 if res.is_json:171 try:172 return Parsed(kind="json", json=res.json())173 except Exception: # noqa: BLE001174 pass175 if res.is_html:176 doc = parse_html(res.content, res.final_url or res.url)177 return Parsed(kind="html", html=doc, text=doc.text)178 text = res.text179 if target.doc_type in ("model_card", "readme", "markdown") or "markdown" in res.content_type or res.url.endswith((".md", ".mdx")):180 md = parse_markdown(text)181 return Parsed(kind="markdown", markdown=md, text=md.text)182 if res.is_xml:183 return Parsed(kind="xml", text=text)184 return Parsed(kind="text", text=text)185186 async def healthcheck(self) -> bool:187 return True188189 # ------------------------------------------------------------------------------------------ orchestration190 async def run(self, *, force: bool = False, reprocess: bool = False, file_overrides: dict[str, str] | None = None,191 max_targets: int | None = None, only_urls: list[str] | None = None) -> RunContext:192 started = datetime.now(UTC)193 run_id = new_id("connector_run")194 async with Fetcher(rate_per_min=self.rate_per_min, robots=self.respect_robots) as fetcher:195 ctx = RunContext(run_id=run_id, connector=self, fetcher=fetcher, started_at=started, force=force, reprocess=reprocess,196 file_overrides=file_overrides or {}, max_targets=max_targets or int(self.config.get("max_targets", 2000)))197 async with transaction() as conn:198 state = await fetch_one(conn, """select c.source_id, c.enabled, c.circuit_open_until, s.key as source_key,199 exists(select 1 from connector_runs r where r.connector_name = c.name and r.status in ('success','unchanged','suspect')) as has_success200 from connectors c left join sources s on s.id = c.source_id where c.name = :n""", n=self.name)201 if state is None:202 raise ConnectorError(f"connector {self.name!r} not registered (run `aia seed`)")203 self.source_id = state["source_id"]204 ctx.source_key = state["source_key"] or self.source_key or None205 ctx.is_first_run = not state["has_success"]206 if not state["enabled"] and not force:207 ctx.log.info("connector disabled, skipping")208 await self._record(conn, ctx, "skipped", error="disabled")209 return ctx210 if state["circuit_open_until"] and state["circuit_open_until"] > started and not force:211 ctx.log.warning("circuit open, skipping", extra={"until": state["circuit_open_until"].isoformat()})212 await self._record(conn, ctx, "skipped", error="circuit open")213 return ctx214 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)215 await execute(conn, "update connectors set last_attempt_at = :t, health = case when health = 'unknown' then 'degraded' else health end where name = :n",216 t=started, n=self.name)217 t0 = time.perf_counter()218 try:219 targets = await self.discover(ctx)220 if only_urls:221 targets = await self._select_targets(targets, only_urls)222 ctx.stats.docs_discovered = len(targets)223 await self._process_all(ctx, targets)224 if ctx.buffered:225 await self._flush_or_quarantine(ctx)226 status = "quarantined" if ctx.quarantined else self._final_status(ctx)227 async with transaction() as conn:228 await self._record(conn, ctx, status)229 changed = ctx.stats.docs_changed > 0230 await execute(conn, """update connectors set consecutive_failures = 0, circuit_open_until = null,231 last_success_at = case when cast(:q as boolean) then last_success_at else now() end,232 last_change_at = case when cast(:changed as boolean) and not cast(:q as boolean) then now() else last_change_at end,233 consecutive_unchanged = case when cast(:changed as boolean) then 0 else consecutive_unchanged + 1 end,234 interval_seconds = cast(:interval as integer), next_run_at = now() + make_interval(secs => cast(:interval as double precision)),235 last_duration_ms = :d, health = :health, updated_at = now() where name = :n""",236 changed=changed, q=ctx.quarantined, interval=await self._next_interval(conn, changed), d=int((time.perf_counter() - t0) * 1000),237 health="degraded" if status in ("suspect", "quarantined") else "ok", n=self.name)238 if status == "suspect":239 await execute(conn, """insert into review_queue (id, kind, entity_ids, payload, reason, dedupe_key)240 values (:id, 'parser_breakage', '{}', cast(:p as jsonb), :r, :d) on conflict (dedupe_key) do nothing""",241 id=new_id("review"), p=jsonb({"connector": self.name, "run_id": run_id, "records": ctx.stats.records, "expected_min": self.expected_min_records}),242 r=f"{self.name}: {ctx.stats.records} records < expected {self.expected_min_records} — connector failure suspected, nothing deleted",243 d=f"parser_breakage:{self.name}:{started.date()}")244 ctx.log.info("run finished", extra={"status": status, **{k: v for k, v in ctx.stats.__dict__.items() if k != "meta"},245 "ms": int((time.perf_counter() - t0) * 1000)})246 except Exception as exc:247 ctx.log.exception("run failed", extra={"error": str(exc)})248 async with transaction() as conn:249 await self._record(conn, ctx, "failed", error=f"{exc.__class__.__name__}: {exc}"[:2000])250 await execute(conn, "insert into connector_errors (connector_name, run_id, error_type, message) values (:n, :r, :t, :m)",251 n=self.name, r=run_id, t=exc.__class__.__name__, m=str(exc)[:4000])252 row = await fetch_one(conn, """update connectors set consecutive_failures = consecutive_failures + 1, last_duration_ms = :d, health = 'failing',253 next_run_at = now() + make_interval(secs => least(interval_seconds, 3600)), updated_at = now()254 where name = :n returning consecutive_failures""", d=int((time.perf_counter() - t0) * 1000), n=self.name)255 if row and row["consecutive_failures"] >= CIRCUIT_FAILURES:256 until = datetime.now(UTC) + timedelta(seconds=CIRCUIT_COOLDOWN_S)257 await execute(conn, "update connectors set circuit_open_until = :u where name = :n", u=until, n=self.name)258 ctx.log.error("circuit opened", extra={"until": until.isoformat()})259 raise260 return ctx261262 async def _select_targets(self, targets: list[Target], only_urls: list[str]) -> list[Target]:263 """`--url` / single-document reprocess: keep the discovered targets for those URLs; for URLs `discover()` no longer lists, rebuild264 the Target from what the first fetch persisted in `documents.meta` (key, doc_type, meta, needs_llm, entity) so extractors that265 branch on `target.key`/`target.meta` behave exactly as they did originally."""266 wanted = {canonicalize_url(u) for u in only_urls}267 selected = [t for t in targets if canonicalize_url(t.url) in wanted]268 missing = wanted - {canonicalize_url(t.url) for t in selected}269 if missing:270 async with transaction() as conn:271 for url in sorted(missing):272 doc = await fetch_one(conn, "select url, doc_type, meta, needs_llm, entity_id, priority from documents where url = :u", u=url)273 if not doc:274 selected.append(Target(url=url))275 continue276 meta = dict(doc["meta"] or {})277 stored = meta.pop("_target", {}) if isinstance(meta.get("_target"), dict) else {}278 entity = None279 if stored.get("entity"):280 try:281 from aiatlas.sdk.facts import _ref_from # noqa: PLC0415282283 entity = _ref_from(stored["entity"])284 except Exception: # noqa: BLE001285 entity = None286 elif doc["entity_id"]:287 row = await fetch_one(conn, "select entity_type, canonical_name from entities where id = :id", id=doc["entity_id"])288 if row:289 entity = EntityRef(entity_type=row["entity_type"], name=row["canonical_name"], id=doc["entity_id"])290 selected.append(Target(url=doc["url"], doc_type=stored.get("doc_type") or doc["doc_type"] or self.default_doc_type, entity=entity,291 meta=stored.get("meta") if isinstance(stored.get("meta"), dict) else meta, key=stored.get("key"),292 needs_llm=bool(stored.get("needs_llm", doc["needs_llm"])), priority=int(doc["priority"] or 2)))293 return selected294295 @staticmethod296 def _target_meta(target: Target, doc_type: str) -> dict[str, Any]:297 """What `documents.meta` remembers about the target that produced the document (rebuilt by `_select_targets`)."""298 stored: dict[str, Any] = {"key": target.key, "doc_type": doc_type, "meta": _jsonable(target.meta), "needs_llm": bool(target.needs_llm)}299 if target.entity is not None:300 from aiatlas.sdk.facts import _encode # noqa: PLC0415301302 stored["entity"] = _encode(target.entity)303 return {**_jsonable(target.meta), "_target": stored}304305 def _full_extraction(self, ctx: RunContext) -> bool:306 s = ctx.stats307 return s.docs_fetched > 0 and s.docs_changed >= s.docs_fetched - s.docs_failed and s.docs_unchanged == 0308309 def _final_status(self, ctx: RunContext) -> str:310 s = ctx.stats311 # Breakage detection only makes sense when every fetched document was (re)extracted: unchanged documents legitimately312 # produce zero records (304 / same hash), so an incremental run is never "suspect".313 full_extraction = self._full_extraction(ctx)314 if self.expected_min_records and full_extraction and s.records < self.expected_min_records and not ctx.reprocess:315 return "suspect"316 if s.docs_fetched and s.docs_failed == s.docs_fetched and s.docs_fetched > 0:317 return "failed" if s.docs_changed == 0 else "success"318 return "unchanged" if s.docs_changed == 0 else "success"319320 async def _next_interval(self, conn: Any, changed: bool) -> int:321 row = await fetch_one(conn, "select interval_seconds, consecutive_unchanged, min_interval_seconds, max_interval_seconds from connectors where name = :n", n=self.name)322 if not row:323 return self.interval_seconds324 cur = row["interval_seconds"] or self.interval_seconds325 lo, hi = row["min_interval_seconds"] or self.min_interval_seconds, row["max_interval_seconds"] or self.max_interval_seconds326 if changed:327 return max(lo, int(cur * 0.7))328 if row["consecutive_unchanged"] >= 3:329 return min(hi, int(cur * 1.5))330 return cur331332 async def _process_all(self, ctx: RunContext, targets: list[Target]) -> None:333 queue: asyncio.Queue[Target] = asyncio.Queue()334 for t in targets:335 queue.put_nowait(t)336 processed = 0337 sem = asyncio.Semaphore(max(1, self.concurrency))338 pending: set[asyncio.Task[None]] = set()339340 async def worker(target: Target) -> None:341 nonlocal processed342 try:343 async with sem:344 followups = await self._process_target(ctx, target)345 except Exception as exc:346 ctx.stats.docs_failed += 1347 ctx.log.exception("target failed", extra={"url": target.url})348 try:349 async with transaction() as conn:350 await execute(conn, "insert into connector_errors (connector_name, run_id, url, error_type, message) values (:n, :r, :u, :t, :m)",351 n=self.name, r=ctx.run_id, u=target.url, t=exc.__class__.__name__, m=str(exc)[:2000])352 except Exception: # noqa: BLE001353 pass354 return355 for f in followups:356 cu = canonicalize_url(f.url)357 if cu not in ctx.seen_urls and processed + queue.qsize() < ctx.max_targets:358 queue.put_nowait(f)359360 while not queue.empty() or pending:361 while not queue.empty() and processed < ctx.max_targets:362 target = queue.get_nowait()363 cu = canonicalize_url(target.url)364 if cu in ctx.seen_urls:365 continue366 ctx.seen_urls.add(cu)367 processed += 1368 task = asyncio.create_task(worker(target))369 pending.add(task)370 task.add_done_callback(pending.discard)371 if pending:372 await asyncio.wait(pending, return_when=asyncio.FIRST_COMPLETED)373 elif queue.empty():374 break375376 async def _process_target(self, ctx: RunContext, target: Target) -> list[Target]:377 url = canonicalize_url(target.url)378 doc_type = target.doc_type or self.default_doc_type379 async with transaction() as conn:380 doc = await fetch_one(conn, "select * from documents where url = :u", u=url)381 if doc is None:382 doc_id = new_id("document")383 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))384 on conflict (url) do nothing""", id=doc_id, s=self.source_id, c=self.name, u=url, t=doc_type, p=target.priority,385 m=jsonb(self._target_meta(target, doc_type)))386 doc = await fetch_one(conn, "select * from documents where url = :u", u=url)387 elif (target.key or target.meta) and not ctx.reprocess and (doc["meta"] or {}).get("_target", {}).get("key") != target.key:388 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"])389 assert doc390 # ---- reprocess mode: replay latest snapshot without network391 if ctx.reprocess:392 async with transaction() as conn:393 snap = await fetch_one(conn, "select * from snapshots where document_id = :d and changed order by observed_at desc limit 1", d=doc["id"])394 if not snap or not snap["raw_path"]:395 return []396 res = FetchResult(url=url, final_url=snap["final_url"] or url, status=snap["http_status"] or 200, headers=snap["headers"] or {},397 content=archive.load_raw(snap["raw_path"]), content_type=snap["content_type"] or "", fetched_at=snap["observed_at"], duration_ms=0,398 transport="file")399 return await self._extract_and_write(ctx, target, doc, snap["id"], res, first_time=False)400 # ---- fetch401 override = ctx.file_overrides.get(target.key or "") or ctx.file_overrides.get(url) or ctx.file_overrides.get(target.url)402 try:403 if override:404 res = file_result(override, url=url, content_type=target.meta.get("content_type", "text/html"))405 else:406 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,407 min_bytes=target.min_bytes, escalate=target.escalate, accept=target.accept, rate_per_min=target.rate_per_min)408 except NotModified:409 ctx.stats.docs_fetched += 1410 ctx.stats.docs_unchanged += 1411 async with transaction() as conn:412 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"])413 return []414 except BlockedError as exc:415 ctx.stats.docs_fetched += 1416 ctx.stats.docs_failed += 1417 ctx.log.warning("blocked", extra={"url": url, "error": str(exc)})418 async with transaction() as conn:419 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",420 s=exc.status, id=doc["id"])421 await execute(conn, "insert into connector_errors (connector_name, run_id, url, error_type, message) values (:n, :r, :u, 'blocked', :m)",422 n=self.name, r=ctx.run_id, u=url, m=str(exc)[:2000])423 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)424 on conflict (dedupe_key) do nothing""", id=new_id("review"), p=jsonb({"url": url, "connector": self.name, "status": exc.status}),425 r=f"{self.name}: access denied for {url}", d=f"blocked:{url}")426 return []427 except FetchError as exc:428 ctx.stats.docs_fetched += 1429 ctx.stats.docs_failed += 1430 ctx.log.warning("fetch failed", extra={"url": url, "error": str(exc)})431 async with transaction() as conn:432 gone = exc.status in (404, 410)433 await execute(conn, """update documents set last_fetched_at = now(), fetch_count = fetch_count + 1, fail_count = fail_count + 1, last_status = :s,434 status = case when :gone and fail_count >= 2 then 'gone' else status end where id = :id""", s=exc.status, gone=gone, id=doc["id"])435 await execute(conn, "insert into connector_errors (connector_name, run_id, url, error_type, message) values (:n, :r, :u, :t, :m)",436 n=self.name, r=ctx.run_id, u=url, t=exc.__class__.__name__, m=str(exc)[:2000])437 return []438 ctx.stats.docs_fetched += 1439 changed = res.sha256 != doc["content_hash"]440 first_time = doc["content_hash"] is None441 if not changed and not ctx.force:442 ctx.stats.docs_unchanged += 1443 async with transaction() as conn:444 await execute(conn, """update documents set last_fetched_at = now(), fetch_count = fetch_count + 1, last_status = :s, fail_count = 0,445 etag = coalesce(:etag, etag), last_modified = coalesce(:lm, last_modified) where id = :id""",446 s=res.status, etag=res.headers.get("etag"), lm=res.headers.get("last-modified"), id=doc["id"])447 return []448 # ---- changed (or forced): archive + snapshot + extract449 parsed = self.parse(target, res)450 raw_path = archive.store_raw(res.content, sha256=res.sha256)451 text_path, text_hash = archive.store_text(parsed.text) if parsed.text else (None, None)452 structured = parsed.structured()453 snapshot_id = new_id("snapshot")454 async with transaction() as conn:455 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"])456 diff = _structural_diff(prev["structured"] if prev else None, structured) if prev else None457 semantic_change = not prev or prev["text_hash"] != text_hash or bool(diff)458 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,459 raw_path, text_path, text_hash, structured, parser_version, connector_version, transport, changed, diff, processing_status)460 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')""",461 id=snapshot_id, d=doc["id"], run=ctx.run_id, u=url, fu=res.final_url, o=res.fetched_at, st=res.status,462 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")}),463 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,464 pv=self.parser_version, cv=self.version, tr=res.transport, changed=changed, diff=jsonb(diff) if diff else None)465 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)466 await execute(conn, """update documents set last_fetched_at = now(), last_changed_at = case when :changed then now() else last_changed_at end,467 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,468 content_hash = :hash, etag = :etag, last_modified = :lm, canonical_url = coalesce(:canon, canonical_url), title = coalesce(:title, title),469 status = 'active', needs_llm = :llm, doc_type = :dt where id = :id""",470 changed=changed, first=first_time, s=res.status, hash=res.sha256, etag=res.headers.get("etag"), lm=res.headers.get("last-modified"),471 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"])472 if changed and not first_time and semantic_change and doc_type not in ("feed", "sitemap", "listing"):473 await execute(conn, """insert into change_events (id, entity_id, event_type, category, summary, importance, observed_at, source_id, snapshot_id, source_url,474 connector_name, dedupe_key, meta) values (:id, :e, 'DOCUMENT_CHANGED', 'source', :s, 0, now(), :src, :snap, :u, :c, :d, cast(:m as jsonb))475 on conflict (dedupe_key) do nothing""",476 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,477 d=f"DOCUMENT_CHANGED:{snapshot_id}", m=jsonb({"doc_type": doc_type, "diff_keys": list(diff)[:20] if diff else []}))478 if changed:479 ctx.stats.docs_changed += 1480 return await self._extract_and_write(ctx, target, doc, snapshot_id, res, first_time=first_time, parsed=parsed)481482 async def _extract_and_write(self, ctx: RunContext, target: Target, doc: dict[str, Any], snapshot_id: str, res: FetchResult, *,483 first_time: bool, parsed: Parsed | None = None) -> list[Target]:484 parsed = parsed or self.parse(target, res)485 try:486 facts = await self.extract(ctx, target, res, parsed)487 except Exception as exc:488 ctx.log.exception("extract failed", extra={"url": res.url})489 async with transaction() as conn:490 await execute(conn, "insert into connector_errors (connector_name, run_id, url, error_type, message) values (:n, :r, :u, :t, :m)",491 n=self.name, r=ctx.run_id, u=res.url, t=f"extract:{exc.__class__.__name__}", m=str(exc)[:2000])492 await execute(conn, "update snapshots set processing_status = 'failed' where id = :id", id=snapshot_id)493 return []494 if facts is None:495 return []496 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)497 ctx.stats.records += len(facts.entities) + len(facts.prices) + len(facts.results)498 if self.quarantine_enabled and not ctx.reprocess:499 ctx.buffered.append(item) # written (or quarantined) at the end of the run, see _flush_or_quarantine500 else:501 await self._write_buffered(ctx, item)502 return facts.targets503504 async def _write_buffered(self, ctx: RunContext, item: Buffered) -> None:505 facts, target, snapshot_id = item.facts, item.target, item.snapshot_id506 async with transaction() as conn:507 writer = FactWriter(conn, source_id=self.source_id, snapshot_id=snapshot_id, source_url=item.source_url, tier=self.tier,508 connector_name=self.name, extractor="deterministic", extractor_version=self.parser_version, observed_at=item.fetched_at,509 run_id=ctx.run_id, source_key=ctx.source_key, is_first_run=ctx.is_first_run)510 ws = await writer.write(facts)511 main = facts.document_entity or target.entity512 if main and main.id is None:513 await writer.resolver.resolve(main)514 if main and main.id:515 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)516 await execute(conn, "update snapshots set processing_status = :st where id = :id",517 st="llm_pending" if (target.needs_llm or self.needs_llm or facts.llm_hint) else "extracted", id=snapshot_id)518 if target.needs_llm or self.needs_llm or facts.llm_hint:519 from aiatlas.services.jobs import enqueue520521 await enqueue(conn, "llm_extract", {"snapshot_id": snapshot_id, "task": facts.llm_hint or target.meta.get("llm_task") or "auto",522 "entity_id": main.id if main else None, "connector": self.name}, priority=4,523 dedupe_key=f"llm_extract:{snapshot_id}")524 s = ctx.stats525 s.entities_created += ws.entities_created526 s.entities_updated += ws.entities_updated527 s.claims_written += ws.claims528 s.relations_written += ws.relations529 s.events_emitted += ws.events530531 # ------------------------------------------------------------------------------------------ quarantine532 async def _observed_counts(self, ctx: RunContext) -> dict[str, int]:533 """Counts of what the buffered facts would write: entity references, prices, results and *new* entity names (refs that resolve to534 nothing today). Resolution runs in a transaction that is rolled back so the check leaves no trace."""535 refs: dict[str, EntityRef] = {}536 prices = results = 0537 for item in ctx.buffered:538 prices += len(item.facts.prices)539 results += len(item.facts.results)540 for ref in item.facts.entities:541 refs.setdefault(ref.key(), ref)542 new_entities = 0543 async with connection() as conn:544 trans = await conn.begin()545 try:546 resolver = Resolver(conn, source_tier=self.tier)547 for ref in refs.values():548 probe = EntityRef(entity_type=ref.entity_type, name=ref.name, identifiers=dict(ref.identifiers), aliases=list(ref.aliases), slug_hint=ref.slug_hint)549 if await resolver.resolve(probe, create=False) is None:550 new_entities += 1551 finally:552 await trans.rollback()553 return {"entities": len(refs), "prices": prices, "results": results, "new_entities": new_entities}554555 @staticmethod556 def _quarantine_reason(observed: dict[str, int], baseline: dict[str, Any] | None, *, full_extraction: bool) -> str | None:557 """Explosion rules always apply; collapse rules only when every document was re-extracted (an incremental run legitimately yields little)."""558 if not baseline or not baseline.get("history"):559 return None560 med = baseline.get("medians") or {}561 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)562 if observed["new_entities"] > max(QUARANTINE_MIN_NEW_ENTITIES, 1.3 * b_new):563 return f"new entities {observed['new_entities']} > max({QUARANTINE_MIN_NEW_ENTITIES}, 1.3 × baseline {b_new:g})"564 if b_pr > 0 and observed["prices"] > 3 * b_pr:565 return f"price rows {observed['prices']} > 3 × baseline {b_pr:g}"566 if full_extraction:567 if b_ent > 0 and observed["entities"] < 0.8 * b_ent:568 return f"entity references {observed['entities']} < 0.8 × baseline {b_ent:g}"569 if b_res > 0 and observed["results"] < 0.5 * b_res:570 return f"benchmark results {observed['results']} < 0.5 × baseline {b_res:g}"571 return None572573 @staticmethod574 def _next_baseline(baseline: dict[str, Any] | None, observed: dict[str, int]) -> dict[str, Any]:575 history = list((baseline or {}).get("history") or [])576 history.append({k: int(observed[k]) for k in ("entities", "prices", "results", "new_entities")})577 history = history[-BASELINE_RUNS:]578 medians = {k: float(statistics.median(h[k] for h in history)) for k in ("entities", "prices", "results", "new_entities")}579 return {"history": history, "medians": medians, "runs": len(history), "updated_at": datetime.now(UTC).isoformat(timespec="seconds")}580581 async def _flush_or_quarantine(self, ctx: RunContext) -> None:582 observed = await self._observed_counts(ctx)583 full = self._full_extraction(ctx)584 async with transaction() as conn:585 row = await fetch_one(conn, "select baseline from connectors where name = :n", n=self.name)586 baseline = (row or {}).get("baseline")587 reason = None if ctx.force and ctx.file_overrides else self._quarantine_reason(observed, baseline, full_extraction=full)588 ctx.stats.meta["quarantine_check"] = {"observed": observed, "baseline": (baseline or {}).get("medians"), "full_extraction": full}589 if reason:590 ctx.quarantined = True591 ctx.quarantine_reason = reason592 ctx.log.warning("run quarantined", extra={"reason": reason, **observed})593 payload = [{"doc_id": b.doc_id, "snapshot_id": b.snapshot_id, "source_url": b.source_url, "fetched_at": b.fetched_at.isoformat(),594 "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)},595 "facts": facts_to_json(b.facts)} for b in ctx.buffered]596 qid = f"quar_{new_id('connector_run').split('_', 1)[1]}"597 async with transaction() as conn:598 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))""",599 id=qid, r=ctx.run_id, c=self.name, why=reason, s=jsonb({"observed": observed, "baseline": baseline, "full_extraction": full, "documents": len(ctx.buffered)}),600 f=jsonb(payload))601 await execute(conn, "update connector_runs set quarantined = true, baseline = cast(:b as jsonb) where id = :id", b=jsonb(baseline), id=ctx.run_id)602 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])603 await execute(conn, """insert into review_queue (id, kind, entity_ids, payload, reason, dedupe_key) values (:id, 'quarantine', '{}', cast(:p as jsonb), :r, :d)604 on conflict (dedupe_key) do nothing""",605 id=new_id("review"), p=jsonb({"connector": self.name, "run_id": ctx.run_id, "quarantine_id": qid, "observed": observed,606 "baseline": (baseline or {}).get("medians")}),607 r=f"{self.name}: run held in quarantine — {reason}; nothing written, nothing deleted", d=f"quarantine:{qid}")608 ctx.buffered.clear()609 return610 for item in ctx.buffered:611 await self._write_buffered(ctx, item)612 ctx.buffered.clear()613 if full and not ctx.reprocess and not ctx.file_overrides:614 async with transaction() as conn:615 nb = self._next_baseline(baseline, observed)616 await execute(conn, "update connectors set baseline = cast(:b as jsonb) where name = :n", b=jsonb(nb), n=self.name)617 await execute(conn, "update connector_runs set baseline = cast(:b as jsonb) where id = :id", b=jsonb(nb), id=ctx.run_id)618619 async def _record(self, conn: Any, ctx: RunContext, status: str, *, error: str | None = None) -> None:620 s = ctx.stats621 await execute(conn, """insert into connector_runs (id, connector_name, started_at, finished_at, status, duration_ms, docs_discovered, docs_fetched, docs_changed,622 docs_unchanged, docs_failed, entities_created, entities_updated, claims_written, relations_written, events_emitted, error, meta)623 values (:id, :c, :started, now(), :status, :dur, :dd, :df, :dc, :du, :dfail, :ec, :eu, :cw, :rw, :ee, :err, cast(:meta as jsonb))624 on conflict (id) do update set finished_at = now(), status = excluded.status, duration_ms = excluded.duration_ms,625 docs_discovered = excluded.docs_discovered, docs_fetched = excluded.docs_fetched, docs_changed = excluded.docs_changed,626 docs_unchanged = excluded.docs_unchanged, docs_failed = excluded.docs_failed, entities_created = excluded.entities_created,627 entities_updated = excluded.entities_updated, claims_written = excluded.claims_written, relations_written = excluded.relations_written,628 events_emitted = excluded.events_emitted, error = excluded.error, meta = excluded.meta""",629 id=ctx.run_id, c=self.name, started=ctx.started_at, status=status, dur=int((datetime.now(UTC) - ctx.started_at).total_seconds() * 1000),630 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,631 cw=s.claims_written, rw=s.relations_written, ee=s.events_emitted, err=error, meta=jsonb({**s.meta, "records": s.records}))632633634def _structural_diff(prev: dict[str, Any] | None, cur: dict[str, Any] | None) -> dict[str, Any] | None:635 if not prev or not cur:636 return None637 diff: dict[str, Any] = {}638 for key in ("title", "description", "headings", "tables", "json_ld", "front_matter", "published_at", "modified_at", "items"):639 a, b = prev.get(key), cur.get(key)640 if a != b:641 if isinstance(a, list) and isinstance(b, list):642 sa = {json.dumps(x, sort_keys=True, default=str) for x in a}643 sb = {json.dumps(x, sort_keys=True, default=str) for x in b}644 added = [json.loads(x) for x in list(sb - sa)[:50]]645 removed = [json.loads(x) for x in list(sa - sb)[:50]]646 if added or removed:647 diff[key] = {"added": added, "removed": removed}648 else:649 diff[key] = {"old": a, "new": b}650 return diff or None651652653def _pdf_text(content: bytes) -> str:654 try:655 import io656657 from pypdf import PdfReader658659 reader = PdfReader(io.BytesIO(content))660 parts = []661 for page in reader.pages[:60]:662 try:663 parts.append(page.extract_text() or "")664 except Exception: # noqa: BLE001665 continue666 return "\n\n".join(parts).strip()667 except Exception: # noqa: BLE001668 return ""669670671def _jsonable(obj: Any) -> Any:672 try:673 json.dumps(obj)674 return obj675 except (TypeError, ValueError):676 return json.loads(json.dumps(obj, default=str))677678679__all__ = ["BaseConnector", "ConnectorError", "Parsed", "RunContext", "RunStats", "Target"]680