spb/company-atlas
Public
Python 66.3%
TypeScript 22.7%
JavaScript 8.6%
HTML 1.4%
CSS 0.7%
1"""Discovery engine (spec §10–11, §101–102): canonical domain → homepage → robots → sitemaps → navigation → ATS / feeds / subdomains →2URL classification → one sensor per surface.34 discover_company(company, fetcher=…, dry_run=False) -> DiscoveryResult (never raises; ~90 s budget per company)5 onboard_pending(limit, concurrency) (consumes queue_jobs kind='discover' + pending companies)67Every request goes through `fetch.Fetcher` (SSRF guard, robots, governor). Discovery is bounded: ≤ `MAX_PROBES` common-path probes,8≤ `MAX_SUBDOMAIN_GETS` subdomain fetches, one sitemap index (+ a few children), one ATS verification. Sensors are inserted with9`status='pending'`, staggered `next_run_at`, and a `quality_score` seeded from confidence × surface importance × fetch reliability.10"""11from __future__ import annotations1213import asyncio14import logging15import random16import re17import socket18import time19from dataclasses import dataclass, field20from datetime import UTC, datetime, timedelta21from typing import Any22from urllib.parse import urlparse2324from companyatlas.config import settings25from companyatlas.connectors._util import ats_sensor_spec26from companyatlas.db import execute, fetch_all, fetch_one, jsonb, transaction27from companyatlas.fetch import BlockedError, Fetcher, FetchError, FetchResult, NotModified28from companyatlas.ids import new_id29from companyatlas.sdk import connector as connectors30from companyatlas.sdk.models import DiscoveredUrl31from companyatlas.taxonomy import (32 SURFACE_BASE_INTERVAL_S,33 SURFACE_IMPORTANCE,34 FailureClass,35 OnboardingStatus,36 SensorStatus,37 Surface,38 tier_for_interval,39)40from companyatlas.urls import (41 absolutize,42 canonicalize_url,43 classify_url,44 detect_ats,45 is_static_asset,46 looks_like_trap,47 registrable_domain,48 same_company_host,49)5051log = logging.getLogger(__name__)5253DISCOVERY_VERSION = "discovery-v1"54COMPANY_BUDGET_S = 90.055HOMEPAGE_MAX_BYTES = 3 * 1024 * 102456PROBE_MAX_BYTES = 512 * 102457MAX_PROBES = 1258MAX_SUBDOMAIN_GETS = 659MAX_SITEMAP_CHILDREN = 460TIER_FACTOR = {1: 0.5, 2: 0.75, 3: 1.0, 4: 1.5}61METHOD_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,62 "jsonld": 0.85, "manual": 1.0}63# Surfaces worth a blind probe when navigation / sitemap did not reveal them: surface → candidate paths (ordered).64PROBE_PATHS: dict[str, tuple[str, ...]] = {65 Surface.CAREERS: ("/careers", "/jobs", "/careers/", "/company/careers", "/about/careers", "/join-us"),66 Surface.PRICING: ("/pricing", "/plans", "/pricing/"),67 Surface.NEWSROOM: ("/news", "/press", "/newsroom", "/press-releases", "/media"),68 Surface.ABOUT: ("/about", "/about-us", "/company", "/company/about"),69 Surface.LEADERSHIP: ("/leadership", "/team", "/about/leadership", "/company/leadership", "/about/team", "/management"),70 Surface.LOCATIONS: ("/locations", "/offices", "/contact", "/company/locations"),71 Surface.BLOG: ("/blog", "/insights"),72 Surface.DOCS: ("/docs", "/documentation"),73 Surface.CHANGELOG: ("/changelog", "/release-notes", "/whats-new"),74 Surface.INVESTOR_RELATIONS: ("/investors", "/investor-relations", "/ir"),75 Surface.LEGAL_TERMS: ("/terms", "/legal/terms", "/terms-of-service", "/legal"),76 Surface.LEGAL_PRIVACY: ("/privacy", "/legal/privacy", "/privacy-policy"),77}78SUBDOMAINS: tuple[tuple[str, str], ...] = (("careers", Surface.CAREERS), ("jobs", Surface.CAREERS), ("news", Surface.NEWSROOM), ("blog", Surface.BLOG),79 ("docs", Surface.DOCS), ("developer", Surface.DEVELOPER), ("developers", Surface.DEVELOPER),80 ("status", Surface.STATUS), ("investors", Surface.INVESTOR_RELATIONS), ("ir", Surface.INVESTOR_RELATIONS),81 ("shop", Surface.PRODUCTS))82# hosts that are never a company's own canonical domain even when the website redirects there83SHARED_HOSTS = ("linkedin.com", "facebook.com", "instagram.com", "twitter.com", "x.com", "youtube.com", "wixsite.com", "squarespace.com", "godaddy.com",84 "hubspot.com", "wordpress.com", "blogspot.com", "google.com", "sedo.com", "hugedomains.com", "dan.com", "afternic.com", "bluehost.com")85SOFT_404_RE = re.compile(r"(page not found|404|not be found|doesn'?t exist|no longer available|nicht gefunden|introuvable)", re.IGNORECASE)86EMBED_ATS_RE = re.compile(87 r"(?:boards|job-boards)\.greenhouse\.io/(?:embed/job_board(?:/js)?\?(?:[^\"'\s]*&)?for=|)([a-z0-9_-]{2,})|"88 r"jobs\.(?:eu\.)?lever\.co/([a-z0-9_-]{2,})|jobs\.ashbyhq\.com/([a-z0-9_.-]{2,})|apply\.workable\.com/([a-z0-9_-]{2,})|"89 r"([a-z0-9-]{2,})\.recruitee\.com|([a-z0-9-]{2,})\.jobs\.personio\.(?:de|com)|([a-z0-9-]{2,})\.teamtailor\.com|"90 r"(?:careers|jobs)\.smartrecruiters\.com/([A-Za-z0-9_-]{2,})|([a-z0-9-]+\.wd\d+\.myworkdayjobs\.com/[^\"'\s<>]+)", re.IGNORECASE)91EMBED_VENDORS = ("greenhouse", "lever", "ashby", "workable", "recruitee", "personio", "teamtailor", "smartrecruiters", "workday")92GH_JID_RE = re.compile(r"[?&]gh_jid=\d+", re.IGNORECASE)939495@dataclass(slots=True)96class Candidate:97 url: str98 surface: str99 confidence: float100 method: str101 anchor: str | None = None102 verified: bool = False # we fetched it successfully during discovery103 connector_id: str | None = None104 config: dict[str, Any] = field(default_factory=dict)105106 @property107 def score(self) -> float:108 depth = max(0, urlparse(self.url).path.strip("/").count("/"))109 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))110111112@dataclass(slots=True)113class DiscoveryResult:114 company_id: str115 website: str116 final_url: str | None = None117 canonical_domain: str = ""118 redirect_domain: str | None = None119 candidates: list[Candidate] = field(default_factory=list)120 sensors: list[dict[str, Any]] = field(default_factory=list)121 status: str = OnboardingStatus.ACTIVE122 error: str | None = None123 notes: list[str] = field(default_factory=list)124 subdomains: list[str] = field(default_factory=list)125 same_as: list[str] = field(default_factory=list)126 requests: int = 0127 duration_ms: int = 0128 sitemaps: list[str] = field(default_factory=list)129 ats: list[dict[str, Any]] = field(default_factory=list)130131 def table(self) -> list[dict[str, Any]]:132 return [{"surface": s["surface"], "url": s["url"], "connector": s["connector_id"], "confidence": s["discovery_confidence"], "method": s["discovery_method"],133 "interval_s": s["base_interval_s"], "tier": s["tier"], "quality": s["quality_score"]} for s in self.sensors]134135136# ------------------------------------------------------------------------------------------------------------ helpers137138139class _Budget:140 def __init__(self, seconds: float):141 self.deadline = time.monotonic() + seconds142143 @property144 def left(self) -> float:145 return self.deadline - time.monotonic()146147 def ok(self, need: float = 3.0) -> bool:148 return self.left > need149150151async def _get(fetcher: Fetcher, url: str, *, max_bytes: int = PROBE_MAX_BYTES, res: DiscoveryResult, respect_robots: bool = True,152 accept: str | None = None) -> FetchResult | None:153 res.requests += 1154 try:155 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)156 except NotModified:157 return None158 except (FetchError, BlockedError) as exc:159 log.debug("discovery fetch failed", extra={"url": url, "failure": str(exc.failure), "error": str(exc)[:200]})160 return None161 except Exception as exc: # noqa: BLE001162 log.debug("discovery fetch error", extra={"url": url, "error": f"{exc.__class__.__name__}: {exc}"[:200]})163 return None164165166def _is_soft_404(result: FetchResult) -> bool:167 if not result.is_html:168 return False169 head = result.text[:4000]170 m = re.search(r"<title[^>]*>(.*?)</title>", head, re.IGNORECASE | re.DOTALL)171 title = m.group(1) if m else ""172 return bool(SOFT_404_RE.search(title)) or (len(result.content) < 600 and bool(SOFT_404_RE.search(head)))173174175async def _resolves(host: str) -> bool:176 def _r() -> bool:177 try:178 socket.getaddrinfo(host, 443, proto=socket.IPPROTO_TCP)179 return True180 except socket.gaierror:181 return False182 try:183 return await asyncio.wait_for(asyncio.to_thread(_r), 4.0)184 except TimeoutError:185 return False186187188def _website_variants(website: str, canonical_domain: str) -> list[str]:189 out: list[str] = []190 w = (website or "").strip()191 if w and not w.startswith(("http://", "https://")):192 w = "https://" + w193 if w:194 out.append(w)195 dom = canonical_domain.lower().removeprefix("www.")196 for u in (f"https://www.{dom}/", f"https://{dom}/", f"http://www.{dom}/", f"http://{dom}/"):197 if u not in out and canonicalize_url(u) not in {canonicalize_url(x) for x in out}:198 out.append(u)199 return out[:4]200201202def _interval_for(surface: str, tier: int) -> int:203 base = int(SURFACE_BASE_INTERVAL_S.get(surface, 86400) * TIER_FACTOR.get(int(tier or 4), 1.0))204 return max(settings.min_interval_s, min(settings.max_interval_s, base))205206207def _quality(conf: float, surface: str, verified: bool) -> float:208 importance = SURFACE_IMPORTANCE.get(surface, 0.3)209 reliability = 1.0 if verified else 0.8210 return round(100.0 * min(1.0, conf) * (0.5 + 0.5 * importance) * reliability, 1)211212213# ------------------------------------------------------------------------------------------------------------ phases214215216def _ats_from_text(text: str, links: list[str]) -> list[tuple[str, str, str]]:217 """(vendor, token, board_url) from links and embedded scripts/iframes."""218 found: dict[tuple[str, str], str] = {}219 for u in links:220 hit = detect_ats(u)221 if hit and hit[0] in EMBED_VENDORS:222 found.setdefault(hit, u)223 for m in EMBED_ATS_RE.finditer(text[:2_000_000]):224 groups = m.groups()225 for vendor, g in zip(EMBED_VENDORS, groups, strict=False):226 if g:227 token = g228 board = m.group(0)229 if vendor == "workday":230 board = "https://" + g if not g.startswith("http") else g231 hit = detect_ats(board)232 token = hit[1] if hit else g.split(".")[0]233 elif vendor == "greenhouse":234 board = f"https://boards.greenhouse.io/{token}"235 elif vendor == "lever":236 board = f"https://jobs.lever.co/{token}"237 elif vendor == "ashby":238 board = f"https://jobs.ashbyhq.com/{token}"239 elif vendor == "workable":240 board = f"https://apply.workable.com/{token}"241 elif vendor == "recruitee":242 board = f"https://{token}.recruitee.com"243 elif vendor == "personio":244 board = f"https://{token}.jobs.personio.de"245 elif vendor == "teamtailor":246 board = f"https://{token}.teamtailor.com"247 elif vendor == "smartrecruiters":248 board = f"https://careers.smartrecruiters.com/{token}"249 if token.lower() in ("embed", "js", "job_board", "www", "api", "boards"):250 continue251 found.setdefault((vendor, token), board)252 return [(v, t, b) for (v, t), b in found.items()][:4]253254255async def _phase_homepage(fetcher: Fetcher, company: dict[str, Any], res: DiscoveryResult, budget: _Budget) -> tuple[FetchResult | None, Any]:256 canonical = str(company.get("canonical_domain") or "")257 for url in _website_variants(str(company.get("website") or ""), canonical):258 if not budget.ok(10):259 break260 res.requests += 1261 try:262 r = await fetcher.get(url, max_bytes=HOMEPAGE_MAX_BYTES, retries=1, rate_per_min=settings.discovery_rate_per_min)263 except BlockedError as exc:264 res.notes.append(f"homepage blocked: {exc.failure}")265 res.error = f"{exc.failure}: {exc}"[:300]266 continue267 except FetchError as exc:268 res.error = f"{exc.failure}: {exc}"[:300]269 if exc.failure in (FailureClass.DNS, FailureClass.BLOCKED_DESTINATION):270 res.notes.append(f"{url}: {exc.failure}")271 continue272 except Exception as exc: # noqa: BLE001273 res.error = f"{exc.__class__.__name__}: {exc}"[:300]274 continue275 if not r.is_html or _is_soft_404(r):276 res.notes.append(f"{url}: not an HTML homepage")277 continue278 page_conn = connectors.get("generic-html-v1")279 sensor_stub = {"url": r.final_url, "surface": Surface.HOMEPAGE, "config": {"canonical_domain": canonical}}280 try:281 extraction = page_conn.extract(sensor_stub, r)282 except Exception as exc: # noqa: BLE001283 res.error = f"homepage parse failed: {exc}"[:300]284 continue285 return r, extraction286 return None, None287288289async def _phase_robots_sitemaps(fetcher: Fetcher, base: str, res: DiscoveryResult, budget: _Budget, canonical_domain: str) -> list[Candidate]:290 origin = f"{urlparse(base).scheme}://{urlparse(base).netloc}"291 sitemap_urls: list[str] = []292 r = await _get(fetcher, f"{origin}/robots.txt", res=res, respect_robots=False, max_bytes=256 * 1024)293 if r is not None and r.status == 200 and not r.is_html:294 for line in r.text.splitlines():295 if line.lower().startswith("sitemap:"):296 u = line.split(":", 1)[1].strip()297 if u.startswith("http") and same_company_host(u, canonical_domain) and u not in sitemap_urls:298 sitemap_urls.append(u)299 if not sitemap_urls:300 sitemap_urls = [f"{origin}/sitemap.xml"]301 cands: list[Candidate] = []302 from companyatlas.connectors.sitemap import _decode, parse_sitemap # local import: connector module303304 seen_urls = 0305 for sm_url in sitemap_urls[:3]:306 if not budget.ok(8):307 break308 r = await _get(fetcher, sm_url, res=res, max_bytes=4 * 1024 * 1024, accept="application/xml,text/xml,*/*;q=0.5")309 if r is None or r.status != 200 or r.is_html:310 continue311 pages, children = parse_sitemap(_decode(r.content))312 res.sitemaps.append(sm_url)313 if children and not pages:314 for child, _lm in children[:MAX_SITEMAP_CHILDREN]:315 if not budget.ok(6) or seen_urls >= settings.discovery_max_sitemap_urls:316 break317 cr = await _get(fetcher, child, res=res, max_bytes=4 * 1024 * 1024, accept="application/xml,text/xml,*/*;q=0.5")318 if cr is None or cr.is_html:319 continue320 p, _c = parse_sitemap(_decode(cr.content))321 pages.extend(p)322 for loc, _lm in pages[: settings.discovery_max_sitemap_urls]:323 seen_urls += 1324 if not same_company_host(loc, canonical_domain) or is_static_asset(loc) or looks_like_trap(loc):325 continue326 surface, conf = classify_url(loc, canonical_domain=canonical_domain)327 if surface in (Surface.OTHER, Surface.HOMEPAGE, Surface.SITEMAP) or conf < settings.discovery_min_confidence:328 continue329 cands.append(Candidate(url=loc, surface=str(surface), confidence=conf, method="sitemap"))330 if pages:331 break332 for sm_url in res.sitemaps[:1]:333 cands.append(Candidate(url=sm_url, surface=str(Surface.SITEMAP), confidence=0.95, method="robots", verified=True, config={"canonical_domain": canonical_domain}))334 return cands335336337async def _phase_probes(fetcher: Fetcher, base: str, have: set[str], res: DiscoveryResult, budget: _Budget, canonical_domain: str) -> list[Candidate]:338 origin = f"{urlparse(base).scheme}://{urlparse(base).netloc}"339 home_canon = canonicalize_url(base)340 cands: list[Candidate] = []341 probes = 0342 for surface, paths in PROBE_PATHS.items():343 if surface in have or probes >= MAX_PROBES or not budget.ok(6):344 continue345 for path in paths[:2]:346 if probes >= MAX_PROBES:347 break348 probes += 1349 url = origin + path350 r = await _get(fetcher, url, res=res)351 if r is None or r.status != 200 or not r.is_html or _is_soft_404(r):352 continue353 if canonicalize_url(r.final_url) == home_canon:354 continue # redirected back to the homepage: the surface does not exist355 if not same_company_host(r.final_url, canonical_domain):356 hit = detect_ats(r.final_url)357 if hit:358 res.notes.append(f"{path} → ATS {hit[0]}")359 continue360 final = canonicalize_url(r.final_url)361 s2, c2 = classify_url(final, canonical_domain=canonical_domain)362 final_path = urlparse(final).path.rstrip("/").lower()363 if not (s2 == surface or final_path.endswith(path.rstrip("/").lower()) or (s2 == Surface.OTHER and final_path)):364 continue # redirected to a *different* known surface (marketing redirect): not this surface365 if s2 == Surface.OTHER and not final_path.endswith(path.rstrip("/").lower()):366 continue367 conf = max(0.6, c2 if s2 == surface else 0.6)368 cands.append(Candidate(url=final, surface=str(surface), confidence=conf, method="probe", verified=True))369 have.add(surface)370 break371 return cands372373374async def _phase_subdomains(fetcher: Fetcher, canonical_domain: str, have: set[str], res: DiscoveryResult, budget: _Budget) -> list[Candidate]:375 dom = canonical_domain.lower().removeprefix("www.")376 cands: list[Candidate] = []377 gets = 0378 checked: set[str] = set()379 for sub, surface in SUBDOMAINS:380 host = f"{sub}.{dom}"381 if host in checked or gets >= MAX_SUBDOMAIN_GETS or not budget.ok(6):382 continue383 checked.add(host)384 if not await _resolves(host):385 continue386 res.subdomains.append(host)387 gets += 1388 if surface == Surface.STATUS:389 r = await _get(fetcher, f"https://{host}/api/v2/summary.json", res=res, accept="application/json", max_bytes=1024 * 1024)390 if r is not None and r.is_json:391 cands.append(Candidate(url=f"https://{host}/api/v2/summary.json", surface=str(Surface.STATUS), confidence=0.95, method="subdomain",392 verified=True, connector_id="statuspage-v1"))393 continue394 r = await _get(fetcher, f"https://{host}/", res=res)395 if r is not None and r.is_html and not _is_soft_404(r):396 cands.append(Candidate(url=r.final_url, surface=str(Surface.STATUS), confidence=0.8, method="subdomain", verified=True))397 continue398 r = await _get(fetcher, f"https://{host}/", res=res)399 if r is None or not r.is_html or _is_soft_404(r):400 continue401 hit = detect_ats(r.final_url)402 if hit:403 spec = ats_sensor_spec(hit[0], hit[1], r.final_url)404 if spec:405 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]))406 continue407 if registrable_domain(r.final_url) != dom:408 continue409 conf = 0.9 if surface not in have else 0.75410 cands.append(Candidate(url=r.final_url, surface=str(surface), confidence=conf, method="subdomain", verified=True))411 return cands412413414async def _phase_ats(fetcher: Fetcher, home: FetchResult, home_links: list[str], careers: Candidate | None, res: DiscoveryResult, budget: _Budget,415 canonical_domain: str) -> list[Candidate]:416 text = home.text417 links = list(home_links)418 careers_res: FetchResult | None = None419 if careers is not None and budget.ok(8):420 careers_res = await _get(fetcher, careers.url, res=res, max_bytes=HOMEPAGE_MAX_BYTES)421 if careers_res is not None and careers_res.is_html:422 careers.verified = True423 text += "\n" + careers_res.text424 hit = detect_ats(careers_res.final_url)425 if hit:426 links.append(careers_res.final_url)427 for m in re.finditer(r"""(?:href|src|action|data-url|data-src)\s*=\s*["']([^"']+)["']""", careers_res.text[:1_500_000], re.IGNORECASE):428 u = absolutize(careers_res.final_url, m.group(1))429 if u:430 links.append(u)431 cands: list[Candidate] = []432 found = _ats_from_text(text, links)433 if not any(v == "greenhouse" for v, _t, _b in found) and GH_JID_RE.search(text):434 # Greenhouse-hosted jobs rendered on the company's own site (`?gh_jid=`): the board token is usually the company's slug / domain label435 label = canonical_domain.removeprefix("www.").split(".")[0].lower()436 if len(label) >= 2:437 found.append(("greenhouse", label, f"https://boards.greenhouse.io/{label}")) # verified below with one request438 for vendor, token, board in found:439 spec = ats_sensor_spec(vendor, token, board)440 if spec is None:441 continue442 api_url, connector_id, config = spec443 config["canonical_domain"] = canonical_domain444 verified = False445 if budget.ok(8) and not any(a["vendor"] == vendor for a in res.ats):446 conn = connectors.get(connector_id)447 try:448 res.requests += 1449 from companyatlas.sdk.connector import ConnectorContext450451 r = await asyncio.wait_for(conn.fetch(ConnectorContext(company={"canonical_domain": canonical_domain}), {"url": api_url, "config": config}, fetcher), 25)452 ex = conn.extract({"url": api_url, "config": config, "surface": Surface.JOBS_BOARD}, r)453 verified = True454 config["verified_job_count"] = len(ex.jobs)455 except Exception as exc: # noqa: BLE001456 res.notes.append(f"ATS {vendor}/{token} not verified: {exc.__class__.__name__}")457 continue458 res.ats.append({"vendor": vendor, "token": token, "board_url": board, "api_url": api_url, "verified": verified})459 cands.append(Candidate(url=api_url, surface=str(Surface.JOBS_BOARD), confidence=0.97 if verified else 0.85, method="ats", verified=verified,460 connector_id=connector_id, config=config))461 return cands462463464# ------------------------------------------------------------------------------------------------------------ selection465466467LOCALE_SEG_RE = re.compile(r"^/([a-z]{2})(?:[-_][a-z]{2})?(?=/|$)", re.IGNORECASE)468469470def _locale_of(url: str) -> str | None:471 m = LOCALE_SEG_RE.match(urlparse(url).path or "")472 return m.group(1).lower() if m else None473474475def select_sensors(cands: list[Candidate], *, company: dict[str, Any], canonical_domain: str, now: datetime, fetch_now: bool = False,476 home_url: str | None = None) -> list[dict[str, Any]]:477 home_locale = _locale_of(home_url) if home_url else None478 best: dict[str, Candidate] = {}479 best_score: dict[str, float] = {}480 for c in cands:481 if c.confidence < settings.discovery_min_confidence and c.method not in ("ats", "robots"):482 continue483 score = c.score484 loc = _locale_of(c.url)485 if home_locale and loc and loc != home_locale:486 score *= 0.85 # prefer the homepage's language edition (/en/ over /jp/)487 if c.surface not in best or score > best_score[c.surface]:488 best[c.surface], best_score[c.surface] = c, score489 # never keep a *separate* feed sensor pointing at the same URL as another surface490 chosen = list(best.values())491 seen_canon: set[str] = set()492 ranked = sorted(chosen, key=lambda c: (SURFACE_IMPORTANCE.get(c.surface, 0.3) * c.score), reverse=True)493 out: list[dict[str, Any]] = []494 tier = int(company.get("tier") or 4)495 importance = float(company.get("importance") or 0.2)496 for c in ranked:497 canon = canonicalize_url(c.url)498 if canon in seen_canon:499 continue500 seen_canon.add(canon)501 connector = connectors.get(c.connector_id) if c.connector_id else connectors.for_surface(c.surface, c.url)502 base = _interval_for(c.surface, tier)503 if connector.meta.default_interval_s and connector.meta.default_interval_s < base and c.surface in (Surface.JOBS_BOARD, Surface.FEED):504 base = max(settings.min_interval_s, connector.meta.default_interval_s)505 cfg = {"canonical_domain": canonical_domain, **c.config, "discovery": {"version": DISCOVERY_VERSION, "method": c.method, "anchor": c.anchor,506 "verified": c.verified, "at": now.isoformat()}}507 out.append({508 "id": new_id("sensor"), "company_id": company["id"], "surface": c.surface, "connector_id": connector.connector_id, "url": c.url,509 "canonical_url": canon, "domain": registrable_domain(c.url), "discovery_confidence": round(min(0.99, c.confidence), 3),510 "discovery_method": c.method, "quality_score": _quality(c.confidence, c.surface, c.verified), "status": SensorStatus.PENDING,511 "tier": tier_for_interval(base), "base_interval_s": base, "current_interval_s": base,512 "next_run_at": now if fetch_now else now + timedelta(seconds=random.uniform(0, base)),513 "priority": round(min(1.0, 0.3 + 0.5 * importance + 0.2 * SURFACE_IMPORTANCE.get(c.surface, 0.3)), 3), "config": cfg,514 })515 if len(out) >= settings.discovery_max_sensors_per_company:516 break517 return out518519520# ------------------------------------------------------------------------------------------------------------ main entry521522523async def discover_company(company: dict[str, Any], *, fetcher: Fetcher, dry_run: bool = False, fetch_now: bool = False,524 budget_s: float = COMPANY_BUDGET_S) -> DiscoveryResult:525 """Full discovery for one company row. Never raises; persists sensors/domains/company status unless `dry_run`."""526 t0 = time.perf_counter()527 res = DiscoveryResult(company_id=str(company["id"]), website=str(company.get("website") or ""), canonical_domain=str(company.get("canonical_domain") or ""))528 try:529 await asyncio.wait_for(_discover(company, fetcher, res, _Budget(budget_s), fetch_now=fetch_now), budget_s + 15)530 except TimeoutError:531 res.notes.append("discovery budget exhausted")532 if not res.sensors:533 res.status, res.error = OnboardingStatus.FAILED, res.error or "timeout"534 except Exception as exc:535 log.exception("discovery crashed", extra={"company_id": company.get("id")})536 res.status, res.error = OnboardingStatus.FAILED, f"{exc.__class__.__name__}: {exc}"[:300]537 res.duration_ms = int((time.perf_counter() - t0) * 1000)538 if not dry_run:539 try:540 await persist(company, res)541 except Exception:542 log.exception("discovery persist failed", extra={"company_id": company.get("id")})543 res.status, res.error = OnboardingStatus.FAILED, "persist failed"544 return res545546547548async def _discover_without_homepage(company: dict[str, Any], fetcher: Fetcher, res: DiscoveryResult, budget: _Budget, canonical: str, *, fetch_now: bool) -> None:549 base = f"https://www.{canonical}" if not canonical.startswith("www.") else f"https://{canonical}"550 res.canonical_domain = canonical551 res.notes.append("homepage blocked — partial onboarding from robots/sitemaps/probes/subdomains")552 cands: list[Candidate] = []553 if budget.ok(10):554 cands.extend(await _phase_robots_sitemaps(fetcher, base, res, budget, canonical))555 have = {c.surface for c in cands if c.confidence >= 0.7}556 if budget.ok(10):557 cands.extend(await _phase_probes(fetcher, base, have, res, budget, canonical))558 have = {c.surface for c in cands if c.confidence >= 0.7}559 if budget.ok(10):560 cands.extend(await _phase_subdomains(fetcher, canonical, have, res, budget))561 res.candidates = cands562 res.sensors = select_sensors(cands, company={**company, "canonical_domain": canonical}, canonical_domain=canonical, now=datetime.now(UTC),563 fetch_now=fetch_now, home_url=base)564 if res.sensors:565 res.status = OnboardingStatus.ACTIVE566 res.notes.append(f"partial: {len(res.sensors)} sensors without homepage")567 else:568 res.status = OnboardingStatus.FAILED569 res.error = res.error or "homepage blocked and no reachable surface"570571572async def _discover(company: dict[str, Any], fetcher: Fetcher, res: DiscoveryResult, budget: _Budget, *, fetch_now: bool) -> None:573 canonical = str(company.get("canonical_domain") or "").lower()574 home, extraction = await _phase_homepage(fetcher, company, res, budget)575 if home is None:576 err = (res.error or "").upper()577 if "DNS" in err or "BLOCKED_DESTINATION" in err or not canonical:578 res.status, res.error = OnboardingStatus.NO_WEBSITE, res.error or "homepage unreachable"579 return580 # Homepage blocked (anti-bot 403, timeout, 5xx…): many such sites still expose robots/sitemaps, careers., investors., news.581 # subdomains or plain paths. Partial onboarding keeps the company observable instead of failing it outright (spec §60, §190).582 await _discover_without_homepage(company, fetcher, res, budget, canonical, fetch_now=fetch_now)583 return584 res.final_url = home.final_url585 final_dom = registrable_domain(home.final_url)586 if final_dom != registrable_domain(canonical):587 res.redirect_domain = final_dom588 res.notes.append(f"website redirects to {final_dom}")589 canonical_domain = final_dom if (res.redirect_domain and not any(final_dom.endswith(h) for h in SHARED_HOSTS)) else canonical590 res.canonical_domain = canonical_domain591 res.same_as = [s for s in (extraction.meta.get("same_as") or []) if isinstance(s, str)][:20]592 now = datetime.now(UTC)593 cands: list[Candidate] = [Candidate(url=home.final_url, surface=str(Surface.HOMEPAGE), confidence=0.99, method="nav", verified=True)]594 home_links: list[str] = []595 for d in extraction.discovered:596 assert isinstance(d, DiscoveredUrl)597 home_links.append(d.url)598 if d.surface == Surface.JOBS_BOARD:599 hit = detect_ats(d.url)600 if hit:601 spec = ats_sensor_spec(hit[0], hit[1], d.url)602 if spec:603 cands.append(Candidate(url=spec[0], surface=str(Surface.JOBS_BOARD), confidence=0.9, method="ats", connector_id=spec[1], config=spec[2]))604 continue605 if d.surface == Surface.FEED:606 cands.append(Candidate(url=d.url, surface=str(Surface.FEED), confidence=d.confidence, method="feed", connector_id="feed-v1"))607 continue608 if not same_company_host(d.url, canonical_domain):609 continue610 cands.append(Candidate(url=d.url, surface=str(d.surface), confidence=d.confidence, method=d.method, anchor=d.anchor))611 # raw homepage links for ATS scanning (including off-domain)612 for m in re.finditer(r"""(?:href|src|action|data-url)\s*=\s*["']([^"']+)["']""", home.text[:1_500_000], re.IGNORECASE):613 u = absolutize(home.final_url, m.group(1))614 if u:615 home_links.append(u)616 if budget.ok(10):617 cands.extend(await _phase_robots_sitemaps(fetcher, home.final_url, res, budget, canonical_domain))618 have = {c.surface for c in cands if c.confidence >= 0.7}619 careers = max((c for c in cands if c.surface == Surface.CAREERS), key=lambda c: c.score, default=None)620 if budget.ok(10):621 cands.extend(await _phase_ats(fetcher, home, home_links, careers, res, budget, canonical_domain))622 if budget.ok(10):623 cands.extend(await _phase_probes(fetcher, home.final_url, have, res, budget, canonical_domain))624 if budget.ok(10):625 cands.extend(await _phase_subdomains(fetcher, canonical_domain, have, res, budget))626 res.candidates = cands627 res.sensors = select_sensors(cands, company={**company, "canonical_domain": canonical_domain}, canonical_domain=canonical_domain, now=now, fetch_now=fetch_now,628 home_url=home.final_url)629 res.status = OnboardingStatus.ACTIVE if res.sensors else OnboardingStatus.FAILED630 if not res.sensors:631 res.error = res.error or "no sensors discovered"632633634# ------------------------------------------------------------------------------------------------------------ persistence635636637async def persist(company: dict[str, Any], res: DiscoveryResult) -> None:638 now = datetime.now(UTC)639 async with transaction() as conn:640 # domains: redirect target / subdomains641 if res.redirect_domain:642 await _upsert_domain(conn, company["id"], res.redirect_domain, "redirect")643 for host in res.subdomains:644 await _upsert_domain(conn, company["id"], host, "subdomain")645 new_canonical = None646 if res.redirect_domain and res.canonical_domain != str(company.get("canonical_domain") or "").lower():647 clash = await fetch_one(conn, "select id from companies where canonical_domain = :d and id <> :id", d=res.canonical_domain, id=company["id"])648 if clash is None:649 new_canonical = res.canonical_domain650 await _upsert_domain(conn, company["id"], str(company.get("canonical_domain")), "former")651 else:652 res.notes.append(f"canonical domain {res.canonical_domain} already belongs to {clash['id']} — kept {company.get('canonical_domain')}")653 inserted = 0654 for s in res.sensors:655 row = await fetch_one(conn, """656 insert into sensors (id, company_id, surface, connector_id, url, canonical_url, domain, discovery_confidence, discovery_method, quality_score,657 status, tier, base_interval_s, current_interval_s, next_run_at, priority, config)658 values (:id, :company_id, :surface, :connector_id, :url, :canonical_url, :domain, :discovery_confidence, :discovery_method, :quality_score,659 :status, :tier, :base_interval_s, :current_interval_s, :next_run_at, :priority, cast(:config as jsonb))660 on conflict (company_id, canonical_url) do update set661 discovery_confidence = greatest(sensors.discovery_confidence, excluded.discovery_confidence),662 connector_id = case when sensors.status = 'retired' then sensors.connector_id else excluded.connector_id end,663 surface = case when sensors.status in ('retired', 'paused') then sensors.surface else excluded.surface end,664 config = sensors.config || excluded.config, updated_at = now()665 returning (xmax = 0) as inserted666 """, **{**s, "config": jsonb(s["config"]), "status": str(s["status"])})667 if row and row.get("inserted"):668 inserted += 1669 stats = {"discovery": {"version": DISCOVERY_VERSION, "at": now.isoformat(), "sensors": len(res.sensors), "inserted": inserted, "requests": res.requests,670 "duration_ms": res.duration_ms, "candidates": len(res.candidates), "sitemaps": res.sitemaps[:3], "ats": res.ats, "notes": res.notes[:10],671 "subdomains": res.subdomains}}672 source_meta_patch: dict[str, Any] = {}673 if res.same_as:674 source_meta_patch["same_as"] = res.same_as675 if res.final_url:676 source_meta_patch["final_url"] = res.final_url677 await execute(conn, """678 update companies set onboarding_status = cast(:st as text), onboarding_error = cast(:err as text), stats = stats || cast(:stats as jsonb),679 source_meta = source_meta || cast(:sm as jsonb), canonical_domain = coalesce(cast(:cd as text), canonical_domain),680 website = case when cast(:final as text) is not null and cast(:st as text) = 'active' then cast(:final as text) else website end,681 updated_at = now()682 where id = :id683 """, st=str(res.status), err=(res.error[:500] if res.error else None), stats=jsonb(stats), sm=jsonb(source_meta_patch), cd=new_canonical,684 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,685 id=company["id"])686687688async def _upsert_domain(conn: Any, company_id: str, domain: str, kind: str) -> None:689 await execute(conn, """690 insert into domains (id, company_id, domain, kind) values (:id, :cid, :domain, :kind)691 on conflict (domain, company_id) do update set last_seen_at = now(), kind = case when domains.kind = 'primary' then domains.kind else excluded.kind end692 """, id=new_id("domain"), cid=company_id, domain=domain.lower(), kind=kind)693694695# ------------------------------------------------------------------------------------------------------------ onboarding worker696697698DISCOVER_CLAIM_TTL_MIN = 20 # a discovery never legitimately runs this long: older `running` claims belong to a dead worker699700701async def claim_discover_jobs(limit: int, worker: str) -> list[dict[str, Any]]:702 async with transaction() as conn:703 await execute(conn, """update queue_jobs set status = 'pending', locked_by = null, locked_at = null, run_at = now()704 where kind = 'discover' and status = 'running' and locked_at < now() - make_interval(mins => :ttl)""", ttl=DISCOVER_CLAIM_TTL_MIN)705 rows = await fetch_all(conn, """706 with due as (707 select id from queue_jobs where kind = 'discover' and status = 'pending' and run_at <= now()708 order by priority desc, run_at limit :limit for update skip locked)709 update queue_jobs q set status = 'running', locked_at = now(), locked_by = :worker, attempts = attempts + 1710 from due where q.id = due.id returning q.id, q.key, q.payload, q.attempts, q.max_attempts711 """, limit=limit, worker=worker)712 return rows713714715async def finish_discover_job(job_id: str, *, ok: bool, error: str | None, attempts: int, max_attempts: int) -> None:716 async with transaction() as conn:717 if ok:718 await execute(conn, "update queue_jobs set status = 'done', finished_at = now(), last_error = null where id = :id", id=job_id)719 elif attempts >= max_attempts:720 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])721 else:722 await execute(conn, """update queue_jobs set status = 'pending', locked_at = null, locked_by = null, last_error = :e,723 run_at = now() + make_interval(mins => :mins) where id = :id""", id=job_id, e=(error or "")[:500], mins=30 * attempts)724725726async def onboard_pending(limit: int = 0, concurrency: int | None = None, *, fetcher: Fetcher | None = None, company_slug: str | None = None,727 fetch_now: bool = False, dry_run: bool = False, worker: str = "onboard") -> dict[str, Any]:728 """Discover pending companies with a rolling pool of `concurrency` workers (no batch barrier): each worker claims ONE target at a729 time — a `discover` queue job first (SKIP LOCKED), else a `companies.onboarding_status='pending'` row — until `limit` companies730 have been processed (`limit <= 0` = everything pending) or nothing is left. Safe to run from several processes/machines."""731 concurrency = concurrency or settings.onboarding_concurrency732 own_fetcher = fetcher is None733 fetcher = fetcher or Fetcher()734 if own_fetcher:735 await fetcher.open()736 stats = {"claimed": 0, "active": 0, "failed": 0, "no_website": 0, "sensors": 0}737 lock = asyncio.Lock()738 budget = {"left": limit if limit and limit > 0 else float("inf")}739740 async def process(comp: dict[str, Any], job: dict[str, Any] | None) -> None:741 res = await discover_company(comp, fetcher=fetcher, dry_run=dry_run, fetch_now=fetch_now)742 async with lock:743 stats[str(res.status)] = stats.get(str(res.status), 0) + 1744 stats["sensors"] += len(res.sensors)745 if job is not None:746 await finish_discover_job(job["id"], ok=res.status == OnboardingStatus.ACTIVE, error=res.error, attempts=job["attempts"], max_attempts=job["max_attempts"])747 log.info("company discovered", extra={"company": comp.get("slug"), "status": str(res.status), "sensors": len(res.sensors), "requests": res.requests,748 "ms": res.duration_ms, "error": res.error})749 if fetch_now and not dry_run and res.sensors:750 from companyatlas.services.pipeline import run_sensor_ids751752 await run_sensor_ids([s["id"] for s in res.sensors], fetcher=fetcher, worker=worker)753754 async def claim_one() -> tuple[dict[str, Any], dict[str, Any] | None] | None:755 async with lock:756 if budget["left"] <= 0:757 return None758 budget["left"] -= 1759 for j in await claim_discover_jobs(1, worker):760 cid = (j.get("payload") or {}).get("company_id") or j["key"].removeprefix("discover:")761 async with transaction() as conn:762 comp = await fetch_one(conn, "select * from companies where id = :id or slug = :id", id=cid)763 if comp is None:764 await finish_discover_job(j["id"], ok=False, error="company not found", attempts=j["attempts"], max_attempts=j["max_attempts"])765 continue766 return comp, j767 async with transaction() as conn:768 row = await fetch_one(conn, """769 update companies set onboarding_status = 'discovering', updated_at = now()770 where id = (select id from companies where onboarding_status = 'pending' order by importance desc, created_at limit 1 for update skip locked)771 returning *""")772 if row is None:773 async with lock:774 budget["left"] += 1775 return None776 return row, None777778 async def worker_loop() -> None:779 while True:780 target = await claim_one()781 if target is None:782 return783 async with lock:784 stats["claimed"] += 1785 try:786 await process(*target)787 except Exception: # one company must never stop the pool788 log.exception("onboarding worker failed", extra={"company": target[0].get("slug")})789790 try:791 if company_slug:792 async with transaction() as conn:793 comp = await fetch_one(conn, "select * from companies where slug = :s or id = :s or canonical_domain = :s", s=company_slug)794 if comp is None:795 raise LookupError(f"company {company_slug!r} not found")796 stats["claimed"] = 1797 await process(comp, None)798 else:799 await asyncio.gather(*(worker_loop() for _ in range(max(1, concurrency))))800 finally:801 if own_fetcher:802 await fetcher.close()803 return stats804805806__all__ = ["DISCOVERY_VERSION", "Candidate", "DiscoveryResult", "claim_discover_jobs", "discover_company", "onboard_pending", "persist", "select_sensors"]807