"""Discovery engine (spec §10–11, §101–102): canonical domain → homepage → robots → sitemaps → navigation → ATS / feeds / subdomains → URL classification → one sensor per surface. discover_company(company, fetcher=…, dry_run=False) -> DiscoveryResult (never raises; ~90 s budget per company) onboard_pending(limit, concurrency) (consumes queue_jobs kind='discover' + pending companies) Every request goes through `fetch.Fetcher` (SSRF guard, robots, governor). Discovery is bounded: ≤ `MAX_PROBES` common-path probes, ≤ `MAX_SUBDOMAIN_GETS` subdomain fetches, one sitemap index (+ a few children), one ATS verification. Sensors are inserted with `status='pending'`, staggered `next_run_at`, and a `quality_score` seeded from confidence × surface importance × fetch reliability. """ from __future__ import annotations import asyncio import logging import random import re import socket import time from dataclasses import dataclass, field from datetime import UTC, datetime, timedelta from typing import Any from urllib.parse import urlparse from companyatlas.config import settings from companyatlas.connectors._util import ats_sensor_spec from companyatlas.db import execute, fetch_all, fetch_one, jsonb, transaction from companyatlas.fetch import BlockedError, Fetcher, FetchError, FetchResult, NotModified from companyatlas.ids import new_id from companyatlas.sdk import connector as connectors from companyatlas.sdk.models import DiscoveredUrl from companyatlas.taxonomy import ( SURFACE_BASE_INTERVAL_S, SURFACE_IMPORTANCE, FailureClass, OnboardingStatus, SensorStatus, Surface, tier_for_interval, ) from companyatlas.urls import ( absolutize, canonicalize_url, classify_url, detect_ats, is_static_asset, looks_like_trap, registrable_domain, same_company_host, ) log = logging.getLogger(__name__) DISCOVERY_VERSION = "discovery-v1" COMPANY_BUDGET_S = 90.0 HOMEPAGE_MAX_BYTES = 3 * 1024 * 1024 PROBE_MAX_BYTES = 512 * 1024 MAX_PROBES = 12 MAX_SUBDOMAIN_GETS = 6 MAX_SITEMAP_CHILDREN = 4 TIER_FACTOR = {1: 0.5, 2: 0.75, 3: 1.0, 4: 1.5} METHOD_WEIGHT = {"ats": 1.0, "feed": 0.95, "nav": 1.0, "link": 0.9, "probe": 0.92, "subdomain": 0.9, "sitemap": 0.85, "robots": 0.9, "pattern": 0.8, "jsonld": 0.85, "manual": 1.0} # Surfaces worth a blind probe when navigation / sitemap did not reveal them: surface → candidate paths (ordered). PROBE_PATHS: dict[str, tuple[str, ...]] = { Surface.CAREERS: ("/careers", "/jobs", "/careers/", "/company/careers", "/about/careers", "/join-us"), Surface.PRICING: ("/pricing", "/plans", "/pricing/"), Surface.NEWSROOM: ("/news", "/press", "/newsroom", "/press-releases", "/media"), Surface.ABOUT: ("/about", "/about-us", "/company", "/company/about"), Surface.LEADERSHIP: ("/leadership", "/team", "/about/leadership", "/company/leadership", "/about/team", "/management"), Surface.LOCATIONS: ("/locations", "/offices", "/contact", "/company/locations"), Surface.BLOG: ("/blog", "/insights"), Surface.DOCS: ("/docs", "/documentation"), Surface.CHANGELOG: ("/changelog", "/release-notes", "/whats-new"), Surface.INVESTOR_RELATIONS: ("/investors", "/investor-relations", "/ir"), Surface.LEGAL_TERMS: ("/terms", "/legal/terms", "/terms-of-service", "/legal"), Surface.LEGAL_PRIVACY: ("/privacy", "/legal/privacy", "/privacy-policy"), } SUBDOMAINS: tuple[tuple[str, str], ...] = (("careers", Surface.CAREERS), ("jobs", Surface.CAREERS), ("news", Surface.NEWSROOM), ("blog", Surface.BLOG), ("docs", Surface.DOCS), ("developer", Surface.DEVELOPER), ("developers", Surface.DEVELOPER), ("status", Surface.STATUS), ("investors", Surface.INVESTOR_RELATIONS), ("ir", Surface.INVESTOR_RELATIONS), ("shop", Surface.PRODUCTS)) # hosts that are never a company's own canonical domain even when the website redirects there SHARED_HOSTS = ("linkedin.com", "facebook.com", "instagram.com", "twitter.com", "x.com", "youtube.com", "wixsite.com", "squarespace.com", "godaddy.com", "hubspot.com", "wordpress.com", "blogspot.com", "google.com", "sedo.com", "hugedomains.com", "dan.com", "afternic.com", "bluehost.com") SOFT_404_RE = re.compile(r"(page not found|404|not be found|doesn'?t exist|no longer available|nicht gefunden|introuvable)", re.IGNORECASE) EMBED_ATS_RE = re.compile( r"(?:boards|job-boards)\.greenhouse\.io/(?:embed/job_board(?:/js)?\?(?:[^\"'\s]*&)?for=|)([a-z0-9_-]{2,})|" r"jobs\.(?:eu\.)?lever\.co/([a-z0-9_-]{2,})|jobs\.ashbyhq\.com/([a-z0-9_.-]{2,})|apply\.workable\.com/([a-z0-9_-]{2,})|" r"([a-z0-9-]{2,})\.recruitee\.com|([a-z0-9-]{2,})\.jobs\.personio\.(?:de|com)|([a-z0-9-]{2,})\.teamtailor\.com|" r"(?:careers|jobs)\.smartrecruiters\.com/([A-Za-z0-9_-]{2,})|([a-z0-9-]+\.wd\d+\.myworkdayjobs\.com/[^\"'\s<>]+)", re.IGNORECASE) EMBED_VENDORS = ("greenhouse", "lever", "ashby", "workable", "recruitee", "personio", "teamtailor", "smartrecruiters", "workday") GH_JID_RE = re.compile(r"[?&]gh_jid=\d+", re.IGNORECASE) @dataclass(slots=True) class Candidate: url: str surface: str confidence: float method: str anchor: str | None = None verified: bool = False # we fetched it successfully during discovery connector_id: str | None = None config: dict[str, Any] = field(default_factory=dict) @property def score(self) -> float: depth = max(0, urlparse(self.url).path.strip("/").count("/")) return self.confidence * METHOD_WEIGHT.get(self.method, 0.8) * (1.0 if self.verified else 0.9) * (1.0 - 0.05 * min(depth, 3)) @dataclass(slots=True) class DiscoveryResult: company_id: str website: str final_url: str | None = None canonical_domain: str = "" redirect_domain: str | None = None candidates: list[Candidate] = field(default_factory=list) sensors: list[dict[str, Any]] = field(default_factory=list) status: str = OnboardingStatus.ACTIVE error: str | None = None notes: list[str] = field(default_factory=list) subdomains: list[str] = field(default_factory=list) same_as: list[str] = field(default_factory=list) requests: int = 0 duration_ms: int = 0 sitemaps: list[str] = field(default_factory=list) ats: list[dict[str, Any]] = field(default_factory=list) def table(self) -> list[dict[str, Any]]: return [{"surface": s["surface"], "url": s["url"], "connector": s["connector_id"], "confidence": s["discovery_confidence"], "method": s["discovery_method"], "interval_s": s["base_interval_s"], "tier": s["tier"], "quality": s["quality_score"]} for s in self.sensors] # ------------------------------------------------------------------------------------------------------------ helpers class _Budget: def __init__(self, seconds: float): self.deadline = time.monotonic() + seconds @property def left(self) -> float: return self.deadline - time.monotonic() def ok(self, need: float = 3.0) -> bool: return self.left > need async def _get(fetcher: Fetcher, url: str, *, max_bytes: int = PROBE_MAX_BYTES, res: DiscoveryResult, respect_robots: bool = True, accept: str | None = None) -> FetchResult | None: res.requests += 1 try: return await fetcher.get(url, max_bytes=max_bytes, respect_robots=respect_robots, retries=0, accept=accept, rate_per_min=settings.discovery_rate_per_min) except NotModified: return None except (FetchError, BlockedError) as exc: log.debug("discovery fetch failed", extra={"url": url, "failure": str(exc.failure), "error": str(exc)[:200]}) return None except Exception as exc: # noqa: BLE001 log.debug("discovery fetch error", extra={"url": url, "error": f"{exc.__class__.__name__}: {exc}"[:200]}) return None def _is_soft_404(result: FetchResult) -> bool: if not result.is_html: return False head = result.text[:4000] m = re.search(r"]*>(.*?)", head, re.IGNORECASE | re.DOTALL) title = m.group(1) if m else "" return bool(SOFT_404_RE.search(title)) or (len(result.content) < 600 and bool(SOFT_404_RE.search(head))) async def _resolves(host: str) -> bool: def _r() -> bool: try: socket.getaddrinfo(host, 443, proto=socket.IPPROTO_TCP) return True except socket.gaierror: return False try: return await asyncio.wait_for(asyncio.to_thread(_r), 4.0) except TimeoutError: return False def _website_variants(website: str, canonical_domain: str) -> list[str]: out: list[str] = [] w = (website or "").strip() if w and not w.startswith(("http://", "https://")): w = "https://" + w if w: out.append(w) dom = canonical_domain.lower().removeprefix("www.") for u in (f"https://www.{dom}/", f"https://{dom}/", f"http://www.{dom}/", f"http://{dom}/"): if u not in out and canonicalize_url(u) not in {canonicalize_url(x) for x in out}: out.append(u) return out[:4] def _interval_for(surface: str, tier: int) -> int: base = int(SURFACE_BASE_INTERVAL_S.get(surface, 86400) * TIER_FACTOR.get(int(tier or 4), 1.0)) return max(settings.min_interval_s, min(settings.max_interval_s, base)) def _quality(conf: float, surface: str, verified: bool) -> float: importance = SURFACE_IMPORTANCE.get(surface, 0.3) reliability = 1.0 if verified else 0.8 return round(100.0 * min(1.0, conf) * (0.5 + 0.5 * importance) * reliability, 1) # ------------------------------------------------------------------------------------------------------------ phases def _ats_from_text(text: str, links: list[str]) -> list[tuple[str, str, str]]: """(vendor, token, board_url) from links and embedded scripts/iframes.""" found: dict[tuple[str, str], str] = {} for u in links: hit = detect_ats(u) if hit and hit[0] in EMBED_VENDORS: found.setdefault(hit, u) for m in EMBED_ATS_RE.finditer(text[:2_000_000]): groups = m.groups() for vendor, g in zip(EMBED_VENDORS, groups, strict=False): if g: token = g board = m.group(0) if vendor == "workday": board = "https://" + g if not g.startswith("http") else g hit = detect_ats(board) token = hit[1] if hit else g.split(".")[0] elif vendor == "greenhouse": board = f"https://boards.greenhouse.io/{token}" elif vendor == "lever": board = f"https://jobs.lever.co/{token}" elif vendor == "ashby": board = f"https://jobs.ashbyhq.com/{token}" elif vendor == "workable": board = f"https://apply.workable.com/{token}" elif vendor == "recruitee": board = f"https://{token}.recruitee.com" elif vendor == "personio": board = f"https://{token}.jobs.personio.de" elif vendor == "teamtailor": board = f"https://{token}.teamtailor.com" elif vendor == "smartrecruiters": board = f"https://careers.smartrecruiters.com/{token}" if token.lower() in ("embed", "js", "job_board", "www", "api", "boards"): continue found.setdefault((vendor, token), board) return [(v, t, b) for (v, t), b in found.items()][:4] async def _phase_homepage(fetcher: Fetcher, company: dict[str, Any], res: DiscoveryResult, budget: _Budget) -> tuple[FetchResult | None, Any]: canonical = str(company.get("canonical_domain") or "") for url in _website_variants(str(company.get("website") or ""), canonical): if not budget.ok(10): break res.requests += 1 try: r = await fetcher.get(url, max_bytes=HOMEPAGE_MAX_BYTES, retries=1, rate_per_min=settings.discovery_rate_per_min) except BlockedError as exc: res.notes.append(f"homepage blocked: {exc.failure}") res.error = f"{exc.failure}: {exc}"[:300] continue except FetchError as exc: res.error = f"{exc.failure}: {exc}"[:300] if exc.failure in (FailureClass.DNS, FailureClass.BLOCKED_DESTINATION): res.notes.append(f"{url}: {exc.failure}") continue except Exception as exc: # noqa: BLE001 res.error = f"{exc.__class__.__name__}: {exc}"[:300] continue if not r.is_html or _is_soft_404(r): res.notes.append(f"{url}: not an HTML homepage") continue page_conn = connectors.get("generic-html-v1") sensor_stub = {"url": r.final_url, "surface": Surface.HOMEPAGE, "config": {"canonical_domain": canonical}} try: extraction = page_conn.extract(sensor_stub, r) except Exception as exc: # noqa: BLE001 res.error = f"homepage parse failed: {exc}"[:300] continue return r, extraction return None, None async def _phase_robots_sitemaps(fetcher: Fetcher, base: str, res: DiscoveryResult, budget: _Budget, canonical_domain: str) -> list[Candidate]: origin = f"{urlparse(base).scheme}://{urlparse(base).netloc}" sitemap_urls: list[str] = [] r = await _get(fetcher, f"{origin}/robots.txt", res=res, respect_robots=False, max_bytes=256 * 1024) if r is not None and r.status == 200 and not r.is_html: for line in r.text.splitlines(): if line.lower().startswith("sitemap:"): u = line.split(":", 1)[1].strip() if u.startswith("http") and same_company_host(u, canonical_domain) and u not in sitemap_urls: sitemap_urls.append(u) if not sitemap_urls: sitemap_urls = [f"{origin}/sitemap.xml"] cands: list[Candidate] = [] from companyatlas.connectors.sitemap import _decode, parse_sitemap # local import: connector module seen_urls = 0 for sm_url in sitemap_urls[:3]: if not budget.ok(8): break r = await _get(fetcher, sm_url, res=res, max_bytes=4 * 1024 * 1024, accept="application/xml,text/xml,*/*;q=0.5") if r is None or r.status != 200 or r.is_html: continue pages, children = parse_sitemap(_decode(r.content)) res.sitemaps.append(sm_url) if children and not pages: for child, _lm in children[:MAX_SITEMAP_CHILDREN]: if not budget.ok(6) or seen_urls >= settings.discovery_max_sitemap_urls: break cr = await _get(fetcher, child, res=res, max_bytes=4 * 1024 * 1024, accept="application/xml,text/xml,*/*;q=0.5") if cr is None or cr.is_html: continue p, _c = parse_sitemap(_decode(cr.content)) pages.extend(p) for loc, _lm in pages[: settings.discovery_max_sitemap_urls]: seen_urls += 1 if not same_company_host(loc, canonical_domain) or is_static_asset(loc) or looks_like_trap(loc): continue surface, conf = classify_url(loc, canonical_domain=canonical_domain) if surface in (Surface.OTHER, Surface.HOMEPAGE, Surface.SITEMAP) or conf < settings.discovery_min_confidence: continue cands.append(Candidate(url=loc, surface=str(surface), confidence=conf, method="sitemap")) if pages: break for sm_url in res.sitemaps[:1]: cands.append(Candidate(url=sm_url, surface=str(Surface.SITEMAP), confidence=0.95, method="robots", verified=True, config={"canonical_domain": canonical_domain})) return cands async def _phase_probes(fetcher: Fetcher, base: str, have: set[str], res: DiscoveryResult, budget: _Budget, canonical_domain: str) -> list[Candidate]: origin = f"{urlparse(base).scheme}://{urlparse(base).netloc}" home_canon = canonicalize_url(base) cands: list[Candidate] = [] probes = 0 for surface, paths in PROBE_PATHS.items(): if surface in have or probes >= MAX_PROBES or not budget.ok(6): continue for path in paths[:2]: if probes >= MAX_PROBES: break probes += 1 url = origin + path r = await _get(fetcher, url, res=res) if r is None or r.status != 200 or not r.is_html or _is_soft_404(r): continue if canonicalize_url(r.final_url) == home_canon: continue # redirected back to the homepage: the surface does not exist if not same_company_host(r.final_url, canonical_domain): hit = detect_ats(r.final_url) if hit: res.notes.append(f"{path} → ATS {hit[0]}") continue final = canonicalize_url(r.final_url) s2, c2 = classify_url(final, canonical_domain=canonical_domain) final_path = urlparse(final).path.rstrip("/").lower() if not (s2 == surface or final_path.endswith(path.rstrip("/").lower()) or (s2 == Surface.OTHER and final_path)): continue # redirected to a *different* known surface (marketing redirect): not this surface if s2 == Surface.OTHER and not final_path.endswith(path.rstrip("/").lower()): continue conf = max(0.6, c2 if s2 == surface else 0.6) cands.append(Candidate(url=final, surface=str(surface), confidence=conf, method="probe", verified=True)) have.add(surface) break return cands async def _phase_subdomains(fetcher: Fetcher, canonical_domain: str, have: set[str], res: DiscoveryResult, budget: _Budget) -> list[Candidate]: dom = canonical_domain.lower().removeprefix("www.") cands: list[Candidate] = [] gets = 0 checked: set[str] = set() for sub, surface in SUBDOMAINS: host = f"{sub}.{dom}" if host in checked or gets >= MAX_SUBDOMAIN_GETS or not budget.ok(6): continue checked.add(host) if not await _resolves(host): continue res.subdomains.append(host) gets += 1 if surface == Surface.STATUS: r = await _get(fetcher, f"https://{host}/api/v2/summary.json", res=res, accept="application/json", max_bytes=1024 * 1024) if r is not None and r.is_json: cands.append(Candidate(url=f"https://{host}/api/v2/summary.json", surface=str(Surface.STATUS), confidence=0.95, method="subdomain", verified=True, connector_id="statuspage-v1")) continue r = await _get(fetcher, f"https://{host}/", res=res) if r is not None and r.is_html and not _is_soft_404(r): cands.append(Candidate(url=r.final_url, surface=str(Surface.STATUS), confidence=0.8, method="subdomain", verified=True)) continue r = await _get(fetcher, f"https://{host}/", res=res) if r is None or not r.is_html or _is_soft_404(r): continue hit = detect_ats(r.final_url) if hit: spec = ats_sensor_spec(hit[0], hit[1], r.final_url) if spec: cands.append(Candidate(url=spec[0], surface=str(Surface.JOBS_BOARD), confidence=0.95, method="ats", verified=False, connector_id=spec[1], config=spec[2])) continue if registrable_domain(r.final_url) != dom: continue conf = 0.9 if surface not in have else 0.75 cands.append(Candidate(url=r.final_url, surface=str(surface), confidence=conf, method="subdomain", verified=True)) return cands async def _phase_ats(fetcher: Fetcher, home: FetchResult, home_links: list[str], careers: Candidate | None, res: DiscoveryResult, budget: _Budget, canonical_domain: str) -> list[Candidate]: text = home.text links = list(home_links) careers_res: FetchResult | None = None if careers is not None and budget.ok(8): careers_res = await _get(fetcher, careers.url, res=res, max_bytes=HOMEPAGE_MAX_BYTES) if careers_res is not None and careers_res.is_html: careers.verified = True text += "\n" + careers_res.text hit = detect_ats(careers_res.final_url) if hit: links.append(careers_res.final_url) for m in re.finditer(r"""(?:href|src|action|data-url|data-src)\s*=\s*["']([^"']+)["']""", careers_res.text[:1_500_000], re.IGNORECASE): u = absolutize(careers_res.final_url, m.group(1)) if u: links.append(u) cands: list[Candidate] = [] found = _ats_from_text(text, links) if not any(v == "greenhouse" for v, _t, _b in found) and GH_JID_RE.search(text): # Greenhouse-hosted jobs rendered on the company's own site (`?gh_jid=`): the board token is usually the company's slug / domain label label = canonical_domain.removeprefix("www.").split(".")[0].lower() if len(label) >= 2: found.append(("greenhouse", label, f"https://boards.greenhouse.io/{label}")) # verified below with one request for vendor, token, board in found: spec = ats_sensor_spec(vendor, token, board) if spec is None: continue api_url, connector_id, config = spec config["canonical_domain"] = canonical_domain verified = False if budget.ok(8) and not any(a["vendor"] == vendor for a in res.ats): conn = connectors.get(connector_id) try: res.requests += 1 from companyatlas.sdk.connector import ConnectorContext r = await asyncio.wait_for(conn.fetch(ConnectorContext(company={"canonical_domain": canonical_domain}), {"url": api_url, "config": config}, fetcher), 25) ex = conn.extract({"url": api_url, "config": config, "surface": Surface.JOBS_BOARD}, r) verified = True config["verified_job_count"] = len(ex.jobs) except Exception as exc: # noqa: BLE001 res.notes.append(f"ATS {vendor}/{token} not verified: {exc.__class__.__name__}") continue res.ats.append({"vendor": vendor, "token": token, "board_url": board, "api_url": api_url, "verified": verified}) cands.append(Candidate(url=api_url, surface=str(Surface.JOBS_BOARD), confidence=0.97 if verified else 0.85, method="ats", verified=verified, connector_id=connector_id, config=config)) return cands # ------------------------------------------------------------------------------------------------------------ selection LOCALE_SEG_RE = re.compile(r"^/([a-z]{2})(?:[-_][a-z]{2})?(?=/|$)", re.IGNORECASE) def _locale_of(url: str) -> str | None: m = LOCALE_SEG_RE.match(urlparse(url).path or "") return m.group(1).lower() if m else None def select_sensors(cands: list[Candidate], *, company: dict[str, Any], canonical_domain: str, now: datetime, fetch_now: bool = False, home_url: str | None = None) -> list[dict[str, Any]]: home_locale = _locale_of(home_url) if home_url else None best: dict[str, Candidate] = {} best_score: dict[str, float] = {} for c in cands: if c.confidence < settings.discovery_min_confidence and c.method not in ("ats", "robots"): continue score = c.score loc = _locale_of(c.url) if home_locale and loc and loc != home_locale: score *= 0.85 # prefer the homepage's language edition (/en/ over /jp/) if c.surface not in best or score > best_score[c.surface]: best[c.surface], best_score[c.surface] = c, score # never keep a *separate* feed sensor pointing at the same URL as another surface chosen = list(best.values()) seen_canon: set[str] = set() ranked = sorted(chosen, key=lambda c: (SURFACE_IMPORTANCE.get(c.surface, 0.3) * c.score), reverse=True) out: list[dict[str, Any]] = [] tier = int(company.get("tier") or 4) importance = float(company.get("importance") or 0.2) for c in ranked: canon = canonicalize_url(c.url) if canon in seen_canon: continue seen_canon.add(canon) connector = connectors.get(c.connector_id) if c.connector_id else connectors.for_surface(c.surface, c.url) base = _interval_for(c.surface, tier) if connector.meta.default_interval_s and connector.meta.default_interval_s < base and c.surface in (Surface.JOBS_BOARD, Surface.FEED): base = max(settings.min_interval_s, connector.meta.default_interval_s) cfg = {"canonical_domain": canonical_domain, **c.config, "discovery": {"version": DISCOVERY_VERSION, "method": c.method, "anchor": c.anchor, "verified": c.verified, "at": now.isoformat()}} out.append({ "id": new_id("sensor"), "company_id": company["id"], "surface": c.surface, "connector_id": connector.connector_id, "url": c.url, "canonical_url": canon, "domain": registrable_domain(c.url), "discovery_confidence": round(min(0.99, c.confidence), 3), "discovery_method": c.method, "quality_score": _quality(c.confidence, c.surface, c.verified), "status": SensorStatus.PENDING, "tier": tier_for_interval(base), "base_interval_s": base, "current_interval_s": base, "next_run_at": now if fetch_now else now + timedelta(seconds=random.uniform(0, base)), "priority": round(min(1.0, 0.3 + 0.5 * importance + 0.2 * SURFACE_IMPORTANCE.get(c.surface, 0.3)), 3), "config": cfg, }) if len(out) >= settings.discovery_max_sensors_per_company: break return out # ------------------------------------------------------------------------------------------------------------ main entry async def discover_company(company: dict[str, Any], *, fetcher: Fetcher, dry_run: bool = False, fetch_now: bool = False, budget_s: float = COMPANY_BUDGET_S) -> DiscoveryResult: """Full discovery for one company row. Never raises; persists sensors/domains/company status unless `dry_run`.""" t0 = time.perf_counter() res = DiscoveryResult(company_id=str(company["id"]), website=str(company.get("website") or ""), canonical_domain=str(company.get("canonical_domain") or "")) try: await asyncio.wait_for(_discover(company, fetcher, res, _Budget(budget_s), fetch_now=fetch_now), budget_s + 15) except TimeoutError: res.notes.append("discovery budget exhausted") if not res.sensors: res.status, res.error = OnboardingStatus.FAILED, res.error or "timeout" except Exception as exc: log.exception("discovery crashed", extra={"company_id": company.get("id")}) res.status, res.error = OnboardingStatus.FAILED, f"{exc.__class__.__name__}: {exc}"[:300] res.duration_ms = int((time.perf_counter() - t0) * 1000) if not dry_run: try: await persist(company, res) except Exception: log.exception("discovery persist failed", extra={"company_id": company.get("id")}) res.status, res.error = OnboardingStatus.FAILED, "persist failed" return res async def _discover_without_homepage(company: dict[str, Any], fetcher: Fetcher, res: DiscoveryResult, budget: _Budget, canonical: str, *, fetch_now: bool) -> None: base = f"https://www.{canonical}" if not canonical.startswith("www.") else f"https://{canonical}" res.canonical_domain = canonical res.notes.append("homepage blocked — partial onboarding from robots/sitemaps/probes/subdomains") cands: list[Candidate] = [] if budget.ok(10): cands.extend(await _phase_robots_sitemaps(fetcher, base, res, budget, canonical)) have = {c.surface for c in cands if c.confidence >= 0.7} if budget.ok(10): cands.extend(await _phase_probes(fetcher, base, have, res, budget, canonical)) have = {c.surface for c in cands if c.confidence >= 0.7} if budget.ok(10): cands.extend(await _phase_subdomains(fetcher, canonical, have, res, budget)) res.candidates = cands res.sensors = select_sensors(cands, company={**company, "canonical_domain": canonical}, canonical_domain=canonical, now=datetime.now(UTC), fetch_now=fetch_now, home_url=base) if res.sensors: res.status = OnboardingStatus.ACTIVE res.notes.append(f"partial: {len(res.sensors)} sensors without homepage") else: res.status = OnboardingStatus.FAILED res.error = res.error or "homepage blocked and no reachable surface" async def _discover(company: dict[str, Any], fetcher: Fetcher, res: DiscoveryResult, budget: _Budget, *, fetch_now: bool) -> None: canonical = str(company.get("canonical_domain") or "").lower() home, extraction = await _phase_homepage(fetcher, company, res, budget) if home is None: err = (res.error or "").upper() if "DNS" in err or "BLOCKED_DESTINATION" in err or not canonical: res.status, res.error = OnboardingStatus.NO_WEBSITE, res.error or "homepage unreachable" return # Homepage blocked (anti-bot 403, timeout, 5xx…): many such sites still expose robots/sitemaps, careers., investors., news. # subdomains or plain paths. Partial onboarding keeps the company observable instead of failing it outright (spec §60, §190). await _discover_without_homepage(company, fetcher, res, budget, canonical, fetch_now=fetch_now) return res.final_url = home.final_url final_dom = registrable_domain(home.final_url) if final_dom != registrable_domain(canonical): res.redirect_domain = final_dom res.notes.append(f"website redirects to {final_dom}") canonical_domain = final_dom if (res.redirect_domain and not any(final_dom.endswith(h) for h in SHARED_HOSTS)) else canonical res.canonical_domain = canonical_domain res.same_as = [s for s in (extraction.meta.get("same_as") or []) if isinstance(s, str)][:20] now = datetime.now(UTC) cands: list[Candidate] = [Candidate(url=home.final_url, surface=str(Surface.HOMEPAGE), confidence=0.99, method="nav", verified=True)] home_links: list[str] = [] for d in extraction.discovered: assert isinstance(d, DiscoveredUrl) home_links.append(d.url) if d.surface == Surface.JOBS_BOARD: hit = detect_ats(d.url) if hit: spec = ats_sensor_spec(hit[0], hit[1], d.url) if spec: cands.append(Candidate(url=spec[0], surface=str(Surface.JOBS_BOARD), confidence=0.9, method="ats", connector_id=spec[1], config=spec[2])) continue if d.surface == Surface.FEED: cands.append(Candidate(url=d.url, surface=str(Surface.FEED), confidence=d.confidence, method="feed", connector_id="feed-v1")) continue if not same_company_host(d.url, canonical_domain): continue cands.append(Candidate(url=d.url, surface=str(d.surface), confidence=d.confidence, method=d.method, anchor=d.anchor)) # raw homepage links for ATS scanning (including off-domain) for m in re.finditer(r"""(?:href|src|action|data-url)\s*=\s*["']([^"']+)["']""", home.text[:1_500_000], re.IGNORECASE): u = absolutize(home.final_url, m.group(1)) if u: home_links.append(u) if budget.ok(10): cands.extend(await _phase_robots_sitemaps(fetcher, home.final_url, res, budget, canonical_domain)) have = {c.surface for c in cands if c.confidence >= 0.7} careers = max((c for c in cands if c.surface == Surface.CAREERS), key=lambda c: c.score, default=None) if budget.ok(10): cands.extend(await _phase_ats(fetcher, home, home_links, careers, res, budget, canonical_domain)) if budget.ok(10): cands.extend(await _phase_probes(fetcher, home.final_url, have, res, budget, canonical_domain)) if budget.ok(10): cands.extend(await _phase_subdomains(fetcher, canonical_domain, have, res, budget)) res.candidates = cands res.sensors = select_sensors(cands, company={**company, "canonical_domain": canonical_domain}, canonical_domain=canonical_domain, now=now, fetch_now=fetch_now, home_url=home.final_url) res.status = OnboardingStatus.ACTIVE if res.sensors else OnboardingStatus.FAILED if not res.sensors: res.error = res.error or "no sensors discovered" # ------------------------------------------------------------------------------------------------------------ persistence async def persist(company: dict[str, Any], res: DiscoveryResult) -> None: now = datetime.now(UTC) async with transaction() as conn: # domains: redirect target / subdomains if res.redirect_domain: await _upsert_domain(conn, company["id"], res.redirect_domain, "redirect") for host in res.subdomains: await _upsert_domain(conn, company["id"], host, "subdomain") new_canonical = None if res.redirect_domain and res.canonical_domain != str(company.get("canonical_domain") or "").lower(): clash = await fetch_one(conn, "select id from companies where canonical_domain = :d and id <> :id", d=res.canonical_domain, id=company["id"]) if clash is None: new_canonical = res.canonical_domain await _upsert_domain(conn, company["id"], str(company.get("canonical_domain")), "former") else: res.notes.append(f"canonical domain {res.canonical_domain} already belongs to {clash['id']} — kept {company.get('canonical_domain')}") inserted = 0 for s in res.sensors: row = await fetch_one(conn, """ insert into sensors (id, company_id, surface, connector_id, url, canonical_url, domain, discovery_confidence, discovery_method, quality_score, status, tier, base_interval_s, current_interval_s, next_run_at, priority, config) values (:id, :company_id, :surface, :connector_id, :url, :canonical_url, :domain, :discovery_confidence, :discovery_method, :quality_score, :status, :tier, :base_interval_s, :current_interval_s, :next_run_at, :priority, cast(:config as jsonb)) on conflict (company_id, canonical_url) do update set discovery_confidence = greatest(sensors.discovery_confidence, excluded.discovery_confidence), connector_id = case when sensors.status = 'retired' then sensors.connector_id else excluded.connector_id end, surface = case when sensors.status in ('retired', 'paused') then sensors.surface else excluded.surface end, config = sensors.config || excluded.config, updated_at = now() returning (xmax = 0) as inserted """, **{**s, "config": jsonb(s["config"]), "status": str(s["status"])}) if row and row.get("inserted"): inserted += 1 stats = {"discovery": {"version": DISCOVERY_VERSION, "at": now.isoformat(), "sensors": len(res.sensors), "inserted": inserted, "requests": res.requests, "duration_ms": res.duration_ms, "candidates": len(res.candidates), "sitemaps": res.sitemaps[:3], "ats": res.ats, "notes": res.notes[:10], "subdomains": res.subdomains}} source_meta_patch: dict[str, Any] = {} if res.same_as: source_meta_patch["same_as"] = res.same_as if res.final_url: source_meta_patch["final_url"] = res.final_url await execute(conn, """ update companies set onboarding_status = cast(:st as text), onboarding_error = cast(:err as text), stats = stats || cast(:stats as jsonb), source_meta = source_meta || cast(:sm as jsonb), canonical_domain = coalesce(cast(:cd as text), canonical_domain), website = case when cast(:final as text) is not null and cast(:st as text) = 'active' then cast(:final as text) else website end, updated_at = now() where id = :id """, st=str(res.status), err=(res.error[:500] if res.error else None), stats=jsonb(stats), sm=jsonb(source_meta_patch), cd=new_canonical, final=res.final_url if res.final_url and (new_canonical or same_company_host(res.final_url, str(company.get("canonical_domain") or ""))) else None, id=company["id"]) async def _upsert_domain(conn: Any, company_id: str, domain: str, kind: str) -> None: await execute(conn, """ insert into domains (id, company_id, domain, kind) values (:id, :cid, :domain, :kind) on conflict (domain, company_id) do update set last_seen_at = now(), kind = case when domains.kind = 'primary' then domains.kind else excluded.kind end """, id=new_id("domain"), cid=company_id, domain=domain.lower(), kind=kind) # ------------------------------------------------------------------------------------------------------------ onboarding worker DISCOVER_CLAIM_TTL_MIN = 20 # a discovery never legitimately runs this long: older `running` claims belong to a dead worker async def claim_discover_jobs(limit: int, worker: str) -> list[dict[str, Any]]: async with transaction() as conn: await execute(conn, """update queue_jobs set status = 'pending', locked_by = null, locked_at = null, run_at = now() where kind = 'discover' and status = 'running' and locked_at < now() - make_interval(mins => :ttl)""", ttl=DISCOVER_CLAIM_TTL_MIN) rows = await fetch_all(conn, """ with due as ( select id from queue_jobs where kind = 'discover' and status = 'pending' and run_at <= now() order by priority desc, run_at limit :limit for update skip locked) update queue_jobs q set status = 'running', locked_at = now(), locked_by = :worker, attempts = attempts + 1 from due where q.id = due.id returning q.id, q.key, q.payload, q.attempts, q.max_attempts """, limit=limit, worker=worker) return rows async def finish_discover_job(job_id: str, *, ok: bool, error: str | None, attempts: int, max_attempts: int) -> None: async with transaction() as conn: if ok: await execute(conn, "update queue_jobs set status = 'done', finished_at = now(), last_error = null where id = :id", id=job_id) elif attempts >= max_attempts: await execute(conn, "update queue_jobs set status = 'dead', finished_at = now(), last_error = :e where id = :id", id=job_id, e=(error or "")[:500]) else: await execute(conn, """update queue_jobs set status = 'pending', locked_at = null, locked_by = null, last_error = :e, run_at = now() + make_interval(mins => :mins) where id = :id""", id=job_id, e=(error or "")[:500], mins=30 * attempts) async def onboard_pending(limit: int = 0, concurrency: int | None = None, *, fetcher: Fetcher | None = None, company_slug: str | None = None, fetch_now: bool = False, dry_run: bool = False, worker: str = "onboard") -> dict[str, Any]: """Discover pending companies with a rolling pool of `concurrency` workers (no batch barrier): each worker claims ONE target at a time — a `discover` queue job first (SKIP LOCKED), else a `companies.onboarding_status='pending'` row — until `limit` companies have been processed (`limit <= 0` = everything pending) or nothing is left. Safe to run from several processes/machines.""" concurrency = concurrency or settings.onboarding_concurrency own_fetcher = fetcher is None fetcher = fetcher or Fetcher() if own_fetcher: await fetcher.open() stats = {"claimed": 0, "active": 0, "failed": 0, "no_website": 0, "sensors": 0} lock = asyncio.Lock() budget = {"left": limit if limit and limit > 0 else float("inf")} async def process(comp: dict[str, Any], job: dict[str, Any] | None) -> None: res = await discover_company(comp, fetcher=fetcher, dry_run=dry_run, fetch_now=fetch_now) async with lock: stats[str(res.status)] = stats.get(str(res.status), 0) + 1 stats["sensors"] += len(res.sensors) if job is not None: await finish_discover_job(job["id"], ok=res.status == OnboardingStatus.ACTIVE, error=res.error, attempts=job["attempts"], max_attempts=job["max_attempts"]) log.info("company discovered", extra={"company": comp.get("slug"), "status": str(res.status), "sensors": len(res.sensors), "requests": res.requests, "ms": res.duration_ms, "error": res.error}) if fetch_now and not dry_run and res.sensors: from companyatlas.services.pipeline import run_sensor_ids await run_sensor_ids([s["id"] for s in res.sensors], fetcher=fetcher, worker=worker) async def claim_one() -> tuple[dict[str, Any], dict[str, Any] | None] | None: async with lock: if budget["left"] <= 0: return None budget["left"] -= 1 for j in await claim_discover_jobs(1, worker): cid = (j.get("payload") or {}).get("company_id") or j["key"].removeprefix("discover:") async with transaction() as conn: comp = await fetch_one(conn, "select * from companies where id = :id or slug = :id", id=cid) if comp is None: await finish_discover_job(j["id"], ok=False, error="company not found", attempts=j["attempts"], max_attempts=j["max_attempts"]) continue return comp, j async with transaction() as conn: row = await fetch_one(conn, """ update companies set onboarding_status = 'discovering', updated_at = now() where id = (select id from companies where onboarding_status = 'pending' order by importance desc, created_at limit 1 for update skip locked) returning *""") if row is None: async with lock: budget["left"] += 1 return None return row, None async def worker_loop() -> None: while True: target = await claim_one() if target is None: return async with lock: stats["claimed"] += 1 try: await process(*target) except Exception: # one company must never stop the pool log.exception("onboarding worker failed", extra={"company": target[0].get("slug")}) try: if company_slug: async with transaction() as conn: comp = await fetch_one(conn, "select * from companies where slug = :s or id = :s or canonical_domain = :s", s=company_slug) if comp is None: raise LookupError(f"company {company_slug!r} not found") stats["claimed"] = 1 await process(comp, None) else: await asyncio.gather(*(worker_loop() for _ in range(max(1, concurrency)))) finally: if own_fetcher: await fetcher.close() return stats __all__ = ["DISCOVERY_VERSION", "Candidate", "DiscoveryResult", "claim_discover_jobs", "discover_company", "onboard_pending", "persist", "select_sensors"]