spb/company-atlas
Public
Python 66.3%
TypeScript 22.7%
JavaScript 8.6%
HTML 1.4%
CSS 0.7%
1"""Sensor auto-repair (spec §60, §189): failing / stale / redirected / PAGE_REMOVED sensors.23 retry ─▶ (redirect target) ─▶ re-fetch sitemap ─▶ rediscover navigation ─▶ candidate URL for the same surface4 ─▶ content-identity check (simhash / fuzzy ratio against the last snapshot text) ─▶ migrate (old retired with5 config.successor_id, new sensor with config.predecessor_id, history kept) or review_queue(kind='sensor_migration').67Registered as a periodic task (every 30 min, bounded batch). Bounded requests per sensor; never raises out of the batch.8"""9from __future__ import annotations1011import logging12import re13from datetime import UTC, datetime14from typing import Any15from urllib.parse import urlparse1617from rapidfuzz import fuzz1819from companyatlas import archive20from companyatlas.config import settings21from companyatlas.db import execute, fetch_all, fetch_one, jsonb, transaction22from companyatlas.fetch import BlockedError, Fetcher, FetchError, FetchResult, NotModified23from companyatlas.ids import new_id24from companyatlas.sdk import connector as connectors25from companyatlas.sdk import normalize26from companyatlas.sdk.normalize import hamming, normalized_text, simhash27from companyatlas.services.periodic import periodic28from companyatlas.taxonomy import SensorStatus, Surface29from companyatlas.urls import canonicalize_url, classify_url, registrable_domain, same_company_host3031log = logging.getLogger(__name__)3233REPAIR_VERSION = "repair-v1"34MAX_REQUESTS_PER_SENSOR = 635IDENTITY_HAMMING = 1236IDENTITY_RATIO = 0.637MIN_CANDIDATE_CONF = 0.738SOFT_404_RE = re.compile(r"(page not found|404|not be found|doesn'?t exist|no longer available)", re.IGNORECASE)394041class _Session:42 def __init__(self, fetcher: Fetcher):43 self.fetcher = fetcher44 self.requests = 04546 async def get(self, url: str, *, max_bytes: int = 2 * 1024 * 1024, accept: str | None = None) -> FetchResult | None:47 if self.requests >= MAX_REQUESTS_PER_SENSOR:48 return None49 self.requests += 150 try:51 return await self.fetcher.get(url, max_bytes=max_bytes, retries=0, accept=accept)52 except NotModified:53 return None54 except (FetchError, BlockedError) as exc:55 log.debug("repair fetch failed", extra={"url": url, "failure": str(exc.failure)})56 return None57 except Exception as exc: # noqa: BLE00158 log.debug("repair fetch error", extra={"url": url, "error": str(exc)[:200]})59 return None606162def _soft_404(res: FetchResult) -> bool:63 m = re.search(r"<title[^>]*>(.*?)</title>", res.text[:4000], re.IGNORECASE | re.DOTALL)64 return bool(m and SOFT_404_RE.search(m.group(1)))656667async def _last_text(sensor: dict[str, Any]) -> str:68 sid = sensor.get("last_snapshot_id")69 if not sid:70 return ""71 async with transaction() as conn:72 row = await fetch_one(conn, "select text_key from snapshots where id = :id", id=sid)73 if row and row.get("text_key") and archive.exists(row["text_key"]):74 try:75 return archive.get_text(row["text_key"])76 except OSError:77 return ""78 return ""798081def content_identity(previous_text: str, candidate_text: str) -> tuple[bool, float]:82 """(same_content, score). simhash distance first, fuzzy ratio second — both on noise-normalised text."""83 if not previous_text or not candidate_text:84 return False, 0.085 d = hamming(simhash(previous_text), simhash(candidate_text))86 ratio = fuzz.ratio(normalized_text(previous_text)[:20000], normalized_text(candidate_text)[:20000]) / 100.087 return (d <= IDENTITY_HAMMING or ratio >= IDENTITY_RATIO), round(max(ratio, 1 - d / 64), 3)888990async def _candidates(session: _Session, sensor: dict[str, Any], company: dict[str, Any]) -> list[tuple[str, float, str]]:91 """(url, confidence, method) for the sensor's surface: redirect target, sitemap entries, navigation links."""92 surface = str(sensor["surface"])93 canonical = str(company.get("canonical_domain") or sensor["domain"])94 out: list[tuple[str, float, str]] = []95 cfg = sensor.get("config") or {}96 if cfg.get("redirect_url"):97 out.append((str(cfg["redirect_url"]), 0.9, "redirect"))98 origin = f"https://{urlparse(str(company.get('website') or 'https://' + canonical)).netloc or canonical}"99 # sitemap100 async with transaction() as conn:101 sm = await fetch_one(conn, "select url from sensors where company_id = :c and surface = 'sitemap' and status <> 'retired' order by quality_score desc limit 1",102 c=company["id"])103 sm_url = str(sm["url"]) if sm else f"{origin}/sitemap.xml"104 res = await session.get(sm_url, max_bytes=4 * 1024 * 1024, accept="application/xml,text/xml,*/*;q=0.5")105 if res is not None and not res.is_html:106 from companyatlas.connectors.sitemap import _decode, parse_sitemap107108 pages, children = parse_sitemap(_decode(res.content))109 if children and not pages:110 child = await session.get(children[0][0], max_bytes=4 * 1024 * 1024, accept="application/xml,text/xml,*/*;q=0.5")111 if child is not None:112 pages, _ = parse_sitemap(_decode(child.content))113 for loc, _lm in pages[: settings.discovery_max_sitemap_urls]:114 if not same_company_host(loc, canonical):115 continue116 s, c = classify_url(loc, canonical_domain=canonical)117 if s == surface and c >= MIN_CANDIDATE_CONF:118 out.append((loc, c * 0.9, "sitemap"))119 # navigation120 home = await session.get(origin + "/")121 if home is not None and home.is_html:122 page = normalize.parse(home.text, url=home.final_url)123 for ln in page.links:124 if not same_company_host(ln.url, canonical):125 continue126 s, c = classify_url(ln.url, anchor=ln.anchor, canonical_domain=canonical)127 if s == surface and c >= MIN_CANDIDATE_CONF:128 out.append((ln.url, c, "nav"))129 # dedupe, exclude the broken URL itself130 seen = {canonicalize_url(str(sensor["url"]))}131 uniq: list[tuple[str, float, str]] = []132 for url, conf, method in sorted(out, key=lambda x: -x[1]):133 cu = canonicalize_url(url)134 if cu in seen:135 continue136 seen.add(cu)137 uniq.append((url, conf, method))138 return uniq[:5]139140141async def migrate_sensor(sensor: dict[str, Any], new_url: str, *, reason: str, confidence: float, identity_score: float) -> str:142 """Retire the old sensor (successor_id) and create the replacement (predecessor_id). Returns the new sensor id."""143 now = datetime.now(UTC)144 surface = str(sensor["surface"])145 connector = connectors.for_surface(surface, new_url)146 new_id_ = new_id("sensor")147 old_cfg = dict(sensor.get("config") or {})148 new_cfg = {k: v for k, v in old_cfg.items() if k in ("canonical_domain", "vendor", "token")}149 new_cfg.update({"predecessor_id": sensor["id"], "repair": {"version": REPAIR_VERSION, "reason": reason, "confidence": confidence, "identity": identity_score,150 "at": now.isoformat(), "from_url": sensor["url"]}})151 async with transaction() as conn:152 existing = await fetch_one(conn, "select id from sensors where company_id = :c and canonical_url = :u", c=sensor["company_id"], u=canonicalize_url(new_url))153 if existing is not None:154 new_id_ = str(existing["id"])155 await execute(conn, "update sensors set status = case when status = 'retired' then 'pending' else status end, next_run_at = now(), updated_at = now() where id = :id",156 id=new_id_)157 else:158 await execute(conn, """159 insert into sensors (id, company_id, surface, connector_id, url, canonical_url, domain, discovery_confidence, discovery_method, quality_score, status, tier,160 base_interval_s, current_interval_s, next_run_at, priority, config)161 values (:id, :c, :surface, :conn, :url, :curl, :domain, :conf, 'repair', :q, 'pending', :tier, :base, :base, now(), :prio, cast(:cfg as jsonb))162 """, id=new_id_, c=sensor["company_id"], surface=surface, conn=connector.connector_id, url=new_url, curl=canonicalize_url(new_url),163 domain=registrable_domain(new_url), conf=round(confidence, 3), q=max(30.0, float(sensor.get("quality_score") or 50) * 0.9),164 tier=sensor.get("tier") or "D", base=int(sensor.get("base_interval_s") or 86400), prio=float(sensor.get("priority") or 0.5), cfg=jsonb(new_cfg))165 await execute(conn, """update sensors set status = 'retired', retired_at = coalesce(retired_at, now()), claimed_by = null, claimed_at = null,166 config = config || cast(:cfg as jsonb), updated_at = now() where id = :id""",167 id=sensor["id"], cfg=jsonb({"successor_id": new_id_, "retired_reason": reason}))168 await execute(conn, "update review_queue set status = 'resolved', resolution = :r, resolved_at = now() where ref_id = :id and status = 'open'",169 id=sensor["id"], r=f"migrated to {new_id_}")170 return new_id_171172173async def repair_sensor(sensor: dict[str, Any], *, fetcher: Fetcher, dry_run: bool = False) -> dict[str, Any]:174 session = _Session(fetcher)175 sid = str(sensor["id"])176 result: dict[str, Any] = {"sensor_id": sid, "surface": sensor["surface"], "url": sensor["url"], "action": "none", "requests": 0}177 async with transaction() as conn:178 company = await fetch_one(conn, "select id, canonical_domain, website from companies where id = :id", id=sensor["company_id"])179 if company is None:180 result["action"] = "orphan"181 return result182 # 1. retry the URL itself183 res = await session.get(str(sensor["url"]))184 if res is not None and res.status == 200 and registrable_domain(res.final_url) == registrable_domain(str(sensor["url"])) and not (res.is_html and _soft_404(res)):185 result["action"] = "recovered"186 if not dry_run:187 async with transaction() as conn:188 await execute(conn, """update sensors set status = 'active', consecutive_failures = 0, next_run_at = now(), claimed_by = null, claimed_at = null,189 config = config - 'redirect_url', updated_at = now() where id = :id""", id=sid)190 result["requests"] = session.requests191 return result192 if res is not None and registrable_domain(res.final_url) != registrable_domain(str(sensor["url"])):193 sensor = {**sensor, "config": {**(sensor.get("config") or {}), "redirect_url": res.final_url}}194 # 2–4. candidates + identity check195 previous = await _last_text(sensor)196 cands = await _candidates(session, sensor, company)197 result["candidates"] = [{"url": u, "confidence": c, "method": m} for u, c, m in cands]198 best: tuple[str, float, str, float] | None = None199 for url, conf, method in cands[:3]:200 r = await session.get(url)201 if r is None or r.status != 200 or (r.is_html and _soft_404(r)):202 continue203 text = normalize.parse(r.text, url=r.final_url).text if r.is_html else r.text204 same, score = content_identity(previous, text)205 result.setdefault("checked", []).append({"url": url, "identity": score, "same": same})206 if same and (best is None or score > best[3]):207 best = (r.final_url, conf, method, score)208 elif best is None and method == "redirect" and conf >= 0.9 and not previous:209 best = (r.final_url, conf, method, 0.0)210 result["requests"] = session.requests211 if best is not None:212 result["action"] = "migrated" if not dry_run else "would_migrate"213 result["new_url"] = best[0]214 if not dry_run:215 result["new_sensor_id"] = await migrate_sensor(sensor, best[0], reason=best[2], confidence=best[1], identity_score=best[3])216 return result217 # uncertain: one strong candidate → human review; none → leave the sensor to its failure policy218 if cands and not dry_run:219 async with transaction() as conn:220 existing = await fetch_one(conn, "select id from review_queue where kind = 'sensor_migration' and ref_id = :r and status = 'open'", r=sid)221 if existing is None:222 await execute(conn, "insert into review_queue (id, kind, ref_id, company_id, payload) values (:id, 'sensor_migration', :r, :c, cast(:p as jsonb))",223 id=new_id("review"), r=sid, c=sensor["company_id"], p=jsonb({"sensor_id": sid, "url": sensor["url"], "surface": sensor["surface"],224 "candidates": result["candidates"], "checked": result.get("checked", [])}))225 result["action"] = "review"226 elif cands:227 result["action"] = "would_review"228 async with transaction() as conn:229 await execute(conn, "update sensors set config = config || cast(:c as jsonb), updated_at = now() where id = :id", id=sid,230 c=jsonb({"last_repair_at": datetime.now(UTC).isoformat(), "last_repair_action": result["action"]}))231 return result232233234async def repair_batch(limit: int = 50, *, fetcher: Fetcher | None = None, dry_run: bool = False) -> dict[str, Any]:235 own = fetcher is None236 fetcher = fetcher or Fetcher()237 if own:238 await fetcher.open()239 stats: dict[str, Any] = {"examined": 0, "recovered": 0, "migrated": 0, "review": 0, "none": 0, "results": []}240 try:241 async with transaction() as conn:242 rows = await fetch_all(conn, """243 select * from sensors244 where (status in ('failing', 'stale', 'redirected') or (status = 'active' and last_failure_class = 'PAGE_REMOVED' and consecutive_failures >= 2))245 and (config->>'last_repair_at' is null or (config->>'last_repair_at')::timestamptz < now() - interval '1 day')246 and surface <> :sitemap247 order by quality_score desc, last_run_at nulls first limit :n248 """, n=limit, sitemap=str(Surface.SITEMAP))249 for row in rows:250 stats["examined"] += 1251 try:252 r = await repair_sensor(row, fetcher=fetcher, dry_run=dry_run)253 except Exception as exc:254 log.exception("repair failed", extra={"sensor_id": row.get("id")})255 r = {"sensor_id": row.get("id"), "action": "error", "error": str(exc)[:200]}256 key = r["action"].removeprefix("would_")257 stats[key] = stats.get(key, 0) + 1258 stats["results"].append(r)259 finally:260 if own:261 await fetcher.close()262 return stats263264265@periodic("sensor-repair", every_s=1800, initial_delay_s=120)266async def repair_task() -> None:267 stats = await repair_batch(limit=50)268 if stats["examined"]:269 log.info("repair batch", extra={k: v for k, v in stats.items() if k != "results"})270271272def sensor_status_values() -> tuple[str, ...]:273 return tuple(s.value for s in SensorStatus)274275276__all__ = ["REPAIR_VERSION", "content_identity", "migrate_sensor", "repair_batch", "repair_sensor", "repair_task"]277