spb/company-atlas
Public
Python 66.3%
TypeScript 22.7%
JavaScript 8.6%
HTML 1.4%
CSS 0.7%
1"""Generic HTML connector (spec §9.1, §107): one connector for every corporate web surface. `sdk.normalize` supplies text, blocks,2metadata, JSON-LD and links; this module adds *surface-aware* typed extraction:34 leadership → people (role category, is_executive) pricing → plans (currency, period, unit, contact-sales, features)5 products/services/solutions → product cards locations → offices/stores (city / country only when stated)6 newsroom/blog/press/changelog/research/IR → news items careers → job listings (list / table / card patterns with job-like anchors)7 legal/docs/homepage/about/… → text + blocks only89Structured data (JSON-LD / microdata) is used first, DOM heuristics second, and nothing is ever invented: no address, no country, no10date that the page does not state. Every surface also yields `discovered` URLs (classified links) for the discovery feedback loop.11Precision rules (`connectors/_precision.py`) reject navigation / CTA / cookie-consent / marketing noise at the point of extraction.12"""13from __future__ import annotations1415import re16from collections.abc import Mapping17from typing import Any18from urllib.parse import urlparse1920from selectolax.lexbor import LexborHTMLParser2122from companyatlas.config import settings23from companyatlas.connectors._precision import (24 STREET_RE,25 is_known_city,26 looks_like_person_name,27 looks_like_role_title,28 news_verdict,29 normalize_location,30 normalize_person,31 plan_name_ok,32 plan_verdict,33 price_text_from,34 product_verdict,35 refine_job,36 role_category,37 states_job_location,38)39from companyatlas.connectors._util import (40 country_code,41 date_from_text,42 date_from_url,43 finish_job,44 job_blocks,45 norm_name,46 parse_date,47 parse_location,48 text_of,49)50from companyatlas.connectors.jsonld_jobs import jobs_from_jsonld51from companyatlas.fetch import FetchResult52from companyatlas.sdk import normalize53from companyatlas.sdk.connector import Connector, ConnectorMeta, register54from companyatlas.sdk.models import (55 Block,56 DiscoveredUrl,57 ExtractedJob,58 ExtractedLocation,59 ExtractedNewsItem,60 ExtractedPerson,61 ExtractedPlan,62 ExtractedProduct,63 Extraction,64)65from companyatlas.sdk.normalize import NormalizedPage, normalize_whitespace66from companyatlas.taxonomy import FetchMode, Surface67from companyatlas.urls import absolutize, canonicalize_url, classify_url, is_static_asset, looks_like_trap, registrable_domain6869MAX_PEOPLE, MAX_PLANS, MAX_LOCATIONS, MAX_NEWS, MAX_JOBS, MAX_PRODUCTS, MAX_DISCOVERED = 200, 16, 300, 80, 300, 120, 12070NEWS_SURFACES = {Surface.NEWSROOM, Surface.BLOG, Surface.CHANGELOG, Surface.RESEARCH, Surface.INVESTOR_RELATIONS}71NEWS_CATEGORY = {Surface.NEWSROOM: "press", Surface.BLOG: "blog", Surface.CHANGELOG: "changelog", Surface.RESEARCH: "research",72 Surface.INVESTOR_RELATIONS: "ir"}73PRODUCT_SURFACES = {Surface.PRODUCTS, Surface.SERVICES, Surface.SOLUTIONS}7475# ------------------------------------------------------------------------------------------------------------ people7677# Role vocabulary, name shape and title cleaning live in `_precision` (shared with the pipeline); re-exported here for callers/tests.78looks_like_name = looks_like_person_name79looks_like_title = looks_like_role_title80PERSON_CARD_SCAN_LINES = 4 # a person card states the name within its first lines81PERSON_TITLE_WINDOW = 3 # …and the title within the next few828384def _card_person(lines: list[str]) -> tuple[str, str | None] | None:85 """(name, title) from a card's lines: the title may precede the name (swapped cards) or follow it; role-like lines win over prose."""86 name_idx = next((i for i, ln in enumerate(lines[:PERSON_CARD_SCAN_LINES]) if looks_like_name(ln)), None)87 if name_idx is None:88 return None89 before = [x for x in lines[:name_idx] if looks_like_title(x)]90 after = [x for x in lines[name_idx + 1:name_idx + 1 + PERSON_TITLE_WINDOW] if not looks_like_name(x)]91 title = (before[-1] if before else None) or next((x for x in after if looks_like_title(x)), None) or (after[0] if after else None)92 return lines[name_idx], title939495def extract_people(page: NormalizedPage) -> list[ExtractedPerson]:96 out: list[ExtractedPerson] = []97 seen: set[str] = set()9899 def add(name: str, title: str | None, url: str | None = None) -> None:100 fixed = normalize_person(name, title)101 if fixed is None:102 return103 name, title = fixed104 key = norm_name(name)105 if not key or key in seen or len(out) >= MAX_PEOPLE:106 return107 seen.add(key)108 cat, is_exec = role_category(title)109 out.append(ExtractedPerson(name=name[:120], title=(title[:160] if title else None), role_category=cat, is_executive=is_exec, url=url))110111 for p in page.jsonld.get("persons", []) + page.microdata.get("persons", []):112 name = text_of(p.get("name"))113 if name:114 add(name, text_of(p.get("jobTitle")) or text_of(p.get("title")), text_of(p.get("url")) if isinstance(p.get("url"), str) else None)115 for b in page.blocks:116 if b.kind != "person":117 continue118 lines = [ln.strip() for ln in b.text.split("\n") if ln.strip()]119 found = _card_person(lines)120 if found is not None:121 add(found[0], found[1], b.attrs.get("href"))122 if len(out) < 2: # fallback: heading = name, next short block = title123 blocks = page.blocks124 for i, b in enumerate(blocks):125 if b.kind == "heading" and int(b.attrs.get("level", 3)) >= 2 and looks_like_name(b.text):126 nxt = next((x for x in blocks[i + 1:i + 3] if x.kind in ("paragraph", "other", "list", "section")), None)127 if nxt is not None and looks_like_title(nxt.text.split("\n")[0]):128 add(b.text, nxt.text.split("\n")[0])129 return out130131132# ------------------------------------------------------------------------------------------------------------ pricing133134CURRENCY_SYMBOLS = {"$": "USD", "US$": "USD", "USD": "USD", "€": "EUR", "EUR": "EUR", "£": "GBP", "GBP": "GBP", "¥": "JPY", "JPY": "JPY", "₹": "INR",135 "INR": "INR", "C$": "CAD", "CA$": "CAD", "CAD": "CAD", "A$": "AUD", "AUD": "AUD", "CHF": "CHF", "SEK": "SEK", "NOK": "NOK",136 "DKK": "DKK", "kr": "SEK", "zł": "PLN", "PLN": "PLN", "R$": "BRL", "BRL": "BRL", "MX$": "MXN", "₩": "KRW", "SGD": "SGD", "S$": "SGD",137 "HK$": "HKD", "NZ$": "NZD", "₺": "TRY", "元": "CNY", "CNY": "CNY", "RMB": "CNY"}138PRICE_RE = re.compile(r"(?:(?P<cur>US\$|CA\$|C\$|A\$|NZ\$|HK\$|S\$|MX\$|R\$|USD|EUR|GBP|CAD|AUD|CHF|JPY|INR|SEK|NOK|DKK|PLN|BRL|CNY|RMB|SGD|[$€£¥₹₩₺])\s?"139 r"(?P<amt>\d{1,3}(?:[,.\s]\d{3})*(?:[.,]\d{1,2})?|\d+(?:[.,]\d{1,2})?))|"140 r"(?:(?P<amt2>\d{1,3}(?:[,.\s]\d{3})*(?:[.,]\d{1,2})?|\d+(?:[.,]\d{1,2})?)\s?(?P<cur2>€|£|USD|EUR|GBP|CHF|kr|zł|元|\$))")141PERIOD_RE = re.compile(r"(?:(?:per|/|a|each|every)\s*(?P<unit>month|mo|year|yr|annum|week|wk|day|hour|hr|user|seat|member|license|licence|agent|editor|"142 r"contact|1,?000|1k|GB|TB|request|transaction|call|minute|mois|an|année|monat|jahr)\b)|(?P<word>monthly|annually|yearly|per annum|"143 r"billed (?:monthly|annually|yearly)|mensuel|annuel|monatlich|jährlich|one[- ]time|lifetime)", re.IGNORECASE)144CONTACT_RE = re.compile(r"\b(contact (us|sales)|talk to (us|sales|an expert)|custom pricing|custom quote|get a quote|request (a )?(quote|demo)|let'?s talk|"145 r"on request|upon request|sur devis|nous contacter|contactez-nous|auf anfrage|individuell|tailored|bespoke|call us)\b", re.IGNORECASE)146FREE_RE = re.compile(r"^(free|gratuit|kostenlos|gratis|\$\s?0(\.00)?|0\s?€|€\s?0)\b", re.IGNORECASE)147FROM_RE = re.compile(r"\b(from|starting at|starts at|as low as|à partir de|ab)\b", re.IGNORECASE)148MONTH_UNITS = {"month", "mo", "monthly", "mensuel", "monatlich", "mois", "monat"}149YEAR_UNITS = {"year", "yr", "annum", "annually", "yearly", "per annum", "annuel", "jährlich", "an", "année", "jahr"}150SEAT_UNITS = {"user", "seat", "member", "license", "licence", "agent", "editor", "contact"}151USAGE_UNITS = {"1,000", "1000", "1k", "gb", "tb", "request", "transaction", "call", "minute", "hour", "hr", "day", "week", "wk"}152153154def parse_price(text: str) -> dict[str, Any] | None:155 """→ {price, currency, billing_period, unit, price_text, contact_sales} or None when the text states no price."""156 t = normalize_whitespace(text)157 if CONTACT_RE.search(t) and not PRICE_RE.search(t):158 return {"price": None, "currency": None, "billing_period": "contact", "unit": None, "price_text": price_text_from(t) or t[:80], "contact_sales": True}159 m = PRICE_RE.search(t)160 if m is None:161 if FREE_RE.match(t):162 return {"price": 0.0, "currency": None, "billing_period": None, "unit": None, "price_text": price_text_from(t) or t[:80], "contact_sales": False}163 return None164 cur = m.group("cur") or m.group("cur2")165 amt = m.group("amt") or m.group("amt2")166 amt_clean = amt.replace(" ", "")167 if re.fullmatch(r"\d{1,3}(,\d{3})+(\.\d{1,2})?", amt_clean):168 amt_clean = amt_clean.replace(",", "")169 elif re.fullmatch(r"\d{1,3}(\.\d{3})+(,\d{1,2})?", amt_clean):170 amt_clean = amt_clean.replace(".", "").replace(",", ".")171 elif re.fullmatch(r"\d+,\d{1,2}", amt_clean):172 amt_clean = amt_clean.replace(",", ".")173 else:174 amt_clean = amt_clean.replace(",", "")175 try:176 price = float(amt_clean)177 except ValueError:178 return None179 tail = t[m.end():m.end() + 60]180 pm = PERIOD_RE.search(tail) or PERIOD_RE.search(t)181 period, unit = None, None182 if pm:183 u = (pm.group("unit") or pm.group("word") or "").lower()184 if u in MONTH_UNITS or u.startswith("billed monthly"):185 period = "month"186 elif u in YEAR_UNITS or "annual" in u or "yearly" in u:187 period = "year"188 elif u in ("one-time", "one time", "lifetime"):189 period = "one_time"190 elif u in SEAT_UNITS:191 unit = u192 elif u in USAGE_UNITS:193 period, unit = "usage", u194 # a second unit ("per user per month")195 for pm2 in PERIOD_RE.finditer(tail):196 u2 = (pm2.group("unit") or pm2.group("word") or "").lower()197 if u2 in SEAT_UNITS and not unit:198 unit = u2199 elif (u2 in MONTH_UNITS) and not period:200 period = "month"201 elif (u2 in YEAR_UNITS) and not period:202 period = "year"203 return {"price": price, "currency": CURRENCY_SYMBOLS.get(cur, cur.upper() if cur and cur.isalpha() else None), "billing_period": period, "unit": unit,204 "price_text": price_text_from(t) or t[m.start():m.end()].strip()[:80], "contact_sales": bool(CONTACT_RE.search(t))}205206207PLAN_NAME_SCAN_LINES = 3 # the plan name is one of the first lines before the price (eyebrows / marketing headings are skipped)208209210def _plan_name_from(lines: list[str], path: str) -> tuple[str | None, list[str]]:211 """(plan name, remaining lines): the first valid short label before the price line (eyebrows such as "Most popular" are skipped).212 Otherwise the heading just above the card (its path tail) names the tier — unless that heading is generic ("Plans", "Pricing") or a213 marketing sentence, in which case the card yields no plan."""214 for i, ln in enumerate(lines[:PLAN_NAME_SCAN_LINES]):215 if plan_name_ok(ln): # "Free" is a valid tier name even though it also reads as a price216 return ln, lines[i + 1:]217 if PRICE_RE.search(ln) or parse_price(ln) is not None:218 break219 tail = path.split(" > ")[-1] if path else ""220 return (tail if plan_name_ok(tail) else None), lines221222223def _plan_from_lines(name: str | None, lines: list[str]) -> ExtractedPlan | None:224 if not name:225 return None226 price_info = None227 features: list[str] = []228 for ln in lines:229 if price_info is None:230 info = parse_price(ln)231 if info is not None:232 price_info = info233 continue234 if 2 <= len(ln) <= 140 and not parse_price(ln) and len(features) < 25:235 features.append(ln)236 if price_info is None and FREE_RE.match(name):237 price_info = parse_price(name) # a tier literally called "Free" with no separate price line238 if price_info is None:239 return None240 return ExtractedPlan(plan_name=name[:80], price=price_info["price"], price_text=price_info["price_text"], currency=price_info["currency"],241 billing_period=price_info["billing_period"], unit=price_info["unit"], features=features, contact_sales=price_info["contact_sales"])242243244def extract_plans(page: NormalizedPage) -> list[ExtractedPlan]:245 out: list[ExtractedPlan] = []246 seen: set[str] = set()247248 def add(plan: ExtractedPlan | None) -> None:249 if plan is None or not plan_verdict(plan).ok:250 return251 key = norm_name(plan.plan_name)252 if not key or key in seen or len(out) >= MAX_PLANS:253 return254 seen.add(key)255 out.append(plan)256257 for b in page.blocks:258 if b.kind != "pricing_plan":259 continue260 lines = [ln.strip() for ln in b.text.split("\n") if ln.strip()]261 if not lines:262 continue263 name, rest = _plan_name_from(lines, b.path)264 add(_plan_from_lines(name, rest))265 if not out: # fallback: heading followed by a price before the next heading (common in hand-rolled pricing tables)266 blocks = page.blocks267 i = 0268 while i < len(blocks):269 b = blocks[i]270 if b.kind == "heading" and int(b.attrs.get("level", 2)) >= 2 and plan_name_ok(b.text):271 j = i + 1272 lines: list[str] = []273 while j < len(blocks) and blocks[j].kind != "heading" and j - i <= 10:274 lines.extend(x.strip() for x in blocks[j].text.split("\n") if x.strip())275 j += 1276 add(_plan_from_lines(b.text, lines))277 i = j278 continue279 i += 1280 if not out: # tables: first cell = plan, another cell = price281 for b in page.blocks:282 if b.kind == "table" and "|" in b.text:283 cells = [c.strip() for c in b.text.split("|")]284 if len(cells) >= 2 and parse_price(cells[0]) is None and any(parse_price(c) for c in cells[1:]):285 add(_plan_from_lines(cells[0], cells[1:]))286 return out287288289# ------------------------------------------------------------------------------------------------------------ locations290291LOCATION_KIND_RULES: list[tuple[str, re.Pattern[str]]] = [292 ("headquarters", re.compile(r"\b(headquarters|head office|hq|global office|corporate office|siège|hauptsitz|sede central)\b", re.IGNORECASE)),293 ("factory", re.compile(r"\b(factory|plant|manufacturing|production site|mill|foundry|usine|werk|fabrik)\b", re.IGNORECASE)),294 ("warehouse", re.compile(r"\b(warehouse|distribution cent(er|re)|fulfil?lment|logistics hub|entrepôt|lager)\b", re.IGNORECASE)),295 ("lab", re.compile(r"\b(lab|laboratory|research cent(er|re)|r&d|innovation cent(er|re))\b", re.IGNORECASE)),296 ("data_center", re.compile(r"\b(data ?cent(er|re)|datacenter|server farm)\b", re.IGNORECASE)),297 ("store", re.compile(r"\b(store|shop|boutique|showroom|outlet|dealer(ship)?|branch|agence|filiale)\b", re.IGNORECASE)),298]299CITY_COUNTRY_RE = re.compile(r"^([A-ZÀ-Ý][\w'’.\- ]{1,40}),\s*([A-Za-zÀ-ÿ .]{2,40})$")300MAX_VENUE_LINE_CHARS = 60 # "Hormuz Grand Hotel" — a venue line under a country heading301302303def _location_from_lines(name: str, lines: list[str], href: str | None = None) -> ExtractedLocation | None:304 text = " \n ".join(lines)305 kind = next((k for k, pat in LOCATION_KIND_RULES if pat.search(name) or pat.search(text[:200])), "office")306 city = region = country = address = None307 for ln in lines:308 street = STREET_RE.search(ln)309 if street and address is None and len(ln) <= 120:310 address = ln311 probe = (ln[:street.start()] + " " + ln[street.end():]).strip(" ,") if street else ln312 if country is None and probe:313 loc = parse_location(probe)314 if loc["country"]: # the line that states the country is the authoritative "City, Country" line315 city, region, country = loc["city"] or city, loc["region"], loc["country"]316 elif loc["region"]:317 city, region = city or loc["city"], loc["region"]318 if country is None:319 c = country_code(name)320 if c:321 country = c322 else:323 m = CITY_COUNTRY_RE.match(name)324 if m and country_code(m.group(2)):325 city, country = m.group(1).strip(), country_code(m.group(2))326 if country_code(name) and city is None:327 # heading is a country ("Oman") — the first short line that is neither an address nor a place is the venue / office name328 venue = next((ln for ln in lines if ln != address and len(ln) <= MAX_VENUE_LINE_CHARS and not parse_location(ln)["country"]329 and not country_code(ln) and not any(ch.isdigit() for ch in ln)), None)330 if venue:331 name = venue332 elif is_known_city(name):333 city = name # city-states: Singapore, Monaco, Hong Kong334 elif city is None and country is not None and looks_like_city(name):335 city = name336 if city is None and country is None and address is None and kind == "office":337 return None338 return ExtractedLocation(name=name[:120], kind=kind, city=city, region=region, country=country, address_text=address)339340341def looks_like_city(text: str) -> bool:342 t = normalize_whitespace(text)343 return 2 <= len(t) <= 40 and not any(ch.isdigit() for ch in t) and t[0].isupper() and not LOCATION_KIND_RULES[0][1].search(t)344345346def extract_locations(page: NormalizedPage) -> list[ExtractedLocation]:347 out: list[ExtractedLocation] = []348 seen: set[str] = set()349350 def add(loc: ExtractedLocation | None) -> None:351 loc = normalize_location(loc) if loc is not None else None352 if loc is None:353 return354 key = norm_name(loc.name)355 if not key or key in seen or len(out) >= MAX_LOCATIONS:356 return357 seen.add(key)358 out.append(loc)359360 for a in page.jsonld.get("addresses", []) + page.jsonld.get("places", []) + page.microdata.get("addresses", []):361 addr = a.get("address") if isinstance(a.get("address"), dict) else a362 city = text_of(addr.get("addressLocality")) if isinstance(addr, dict) else None363 country = country_code(text_of(addr.get("addressCountry"))) if isinstance(addr, dict) else None364 name = text_of(a.get("name")) or ", ".join(x for x in (city, country) if x)365 if name and (city or country):366 street = text_of(addr.get("streetAddress")) if isinstance(addr, dict) else None367 add(ExtractedLocation(name=name[:120], kind="office", city=city, region=text_of(addr.get("addressRegion")) if isinstance(addr, dict) else None,368 country=country, address_text=street))369 for b in page.blocks:370 if b.kind != "location":371 continue372 lines = [ln.strip() for ln in b.text.split("\n") if ln.strip()]373 if lines:374 add(_location_from_lines(lines[0], lines[1:], b.attrs.get("href")))375 if len(out) < 2: # fallback: headings naming a city/country followed by address-like lines376 blocks = page.blocks377 for i, b in enumerate(blocks):378 if b.kind == "heading" and int(b.attrs.get("level", 3)) >= 2 and (looks_like_city(b.text) or country_code(b.text)):379 lines: list[str] = []380 for nb in blocks[i + 1:i + 4]:381 if nb.kind == "heading":382 break383 lines.extend(x.strip() for x in nb.text.split("\n") if x.strip())384 add(_location_from_lines(b.text, lines))385 return out386387388# ------------------------------------------------------------------------------------------------------------ news389390NEWS_NOISE_ANCHOR = re.compile(r"^(read more|learn more|more|continue reading|view all|see all|all news|all posts|next|previous|older|newer|\d+)$", re.IGNORECASE)391MIN_NEWS_TITLE_CHARS = 3 # anything shorter is a glyph or a counter; the precision rule decides the rest (words / date)392393394def _time_index(html: str, base_url: str) -> dict[str, Any]:395 """canonical href → datetime for <time datetime> elements near a link (dates that the page states)."""396 out: dict[str, Any] = {}397 try:398 tree = LexborHTMLParser(html)399 except Exception: # noqa: BLE001400 return out401 for t in tree.css("time[datetime]")[:400]:402 dt = parse_date(t.attributes.get("datetime"))403 if dt is None:404 continue405 node = t406 for _ in range(7):407 if node is None or node.tag in ("body", "html"):408 break409 a = node.css_first("a[href]") if node.tag != "a" else node410 if a is not None:411 url = absolutize(base_url, a.attributes.get("href") or "")412 if url:413 out.setdefault(canonicalize_url(url), dt)414 break415 node = node.parent416 return out417418419def extract_news(page: NormalizedPage, html: str, *, surface: str, base_url: str) -> list[ExtractedNewsItem]:420 category = NEWS_CATEGORY.get(surface, "other") # type: ignore[call-overload]421 times = _time_index(html, base_url)422 out: list[ExtractedNewsItem] = []423 seen: set[str] = set()424 site = registrable_domain(base_url)425426 def add(title: str, url: str, published: Any, summary: str | None = None) -> None:427 title = normalize_whitespace(title)428 if len(title) < MIN_NEWS_TITLE_CHARS or NEWS_NOISE_ANCHOR.match(title) or len(out) >= MAX_NEWS:429 return430 canon = canonicalize_url(url)431 if canon in seen or canon == canonicalize_url(base_url) or is_static_asset(url):432 return433 published = published or times.get(canon)434 if not news_verdict(title, url=url, published_at=published).ok:435 return436 seen.add(canon)437 out.append(ExtractedNewsItem(title=title[:300], url=url, published_at=published, summary=(summary or None), category=category, language=page.lang))438439 for a in page.jsonld.get("articles", []) + page.microdata.get("articles", []):440 title, url = text_of(a.get("headline")) or text_of(a.get("name")), a.get("url") or (a.get("mainEntityOfPage") if isinstance(a.get("mainEntityOfPage"), str) else None)441 if title and isinstance(url, str):442 add(title, url, parse_date(a.get("datePublished")), text_of(a.get("description")))443 for b in page.blocks:444 href = b.attrs.get("href")445 if b.kind != "news_item" or not href:446 continue447 lines = [ln.strip() for ln in b.text.split("\n") if ln.strip()]448 undated = [ln for ln in lines if not (len(ln) <= 32 and date_from_text(ln))]449 title = next((ln for ln in undated if len(ln) >= 12), undated[0] if undated else "")450 rest = [ln for ln in undated if ln != title]451 add(title, href, date_from_text(b.text) or times.get(canonicalize_url(href)) or date_from_url(href), " ".join(rest[:2])[:300] or None)452 if len(out) < 3: # fallback: article-like links in the main region (same site, deeper path, long anchor)453 base_depth = urlparse(base_url).path.strip("/").count("/")454 for ln in page.links:455 if ln.region != "main" or registrable_domain(ln.url) != site or looks_like_trap(ln.url):456 continue457 depth = urlparse(ln.url).path.strip("/").count("/")458 dated = date_from_url(ln.url) or times.get(canonicalize_url(ln.url))459 if (len(ln.anchor) >= 25 and depth >= base_depth) or (dated and len(ln.anchor) >= 12):460 add(ln.anchor, ln.url, dated)461 out.sort(key=lambda n: (n.published_at is None, -(n.published_at.timestamp() if n.published_at else 0)))462 return out463464465# ------------------------------------------------------------------------------------------------------------ careers (HTML)466467JOB_URL_RE = re.compile(r"(/jobs?/|/careers?/[^/]+|/positions?/|/openings?/|/vacanc|/opportunit|/stellen|/emplois?/|/offres?/|gh_jid=|/job-|lever\.co/|"468 r"greenhouse\.io/|ashbyhq\.com/|myworkdayjobs\.com/.+/job/|smartrecruiters\.com/|workable\.com/j/|recruitee\.com/o/|"469 r"personio\.de/job/|teamtailor\.com/jobs/|bamboohr\.com/careers/\d|jobvite\.com/|icims\.com/jobs/|taleo\.net/.+job)", re.IGNORECASE)470JOB_NOISE_ANCHOR = re.compile(r"^(apply( now)?|view( all)?( jobs| openings| roles| positions)?|see (all|more|open)( jobs| roles| positions)?|learn more|read more|"471 r"careers?|jobs?|join (us|the team|our team)|open (roles|positions)|all (jobs|roles|positions|openings)|search jobs|"472 r"browse jobs|explore|more|back|home|filter|next|previous|\d+)$", re.IGNORECASE)473474475_states_location = states_job_location476477478def extract_jobs_html(page: NormalizedPage, *, base_url: str) -> list[ExtractedJob]:479 jobs: list[ExtractedJob] = []480 seen: set[str] = set()481482 def add(title: str, url: str | None, location: str | None, department: str | None = None) -> None:483 title = normalize_whitespace(title).strip(" -–—|·")484 if not (4 <= len(title) <= 140) or JOB_NOISE_ANCHOR.match(title) or len(jobs) >= MAX_JOBS:485 return486 job = refine_job(ExtractedJob(title=title, url=url, location_text=location, department=department))487 if job is None:488 return489 key = (canonicalize_url(url) if url else "") + "|" + job.title.lower()490 if key in seen:491 return492 seen.add(key)493 jobs.append(finish_job(job))494495 jl_jobs = jobs_from_jsonld(list(page.jsonld.get("job_postings", [])) + list(page.microdata.get("job_postings", [])), page_url=base_url)496 for j in jl_jobs:497 add(j.title, j.url, j.location_text, j.department)498 if jobs and jobs[-1].title == normalize_whitespace(j.title):499 jobs[-1] = j500 block_jobs: list[ExtractedJob] = []501 for b in page.blocks:502 href = b.attrs.get("href")503 if b.kind != "job_listing" or not href or not JOB_URL_RE.search(href) and registrable_domain(href) == registrable_domain(base_url) and urlparse(href).path.count("/") < 2:504 continue505 lines = [ln.strip() for ln in b.text.split("\n") if ln.strip()]506 if lines and " | " in lines[0]: # table row: cells joined by " | "507 lines = [c.strip() for c in lines[0].split(" | ") if c.strip()] + lines[1:]508 if not lines:509 continue510 title = lines[0]511 loc = next((ln for ln in lines[1:6] if _states_location(ln) or is_known_city(ln)), None)512 dept = next((ln for ln in lines[1:6] if ln != loc and 2 <= len(ln) <= 40 and not _states_location(ln) and not date_from_text(ln)), None)513 before = len(jobs)514 add(title, href, loc, dept)515 if len(jobs) > before:516 block_jobs.append(jobs[-1])517 if len(jobs) < 3: # fallback: job-like anchors anywhere in the main region518 cands = [ln for ln in page.links if ln.region == "main" and JOB_URL_RE.search(ln.url) and 4 <= len(ln.anchor) <= 140519 and not JOB_NOISE_ANCHOR.match(ln.anchor) and not is_static_asset(ln.url)]520 if len(cands) >= 3:521 for ln in cands:522 add(ln.anchor, ln.url, None)523 if len(jobs) < 3 and not jl_jobs:524 return [] # not a trustworthy listing — the pipeline must not mark jobs removed on this525 return jobs526527528# ------------------------------------------------------------------------------------------------------------ products529530531def extract_products(page: NormalizedPage) -> list[ExtractedProduct]:532 out: list[ExtractedProduct] = []533 seen: set[str] = set()534535 def add(name: str, url: str | None, description: str | None, category: str | None = None) -> None:536 name = normalize_whitespace(name).strip(" -–—|·")537 key = norm_name(name)538 if not (2 <= len(name) <= 90) or not key or key in seen or len(out) >= MAX_PRODUCTS or NEWS_NOISE_ANCHOR.match(name):539 return540 if not product_verdict(name).ok:541 return542 seen.add(key)543 out.append(ExtractedProduct(name=name, url=url, category=category, description=(description or None)))544545 for p in page.jsonld.get("products", []) + page.microdata.get("products", []):546 name = text_of(p.get("name"))547 if name:548 add(name, p.get("url") if isinstance(p.get("url"), str) else None, text_of(p.get("description")), text_of(p.get("category")))549 for b in page.blocks:550 if b.kind != "product_card":551 continue552 lines = [ln.strip() for ln in b.text.split("\n") if ln.strip()]553 if lines:554 add(lines[0], b.attrs.get("href"), " ".join(lines[1:4])[:300], b.path.split(" > ")[-1] if b.path else None)555 return out556557558# ------------------------------------------------------------------------------------------------------------ discovery links559560561def discovered_links(page: NormalizedPage, *, base_url: str, canonical_domain: str | None) -> list[DiscoveredUrl]:562 site = registrable_domain(canonical_domain or base_url)563 out: list[DiscoveredUrl] = []564 seen: set[str] = set()565 for ln in page.links:566 if registrable_domain(ln.url) != site or is_static_asset(ln.url) or looks_like_trap(ln.url):567 continue568 canon = canonicalize_url(ln.url)569 if canon in seen:570 continue571 surface, conf = classify_url(ln.url, anchor=ln.anchor, canonical_domain=canonical_domain or site)572 if surface in (Surface.OTHER, Surface.HOMEPAGE) or conf < settings.discovery_min_confidence:573 continue574 seen.add(canon)575 out.append(DiscoveredUrl(url=ln.url, surface=surface, confidence=conf, anchor=ln.anchor[:120] or None, method="nav" if ln.region in ("nav", "header", "footer") else "link"))576 if len(out) >= MAX_DISCOVERED:577 break578 for feed in page.feeds:579 out.append(DiscoveredUrl(url=feed, surface=Surface.FEED, confidence=0.9, method="feed"))580 return out581582583# ------------------------------------------------------------------------------------------------------------ connector584585586@register587class GenericHtmlConnector(Connector):588 meta = ConnectorMeta(connector_id="generic-html-v1", name="Generic HTML surface", version="1", category=Surface.OTHER, fetch_mode=FetchMode.HTTP,589 supports_discovery=True, default_interval_s=24 * 3600, surfaces=("*",), priority=1,590 description="Surface-aware extraction for any corporate web page")591592 def extract(self, sensor: Mapping[str, Any], result: FetchResult) -> Extraction:593 surface = str(sensor.get("surface") or Surface.OTHER)594 html = result.text595 if not result.is_html and result.is_json:596 raise ValueError("generic HTML connector received JSON")597 page = normalize.parse(html, url=result.final_url, surface=surface)598 ex = page.to_extraction()599 cfg = sensor.get("config") or {}600 canonical_domain = str(cfg.get("canonical_domain") or "") or None601 if surface == Surface.LEADERSHIP or surface == Surface.ABOUT:602 ex.people = extract_people(page)603 if surface == Surface.PRICING:604 ex.plans = extract_plans(page)605 if surface == Surface.LOCATIONS or surface == Surface.CONTACT:606 ex.locations = extract_locations(page)607 if surface in NEWS_SURFACES:608 ex.news = extract_news(page, html, surface=surface, base_url=result.final_url)609 if surface in (Surface.CAREERS, Surface.JOBS_BOARD):610 ex.jobs = extract_jobs_html(page, base_url=result.final_url)611 if ex.jobs:612 ex.blocks = [b for b in ex.blocks if b.kind != "job_listing"] + job_blocks(ex.jobs, path="Jobs")613 if surface in PRODUCT_SURFACES:614 ex.products = extract_products(page)615 ex.discovered = discovered_links(page, base_url=result.final_url, canonical_domain=canonical_domain)616 ex.meta.update({"surface": surface, "title": page.title, "headings": [h[1][:80] for h in page.headings[:30]], "link_count": len(page.links),617 "structured": bool(ex.jobs or ex.plans or ex.people or ex.locations or ex.news or ex.products)})618 return ex619620621def blocks_summary(blocks: list[Block]) -> dict[str, int]:622 out: dict[str, int] = {}623 for b in blocks:624 out[b.kind] = out.get(b.kind, 0) + 1625 return out626627628__all__ = ["GenericHtmlConnector", "blocks_summary", "discovered_links", "extract_jobs_html", "extract_locations", "extract_news", "extract_people",629 "extract_plans", "extract_products", "looks_like_name", "parse_price", "role_category"]630