spb/company-atlas
Public
Python 66.3%
TypeScript 22.7%
JavaScript 8.6%
HTML 1.4%
CSS 0.7%
1"""The sensor pipeline (spec §17–20, §26, §60–64): one run of one sensor.23 run_sensor(sensor_row, fetcher=…, worker=…) -> RunOutcome45 fetch (conditional) ──▶ NotModified ─▶ observation(not_modified) + interval growth6 ──▶ failure ─▶ observation(failure_class) + failures row + policy backoff + status transitions (+ review)7 ──▶ success ─▶ archive raw ─▶ connector.extract ─▶ hashes8 unchanged ─▶ observation(changed=false) + interval growth9 changed ─▶ snapshot (version, objects, extracted) ─▶ entity reconciliation ─▶ structured_delta10 ─▶ block diff vs previous snapshot ─▶ changes row (significance, kind, status)11 ─▶ sensor counters / validators / adaptive interval / quality ─▶ company + ledgers1213Everything after the network happens in ONE transaction per sensor. Raw bytes, normalized text and blocks live in the object store14(content-addressed); rows only reference them. Nothing is ever overwritten: new snapshot versions, status columns, removed_at.15"""16from __future__ import annotations1718import asyncio19import json20import logging21import random22import re23import time24from dataclasses import dataclass, field25from datetime import UTC, datetime, timedelta26from typing import Any2728from companyatlas import archive29from companyatlas.config import settings30from companyatlas.connectors._precision import HTML_JOB_CONNECTORS, apply_precision31from companyatlas.connectors._util import is_engineering, job_fingerprint, norm_name32from companyatlas.db import execute, fetch_all, fetch_one, jsonb, transaction33from companyatlas.fetch import (34 BlockedError,35 Fetcher,36 FetchError,37 FetchResult,38 NotModified,39 classify_exception,40 file_result,41 text_quality,42)43from companyatlas.ids import new_id44from companyatlas.sdk import connector as connectors45from companyatlas.sdk.diff import DIFF_VERSION, compare46from companyatlas.sdk.models import Block, BlockDiff, Extraction, StructuredDelta47from companyatlas.sdk.normalize import structural_hash, text_hash48from companyatlas.taxonomy import (49 AI_KEYWORDS,50 FAILURE_POLICY,51 SURFACE_IMPORTANCE,52 ChangeKind,53 FailureClass,54 SensorStatus,55 Surface,56 change_kind,57 tier_for_interval,58)59from companyatlas.urls import canonicalize_url, registrable_domain6061log = logging.getLogger(__name__)6263PIPELINE_VERSION = "pipeline-v1"64DELTA_LIST_LIMIT = 20065EXTRACTED_LIST_LIMIT = 30066EXTRACTED_MAX_BYTES = 900_00067JITTER = 0.1 # ± on next_run_at68QUALITY_ALPHA = 0.15 # EMA weight for quality_score updates69HTML_SUSPICIOUS_DROP = 0.7 # HTML listings losing > 70 % of ≥ 10 entities are not trusted for removals70MIN_PREVIOUS_FOR_DROP_GUARD = 1071FETCH_TIMEOUT_FACTOR = 6 # connector.fetch (incl. pagination) may take this × http timeout727374@dataclass(slots=True)75class RunOutcome:76 sensor_id: str77 status: str # ok | not_modified | unchanged | changed | failed | skipped | redirected78 observation_id: str | None = None79 snapshot_id: str | None = None80 change_id: str | None = None81 failure_class: str | None = None82 error: str | None = None83 significance: float | None = None84 kind: str | None = None85 duration_ms: int = 086 next_run_at: datetime | None = None87 interval_s: int | None = None88 delta_counts: dict[str, int] = field(default_factory=dict)89 sensor_status: str | None = None9091 @property92 def ok(self) -> bool:93 return self.status in ("ok", "not_modified", "unchanged", "changed")949596# ------------------------------------------------------------------------------------------------------------ scheduling maths979899def _clamp(v: float) -> int:100 return int(max(settings.min_interval_s, min(settings.max_interval_s, v)))101102103def next_interval_unchanged(current: int, base: int) -> int:104 """Burst decay back to base, then slow growth towards the max (spec §16)."""105 if current < base:106 return _clamp(min(base, current * settings.burst_decay))107 return _clamp(current * settings.stability_growth)108109110def next_interval_changed(kind: ChangeKind, current: int, base: int) -> int:111 if kind in (ChangeKind.MEANINGFUL, ChangeKind.MAJOR, ChangeKind.CRITICAL):112 return _clamp(settings.burst_interval_s)113 if kind == ChangeKind.MINOR:114 return _clamp(min(current, base))115 return next_interval_unchanged(current, base)116117118def next_interval_failed(failure: str, current: int) -> int:119 mult = FAILURE_POLICY.get(failure, FAILURE_POLICY[FailureClass.UNKNOWN])[0]120 return _clamp(current * mult)121122123def _next_run(now: datetime, interval: int) -> datetime:124 return now + timedelta(seconds=interval * random.uniform(1 - JITTER, 1 + JITTER))125126127def _quality_update(current: float, target: float) -> float:128 return round(max(0.0, min(100.0, current * (1 - QUALITY_ALPHA) + target * QUALITY_ALPHA)), 2)129130131def is_ai_title(*parts: str | None) -> bool:132 text = " " + " ".join(p.lower() for p in parts if p) + " "133 return any(k in text for k in AI_KEYWORDS)134135136# ------------------------------------------------------------------------------------------------------------ helpers137138139def _structured_hash(ex: Extraction) -> str:140 """Hash of the typed payload only (jobs/people/products/plans/locations/news), order-insensitive."""141 import hashlib142143 parts: list[str] = []144 for j in ex.jobs:145 parts.append("job|" + job_fingerprint(j.title, j.location_text, j.external_id, j.url))146 for p in ex.people:147 parts.append(f"person|{norm_name(p.name)}|{(p.title or '').lower()}")148 for pr in ex.products:149 parts.append(f"product|{norm_name(pr.name)}")150 for pl in ex.plans:151 parts.append(f"plan|{norm_name(pl.plan_name)}|{pl.price}|{pl.currency}|{pl.billing_period}|{pl.unit}|{pl.contact_sales}|{'/'.join(pl.features[:25])}")152 for loc in ex.locations:153 parts.append(f"loc|{norm_name(loc.name)}|{loc.city}|{loc.country}")154 for n in ex.news:155 parts.append(f"news|{canonicalize_url(n.url)}")156 if not parts:157 return ""158 return hashlib.sha256("\n".join(sorted(parts)).encode("utf-8")).hexdigest()159160161def _bounded_extracted(ex: Extraction) -> dict[str, Any]:162 payload = ex.structured_payload()163 for k in ("jobs", "people", "products", "plans", "locations", "news"):164 lst = payload.get(k) or []165 if len(lst) > EXTRACTED_LIST_LIMIT:166 payload[k] = lst[:EXTRACTED_LIST_LIMIT]167 payload.setdefault("truncated", {})[k] = len(lst)168 meta = dict(payload.get("meta") or {})169 if isinstance(meta.get("urls"), list) and len(meta["urls"]) > 500:170 meta["urls"] = meta["urls"][:500]171 meta["urls_truncated"] = True172 payload["meta"] = meta173 raw = jsonb(payload)174 if len(raw) > EXTRACTED_MAX_BYTES:175 for k in ("jobs", "people", "products", "plans", "locations", "news"):176 payload[k] = (payload.get(k) or [])[:50]177 meta.pop("urls", None)178 payload["meta"] = meta179 payload["truncated"] = {**payload.get("truncated", {}), "reason": "size"}180 return payload181182183def _blocks_json(blocks: list[Block]) -> str:184 return json.dumps([b.to_json() for b in blocks], ensure_ascii=False, default=str)185186187def _blocks_from_json(raw: str) -> list[Block]:188 out: list[Block] = []189 try:190 data = json.loads(raw)191 except json.JSONDecodeError:192 return out193 for d in data if isinstance(data, list) else []:194 try:195 out.append(Block(key=d["key"], kind=d.get("kind", "other"), text=d.get("text", ""), path=d.get("path", ""), hash=d.get("hash", ""),196 simhash=int(d.get("simhash") or 0), weight=float(d.get("weight") or 1.0), order=int(d.get("order") or 0), attrs=d.get("attrs") or {}))197 except (KeyError, TypeError, ValueError):198 continue199 return out200201202def _guess_content_type(path: str) -> str:203 p = path.lower()204 if p.endswith(".json"):205 return "application/json; charset=utf-8"206 if p.endswith((".xml", ".rss", ".atom")):207 return "application/xml; charset=utf-8"208 return "text/html; charset=utf-8"209210211# ------------------------------------------------------------------------------------------------------------ domain budgets212213214async def _check_domain_budget(conn: Any, domain: str, now: datetime) -> tuple[bool, datetime | None, str | None]:215 """(allowed, resume_at, reason). Creates the row on first sight; resets `used_today` on a new day."""216 row = await fetch_one(conn, """217 insert into domain_budgets (domain, max_concurrency, requests_per_minute, daily_budget)218 values (:d, :mc, :rpm, :daily)219 on conflict (domain) do update set used_today = case when domain_budgets.budget_day < current_date then 0 else domain_budgets.used_today end,220 budget_day = greatest(domain_budgets.budget_day, current_date), updated_at = now()221 returning daily_budget, used_today, blocked_until, block_reason222 """, d=domain, mc=settings.domain_max_concurrency, rpm=settings.default_rate_per_min, daily=settings.domain_daily_budget)223 if row is None:224 return True, None, None225 if row["blocked_until"] and row["blocked_until"] > now:226 return False, row["blocked_until"], row.get("block_reason") or "domain blocked"227 if row["used_today"] >= row["daily_budget"]:228 tomorrow = datetime.combine(datetime.now(UTC).date() + timedelta(days=1), datetime.min.time(), tzinfo=UTC)229 return False, tomorrow, "daily budget exhausted"230 return True, None, None231232233async def _consume_domain_budget(conn: Any, domain: str, units: int) -> None:234 await execute(conn, "update domain_budgets set used_today = used_today + :u, updated_at = now() where domain = :d", u=units, d=domain)235236237async def _ledger(conn: Any, company_id: str, connector_id: str, units: int, size_bytes: int) -> None:238 await execute(conn, """239 insert into cost_ledger (day, dimension, key, units) values (current_date, 'fetch', :company, :u)240 on conflict (day, dimension, key) do update set units = cost_ledger.units + excluded.units241 """, company=company_id, u=units)242 await execute(conn, """243 insert into cost_ledger (day, dimension, key, units) values (current_date, 'fetch', :conn, :u)244 on conflict (day, dimension, key) do update set units = cost_ledger.units + excluded.units245 """, conn=f"connector:{connector_id}", u=units)246 if size_bytes:247 await execute(conn, """248 insert into cost_ledger (day, dimension, key, units) values (current_date, 'storage_gb', '', :gb)249 on conflict (day, dimension, key) do update set units = cost_ledger.units + excluded.units250 """, gb=size_bytes / 1e9)251252253# ------------------------------------------------------------------------------------------------------------ entity reconciliation254255256def _trustworthy(ex: Extraction, count: int) -> bool:257 return count >= 1 or bool(ex.meta.get("structured"))258259260def _suspicious_drop(ex: Extraction, previous: int, current: int) -> bool:261 if ex.meta.get("structured"):262 return False263 return previous >= MIN_PREVIOUS_FOR_DROP_GUARD and current < previous * (1 - HTML_SUSPICIOUS_DROP)264265266async def reconcile_jobs(conn: Any, *, company_id: str, sensor_id: str, ex: Extraction, now: datetime, delta: StructuredDelta) -> None:267 existing = await fetch_all(conn, "select id, fingerprint, status, title from jobs where company_id = :c and sensor_id = :s", c=company_id, s=sensor_id)268 by_fp = {r["fingerprint"]: r for r in existing}269 open_before = sum(1 for r in existing if r["status"] == "open")270 seen: set[str] = set()271 added: list[dict[str, Any]] = []272 for j in ex.jobs:273 fp = job_fingerprint(j.title, j.location_text, j.external_id, j.url)274 if fp in seen:275 continue276 seen.add(fp)277 ai = is_ai_title(j.title, j.department, j.team)278 row = by_fp.get(fp)279 params = {"c": company_id, "s": sensor_id, "fp": fp, "title": j.title[:300], "dept": j.department, "team": j.team, "loc": j.location_text,280 "city": j.city, "region": j.region, "country": (j.country or None), "remote": j.remote, "et": j.employment_type, "sen": j.seniority,281 "skills": list(j.skills or [])[:30], "smin": j.salary_min, "smax": j.salary_max, "scur": j.salary_currency, "sper": j.salary_period,282 "url": j.url, "dh": j.description_hash, "posted": j.posted_at, "ai": ai, "eng": is_engineering(j.title), "raw": jsonb(j.raw or {}),283 "ext": j.external_id, "now": now}284 if row is None:285 await execute(conn, """286 insert into jobs (id, company_id, sensor_id, external_id, fingerprint, title, department, team, location_text, city, region, country, remote,287 employment_type, seniority, skills, salary_min, salary_max, salary_currency, salary_period, url, description_hash, posted_at,288 first_seen_at, last_seen_at, status, is_ai, is_engineering, raw)289 values (:id, :c, :s, :ext, :fp, :title, :dept, :team, :loc, :city, :region, :country, :remote, :et, :sen, cast(:skills as text[]), :smin, :smax,290 :scur, :sper, :url, :dh, :posted, :now, :now, 'open', :ai, :eng, cast(:raw as jsonb))291 on conflict (company_id, fingerprint) do update set sensor_id = excluded.sensor_id, last_seen_at = excluded.last_seen_at, status = 'open',292 removed_at = null, url = coalesce(excluded.url, jobs.url), is_ai = excluded.is_ai293 """, id=new_id("job"), **params)294 added.append({"title": j.title, "url": j.url, "location_text": j.location_text, "country": j.country, "remote": j.remote, "department": j.department, "is_ai": ai})295 else:296 await execute(conn, """297 update jobs set last_seen_at = :now, status = 'open', removed_at = null, title = :title, location_text = coalesce(:loc, location_text),298 city = coalesce(:city, city), region = coalesce(:region, region), country = coalesce(:country, country), remote = coalesce(:remote, remote),299 url = coalesce(:url, url), is_ai = :ai, salary_min = coalesce(:smin, salary_min), salary_max = coalesce(:smax, salary_max),300 posted_at = coalesce(posted_at, :posted)301 where id = :id302 """, id=row["id"], **params)303 if row["status"] != "open":304 added.append({"title": j.title, "url": j.url, "location_text": j.location_text, "country": j.country, "remote": j.remote, "department": j.department,305 "is_ai": ai, "relisted": True})306 removed: list[dict[str, Any]] = []307 missing = [r for r in existing if r["status"] == "open" and r["fingerprint"] not in seen]308 if missing and _trustworthy(ex, len(seen)) and not _suspicious_drop(ex, open_before, len(seen)):309 rows = await fetch_all(conn, """310 update jobs set status = 'no_longer_listed', removed_at = :now, last_seen_at = last_seen_at311 where id = any(cast(:ids as text[])) returning title, url, location_text, country, remote, department, is_ai312 """, ids=[r["id"] for r in missing], now=now)313 removed = [dict(r) for r in rows]314 elif missing:315 delta.setdefault("notes", []).append(f"jobs: {len(missing)} missing not marked removed (untrusted extraction)")316 if added or removed or open_before != len(seen):317 delta["jobs"] = {"added": added[:DELTA_LIST_LIMIT], "removed": removed[:DELTA_LIST_LIMIT], "open_before": open_before, "open_after": len(seen),318 "ai_added": sum(1 for a in added if a.get("is_ai")), "ai_removed": sum(1 for r in removed if r.get("is_ai"))}319320321async def reconcile_named(conn: Any, *, table: str, company_id: str, sensor_id: str, sensor_url: str, items: list[Any], ex: Extraction, now: datetime,322 delta: StructuredDelta, key: str) -> None:323 """people / products / locations share the same shape: unique (company_id, name_norm), first/last_seen, removed_at, status."""324 listed_status = "listed"325 existing = await fetch_all(conn, f"select id, name_norm, status, title from {table} where company_id = :c and sensor_id = :s"326 if table == "people" else f"select id, name_norm, status from {table} where company_id = :c and sensor_id = :s",327 c=company_id, s=sensor_id)328 by_norm = {r["name_norm"]: r for r in existing}329 seen: set[str] = set()330 added: list[dict[str, Any]] = []331 title_changed: list[dict[str, Any]] = []332 for it in items:333 name = getattr(it, "name", None)334 if not name:335 continue336 nn = norm_name(name)337 if not nn or nn in seen:338 continue339 seen.add(nn)340 row = by_norm.get(nn)341 if table == "people":342 payload = {"name": name, "title": it.title, "role_category": it.role_category, "is_executive": it.is_executive}343 if row is None:344 await execute(conn, """345 insert into people (id, company_id, sensor_id, name, name_norm, title, role_category, is_executive, first_seen_at, last_seen_at, status, source_url)346 values (:id, :c, :s, :name, :nn, :title, :rc, :ex, :now, :now, 'listed', :url)347 on conflict (company_id, name_norm) do update set sensor_id = excluded.sensor_id, last_seen_at = excluded.last_seen_at, status = 'listed',348 removed_at = null, title = coalesce(excluded.title, people.title), role_category = coalesce(excluded.role_category, people.role_category),349 is_executive = excluded.is_executive350 """, id=new_id("person"), c=company_id, s=sensor_id, name=name[:200], nn=nn, title=it.title, rc=it.role_category, ex=bool(it.is_executive), now=now,351 url=it.url or sensor_url)352 added.append(payload)353 else:354 if row["status"] != listed_status:355 added.append({**payload, "relisted": True})356 elif it.title and row.get("title") and norm_name(it.title) != norm_name(row["title"]):357 title_changed.append({"name": name, "before": row["title"], "after": it.title})358 await execute(conn, """update people set last_seen_at = :now, status = 'listed', removed_at = null, title = coalesce(:title, title),359 role_category = coalesce(:rc, role_category), is_executive = :ex where id = :id""",360 id=row["id"], now=now, title=it.title, rc=it.role_category, ex=bool(it.is_executive))361 elif table == "products":362 payload = {"name": name, "url": it.url}363 if row is None:364 await execute(conn, """365 insert into products (id, company_id, sensor_id, name, name_norm, category, description, url, first_seen_at, last_seen_at, status)366 values (:id, :c, :s, :name, :nn, :cat, :desc, :url, :now, :now, 'listed')367 on conflict (company_id, name_norm) do update set sensor_id = excluded.sensor_id, last_seen_at = excluded.last_seen_at, status = 'listed',368 removed_at = null, url = coalesce(excluded.url, products.url), description = coalesce(excluded.description, products.description)369 """, id=new_id("product"), c=company_id, s=sensor_id, name=name[:200], nn=nn, cat=it.category, desc=(it.description or None), url=it.url, now=now)370 added.append(payload)371 else:372 if row["status"] != listed_status:373 added.append({**payload, "relisted": True})374 await execute(conn, "update products set last_seen_at = :now, status = 'listed', removed_at = null, url = coalesce(:url, url) where id = :id",375 id=row["id"], now=now, url=it.url)376 else: # locations377 payload = {"name": name, "city": it.city, "country": it.country, "kind": it.kind}378 if row is None:379 await execute(conn, """380 insert into locations (id, company_id, sensor_id, kind, name, name_norm, city, region, country, first_seen_at, last_seen_at, status, source_url)381 values (:id, :c, :s, :kind, :name, :nn, :city, :region, :country, :now, :now, 'listed', :url)382 on conflict (company_id, name_norm) do update set sensor_id = excluded.sensor_id, last_seen_at = excluded.last_seen_at, status = 'listed',383 removed_at = null, city = coalesce(excluded.city, locations.city), country = coalesce(excluded.country, locations.country)384 """, id=new_id("location"), c=company_id, s=sensor_id, kind=it.kind or "office", name=name[:200], nn=nn, city=it.city, region=it.region,385 country=(it.country or None), now=now, url=sensor_url)386 added.append(payload)387 else:388 if row["status"] != listed_status:389 added.append({**payload, "relisted": True})390 await execute(conn, "update locations set last_seen_at = :now, status = 'listed', removed_at = null where id = :id", id=row["id"], now=now)391 removed: list[dict[str, Any]] = []392 missing = [r for r in existing if r["status"] == listed_status and r["name_norm"] not in seen]393 if missing and _trustworthy(ex, len(seen)) and not _suspicious_drop(ex, len(existing), len(seen)):394 cols = "name, title, role_category, is_executive" if table == "people" else ("name, url" if table == "products" else "name, city, country, kind")395 rows = await fetch_all(conn, f"update {table} set status = 'no_longer_listed', removed_at = :now where id = any(cast(:ids as text[])) returning {cols}",396 ids=[r["id"] for r in missing], now=now)397 removed = [dict(r) for r in rows]398 elif missing:399 delta.setdefault("notes", []).append(f"{key}: {len(missing)} missing not marked removed (untrusted extraction)")400 if added or removed or title_changed:401 entry: dict[str, Any] = {"added": added[:DELTA_LIST_LIMIT], "removed": removed[:DELTA_LIST_LIMIT]}402 if table == "people" and title_changed:403 entry["title_changed"] = title_changed[:DELTA_LIST_LIMIT]404 if table == "locations":405 prior = await fetch_all(conn, "select distinct country from locations where company_id = :c and country is not null and status = 'listed' and first_seen_at < :now",406 c=company_id, now=now)407 known = {r["country"] for r in prior}408 new_countries = sorted({a["country"] for a in added if a.get("country") and a["country"] not in known})409 if new_countries:410 entry["new_countries"] = new_countries411 delta[key] = entry412413414async def reconcile_plans(conn: Any, *, company_id: str, sensor_id: str, sensor_url: str, ex: Extraction, now: datetime, delta: StructuredDelta) -> None:415 current = await fetch_all(conn, """select id, plan_norm, plan_name, currency, billing_period, price, unit, features, contact_sales, version_no416 from pricing_plans where company_id = :c and sensor_id = :s and status = 'current'""", c=company_id, s=sensor_id)417 by_norm = {r["plan_norm"]: r for r in current}418 seen: set[str] = set()419 added: list[dict[str, Any]] = []420 price_changed: list[dict[str, Any]] = []421 for pl in ex.plans:422 nn = norm_name(pl.plan_name)423 if not nn or nn in seen:424 continue425 seen.add(nn)426 row = by_norm.get(nn)427 feats = list(pl.features or [])[:40]428 same = row is not None and (row["price"] is None) == (pl.price is None) and (row["price"] is None or abs(float(row["price"]) - float(pl.price)) < 1e-9) \429 and (row["currency"] or None) == (pl.currency or None) and (row["billing_period"] or None) == (pl.billing_period or None) \430 and bool(row["contact_sales"]) == bool(pl.contact_sales) and list(row["features"] or []) == feats431 if same:432 await execute(conn, "update pricing_plans set last_seen_at = :now where id = :id", id=row["id"], now=now)433 continue434 version = 1435 if row is not None:436 version = int(row["version_no"]) + 1437 await execute(conn, "update pricing_plans set status = 'superseded', valid_to = :now where id = :id", id=row["id"], now=now)438 if (row["price"] is None) != (pl.price is None) or (row["price"] is not None and abs(float(row["price"]) - float(pl.price)) >= 1e-9):439 before = float(row["price"]) if row["price"] is not None else None440 pct = round((pl.price - before) / before * 100, 2) if (before and pl.price is not None) else None441 price_changed.append({"plan_name": pl.plan_name, "before": before, "after": pl.price, "currency": pl.currency or row["currency"],442 "billing_period": pl.billing_period or row["billing_period"], "pct": pct})443 else:444 added.append({"plan_name": pl.plan_name, "price": pl.price, "currency": pl.currency, "billing_period": pl.billing_period})445 await execute(conn, """446 insert into pricing_plans (id, company_id, sensor_id, plan_name, plan_norm, currency, billing_period, price, price_text, unit, features, contact_sales,447 version_no, valid_from, first_seen_at, last_seen_at, status, source_url)448 values (:id, :c, :s, :name, :nn, :cur, :bp, :price, :pt, :unit, cast(:feats as jsonb), :cs, :v, :now, :now, :now, 'current', :url)449 """, id=new_id("plan"), c=company_id, s=sensor_id, name=pl.plan_name[:200], nn=nn, cur=pl.currency, bp=pl.billing_period, price=pl.price, pt=pl.price_text,450 unit=pl.unit, feats=jsonb(feats), cs=bool(pl.contact_sales), v=version, now=now, url=sensor_url)451 removed: list[dict[str, Any]] = []452 missing = [r for r in current if r["plan_norm"] not in seen]453 if missing and _trustworthy(ex, len(seen)):454 rows = await fetch_all(conn, """update pricing_plans set status = 'removed', valid_to = :now where id = any(cast(:ids as text[]))455 returning plan_name, price, currency, billing_period""", ids=[r["id"] for r in missing], now=now)456 removed = [dict(r) for r in rows]457 if added or removed or price_changed:458 delta["plans"] = {"added": added, "removed": removed, "price_changed": price_changed}459460461async def reconcile_news(conn: Any, *, company_id: str, sensor_id: str, ex: Extraction, now: datetime, delta: StructuredDelta) -> None:462 added: list[dict[str, Any]] = []463 seen: set[str] = set()464 for n in ex.news[:500]:465 canon = canonicalize_url(n.url)466 if canon in seen:467 continue468 seen.add(canon)469 row = await fetch_one(conn, """470 insert into news_items (id, company_id, sensor_id, url, canonical_url, title, summary, category, published_at, first_seen_at, language)471 values (:id, :c, :s, :url, :canon, :title, :summary, :cat, :pub, :now, :lang)472 on conflict (company_id, canonical_url) do nothing returning id473 """, id=new_id("news"), c=company_id, s=sensor_id, url=n.url[:2000], canon=canon[:2000], title=n.title[:300], summary=(n.summary or None), cat=n.category,474 pub=n.published_at, now=now, lang=n.language)475 if row is not None:476 added.append({"title": n.title, "url": n.url, "published_at": n.published_at.isoformat() if n.published_at else None, "category": n.category})477 if added:478 delta["news"] = {"added": added[:DELTA_LIST_LIMIT], "count": len(added)}479480481async def reconcile(conn: Any, *, company_id: str, sensor: dict[str, Any], ex: Extraction, now: datetime, previous_meta: dict[str, Any] | None) -> StructuredDelta:482 delta: StructuredDelta = {}483 sid, url = str(sensor["id"]), str(sensor["url"])484 surface = str(sensor.get("surface") or "")485 if ex.jobs or surface in (Surface.JOBS_BOARD, Surface.CAREERS):486 await reconcile_jobs(conn, company_id=company_id, sensor_id=sid, ex=ex, now=now, delta=delta)487 if ex.people or surface == Surface.LEADERSHIP:488 await reconcile_named(conn, table="people", company_id=company_id, sensor_id=sid, sensor_url=url, items=ex.people, ex=ex, now=now, delta=delta, key="people")489 if ex.products or surface in (Surface.PRODUCTS, Surface.SERVICES, Surface.SOLUTIONS):490 await reconcile_named(conn, table="products", company_id=company_id, sensor_id=sid, sensor_url=url, items=ex.products, ex=ex, now=now, delta=delta, key="products")491 if ex.locations or surface == Surface.LOCATIONS:492 await reconcile_named(conn, table="locations", company_id=company_id, sensor_id=sid, sensor_url=url, items=ex.locations, ex=ex, now=now, delta=delta, key="locations")493 if ex.plans or surface == Surface.PRICING:494 await reconcile_plans(conn, company_id=company_id, sensor_id=sid, sensor_url=url, ex=ex, now=now, delta=delta)495 if ex.news:496 await reconcile_news(conn, company_id=company_id, sensor_id=sid, ex=ex, now=now, delta=delta)497 meta: dict[str, Any] = {"language": ex.language}498 if previous_meta:499 if (previous_meta.get("title") or None) != (ex.title or None) and (previous_meta.get("title") or ex.title):500 meta["title_changed"] = {"before": previous_meta.get("title"), "after": ex.title}501 if (previous_meta.get("description") or None) != (ex.meta.get("description") or None):502 meta["description_changed"] = True503 delta["meta"] = meta504 if int(sensor.get("snapshot_count") or 0) == 0:505 # First snapshot of this sensor: everything it lists pre-dates our observation. Flag it so "new" counts stay honest (spec §145).506 for table in ("jobs", "people", "products", "locations", "news_items", "pricing_plans"):507 await execute(conn, f"update {table} set baseline = true where sensor_id = :sid and first_seen_at = :now", sid=sid, now=now)508 delta["baseline"] = True509 return delta510511512def _delta_counts(delta: StructuredDelta) -> dict[str, int]:513 out: dict[str, int] = {}514 for k, v in delta.items():515 if isinstance(v, dict):516 for kk in ("added", "removed", "price_changed", "title_changed", "new_countries"):517 if isinstance(v.get(kk), list) and v[kk]:518 out[f"{k}_{kk}"] = len(v[kk])519 return out520521522def _clean_label(value: str | None) -> str | None:523 if value is None:524 return None525 v = re.sub(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]", "", value)526 v = re.sub(r"\s+", " ", v).strip()527 return v or None528529530def _drop_corrupt_entities(ex: Extraction, *, connector_id: str | None = None) -> None:531 """Remove typed items whose labels carry replacement characters or no letters at all (mis-decoded or empty extractions), then apply the532 shared precision rules (`connectors/_precision`) as the last line of defence: CTA/nav anchors, cookie categories, marketing headings,533 swapped person cards. Job rules apply to HTML-scraped listings only — structured boards (ATS) are trusted."""534 def ok(label: str | None) -> bool:535 return bool(label) and "\ufffd" not in label and re.search(r"[^\W\d_]", label) is not None536537 for j in ex.jobs:538 j.title = _clean_label(j.title) or j.title539 ex.jobs = [j for j in ex.jobs if ok(j.title)]540 for p in ex.people:541 p.name = _clean_label(p.name) or p.name542 ex.people = [p for p in ex.people if ok(p.name)]543 for p in ex.products:544 p.name = _clean_label(p.name) or p.name545 ex.products = [p for p in ex.products if ok(p.name)]546 for p in ex.plans:547 p.plan_name = _clean_label(p.plan_name) or p.plan_name548 ex.plans = [p for p in ex.plans if ok(p.plan_name)]549 for loc in ex.locations:550 loc.name = _clean_label(loc.name) or loc.name551 ex.locations = [loc for loc in ex.locations if ok(loc.name)]552 for n in ex.news:553 n.title = _clean_label(n.title) or n.title554 ex.news = [n for n in ex.news if ok(n.title) and "\ufffd" not in (n.url or "")]555 if ex.title:556 ex.title = _clean_label(ex.title)557 dropped = apply_precision(ex, html_jobs=connector_id in HTML_JOB_CONNECTORS)558 if dropped:559 log.debug("precision rules dropped items", extra={"connector_id": connector_id, "dropped": dropped})560561562# ------------------------------------------------------------------------------------------------------------ the run563564565async def run_sensor(sensor: dict[str, Any], *, fetcher: Fetcher, worker: str = "local", force: bool = False, result: FetchResult | None = None,566 collection_method: str = "live") -> RunOutcome:567 """One complete run. Never raises for fetch/extract problems (they become observations); DB errors propagate."""568 t0 = time.perf_counter()569 sid = str(sensor["id"])570 now = datetime.now(UTC)571 outcome = RunOutcome(sensor_id=sid, status="failed")572 domain = str(sensor.get("domain") or registrable_domain(str(sensor["url"])))573 company_id = str(sensor["company_id"])574 cfg = dict(sensor.get("config") or {})575 connector = connectors.get(str(sensor["connector_id"]))576577 # ---- admission: domain budget / block (own short transaction)578 if result is None:579 async with transaction() as conn:580 allowed, resume_at, reason = await _check_domain_budget(conn, domain, now)581 if not allowed:582 resume = (resume_at or now + timedelta(hours=1)) + timedelta(seconds=random.uniform(60, 900))583 await execute(conn, "update sensors set next_run_at = :n, claimed_by = null, claimed_at = null, updated_at = now() where id = :id", n=resume, id=sid)584 outcome.status, outcome.error, outcome.next_run_at = "skipped", reason, resume585 outcome.duration_ms = int((time.perf_counter() - t0) * 1000)586 return outcome587 company = await fetch_one(conn, "select id, canonical_domain, tier, importance, slug from companies where id = :id", id=company_id)588 else:589 async with transaction() as conn:590 company = await fetch_one(conn, "select id, canonical_domain, tier, importance, slug from companies where id = :id", id=company_id)591 ctx = connectors.ConnectorContext(company=company or {"id": company_id})592593 # ---- network (outside any transaction)594 fetched: FetchResult | None = result595 failure: tuple[str, str, int | None] | None = None596 not_modified = False597 fetch_ms = 0598 if fetched is None:599 req_sensor = {**sensor, "etag": None, "last_modified": None} if force else sensor600 tf = time.perf_counter()601 try:602 fetched = await asyncio.wait_for(connector.fetch(ctx, req_sensor, fetcher), settings.http_timeout_s * FETCH_TIMEOUT_FACTOR)603 except NotModified as nm:604 not_modified = True605 fetch_ms = nm.duration_ms606 except (FetchError, BlockedError) as exc:607 failure = (str(exc.failure), str(exc)[:500], exc.status)608 except TimeoutError:609 failure = (str(FailureClass.TIMEOUT), "connector fetch timed out", None)610 except Exception as exc: # noqa: BLE001611 failure = (str(classify_exception(exc)), f"{exc.__class__.__name__}: {exc}"[:500], None)612 fetch_ms = fetch_ms or int((time.perf_counter() - tf) * 1000)613614 # ---- redirect to another registrable domain (keeps the observation, parks the sensor)615 if fetched is not None and result is None and registrable_domain(fetched.final_url) != registrable_domain(str(sensor["url"])):616 failure = (str(FailureClass.REDIRECT), f"redirected off-domain to {fetched.final_url}", fetched.status)617 cfg["redirect_url"] = fetched.final_url618619 async with transaction() as conn:620 base_interval = int(sensor.get("base_interval_s") or 86400)621 current = int(sensor.get("current_interval_s") or base_interval)622 quality = float(sensor.get("quality_score") or 50.0)623 obs_id = new_id("observation")624 outcome.observation_id = obs_id625626 if not_modified:627 interval = next_interval_unchanged(current, base_interval)628 nxt = _next_run(now, interval)629 await execute(conn, """630 insert into observations (id, sensor_id, company_id, fetched_at, status_code, duration_ms, transport, final_url, not_modified, changed,631 connector_version, collection_method, worker)632 values (:id, :sid, :cid, :now, 304, :ms, 'http', :url, true, false, :cv, :cm, :worker)633 """, id=obs_id, sid=sid, cid=company_id, now=now, ms=fetch_ms, url=sensor["url"], cv=connector.connector_id, cm=collection_method, worker=worker)634 quality = _quality_update(quality, 100.0)635 await execute(conn, """636 update sensors set status = 'active', last_run_at = :now, last_success_at = :now, last_status = 304, last_failure_class = null, last_error = null,637 consecutive_failures = 0, consecutive_unchanged = consecutive_unchanged + 1, observation_count = observation_count + 1,638 current_interval_s = :iv, tier = :tier, next_run_at = :nxt, quality_score = :q, claimed_by = null, claimed_at = null, updated_at = now()639 where id = :id640 """, now=now, iv=interval, tier=tier_for_interval(interval), nxt=nxt, q=quality, id=sid)641 await execute(conn, "update companies set last_observed_at = :now, first_observed_at = coalesce(first_observed_at, :now) where id = :id", now=now, id=company_id)642 await _consume_domain_budget(conn, domain, 1)643 await _ledger(conn, company_id, connector.connector_id, 1, 0)644 outcome.status, outcome.next_run_at, outcome.interval_s, outcome.sensor_status = "not_modified", nxt, interval, "active"645 outcome.duration_ms = int((time.perf_counter() - t0) * 1000)646 return outcome647648 if failure is not None:649 await _record_failure(conn, sensor=sensor, cfg=cfg, company_id=company_id, obs_id=obs_id, failure=failure, now=now, fetch_ms=fetch_ms,650 connector_id=connector.connector_id, worker=worker, current=current, quality=quality, outcome=outcome,651 collection_method=collection_method, final_url=fetched.final_url if fetched else None)652 await _consume_domain_budget(conn, domain, 1)653 await _ledger(conn, company_id, connector.connector_id, 1, 0)654 outcome.duration_ms = int((time.perf_counter() - t0) * 1000)655 return outcome656657 assert fetched is not None658 pages = int(fetched.headers.get("x-companyatlas-pages", "1") or 1)659 object_key, _stored, _created = archive.put_bytes(fetched.content)660661 # ---- extract662 try:663 ex = connector.extract(sensor, fetched)664 except Exception as exc: # noqa: BLE001665 failure = (str(FailureClass.PARSING), f"extract failed: {exc.__class__.__name__}: {exc}"[:500], fetched.status)666 await _record_failure(conn, sensor=sensor, cfg=cfg, company_id=company_id, obs_id=obs_id, failure=failure, now=now, fetch_ms=fetch_ms,667 connector_id=connector.connector_id, worker=worker, current=current, quality=quality, outcome=outcome,668 collection_method=collection_method, final_url=fetched.final_url, object_key=object_key, size=len(fetched.content),669 content_type=fetched.content_type)670 await _consume_domain_budget(conn, domain, pages)671 await _ledger(conn, company_id, connector.connector_id, pages, len(fetched.content))672 outcome.duration_ms = int((time.perf_counter() - t0) * 1000)673 return outcome674675 # ---- corruption guard (spec: never fabricate): mis-decoded bodies never become snapshots, entities or events676 quality = text_quality(ex.text) if ex.text else 1.0677 if quality < 0.99 or (ex.title and "\ufffd" in ex.title):678 failure = (str(FailureClass.PARSING), f"extracted text is corrupt (quality {quality:.3f}, content-type {fetched.content_type[:40]!r})", fetched.status)679 await _record_failure(conn, sensor=sensor, cfg=cfg, company_id=company_id, obs_id=obs_id, failure=failure, now=now, fetch_ms=fetch_ms,680 connector_id=connector.connector_id, worker=worker, current=current, quality=quality, outcome=outcome,681 collection_method=collection_method, final_url=fetched.final_url, object_key=object_key, size=len(fetched.content),682 content_type=fetched.content_type)683 await _consume_domain_budget(conn, domain, pages)684 await _ledger(conn, company_id, connector.connector_id, pages, len(fetched.content))685 outcome.duration_ms = int((time.perf_counter() - t0) * 1000)686 return outcome687 _drop_corrupt_entities(ex, connector_id=connector.connector_id)688 ex.normalized_hash = ex.normalized_hash or text_hash(ex.text)689 ex.structured_hash = ex.structured_hash or _structured_hash(ex)690 struct_hash = structural_hash(ex.blocks)691 unchanged = (not force and sensor.get("last_normalized_hash") == ex.normalized_hash and (cfg.get("last_structured_hash") or "") == ex.structured_hash692 and sensor.get("last_snapshot_id") is not None)693 etag, last_mod = fetched.etag, fetched.last_modified694695 await execute(conn, """696 insert into observations (id, sensor_id, company_id, fetched_at, status_code, duration_ms, transport, final_url, redirects, not_modified, changed,697 content_hash, normalized_hash, structural_hash, object_key, size_bytes, content_type, connector_version, collection_method, worker)698 values (:id, :sid, :cid, :now, :st, :ms, :tr, :url, :redir, false, :changed, :ch, :nh, :sh, :ok, :size, :ct, :cv, :cm, :worker)699 """, id=obs_id, sid=sid, cid=company_id, now=now, st=fetched.status, ms=fetch_ms, tr=fetched.transport, url=fetched.final_url, redir=fetched.redirects,700 changed=not unchanged, ch=fetched.sha256, nh=ex.normalized_hash, sh=struct_hash, ok=object_key, size=len(fetched.content),701 ct=fetched.content_type[:100], cv=connector.connector_id, cm=collection_method, worker=worker)702 await _consume_domain_budget(conn, domain, pages)703 await _ledger(conn, company_id, connector.connector_id, pages, len(fetched.content))704 extraction_conf = 100.0 if (ex.meta.get("structured") or len(ex.text) > 200) else 60.0705706 if unchanged:707 interval = next_interval_unchanged(current, base_interval)708 nxt = _next_run(now, interval)709 quality = _quality_update(quality, extraction_conf)710 await execute(conn, """711 update sensors set status = 'active', last_run_at = :now, last_success_at = :now, last_status = :st, last_failure_class = null, last_error = null,712 consecutive_failures = 0, consecutive_unchanged = consecutive_unchanged + 1, observation_count = observation_count + 1,713 current_interval_s = :iv, tier = :tier, next_run_at = :nxt, quality_score = :q, etag = coalesce(:etag, etag),714 last_modified = coalesce(:lm, last_modified), last_content_hash = :ch, claimed_by = null, claimed_at = null, updated_at = now()715 where id = :id716 """, now=now, st=fetched.status, iv=interval, tier=tier_for_interval(interval), nxt=nxt, q=quality, etag=etag, lm=last_mod, ch=fetched.sha256, id=sid)717 await execute(conn, "update companies set last_observed_at = :now, first_observed_at = coalesce(first_observed_at, :now) where id = :id", now=now, id=company_id)718 outcome.status, outcome.next_run_at, outcome.interval_s, outcome.sensor_status = "unchanged", nxt, interval, "active"719 outcome.duration_ms = int((time.perf_counter() - t0) * 1000)720 return outcome721722 # ---- changed (or first run / forced): snapshot + reconciliation + diff723 prev_id = sensor.get("last_snapshot_id")724 prev = await fetch_one(conn, "select id, version_no, blocks_key, text_key, title, extracted from snapshots where id = :id", id=prev_id) if prev_id else None725 prev_blocks: list[Block] = []726 prev_text = ""727 if prev is not None:728 try:729 if prev.get("blocks_key") and archive.exists(prev["blocks_key"]):730 prev_blocks = _blocks_from_json(archive.get_text(prev["blocks_key"]))731 if prev.get("text_key") and archive.exists(prev["text_key"]):732 prev_text = archive.get_text(prev["text_key"])733 except OSError:734 log.warning("previous snapshot objects unreadable", extra={"sensor_id": sid, "snapshot_id": prev_id})735 prev_meta = None736 if prev is not None:737 pm = (prev.get("extracted") or {}).get("meta") if isinstance(prev.get("extracted"), dict) else {}738 prev_meta = {"title": prev.get("title"), "description": (pm or {}).get("description")}739740 delta = await reconcile(conn, company_id=company_id, sensor=sensor, ex=ex, now=now, previous_meta=prev_meta)741 history = {"consecutive_unchanged": sensor.get("consecutive_unchanged") or 0, "observation_count": sensor.get("observation_count") or 0,742 "change_count": sensor.get("change_count") or 0}743 diff: BlockDiff = compare(prev_blocks, ex.blocks, surface=str(sensor["surface"]), before_text=prev_text, after_text=ex.text,744 structured_delta=delta, history=history) if prev is not None else BlockDiff()745 kind = change_kind(diff.significance, noise=settings.noise_threshold, meaningful=settings.meaningful_threshold, major=settings.major_threshold,746 critical=settings.critical_threshold) if prev is not None else ChangeKind.NOISE747 typed = any(k in delta and any(isinstance(v, list) and v for v in delta[k].values()) for k in ("jobs", "people", "products", "plans", "locations", "news")748 if isinstance(delta.get(k), dict))749 keep_snapshot = prev is None or kind != ChangeKind.NOISE or typed or settings.keep_noise_snapshots750 snap_id: str | None = None751 change_id: str | None = None752 version_no = int(sensor.get("snapshot_count") or 0) + 1753 if keep_snapshot:754 snap_id = new_id("snapshot")755 text_key, _s, _c = archive.put_text(ex.text)756 blocks_key, _s, _c = archive.put_text(_blocks_json(ex.blocks))757 extracted = _bounded_extracted(ex)758 await execute(conn, """759 insert into snapshots (id, sensor_id, company_id, observation_id, previous_snapshot_id, version_no, fetched_at, content_hash, normalized_hash,760 structural_hash, object_key, text_key, blocks_key, extracted, extracted_summary, title, language, size_bytes, text_length,761 block_count, content_type, connector_version, collection_method)762 values (:id, :sid, :cid, :oid, :prev, :v, :now, :ch, :nh, :sh, :ok, :tk, :bk, cast(:ex as jsonb), cast(:sum as jsonb), :title, :lang, :size, :tl, :bc,763 :ct, :cv, :cm)764 """, id=snap_id, sid=sid, cid=company_id, oid=obs_id, prev=prev_id, v=version_no, now=now, ch=fetched.sha256, nh=ex.normalized_hash, sh=struct_hash,765 ok=object_key, tk=text_key, bk=blocks_key, ex=jsonb(extracted), sum=jsonb(ex.summary()), title=(ex.title or None), lang=ex.language,766 size=len(fetched.content), tl=len(ex.text), bc=len(ex.blocks), ct=fetched.content_type[:100], cv=connector.connector_id, cm=collection_method)767 if prev is not None:768 change_id = new_id("change")769 status = "pending" if kind in (ChangeKind.MEANINGFUL, ChangeKind.MAJOR, ChangeKind.CRITICAL) else "archived"770 await execute(conn, """771 insert into changes (id, sensor_id, company_id, surface, snapshot_before, snapshot_after, detected_at, significance, kind, blocks_added,772 blocks_removed, blocks_modified, blocks_moved, text_delta_ratio, similarity, diff, structured_delta, status, diff_version)773 values (:id, :sid, :cid, :surface, :before, :after, :now, :sig, :kind, :ba, :br, :bm, :bmv, :tdr, :sim, cast(:diff as jsonb), cast(:sd as jsonb),774 :status, :dv)775 """, id=change_id, sid=sid, cid=company_id, surface=str(sensor["surface"]), before=prev_id, after=snap_id, now=now, sig=diff.significance,776 kind=str(kind), ba=len(diff.added), br=len(diff.removed), bm=len(diff.modified), bmv=len(diff.moved), tdr=diff.text_delta_ratio,777 sim=diff.similarity, diff=jsonb(diff.to_json()), sd=jsonb(delta), status=status, dv=DIFF_VERSION)778 meaningful = prev is not None and kind in (ChangeKind.MEANINGFUL, ChangeKind.MAJOR, ChangeKind.CRITICAL)779 if prev is None:780 interval = current781 else:782 interval = next_interval_changed(kind, current, base_interval)783 nxt = _next_run(now, interval)784 quality = _quality_update(quality, extraction_conf)785 cfg["last_structured_hash"] = ex.structured_hash786 cfg.pop("redirect_url", None)787 await execute(conn, """788 update sensors set status = 'active', last_run_at = :now, last_success_at = :now, last_status = :st, last_failure_class = null, last_error = null,789 consecutive_failures = 0, consecutive_unchanged = case when :meaningful then 0 else consecutive_unchanged + 1 end,790 observation_count = observation_count + 1, snapshot_count = case when :kept then snapshot_count + 1 else snapshot_count end,791 change_count = case when :haschange then change_count + 1 else change_count end,792 meaningful_change_count = case when :meaningful then meaningful_change_count + 1 else meaningful_change_count end,793 last_change_at = case when :haschange then :now else last_change_at end,794 last_meaningful_change_at = case when :meaningful then :now else last_meaningful_change_at end,795 last_content_hash = :ch, last_normalized_hash = :nh, last_structural_hash = :sh, last_snapshot_id = coalesce(:snap, last_snapshot_id),796 etag = :etag, last_modified = :lm, current_interval_s = :iv, tier = :tier, next_run_at = :nxt, quality_score = :q,797 config = config || cast(:cfg as jsonb), claimed_by = null, claimed_at = null, updated_at = now()798 where id = :id799 """, now=now, st=fetched.status, meaningful=meaningful, kept=keep_snapshot, haschange=change_id is not None, ch=fetched.sha256, nh=ex.normalized_hash,800 sh=struct_hash, snap=snap_id, etag=etag, lm=last_mod, iv=interval, tier=tier_for_interval(interval), nxt=nxt, q=quality,801 cfg=jsonb({"last_structured_hash": ex.structured_hash, "redirect_url": None}), id=sid)802 await execute(conn, """803 update companies set last_observed_at = :now, first_observed_at = coalesce(first_observed_at, :now),804 last_change_at = case when :meaningful then :now else last_change_at end,805 stats = stats || jsonb_build_object('observations', coalesce((stats->>'observations')::int, 0) + 1,806 'snapshots', coalesce((stats->>'snapshots')::int, 0) + case when :kept then 1 else 0 end,807 'changes', coalesce((stats->>'changes')::int, 0) + case when :haschange then 1 else 0 end,808 'meaningful_changes', coalesce((stats->>'meaningful_changes')::int, 0) + case when :meaningful then 1 else 0 end),809 updated_at = now()810 where id = :id811 """, now=now, meaningful=meaningful, kept=keep_snapshot, haschange=change_id is not None, id=company_id)812 outcome.status = "changed" if prev is not None else "ok"813 outcome.snapshot_id, outcome.change_id = snap_id, change_id814 outcome.significance = diff.significance if prev is not None else None815 outcome.kind = str(kind) if prev is not None else None816 outcome.next_run_at, outcome.interval_s, outcome.sensor_status = nxt, interval, "active"817 outcome.delta_counts = _delta_counts(delta)818 outcome.duration_ms = int((time.perf_counter() - t0) * 1000)819 log.info("sensor run", extra={"sensor_id": sid, "surface": sensor.get("surface"), "status": outcome.status, "kind": outcome.kind, "significance": outcome.significance,820 "delta": outcome.delta_counts, "interval_s": interval, "ms": outcome.duration_ms})821 return outcome822823824async def _record_failure(conn: Any, *, sensor: dict[str, Any], cfg: dict[str, Any], company_id: str, obs_id: str, failure: tuple[str, str, int | None], now: datetime,825 fetch_ms: int, connector_id: str, worker: str, current: int, quality: float, outcome: RunOutcome, collection_method: str,826 final_url: str | None = None, object_key: str | None = None, size: int | None = None, content_type: str | None = None) -> None:827 fclass, message, status_code = failure828 sid = str(sensor["id"])829 await execute(conn, """830 insert into observations (id, sensor_id, company_id, fetched_at, status_code, duration_ms, transport, final_url, not_modified, changed, failure_class, error,831 object_key, size_bytes, content_type, connector_version, collection_method, worker)832 values (:id, :sid, :cid, :now, :st, :ms, 'http', :url, false, false, :fc, :err, :ok, :size, :ct, :cv, :cm, :worker)833 """, id=obs_id, sid=sid, cid=company_id, now=now, st=status_code, ms=fetch_ms, url=final_url or sensor["url"], fc=fclass, err=message, ok=object_key, size=size,834 ct=(content_type or "")[:100] or None, cv=connector_id, cm=collection_method, worker=worker)835 await execute(conn, "insert into failures (id, sensor_id, company_id, at, failure_class, status_code, message, url) values (:id, :sid, :cid, :now, :fc, :st, :msg, :url)",836 id=new_id("failure"), sid=sid, cid=company_id, now=now, fc=fclass, st=status_code, msg=message[:500], url=sensor["url"])837 failures = int(sensor.get("consecutive_failures") or 0) + 1838 _mult, threshold = FAILURE_POLICY.get(fclass, FAILURE_POLICY[FailureClass.UNKNOWN])839 prev_status = str(sensor.get("status") or "active")840 status = prev_status if prev_status in (SensorStatus.PAUSED, SensorStatus.RETIRED) else SensorStatus.ACTIVE841 review_kind: str | None = None842 if fclass == FailureClass.REDIRECT and cfg.get("redirect_url"):843 status, review_kind = SensorStatus.REDIRECTED, "sensor_migration"844 elif fclass == FailureClass.ROBOTS:845 status, review_kind = SensorStatus.BLOCKED, "blocked_source"846 elif fclass == FailureClass.BOT_CHALLENGE:847 status = SensorStatus.FAILING if failures >= threshold else status848 review_kind = "blocked_source" if failures >= threshold else None849 elif failures >= settings.retire_after_failures:850 status = SensorStatus.RETIRED851 elif failures >= settings.stale_after_failures:852 status = SensorStatus.STALE853 elif failures >= threshold:854 status = SensorStatus.FAILING855 interval = next_interval_failed(fclass, current)856 nxt = _next_run(now, interval)857 quality = _quality_update(quality, 0.0)858 retire = status == SensorStatus.RETIRED859 await execute(conn, """860 update sensors set status = :status, last_run_at = :now, last_status = :st, last_failure_class = :fc, last_error = :err, consecutive_failures = :cf,861 consecutive_unchanged = 0, observation_count = observation_count + 1, current_interval_s = :iv, tier = :tier, next_run_at = :nxt, quality_score = :q,862 config = config || cast(:cfg as jsonb), retired_at = case when :retire then coalesce(retired_at, :now) else retired_at end,863 claimed_by = null, claimed_at = null, updated_at = now()864 where id = :id865 """, status=str(status), now=now, st=status_code, fc=fclass, err=message[:500], cf=failures, iv=interval, tier=tier_for_interval(interval), nxt=nxt, q=quality,866 cfg=jsonb({k: v for k, v in cfg.items() if k in ("redirect_url",)}), retire=retire, id=sid)867 if review_kind:868 existing = await fetch_one(conn, "select id from review_queue where kind = :k and ref_id = :r and status = 'open'", k=review_kind, r=sid)869 if existing is None:870 await execute(conn, """insert into review_queue (id, kind, ref_id, company_id, payload) values (:id, :k, :r, :cid, cast(:p as jsonb))""",871 id=new_id("review"), k=review_kind, r=sid, cid=company_id,872 p=jsonb({"sensor_id": sid, "url": sensor["url"], "surface": sensor.get("surface"), "failure_class": fclass, "message": message[:300],873 "redirect_url": cfg.get("redirect_url"), "consecutive_failures": failures}))874 await execute(conn, "update companies set last_observed_at = :now where id = :id", now=now, id=company_id)875 outcome.status = "redirected" if status == SensorStatus.REDIRECTED else "failed"876 outcome.failure_class, outcome.error, outcome.next_run_at, outcome.interval_s, outcome.sensor_status = fclass, message, nxt, interval, str(status)877878879# ------------------------------------------------------------------------------------------------------------ convenience entry points880881882async def load_sensor(sensor_ref: str) -> dict[str, Any] | None:883 async with transaction() as conn:884 row = await fetch_one(conn, "select * from sensors where id = :ref", ref=sensor_ref)885 if row is None:886 row = await fetch_one(conn, "select * from sensors where canonical_url = :c order by created_at limit 1", c=canonicalize_url(sensor_ref))887 if row is None:888 row = await fetch_one(conn, "select * from sensors where url = :u order by created_at limit 1", u=sensor_ref)889 return row890891892async def run_sensor_ids(ids: list[str], *, fetcher: Fetcher, worker: str = "local", force: bool = False, concurrency: int | None = None) -> list[RunOutcome]:893 sem = asyncio.Semaphore(max(1, concurrency or settings.fetch_concurrency))894 out: list[RunOutcome] = []895896 async def one(sid: str) -> None:897 row = await load_sensor(sid)898 if row is None:899 out.append(RunOutcome(sensor_id=sid, status="failed", error="sensor not found"))900 return901 async with sem:902 try:903 out.append(await run_sensor(row, fetcher=fetcher, worker=worker, force=force))904 except Exception as exc:905 log.exception("sensor run crashed", extra={"sensor_id": sid})906 out.append(RunOutcome(sensor_id=sid, status="failed", error=f"{exc.__class__.__name__}: {exc}"[:300]))907 async with transaction() as conn:908 await execute(conn, "update sensors set claimed_by = null, claimed_at = null, next_run_at = now() + interval '30 minutes' where id = :id", id=sid)909910 await asyncio.gather(*(one(s) for s in ids))911 return out912913914async def run_sensor_by_url_with_file(sensor_ref: str, path: str, *, worker: str = "file", force: bool = False, content_type: str | None = None) -> RunOutcome:915 """Run a sensor against a fixture file instead of the network (`catlas run-sensor --file`)."""916 row = await load_sensor(sensor_ref)917 if row is None:918 raise LookupError(f"sensor {sensor_ref!r} not found")919 res = file_result(path, url=str(row["url"]), content_type=content_type or _guess_content_type(path))920 return await run_sensor(row, fetcher=Fetcher(), worker=worker, force=force, result=res)921922923def surface_importance(surface: str) -> float:924 return float(SURFACE_IMPORTANCE.get(surface, 0.3))925926927__all__ = ["PIPELINE_VERSION", "RunOutcome", "is_ai_title", "load_sensor", "next_interval_changed", "next_interval_failed", "next_interval_unchanged", "reconcile",928 "run_sensor", "run_sensor_by_url_with_file", "run_sensor_ids"]929