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 time15from dataclasses import dataclass, field16from datetime import UTC, datetime, timedelta17from typing import Any1819from aiatlas.config import settings20from aiatlas.db import 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 Facts, Target27from aiatlas.sdk.fetch import BlockedError, FetchError, Fetcher, FetchResult, NotModified, canonicalize_url, file_result28from aiatlas.sdk.writer import FactWriter2930log = logging.getLogger(__name__)3132CIRCUIT_FAILURES = 333CIRCUIT_COOLDOWN_S = 1800343536class ConnectorError(Exception):37 pass383940@dataclass41class Parsed:42 kind: str # html|markdown|feed|json|pdf|text|xml43 html: HtmlDoc | None = None44 markdown: MarkdownDoc | None = None45 feed_meta: dict[str, Any] | None = None46 feed_items: list[FeedItem] = field(default_factory=list)47 json: Any = None48 text: str = ""4950 def structured(self) -> dict[str, Any] | None:51 if self.html:52 return self.html.structured()53 if self.markdown:54 return {"front_matter": _jsonable(self.markdown.front_matter), "headings": self.markdown.headings[:200], "tables": self.markdown.tables[:40],55 "link_count": len(self.markdown.links), "text_length": len(self.markdown.text)}56 if self.kind == "feed":57 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}58 for i in self.feed_items[:200]]}59 if self.kind == "json":60 return {"json_keys": list(self.json)[:100] if isinstance(self.json, dict) else None,61 "json_length": len(self.json) if isinstance(self.json, (list, dict)) else None}62 return {"text_length": len(self.text)} if self.text else None636465@dataclass66class RunStats:67 docs_discovered: int = 068 docs_fetched: int = 069 docs_changed: int = 070 docs_unchanged: int = 071 docs_failed: int = 072 entities_created: int = 073 entities_updated: int = 074 claims_written: int = 075 relations_written: int = 076 events_emitted: int = 077 records: int = 0 # connector-defined "records" for breakage detection (models, papers, prices…)78 meta: dict[str, Any] = field(default_factory=dict)798081@dataclass82class RunContext:83 run_id: str84 connector: BaseConnector85 fetcher: Fetcher86 started_at: datetime87 force: bool = False88 reprocess: bool = False # re-extract from stored snapshots without fetching89 file_overrides: dict[str, str] = field(default_factory=dict) # target.key or url → local path90 max_targets: int = 200091 stats: RunStats = field(default_factory=RunStats)92 seen_urls: set[str] = field(default_factory=set)93 log: logging.LoggerAdapter[logging.Logger] = field(init=False)9495 def __post_init__(self) -> None:96 self.log = logging.LoggerAdapter(logging.getLogger(f"connector.{self.connector.name}"), {"connector": self.connector.name, "run_id": self.run_id})979899class BaseConnector:100 # ------------------------------------------------------------------------------------------ identity & policy101 name: str = ""102 label: str = ""103 description: str = ""104 source_key: str = "" # sources.key (seeded from registry/sources.yaml)105 version: str = "1" # connector version106 parser_version: str = "1" # bump when extraction improves → `aia reprocess <connector>` replays snapshots107 interval_seconds: int = 3600108 min_interval_seconds: int = 900109 max_interval_seconds: int = 7 * 86400110 rate_per_min: int = 30111 respect_robots: bool = True112 expected_min_records: int = 0113 priority: int = 2114 tier: int = 1115 concurrency: int = 3116 default_doc_type: str = "page"117 needs_llm: bool = False # queue documents for LLM extraction after deterministic extraction118119 def __init__(self, config: dict[str, Any] | None = None):120 self.config = config or {}121 self.source_id: str | None = None122123 # ------------------------------------------------------------------------------------------ to implement124 async def discover(self, ctx: RunContext) -> list[Target]:125 """Return the targets for this run (seed pages, feeds, sitemaps, org pages…)."""126 seeds = self.config.get("seeds") or getattr(self, "seeds", [])127 return [Target(url=s) if isinstance(s, str) else Target(**s) for s in seeds]128129 async def extract(self, ctx: RunContext, target: Target, res: FetchResult, parsed: Parsed) -> Facts:130 """Turn a parsed document into facts. Deterministic only; LLM extraction is queued separately."""131 return Facts()132133 def parse(self, target: Target, res: FetchResult) -> Parsed:134 """Default parser by content type; connectors may override for special formats."""135 if res.is_pdf:136 return Parsed(kind="pdf", text=_pdf_text(res.content))137 if target.doc_type == "feed" or (res.is_xml and b"<rss" in res.content[:2000] or b"<feed" in res.content[:2000]):138 meta, items = parse_feed(res.content)139 if items or meta.get("title"):140 return Parsed(kind="feed", feed_meta=meta, feed_items=items)141 if res.is_json:142 try:143 return Parsed(kind="json", json=res.json())144 except Exception: # noqa: BLE001145 pass146 if res.is_html:147 doc = parse_html(res.content, res.final_url or res.url)148 return Parsed(kind="html", html=doc, text=doc.text)149 text = res.text150 if target.doc_type in ("model_card", "readme", "markdown") or "markdown" in res.content_type or res.url.endswith((".md", ".mdx")):151 md = parse_markdown(text)152 return Parsed(kind="markdown", markdown=md, text=md.text)153 if res.is_xml:154 return Parsed(kind="xml", text=text)155 return Parsed(kind="text", text=text)156157 async def healthcheck(self) -> bool:158 return True159160 # ------------------------------------------------------------------------------------------ orchestration161 async def run(self, *, force: bool = False, reprocess: bool = False, file_overrides: dict[str, str] | None = None,162 max_targets: int | None = None, only_urls: list[str] | None = None) -> RunContext:163 started = datetime.now(UTC)164 run_id = new_id("connector_run")165 async with Fetcher(rate_per_min=self.rate_per_min, robots=self.respect_robots) as fetcher:166 ctx = RunContext(run_id=run_id, connector=self, fetcher=fetcher, started_at=started, force=force, reprocess=reprocess,167 file_overrides=file_overrides or {}, max_targets=max_targets or int(self.config.get("max_targets", 2000)))168 async with transaction() as conn:169 state = await fetch_one(conn, "select source_id, enabled, circuit_open_until from connectors where name = :n", n=self.name)170 if state is None:171 raise ConnectorError(f"connector {self.name!r} not registered (run `aia seed`)")172 self.source_id = state["source_id"]173 if not state["enabled"] and not force:174 ctx.log.info("connector disabled, skipping")175 await self._record(conn, ctx, "skipped", error="disabled")176 return ctx177 if state["circuit_open_until"] and state["circuit_open_until"] > started and not force:178 ctx.log.warning("circuit open, skipping", extra={"until": state["circuit_open_until"].isoformat()})179 await self._record(conn, ctx, "skipped", error="circuit open")180 return ctx181 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)182 await execute(conn, "update connectors set last_attempt_at = :t, health = case when health = 'unknown' then 'degraded' else health end where name = :n",183 t=started, n=self.name)184 t0 = time.perf_counter()185 try:186 targets = await self.discover(ctx)187 if only_urls:188 wanted = {canonicalize_url(u) for u in only_urls}189 targets = [t for t in targets if canonicalize_url(t.url) in wanted] or [Target(url=u) for u in only_urls]190 ctx.stats.docs_discovered = len(targets)191 await self._process_all(ctx, targets)192 status = self._final_status(ctx)193 async with transaction() as conn:194 await self._record(conn, ctx, status)195 changed = ctx.stats.docs_changed > 0196 await execute(conn, """update connectors set consecutive_failures = 0, circuit_open_until = null, last_success_at = now(),197 last_change_at = case when cast(:changed as boolean) then now() else last_change_at end,198 consecutive_unchanged = case when cast(:changed as boolean) then 0 else consecutive_unchanged + 1 end,199 interval_seconds = cast(:interval as integer), next_run_at = now() + make_interval(secs => cast(:interval as double precision)),200 last_duration_ms = :d, health = :health, updated_at = now() where name = :n""",201 changed=changed, interval=await self._next_interval(conn, changed), d=int((time.perf_counter() - t0) * 1000),202 health="degraded" if status == "suspect" else "ok", n=self.name)203 if status == "suspect":204 await execute(conn, """insert into review_queue (id, kind, entity_ids, payload, reason, dedupe_key)205 values (:id, 'parser_breakage', '{}', cast(:p as jsonb), :r, :d) on conflict (dedupe_key) do nothing""",206 id=new_id("review"), p=jsonb({"connector": self.name, "run_id": run_id, "records": ctx.stats.records, "expected_min": self.expected_min_records}),207 r=f"{self.name}: {ctx.stats.records} records < expected {self.expected_min_records} — connector failure suspected, nothing deleted",208 d=f"parser_breakage:{self.name}:{started.date()}")209 ctx.log.info("run finished", extra={"status": status, **{k: v for k, v in ctx.stats.__dict__.items() if k != "meta"},210 "ms": int((time.perf_counter() - t0) * 1000)})211 except Exception as exc: # noqa: BLE001212 ctx.log.exception("run failed", extra={"error": str(exc)})213 async with transaction() as conn:214 await self._record(conn, ctx, "failed", error=f"{exc.__class__.__name__}: {exc}"[:2000])215 await execute(conn, "insert into connector_errors (connector_name, run_id, error_type, message) values (:n, :r, :t, :m)",216 n=self.name, r=run_id, t=exc.__class__.__name__, m=str(exc)[:4000])217 row = await fetch_one(conn, """update connectors set consecutive_failures = consecutive_failures + 1, last_duration_ms = :d, health = 'failing',218 next_run_at = now() + make_interval(secs => least(interval_seconds, 3600)), updated_at = now()219 where name = :n returning consecutive_failures""", d=int((time.perf_counter() - t0) * 1000), n=self.name)220 if row and row["consecutive_failures"] >= CIRCUIT_FAILURES:221 until = datetime.now(UTC) + timedelta(seconds=CIRCUIT_COOLDOWN_S)222 await execute(conn, "update connectors set circuit_open_until = :u where name = :n", u=until, n=self.name)223 ctx.log.error("circuit opened", extra={"until": until.isoformat()})224 raise225 return ctx226227 def _final_status(self, ctx: RunContext) -> str:228 s = ctx.stats229 if self.expected_min_records and s.records < self.expected_min_records and not ctx.reprocess:230 return "suspect"231 if s.docs_fetched and s.docs_failed == s.docs_fetched and s.docs_fetched > 0:232 return "failed" if s.docs_changed == 0 else "success"233 return "unchanged" if s.docs_changed == 0 else "success"234235 async def _next_interval(self, conn: Any, changed: bool) -> int:236 row = await fetch_one(conn, "select interval_seconds, consecutive_unchanged, min_interval_seconds, max_interval_seconds from connectors where name = :n", n=self.name)237 if not row:238 return self.interval_seconds239 cur = row["interval_seconds"] or self.interval_seconds240 lo, hi = row["min_interval_seconds"] or self.min_interval_seconds, row["max_interval_seconds"] or self.max_interval_seconds241 if changed:242 return max(lo, int(cur * 0.7))243 if row["consecutive_unchanged"] >= 3:244 return min(hi, int(cur * 1.5))245 return cur246247 async def _process_all(self, ctx: RunContext, targets: list[Target]) -> None:248 queue: asyncio.Queue[Target] = asyncio.Queue()249 for t in targets:250 queue.put_nowait(t)251 processed = 0252 sem = asyncio.Semaphore(max(1, self.concurrency))253 pending: set[asyncio.Task[None]] = set()254255 async def worker(target: Target) -> None:256 nonlocal processed257 async with sem:258 followups = await self._process_target(ctx, target)259 for f in followups:260 cu = canonicalize_url(f.url)261 if cu not in ctx.seen_urls and processed + queue.qsize() < ctx.max_targets:262 queue.put_nowait(f)263264 while not queue.empty() or pending:265 while not queue.empty() and processed < ctx.max_targets:266 target = queue.get_nowait()267 cu = canonicalize_url(target.url)268 if cu in ctx.seen_urls:269 continue270 ctx.seen_urls.add(cu)271 processed += 1272 task = asyncio.create_task(worker(target))273 pending.add(task)274 task.add_done_callback(pending.discard)275 if pending:276 await asyncio.wait(pending, return_when=asyncio.FIRST_COMPLETED)277 elif queue.empty():278 break279280 async def _process_target(self, ctx: RunContext, target: Target) -> list[Target]:281 url = canonicalize_url(target.url)282 doc_type = target.doc_type or self.default_doc_type283 async with transaction() as conn:284 doc = await fetch_one(conn, "select * from documents where url = :u", u=url)285 if doc is None:286 doc_id = new_id("document")287 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))288 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))289 doc = await fetch_one(conn, "select * from documents where url = :u", u=url)290 assert doc291 # ---- reprocess mode: replay latest snapshot without network292 if ctx.reprocess:293 async with transaction() as conn:294 snap = await fetch_one(conn, "select * from snapshots where document_id = :d and changed order by observed_at desc limit 1", d=doc["id"])295 if not snap or not snap["raw_path"]:296 return []297 res = FetchResult(url=url, final_url=snap["final_url"] or url, status=snap["http_status"] or 200, headers=snap["headers"] or {},298 content=archive.load_raw(snap["raw_path"]), content_type=snap["content_type"] or "", fetched_at=snap["observed_at"], duration_ms=0,299 transport="file")300 return await self._extract_and_write(ctx, target, doc, snap["id"], res, first_time=False)301 # ---- fetch302 override = ctx.file_overrides.get(target.key or "") or ctx.file_overrides.get(url) or ctx.file_overrides.get(target.url)303 try:304 if override:305 res = file_result(override, url=url, content_type=target.meta.get("content_type", "text/html"))306 else:307 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,308 min_bytes=target.min_bytes, escalate=target.escalate, accept=target.accept, rate_per_min=target.rate_per_min)309 except NotModified:310 ctx.stats.docs_fetched += 1311 ctx.stats.docs_unchanged += 1312 async with transaction() as conn:313 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"])314 return []315 except BlockedError as exc:316 ctx.stats.docs_fetched += 1317 ctx.stats.docs_failed += 1318 ctx.log.warning("blocked", extra={"url": url, "error": str(exc)})319 async with transaction() as conn:320 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",321 s=exc.status, id=doc["id"])322 await execute(conn, "insert into connector_errors (connector_name, run_id, url, error_type, message) values (:n, :r, :u, 'blocked', :m)",323 n=self.name, r=ctx.run_id, u=url, m=str(exc)[:2000])324 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)325 on conflict (dedupe_key) do nothing""", id=new_id("review"), p=jsonb({"url": url, "connector": self.name, "status": exc.status}),326 r=f"{self.name}: access denied for {url}", d=f"blocked:{url}")327 return []328 except FetchError as exc:329 ctx.stats.docs_fetched += 1330 ctx.stats.docs_failed += 1331 ctx.log.warning("fetch failed", extra={"url": url, "error": str(exc)})332 async with transaction() as conn:333 gone = exc.status in (404, 410)334 await execute(conn, """update documents set last_fetched_at = now(), fetch_count = fetch_count + 1, fail_count = fail_count + 1, last_status = :s,335 status = case when :gone and fail_count >= 2 then 'gone' else status end where id = :id""", s=exc.status, gone=gone, id=doc["id"])336 await execute(conn, "insert into connector_errors (connector_name, run_id, url, error_type, message) values (:n, :r, :u, :t, :m)",337 n=self.name, r=ctx.run_id, u=url, t=exc.__class__.__name__, m=str(exc)[:2000])338 return []339 ctx.stats.docs_fetched += 1340 changed = res.sha256 != doc["content_hash"]341 first_time = doc["content_hash"] is None342 if not changed and not ctx.force:343 ctx.stats.docs_unchanged += 1344 async with transaction() as conn:345 await execute(conn, """update documents set last_fetched_at = now(), fetch_count = fetch_count + 1, last_status = :s, fail_count = 0,346 etag = coalesce(:etag, etag), last_modified = coalesce(:lm, last_modified) where id = :id""",347 s=res.status, etag=res.headers.get("etag"), lm=res.headers.get("last-modified"), id=doc["id"])348 return []349 # ---- changed (or forced): archive + snapshot + extract350 parsed = self.parse(target, res)351 raw_path = archive.store_raw(res.content, sha256=res.sha256)352 text_path, text_hash = archive.store_text(parsed.text) if parsed.text else (None, None)353 structured = parsed.structured()354 snapshot_id = new_id("snapshot")355 async with transaction() as conn:356 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"])357 diff = _structural_diff(prev["structured"] if prev else None, structured) if prev else None358 semantic_change = not prev or prev["text_hash"] != text_hash or bool(diff)359 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,360 raw_path, text_path, text_hash, structured, parser_version, connector_version, transport, changed, diff, processing_status)361 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')""",362 id=snapshot_id, d=doc["id"], run=ctx.run_id, u=url, fu=res.final_url, o=res.fetched_at, st=res.status,363 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")}),364 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,365 pv=self.parser_version, cv=self.version, tr=res.transport, changed=changed, diff=jsonb(diff) if diff else None)366 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)367 await execute(conn, """update documents set last_fetched_at = now(), last_changed_at = case when :changed then now() else last_changed_at end,368 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,369 content_hash = :hash, etag = :etag, last_modified = :lm, canonical_url = coalesce(:canon, canonical_url), title = coalesce(:title, title),370 status = 'active', needs_llm = :llm, doc_type = :dt where id = :id""",371 changed=changed, first=first_time, s=res.status, hash=res.sha256, etag=res.headers.get("etag"), lm=res.headers.get("last-modified"),372 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"])373 if changed and not first_time and semantic_change and doc_type not in ("feed", "sitemap", "listing"):374 await execute(conn, """insert into change_events (id, entity_id, event_type, category, summary, importance, observed_at, source_id, snapshot_id, source_url,375 connector_name, dedupe_key, meta) values (:id, :e, 'DOCUMENT_CHANGED', 'source', :s, 0, now(), :src, :snap, :u, :c, :d, cast(:m as jsonb))376 on conflict (dedupe_key) do nothing""",377 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,378 d=f"DOCUMENT_CHANGED:{snapshot_id}", m=jsonb({"doc_type": doc_type, "diff_keys": list(diff)[:20] if diff else []}))379 if changed:380 ctx.stats.docs_changed += 1381 return await self._extract_and_write(ctx, target, doc, snapshot_id, res, first_time=first_time, parsed=parsed)382383 async def _extract_and_write(self, ctx: RunContext, target: Target, doc: dict[str, Any], snapshot_id: str, res: FetchResult, *,384 first_time: bool, parsed: Parsed | None = None) -> list[Target]:385 parsed = parsed or self.parse(target, res)386 try:387 facts = await self.extract(ctx, target, res, parsed)388 except Exception as exc: # noqa: BLE001389 ctx.log.exception("extract failed", extra={"url": res.url})390 async with transaction() as conn:391 await execute(conn, "insert into connector_errors (connector_name, run_id, url, error_type, message) values (:n, :r, :u, :t, :m)",392 n=self.name, r=ctx.run_id, u=res.url, t=f"extract:{exc.__class__.__name__}", m=str(exc)[:2000])393 await execute(conn, "update snapshots set processing_status = 'failed' where id = :id", id=snapshot_id)394 return []395 if facts is None:396 return []397 async with transaction() as conn:398 writer = FactWriter(conn, source_id=self.source_id, snapshot_id=snapshot_id, source_url=res.final_url or res.url, tier=self.tier,399 connector_name=self.name, extractor="deterministic", extractor_version=self.parser_version, observed_at=res.fetched_at)400 ws = await writer.write(facts)401 main = facts.document_entity or target.entity402 if main and main.id is None:403 await writer.resolver.resolve(main)404 if main and main.id:405 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"])406 await execute(conn, "update snapshots set processing_status = :st where id = :id",407 st="llm_pending" if (target.needs_llm or self.needs_llm or facts.llm_hint) else "extracted", id=snapshot_id)408 if target.needs_llm or self.needs_llm or facts.llm_hint:409 from aiatlas.services.jobs import enqueue410411 await enqueue(conn, "llm_extract", {"snapshot_id": snapshot_id, "task": facts.llm_hint or target.meta.get("llm_task") or "auto",412 "entity_id": main.id if main else None, "connector": self.name}, priority=4,413 dedupe_key=f"llm_extract:{snapshot_id}")414 s = ctx.stats415 s.entities_created += ws.entities_created416 s.entities_updated += ws.entities_updated417 s.claims_written += ws.claims418 s.relations_written += ws.relations419 s.events_emitted += ws.events420 s.records += len(facts.entities) + len(facts.prices) + len(facts.results)421 return facts.targets422423 async def _record(self, conn: Any, ctx: RunContext, status: str, *, error: str | None = None) -> None:424 s = ctx.stats425 await execute(conn, """insert into connector_runs (id, connector_name, started_at, finished_at, status, duration_ms, docs_discovered, docs_fetched, docs_changed,426 docs_unchanged, docs_failed, entities_created, entities_updated, claims_written, relations_written, events_emitted, error, meta)427 values (:id, :c, :started, now(), :status, :dur, :dd, :df, :dc, :du, :dfail, :ec, :eu, :cw, :rw, :ee, :err, cast(:meta as jsonb))428 on conflict (id) do update set finished_at = now(), status = excluded.status, duration_ms = excluded.duration_ms,429 docs_discovered = excluded.docs_discovered, docs_fetched = excluded.docs_fetched, docs_changed = excluded.docs_changed,430 docs_unchanged = excluded.docs_unchanged, docs_failed = excluded.docs_failed, entities_created = excluded.entities_created,431 entities_updated = excluded.entities_updated, claims_written = excluded.claims_written, relations_written = excluded.relations_written,432 events_emitted = excluded.events_emitted, error = excluded.error, meta = excluded.meta""",433 id=ctx.run_id, c=self.name, started=ctx.started_at, status=status, dur=int((datetime.now(UTC) - ctx.started_at).total_seconds() * 1000),434 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,435 cw=s.claims_written, rw=s.relations_written, ee=s.events_emitted, err=error, meta=jsonb({**s.meta, "records": s.records}))436437438def _structural_diff(prev: dict[str, Any] | None, cur: dict[str, Any] | None) -> dict[str, Any] | None:439 if not prev or not cur:440 return None441 diff: dict[str, Any] = {}442 for key in ("title", "description", "headings", "tables", "json_ld", "front_matter", "published_at", "modified_at", "items"):443 a, b = prev.get(key), cur.get(key)444 if a != b:445 if isinstance(a, list) and isinstance(b, list):446 sa = {json.dumps(x, sort_keys=True, default=str) for x in a}447 sb = {json.dumps(x, sort_keys=True, default=str) for x in b}448 added = [json.loads(x) for x in list(sb - sa)[:50]]449 removed = [json.loads(x) for x in list(sa - sb)[:50]]450 if added or removed:451 diff[key] = {"added": added, "removed": removed}452 else:453 diff[key] = {"old": a, "new": b}454 return diff or None455456457def _pdf_text(content: bytes) -> str:458 try:459 import io460461 from pypdf import PdfReader462463 reader = PdfReader(io.BytesIO(content))464 parts = []465 for page in reader.pages[:60]:466 try:467 parts.append(page.extract_text() or "")468 except Exception: # noqa: BLE001469 continue470 return "\n\n".join(parts).strip()471 except Exception: # noqa: BLE001472 return ""473474475def _jsonable(obj: Any) -> Any:476 try:477 json.dumps(obj)478 return obj479 except (TypeError, ValueError):480 return json.loads(json.dumps(obj, default=str))481482483__all__ = ["BaseConnector", "ConnectorError", "RunContext", "RunStats", "Parsed", "Target"]484