"""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 time from dataclasses import dataclass, field from datetime import UTC, datetime, timedelta from typing import Any from aiatlas.config import settings from aiatlas.db import 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 Facts, Target from aiatlas.sdk.fetch import BlockedError, FetchError, Fetcher, FetchResult, NotModified, canonicalize_url, file_result from aiatlas.sdk.writer import FactWriter log = logging.getLogger(__name__) CIRCUIT_FAILURES = 3 CIRCUIT_COOLDOWN_S = 1800 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 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) 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 def __init__(self, config: dict[str, Any] | None = None): self.config = config or {} self.source_id: str | None = None # ------------------------------------------------------------------------------------------ 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 source_id, enabled, circuit_open_until from connectors where 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"] 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: wanted = {canonicalize_url(u) for u in only_urls} targets = [t for t in targets if canonicalize_url(t.url) in wanted] or [Target(url=u) for u in only_urls] ctx.stats.docs_discovered = len(targets) await self._process_all(ctx, targets) status = 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 = now(), last_change_at = case when cast(:changed 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, interval=await self._next_interval(conn, changed), d=int((time.perf_counter() - t0) * 1000), health="degraded" if status == "suspect" 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: # noqa: BLE001 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 def _final_status(self, ctx: RunContext) -> str: s = ctx.stats if self.expected_min_records 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 async with sem: followups = await self._process_target(ctx, target) 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(target.meta)) doc = await fetch_one(conn, "select * from documents where url = :u", u=url) 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: # noqa: BLE001 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 [] async with transaction() as conn: writer = FactWriter(conn, source_id=self.source_id, snapshot_id=snapshot_id, source_url=res.final_url or res.url, tier=self.tier, connector_name=self.name, extractor="deterministic", extractor_version=self.parser_version, observed_at=res.fetched_at) 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=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 s.records += len(facts.entities) + len(facts.prices) + len(facts.results) return facts.targets 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", "RunContext", "RunStats", "Parsed", "Target"]