Contact Scraper
Pricing
from $100.00 / 1,000 site analyzeds
Pricing
from $100.00 / 1,000 site analyzeds
Rating
0.0
(0)
Developer
Chel
Maintained by CommunityActor stats
0
Bookmarked
7
Total users
1
Monthly active users
9 days ago
Last modified
Categories
Share
1apify>=3.3.0,<4.0.02beautifulsoup4>=4.12.0,<5.0.03dnspython>=2.6.0,<3.0.04phonenumbers>=8.13.0,<9.0.05playwright==1.56.06requests>=2.31.0,<3.0.07tldextract>=5.1.0,<6.0.0FROM apify/actor-python-playwright:3.11-1.56.0
WORKDIR /usr/src/app
COPY requirements.txt ./RUN python -m pip install --no-cache-dir -r requirements.txt
COPY . ./
CMD ["python", "-u", "main.py"]1#!/usr/bin/env python32from __future__ import annotations3
4import asyncio5import json6import re7import time8from collections import deque9from dataclasses import dataclass10from typing import Any, Dict, Iterable, List, Optional, Set, Tuple11from urllib.parse import parse_qsl, unquote, urlencode, urljoin, urlparse, urlunparse12
13import requests14from apify import Actor15from bs4 import BeautifulSoup, Comment16
17try: # pragma: no cover - optional dependency18 import tldextract # type: ignore19 _TLD_EXTRACT = tldextract.TLDExtract(suffix_list_urls=None)20except Exception: # pragma: no cover21 tldextract = None22 _TLD_EXTRACT = None23
24try: # pragma: no cover - optional dependency25 import dns.resolver # type: ignore26 dns = dns.resolver27except Exception: # pragma: no cover28 dns = None29
30try: # pragma: no cover - optional dependency31 import phonenumbers # type: ignore32except Exception: # pragma: no cover33 phonenumbers = None34
35try: # pragma: no cover - optional dependency36 from playwright.sync_api import TimeoutError as PlaywrightTimeoutError # type: ignore37 from playwright.sync_api import sync_playwright # type: ignore38except Exception: # pragma: no cover39 PlaywrightTimeoutError = TimeoutError40 sync_playwright = None41
42
43DEFAULT_HEADERS = {44 "User-Agent": (45 "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "46 "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36"47 ),48 "Accept-Language": "en-US,en;q=0.9",49}50
51SITE_ANALYZED_EVENT_NAME = "site-analyzed"52SITE_ANALYZED_STATE_KEY = "__site_analyzed_ppe_state"53
54
55EMAIL_RE = re.compile(r"(?<![A-Za-z0-9._%+-])([A-Za-z0-9._%+-]{2,64}@[A-Za-z0-9.-]+\.[A-Za-z]{2,63})(?![A-Za-z0-9._%+-])")56PHONE_RE = re.compile(r"\+?\d[\d\-\.\(\)\s]{7,}\d")57
58SCRIPT_EMAIL_RE = re.compile(59 r"""(?is)60 (?:(?:contact|support|sales|info|hello|admin|email|mail)[-_]?(?:email|address)?)\s*[:=]\s*61 ['"]([A-Za-z0-9._%+-]{2,64}@[A-Za-z0-9.-]+\.[A-Za-z]{2,63})['"]62 """,63 re.VERBOSE,64)65
66SCRIPT_PHONE_RE = re.compile(67 r"""(?is)68 (?:(?:contact|support|sales|office|help|customer)[-_]?)?69 (?:(?:phone|telephone|mobile|tel)[-_]?(?:number)?)\s*[:=]\s*70 ['"](\+?[0-9][0-9\-\.\(\)\s]{7,}[0-9])['"]71 """,72 re.VERBOSE,73)74
75CONTACT_KEYWORDS = [76 "contact us",77 "get in touch",78 "reach out",79 "customer support",80 "contact",81 "support",82 "speak with us",83 "talk to us",84 "reach us",85]86
87EMAIL_CONTEXT_LABELS = [88 "contact",89 "get in touch",90 "reach us",91 "reach out",92 "email us",93 "support",94 "sales",95 "help",96 "customer support",97]98
99EMAIL_CONTAINER_HINTS = [100 "footer",101 "header",102 "nav",103 "main",104 "article",105 "aside",106 "contact",107 "support",108 "sales",109 "help",110 "get in touch",111 "get-in-touch",112 "reach us",113 "reach-us",114]115
116EMAIL_REGION_HINTS = [117 "footer",118 "header",119 "nav",120 "main",121 "article",122 "aside",123 "contact",124 "support",125 "sales",126 "help",127 "get in touch",128 "get-in-touch",129 "reach us",130 "reach-us",131 "talk to us",132 "talk to sales",133 "request quote",134]135
136CONTACT_URL_PROBES = [137 "/contact-us",138 "/contact-us/",139 "/contact",140 "/contact/",141 "/get-in-touch",142 "/get-in-touch/",143]144
145CONTACT_KEYWORD_PATTERN = re.compile(146 "(?:" + "|".join(re.escape(term) for term in sorted(CONTACT_KEYWORDS, key=len, reverse=True)) + ")",147 re.IGNORECASE,148)149
150CONTACT_INTENT_HINTS = {151 "contact",152 "contactus",153 "get-in-touch",154 "getintouch",155 "talk-to-us",156 "support",157}158
159WEAK_PATH_HINTS = {160 "locations",161 "privacy",162 "terms",163 "legal",164 "blog",165 "docs",166 "documentation",167 "help-article",168 "help",169 "partner",170}171
172EMAIL_ASSET_EXTENSIONS = {".png", ".jpg", ".jpeg", ".gif", ".webp", ".svg", ".pdf", ".js", ".css"}173EMAIL_ASSET_STRINGS = {174 "300x250",175 "768x",176 "1024x",177 "300x",178 "@2x",179 "-min",180 "sprite",181 "image",182 "img",183 "banner",184 "icon",185}186
187PHONE_CONTEXT_HINTS = {"phone", "call", "support", "sales", "whatsapp", "contact", "call us", "tel", "mobile"}188PHONE_STRONG_CONTEXT_HINTS = {189 "call",190 "call us",191 "contact us",192 "get in touch",193 "support",194 "sales",195 "whatsapp",196 "phone",197 "tel",198 "call now",199 "office",200 "reach",201}202PHONE_EXCLUDE_CONTEXT_HINTS = {"timestamp", "tracking", "pixel", "id", "ref", "order", "tracking number"}203PRODUCT_IDENTIFIER_HINTS = {204 "sku",205 "item no",206 "item number",207 "article no",208 "article number",209 "style no",210 "style number",211 "product code",212 "model no",213 "model number",214 "part no",215 "part number",216 "ean",217 "upc",218}219HUMAN_READABLE_ATTRS = {"aria-label", "title", "alt"}220EMAIL_STRONG_CONTACT_LABELS = {"support", "contact", "get in touch", "email", "sales", "reach us", "reach out", "customer support"}221PHONE_PRODUCT_NOISE_HINTS = {"cloud phone", "mobile profiles", "virtual phone", "phone profiles", "virtual browser", "residential proxies"}222
223WIDGET_LAUNCHER_TAGS = {"button", "a", "div", "span"}224WIDGET_LAUNCHER_PHRASES = [225 "live chat",226 "contact us",227 "send us a message",228 "message us",229 "talk to us",230 "customer support",231 "need help",232 "whatsapp",233 "support",234 "contact",235 "chat",236 "help",237]238WIDGET_MATCH_TERMS = {239 "bubble",240 "chat",241 "contact",242 "customerchat",243 "floating",244 "help",245 "launcher",246 "messenger",247 "support",248 "wa",249 "whatsapp",250 "widget",251}252WIDGET_FIXED_POSITION_HINTS = {"fixed", "floating", "sticky", "launcher", "bubble", "fab"}253WIDGET_CORNER_HINTS = {"top", "bottom", "left", "right", "corner"}254GENERIC_WIDGET_PROVIDER = "Generic contact widget"255RENDERED_WIDGET_EXTRA_WAIT_MS = 1500256
257_PLAYWRIGHT_DISABLED_REASON: Optional[str] = None258
259FORM_REQUIRED_ROLES = {"name", "email", "phone", "company", "message", "subject"}260FORM_MESSAGE_ROLES = {"message", "inquiry", "enquiry", "subject", "details", "description"}261FORM_INFO_ONLY_ROLES = {"name", "email", "phone", "company"}262FORM_REVIEW_FIELD_HINTS = {263 "review",264 "write review",265 "submit review",266 "rate and review",267 "pros",268 "cons",269 "recommend",270 "rating",271 "star rating",272 "stars",273 "linkedin profile",274 "your experience",275 "quality",276 "installation",277}278FORM_REVIEW_HINTS = FORM_REVIEW_FIELD_HINTS | {"review form"}279FORM_CONTACT_SUBMIT_HINTS = {280 "send",281 "send message",282 "send email",283 "contact us",284 "get in touch",285 "reach out",286 "support",287 "submit message",288 "submit inquiry",289}290FORM_CONTACT_ACTION_HINTS = {291 "contact",292 "contact us",293 "get in touch",294 "support",295 "help",296 "feedback",297}298FORM_LEAD_CAPTURE_HINTS = {299 "subscribe",300 "subscription",301 "newsletter",302 "watch the demo",303 "book a demo",304 "request a demo",305 "get a demo",306 "get demo",307 "demo",308 "get ebook",309 "ebook",310 "whitepaper",311 "download",312 "free trial",313 "start free",314 "sign up",315 "signup",316 "join now",317 "contact sales",318 "talk to sales",319 "request quote",320 "request a quote",321 "get quote",322 "book a call",323 "schedule a call",324 "book a meeting",325 "schedule a meeting",326}327FORM_HARD_REJECT_HINTS = {328 "watch the demo",329 "book a demo",330 "request a demo",331 "get a demo",332 "get demo",333 "demo",334 "newsletter",335 "subscribe",336 "subscription",337 "contact sales",338 "talk to sales",339 "book a call",340 "schedule a call",341 "request quote",342 "request a quote",343 "get quote",344 "falar com vendas",345}346FORM_SALES_FLOW_HINTS = {347 "talk to sales",348 "contact sales",349 "contact-sales",350 "talk-to-sales",351 "request demo",352 "get demo",353 "get a demo",354 "book demo",355 "schedule demo",356 "book a call",357 "schedule a call",358 "schedule call",359 "book a meeting",360 "schedule a meeting",361 "speak to sales",362 "start free trial",363 "watch demo",364 "price quote",365 "free quote",366 "get quote",367 "request quote",368 "request a quote",369 "book-a-call",370 "request-quote",371 "falar com vendas",372 "fale com vendas",373}374FORM_INTERACTION_GATE_HINTS = {"popup", "modal", "dialog", "drawer", "accordion"}375FORM_CONTACT_TEXT_HINTS = {376 "contact us",377 "customer support",378 "contact support",379 "get in touch",380 "reach out",381 "feedback",382 "help",383 "support",384 "send us a message",385 "message us",386 "general inquiry",387 "general enquiry",388}389FORM_SALES_FIELD_HINTS = {390 "business email",391 "work email",392 "corporate email",393 "company size",394 "team size",395 "employee count",396 "number of employees",397 "company url",398 "company website",399 "website",400 "web site",401 "number of browser profiles",402 "browser profiles",403 "quantity",404 "qty",405 "use case",406 "preferred contact method",407 "best way to contact",408}409FORM_SALES_QUALIFICATION_ROLES = {410 "business_email",411 "company_size",412 "company_url",413 "browser_profiles",414 "quantity",415 "use_case",416 "preferred_contact_method",417}418WIDGET_SUPPORT_INTENT_HINTS = {419 "support",420 "help",421 "chat",422 "message us",423 "send us a message",424 "customer support",425 "contact support",426}427LEAD_GEN_WIDGET_ACTION_HINTS = {428 "book",429 "book demo",430 "book a demo",431 "book a call",432 "schedule",433 "schedule a call",434 "schedule a meeting",435 "meeting",436 "demo",437 "request demo",438 "request a demo",439 "appointment",440 "contact us",441 "talk to us",442}443LEAD_GEN_WIDGET_PROVIDERS = {"Calendly", "HubSpot chat/forms", "Typeform", "Jotform"}444FORM_REJECT_HINTS = {445 "plan",446 "pricing",447 "number of proxies",448 "quantity",449 "coupon",450 "billing",451 "login",452 "signup",453 "newsletter",454 "subscribe",455 "search",456 "filter",457 "country",458}459SNAPSHOT_REF_USAGE_NOTES = [460 "Open the page_url and run a fresh browser snapshot before interacting.",461 "Match snapshot elements to these locator hints by role, label, name, id, placeholder, or aria-label.",462 "Use the matched snapshot ref for fill and click actions.",463 "Do not pass scraper CSS paths directly into browser act commands.",464]465
466TLD_PHONE_REGION_MAP = {467 "us": "US",468 "uk": "GB",469 "ca": "CA",470 "au": "AU",471 "de": "DE",472 "fr": "FR",473 "es": "ES",474 "it": "IT",475 "nl": "NL",476 "se": "SE",477 "no": "NO",478 "dk": "DK",479 "fi": "FI",480 "ch": "CH",481 "be": "BE",482 "in": "IN",483 "sg": "SG",484 "au": "AU",485 "br": "BR",486 "mx": "MX",487 "jp": "JP",488 "za": "ZA",489 "ru": "RU",490 "cn": "CN",491 "nz": "NZ",492}493
494MAIL_PROVIDER_HINTS = [495 ("Google Workspace", {"_aspmx.l.google.com", "aspmx", "alt1.aspmx", "smtp-relay.gmail.com", "_spf.google.com", "_spf1.google.com"}),496 ("Microsoft 365 / Outlook", {"protection.outlook.com", "mail.protection.outlook.com", "protection.outlook365.com", "mail.windows365.com", "outlook365.com"}),497 ("Zoho Mail", {"zoho.com", "zohomail", "zohosecuremail", "zoho.in"}),498 ("Proton Mail", {"protonmail", "protonmail.ch", "mail.protonmail.com"}),499 ("Fastmail", {"fastmail.com", "messagingengine.com"}),500 ("Amazon SES", {"amazonses", "amazonses.com", "email-smtp.", "email-smtp"}),501 ("Mailgun", {"mailgun.org", "mg.mailgun.org", "email.mailgun.org", "mailgun"}),502]503
504
505
506
507def _contains_token(context: str, token: str) -> bool:508 lowered = (context or "").lower()509 token_l = token.lower()510 if not token_l:511 return False512 if " " in token_l:513 return token_l in lowered514 return bool(re.search(rf"\b{re.escape(token_l)}\b", lowered))515
516
517def _normalize_intent_text(text: str) -> str:518 lowered = (text or "").lower()519 lowered = re.sub(r"[_/\-]+", " ", lowered)520 lowered = re.sub(r"\s+", " ", lowered)521 return lowered.strip()522
523
524def _matching_intent_hints(text: str, hints: Iterable[str]) -> List[str]:525 normalized_text = _normalize_intent_text(text)526 if not normalized_text:527 return []528 hits = []529 for hint in hints:530 normalized_hint = _normalize_intent_text(hint)531 if normalized_hint and _contains_token(normalized_text, normalized_hint):532 hits.append(normalized_hint)533 return sorted(set(hits), key=lambda value: (-len(value), value))534
535
536WIDGET_PROVIDERS = {537 "Intercom": {538 "host": {"widget.intercom.io", "intercomcdn.com", "intercom.com"},539 "inline": {"intercom", "intercomsettings", "intercom.boot"},540 "selector": {"intercom-lightweight-app", "intercom-app"},541 "open_channel": True,542 },543 "Drift": {544 "host": {"js.driftt.com", "drift.com"},545 "inline": {"drift", "drift-api", "driftt"},546 "selector": {"drift-frame-controller", "drift-conversations"},547 "open_channel": True,548 },549 "Crisp": {550 "host": {"go.crisp.chat", "crisp.chat"},551 "inline": {"crisp", "CRISP_WEBSITE_ID"},552 "selector": {"crisp-client"},553 "open_channel": True,554 },555 "Zendesk": {556 "host": {"static.zdassets.com", "zendesk.com", "zopim.com", "zendesk.groovehq.com"},557 "inline": {"zopim", "zendesk"},558 "selector": {"zopim-widget", "zendesk-chat"},559 "open_channel": True,560 },561 "Tidio": {562 "host": {"code.tidio.co", "tidio.co"},563 "inline": {"tidio", "tidioChatApi"},564 "selector": {"tidio-chat", "tidio-widget"},565 "open_channel": True,566 },567 "LiveChat": {568 "host": {"livechatinc.com", "livechat.com"},569 "inline": {"livechat", "LiveChatWidget", "widget.livechatinc.com"},570 "selector": {"lc-frame", "livechat_widget"},571 "open_channel": True,572 },573 "HubSpot chat/forms": {574 "host": {"js.hs-scripts.com", "js.hsleadflows.net", "js.hsforms.net", "forms.hubspot.com", "app.hubspot.com"},575 "inline": {"hbspt.forms", "hubspot", "_hsq"},576 "selector": {"hs-form", "hubspot-messages-iframe-container", "hs-chat-widget"},577 "open_channel": True,578 },579 "Calendly": {580 "host": {"calendly.com", "assets.calendly.com"},581 "inline": {"calendly", "Calendly.initInlineWidget"},582 "selector": {"calendly-inline-widget"},583 "open_channel": True,584 },585 "Typeform": {586 "host": {"embed.typeform.com", "typeform.com"},587 "inline": {"typeform", "new Typeform", "typeformEmbed"},588 "selector": {"typeform-widget"},589 "open_channel": True,590 },591 "Jotform": {592 "host": {"form.jotform.com", "js.jotform.com"},593 "inline": {"jotform", "JotForm"},594 "selector": {"jotform-form"},595 "open_channel": True,596 },597 "Freshchat": {598 "host": {"wchat.freshchat.com", "freshchat.com"},599 "inline": {"freshchat", "window.Freshchat"},600 "selector": {"freshchat-widget"},601 "open_channel": True,602 },603 "Chatwoot": {604 "host": {"chatwoot.com"},605 "inline": {"chatwoot", "chatwootSDK"},606 "selector": {"woot-widget", "chatwoot-widget"},607 "open_channel": True,608 },609 "HelpCrunch": {610 "host": {"app.helpcrunch.com"},611 "inline": {"helpcrunch", "helpcrunch-widget", "helpCrunch"},612 "selector": {"helpcrunch-widget"},613 "open_channel": True,614 },615 "Olark": {616 "host": {"static.olark.com", "olark.com"},617 "inline": {"olark", "olark.identify"},618 "selector": {"olark-widget"},619 "open_channel": True,620 },621 "Userlike": {622 "host": {"widget.userlike.com", "userlike.com"},623 "inline": {"userlike", "Userlike"},624 "selector": {"userlike-widget", "userlike-web-messenger"},625 "open_channel": True,626 },627 "WhatsApp widget": {628 "host": {"wa.me", "api.whatsapp.com", "chat.whatsapp.com"},629 "inline": {"wa.me", "whatsapp", "wa-widget"},630 "selector": {"whatsapp-widget", "whatsapp-chat", "wa-chat", "wa-button"},631 "open_channel": True,632 },633 "Facebook Messenger customer chat": {634 "host": {"connect.facebook.net", "facebook.com/customerchat", "messenger.com"},635 "inline": {"customerchat", "M.E.CUSTOMER_CHAT", "facebook customerchat"},636 "selector": {"fb-customerchat", "fb-messenger-checkbox"},637 "open_channel": True,638 },639}640
641SOCIAL_DOMAINS = {642 "linkedin": ["linkedin.com/company", "linkedin.com/in", "linkedin.com/school"],643 "x": ["x.com", "twitter.com"],644 "facebook": ["facebook.com", "fb.com", "facebook.net", "m.me"],645 "instagram": ["instagram.com"],646 "youtube": ["youtube.com", "youtu.be"],647 "tiktok": ["tiktok.com", "vm.tiktok.com"],648 "telegram": ["t.me", "telegram.me", "telegram.org"],649 "whatsapp": ["wa.me", "api.whatsapp.com", "chat.whatsapp.com"],650 "messenger": ["m.me", "www.messenger.com", "facebook.com/messages", "facebook.com/dialog/send"],651}652
653
654CONFIDENCE_ORDER = {"VERIFIED": 3, "STRONG": 2, "PLAUSIBLE": 1}655
656
657def _normalize_query(query: str) -> str:658 pairs = [659 (k, v)660 for k, v in parse_qsl(query, keep_blank_values=True)661 if not k.lower().startswith("utm_") and k.lower() not in {"fbclid", "gclid", "yclid", "msclkid"}662 ]663 return urlencode(pairs, doseq=True)664
665
666def normalize_url(url: str, *, base: str | None = None) -> str:667 if not url:668 return ""669 raw = url.strip()670 if raw.startswith("//"):671 raw = "https:" + raw672 if base and not re.match(r"^[a-z][a-z0-9+\-.]*://", raw, flags=re.I):673 raw = urljoin(base, raw)674 if not re.match(r"^[a-z][a-z0-9+\-.]*://", raw, flags=re.I):675 raw = "https://" + raw676
677 parsed = urlparse(raw)678 if not parsed.scheme or not parsed.netloc:679 return ""680 scheme = parsed.scheme.lower()681 netloc = parsed.netloc.lower().split("@")[-1]682 if ":" in netloc and not re.fullmatch(r".+:\d{2,5}", netloc):683 netloc = netloc.split(":", 1)[0]684 path = (parsed.path or "/").rstrip("/") or "/"685 query = _normalize_query(parsed.query)686 return urlunparse((scheme, netloc, path, "", query, ""))687
688
689def get_registrable_domain(url: str) -> str:690 host = urlparse(url).hostname or ""691 host = host.lower()692 if not host:693 return ""694 if _TLD_EXTRACT is None:695 parts = host.split(".")696 return ".".join(parts[-2:]) if len(parts) >= 2 else host697 ext = _TLD_EXTRACT(host)698 if ext.domain and ext.suffix:699 return f"{ext.domain}.{ext.suffix}"700 return host701
702
703def same_site(url_a: str, url_b: str) -> bool:704 return get_registrable_domain(url_a) == get_registrable_domain(url_b)705
706
707def _strip_text(text: str) -> str:708 return " ".join((text or "").split())709
710def _page_host_block(path_or_url: str) -> str:711 return (path_or_url or "").lower().strip("/")712
713
714def _has_contact_phrase(text: str) -> bool:715 return bool(_matching_intent_hints(text, CONTACT_KEYWORDS))716
717
718def _is_contact_url_hint(path: str) -> bool:719 normalized = _normalize_intent_text(path or "")720 has_contact_hint = bool(_matching_intent_hints(normalized, CONTACT_INTENT_HINTS))721 if not has_contact_hint:722 return False723 if _matching_intent_hints(normalized, WEAK_PATH_HINTS):724 return True725 return has_contact_hint726
727
728def _contact_phrase_snippet(text: str) -> bool:729 return bool(_matching_intent_hints(text, CONTACT_KEYWORDS))730
731
732def _iter_visible_text_nodes(soup: BeautifulSoup) -> Iterable[Tuple[object, str]]:733 for node in soup.find_all(string=True):734 if isinstance(node, Comment):735 continue736 if not isinstance(node, str):737 continue738 parent = node.parent739 if parent is None:740 continue741 if parent.name and parent.name.lower() in {"script", "style", "noscript"}:742 continue743 text = _strip_text(node)744 if text:745 yield parent, text746
747
748def _stringify_attr_value(value: Any) -> str:749 if isinstance(value, list):750 return _strip_text(" ".join(str(part) for part in value if part))751 return _strip_text(str(value or ""))752
753
754def _simple_css_token(value: str) -> str:755 return value if value and re.fullmatch(r"[A-Za-z_][A-Za-z0-9_-]*", value) else ""756
757
758def _css_selector(node: Any, *, max_depth: int = 6) -> str:759 if not getattr(node, "name", None):760 return ""761 parts: List[str] = []762 current = node763 depth = 0764 while current is not None and getattr(current, "name", None) and current.name != "[document]" and depth < max_depth:765 tag = current.name.lower()766 part = tag767 node_id = _simple_css_token(current.get("id") or "")768 if node_id:769 part = f"{tag}#{node_id}"770 parts.append(part)771 break772 classes = [_simple_css_token(cls) for cls in (current.get("class") or [])]773 classes = [cls for cls in classes if cls][:2]774 if classes:775 part += "".join(f".{cls}" for cls in classes)776 parent = getattr(current, "parent", None)777 if parent is not None and getattr(parent, "find_all", None):778 siblings = [sib for sib in parent.find_all(current.name, recursive=False)]779 if len(siblings) > 1:780 index = siblings.index(current) + 1781 part += f":nth-of-type({index})"782 parts.append(part)783 current = parent784 depth += 1785 return " > ".join(reversed(parts))786
787
788def _is_human_readable_attr(name: str, value: str) -> bool:789 attr_name = (name or "").strip().lower()790 text = _strip_text(value)791 if not attr_name or not text or len(text) > 240:792 return False793 if attr_name in HUMAN_READABLE_ATTRS:794 return True795 if not attr_name.startswith("data-"):796 return False797 lowered = text.lower()798 if lowered.startswith(("http://", "https://", "//", "{", "[")):799 return False800 if "=" in text and "@" not in text and not PHONE_RE.search(text):801 return False802 if any(marker in lowered for marker in ("function(", "</", "/>", "javascript:", "display:", "position:")):803 return False804 if not any(ch.isalpha() for ch in text) and "@" not in text and not PHONE_RE.search(text):805 return False806 return True807
808
809def _iter_human_readable_attributes(soup: BeautifulSoup) -> Iterable[Tuple[Any, str, str]]:810 for tag in soup.find_all(True):811 for attr_name, raw_value in tag.attrs.items():812 if attr_name == "href":813 continue814 text = _stringify_attr_value(raw_value)815 if _is_human_readable_attr(attr_name, text):816 yield tag, attr_name, text817
818
819def _decode_cfemail(encoded: str) -> str:820 value = (encoded or "").strip()821 if len(value) < 4 or len(value) % 2 != 0 or not re.fullmatch(r"[0-9a-fA-F]+", value):822 return ""823 try:824 key = int(value[:2], 16)825 chars = [chr(int(value[i : i + 2], 16) ^ key) for i in range(2, len(value), 2)]826 except ValueError:827 return ""828 decoded = "".join(chars).strip().lower()829 return decoded if EMAIL_RE.fullmatch(decoded) else ""830
831
832def _normalize_region_hint(value: str) -> str:833 return (value or "").strip().lower().replace("_", "-")834
835
836def _has_email_region_hint(value: str) -> str:837 lowered = _normalize_region_hint(value)838 for hint in EMAIL_REGION_HINTS:839 hint_norm = _normalize_region_hint(hint)840 if hint_norm in lowered:841 return hint_norm842 return ""843
844
845def _infer_email_region(node: Any) -> str:846 if not node:847 return "text"848
849 default_region = "text"850 cursor = node if hasattr(node, "name") else getattr(node, "parent", None)851 if not cursor:852 return "text"853
854 for idx, ancestor in enumerate([cursor] + list(getattr(cursor, "parents", []))):855 if not getattr(ancestor, "name", None):856 continue857 if idx > 20:858 break859 tag = (ancestor.name or "").lower()860 if tag in {"footer", "header", "nav", "main", "article", "aside"}:861 return tag862 if tag in {"section", "div", "article", "aside"}:863 default_region = tag864
865 attrs = " ".join(866 part867 for part in (868 ancestor.get("id") or "",869 " ".join(ancestor.get("class") or []),870 ancestor.get("role") or "",871 ancestor.get("aria-label") or "",872 )873 if part874 )875 hint = _has_email_region_hint(attrs)876 if hint:877 return hint878
879 return default_region880
881
882def _collect_email_context(node: Any, snippet: str) -> Tuple[str, str, List[str], str]:883 region = _infer_email_region(node)884 context_parts: List[str] = [snippet]885 heading = ""886
887 if node is not None and hasattr(node, "find_previous"):888 heading_node = node.find_previous(["h1", "h2", "h3", "h4", "h5", "h6"])889 if heading_node:890 heading = _strip_text(heading_node.get_text(" ", strip=True))891 if heading:892 context_parts.append(heading)893
894 cursor = node if hasattr(node, "name") else getattr(node, "parent", None)895 if cursor is not None:896 for idx, ancestor in enumerate([cursor] + list(getattr(cursor, "parents", []))):897 if idx > 4:898 break899 if not ancestor or not getattr(ancestor, "get_text", None):900 continue901 text = _strip_text(ancestor.get_text(" ", strip=True))902 if text:903 context_parts.append(text)904
905 combined = " ".join(context_parts)906 combined = combined[:900]907 nearby_labels = _matching_intent_hints(combined, EMAIL_CONTEXT_LABELS)908 if _has_contact_phrase(combined):909 for term in CONTACT_KEYWORDS:910 if _contains_token(combined, term) and term not in nearby_labels:911 nearby_labels.append(term)912 nearby_labels = sorted(set(nearby_labels))913 return region, heading, nearby_labels, combined914
915
916def _collect_phone_context(node: Any, snippet: str) -> Tuple[str, str, List[str], str]:917 region = _infer_email_region(node)918 context_parts: List[str] = [snippet]919 heading = ""920
921 if node is not None and hasattr(node, "find_previous"):922 heading_node = node.find_previous(["h1", "h2", "h3", "h4", "h5", "h6"])923 if heading_node:924 heading = _strip_text(heading_node.get_text(" ", strip=True))925 if heading:926 context_parts.append(heading)927
928 cursor = node if hasattr(node, "name") else getattr(node, "parent", None)929 if cursor is not None:930 for idx, ancestor in enumerate([cursor] + list(getattr(cursor, "parents", []))):931 if idx > 4:932 break933 if not ancestor or not getattr(ancestor, "get_text", None):934 continue935 text = _strip_text(ancestor.get_text(" ", strip=True))936 if text:937 context_parts.append(text)938
939 combined = " ".join(context_parts)[:900]940 labels = set(PHONE_CONTEXT_HINTS) | set(PHONE_STRONG_CONTEXT_HINTS) | {"sms", "text", "text us"}941 nearby_labels = _matching_intent_hints(combined, labels)942 if _has_contact_phrase(combined):943 for term in CONTACT_KEYWORDS:944 if _contains_token(combined, term) and term not in nearby_labels:945 nearby_labels.append(term)946 return region, heading, sorted(set(nearby_labels)), combined947
948
949def _register_email_candidate(950 candidates: Dict[str, Dict[str, Any]],951 email: str,952 candidate: Dict[str, Any],953) -> None:954 existing = candidates.get(email)955 if not existing:956 candidates[email] = candidate957 return958 if _email_candidate_weight(candidate) <= _email_candidate_weight(existing):959 return960 # Preserve best-scoring candidate when context evidence is stronger.961 candidates[email] = candidate962
963
964def _email_candidate_weight(candidate: Dict[str, Any]) -> int:965 source_type = candidate.get("source_type", "")966 weight = {"mailto": 120, "jsonld": 110, "visible_text": 90, "script": 70}.get(source_type, 50)967 region = (candidate.get("dom_region") or "").lower()968 if region in {"footer", "header", "nav", "main", "article", "aside"}:969 weight += 35970 if any(token in region for token in ("contact", "support", "sales", "help", "get-in-touch", "reach-us", "reach us", "get in touch")):971 weight += 45972 nearby_labels = candidate.get("nearby_labels") or []973 weight += min(len(nearby_labels), 4) * 12974 if candidate.get("in_contact_block"):975 weight += 18976 return weight977
978
979def _email_dedupe_weight(item: Dict[str, Any]) -> int:980 band = CONFIDENCE_ORDER.get(item.get("confidence_band", ""), 0)981 region = _normalize_region_hint(item.get("dom_region") or "")982 labels = item.get("nearby_labels") or []983 base = band * 200984 base += 40 if any(token in region for token in ("footer", "header", "nav", "main", "article", "contact", "support", "sales", "help")) else 0985 base += min(len(labels), 5) * 8986 return base987
988
989def _phone_dedupe_weight(item: Dict[str, Any]) -> int:990 band = CONFIDENCE_ORDER.get(item.get("confidence_band", ""), 0) * 100991 source_type = item.get("source_type")992 if source_type == "tel":993 band += 50994 elif source_type in {"callto", "sms"}:995 band += 40996 elif source_type == "jsonld":997 band += 30998 elif source_type == "visible_text":999 band += 101000 return band1001
1002
1003def _node_text_snippet(node: object, token: str, start: int, end: int) -> str:1004 if not hasattr(node, "parent") and not isinstance(node, str):1005 return token1006 raw = ""1007 if isinstance(node, str):1008 raw = node1009 else:1010 raw = str(node)1011 raw = _strip_text(raw)1012 if not raw:1013 return token1014 if start == 0 and end == 0:1015 return raw[:180]1016 window_start = max(0, start - 70)1017 window_end = min(len(raw), end + 70)1018 return raw[window_start:window_end].strip()1019
1020
1021def _normalize_phone(number: str) -> str:1022 cleaned = "".join(ch for ch in (number or "") if ch.isdigit() or ch == "+")1023 if cleaned.startswith("00"):1024 cleaned = "+" + cleaned[2:]1025 return cleaned1026
1027
1028def _infer_region_from_url(url: str) -> Optional[str]:1029 host = urlparse(url).hostname or ""1030 if not host:1031 return None1032 parts = host.lower().split(".")1033 if len(parts) < 2:1034 return None1035 tld = parts[-1]1036 return TLD_PHONE_REGION_MAP.get(tld)1037
1038
1039def normalize_phone_number(raw: str, region: Optional[str]) -> Dict[str, Any]:1040 cleaned = _normalize_phone(raw or "")1041 if not cleaned:1042 return {"raw": raw, "normalized": "", "valid": False, "type_hint": "unknown", "region": region}1043 if phonenumbers is None:1044 return {1045 "raw": raw,1046 "normalized": cleaned if 7 <= len(cleaned) <= 16 else "",1047 "valid": 7 <= len(cleaned) <= 16,1048 "type_hint": "unknown",1049 "region": region,1050 }1051
1052 try:1053 parsed = phonenumbers.parse(raw, region or None)1054 except Exception:1055 return {1056 "raw": raw,1057 "normalized": cleaned if 7 <= len(cleaned) <= 16 else "",1058 "valid": 7 <= len(cleaned) <= 16,1059 "type_hint": "unknown",1060 "region": region,1061 }1062
1063 if phonenumbers.is_possible_number(parsed) and phonenumbers.is_valid_number(parsed):1064 return {1065 "raw": raw,1066 "normalized": phonenumbers.format_number(parsed, phonenumbers.PhoneNumberFormat.E164),1067 "valid": True,1068 "type_hint": "unknown",1069 "region": region,1070 }1071
1072 return {1073 "raw": raw,1074 "normalized": cleaned if 7 <= len(cleaned) <= 16 else "",1075 "valid": False,1076 "type_hint": "unknown",1077 "region": region,1078 }1079
1080
1081def classify_phone_context(snippet: str) -> str:1082 if _matching_intent_hints(snippet, {"whatsapp"}) or "wa.me" in (snippet or "").lower():1083 return "whatsapp"1084 if _matching_intent_hints(snippet, {"sales", "quote"}):1085 return "sales"1086 if _matching_intent_hints(snippet, {"support", "help"}):1087 return "support"1088 if _matching_intent_hints(snippet, {"office"}):1089 return "office"1090 return "unknown"1091
1092
1093def _field_placeholder_looks_like_contact_hint(field: Dict[str, Any]) -> bool:1094 placeholder = _strip_text(field.get("placeholder") or "")1095 if not placeholder:1096 return False1097
1098 role = (field.get("role") or "").lower()1099 input_type = (field.get("type") or "").lower()1100 if role in {"email", "business_email", "phone"} or input_type in {"email", "tel"}:1101 return True1102
1103 if EMAIL_RE.search(placeholder) or PHONE_RE.search(placeholder):1104 return True1105
1106 compact = re.sub(r"[\s\-\.\(\)\+]+", "", placeholder.lower())1107 if compact and len(compact) >= 6 and set(compact) <= {"x", "0"}:1108 return True1109 return False1110
1111
1112def _looks_like_templated_email(email: str) -> bool:1113 local, domain = email.split("@", 1)1114 lowered = email.lower()1115 if any(marker in lowered for marker in (f".{ext}" for ext in EMAIL_ASSET_EXTENSIONS)):1116 return True1117 if any(marker in local.lower() for marker in EMAIL_ASSET_STRINGS):1118 return True1119 if any(marker in domain.lower() for marker in EMAIL_ASSET_STRINGS):1120 return True1121 if local.count(".") > 1:1122 return True1123 if ".." in local:1124 return True1125 return False1126
1127
1128def _in_media_url_context(source: Optional[str], text: str) -> bool:1129 if not source:1130 return False1131 lowered_source = source.lower()1132 if any(ext in lowered_source for ext in [".png", ".jpg", ".jpeg", ".gif", ".webp", ".svg", ".pdf", ".js", ".css"]):1133 return True1134 # If the email appears near an image/script style asset URL.1135 return any(token in text.lower() for token in (".png", ".jpg", ".jpeg", ".gif", ".webp", ".svg", ".pdf", ".js", ".css"))1136
1137
1138def _has_contact_context(text: str) -> bool:1139 return bool(_matching_intent_hints(text, PHONE_CONTEXT_HINTS | set(CONTACT_KEYWORDS)))1140
1141
1142def _has_strong_email_contact_wording(text: str, labels: Optional[List[str]] = None) -> bool:1143 if _matching_intent_hints(text, EMAIL_STRONG_CONTACT_LABELS):1144 return True1145 if labels and any(_contains_token(" ".join(labels), label) for label in EMAIL_STRONG_CONTACT_LABELS):1146 return True1147 return False1148
1149
1150def _looks_like_phone_token_noise(raw: str, snippet: str, nearby_text: str) -> bool:1151 combined = " ".join(part for part in [raw, snippet, nearby_text] if part).lower()1152 compact = combined.replace(" ", "")1153 if re.search(r"\b[0-9a-f]{16,}\b", compact):1154 return True1155 if re.search(r"\b[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}\b", combined):1156 return True1157 if any(token in combined for token in ("uuid", "token", "hash", "checksum", "trace", "session")):1158 return True1159 return False1160
1161
1162def _looks_like_product_identifier_number(raw: str, snippet: str, nearby_text: str) -> bool:1163 combined = " ".join(part for part in [snippet, nearby_text] if part)1164 if not _matching_intent_hints(combined, PRODUCT_IDENTIFIER_HINTS):1165 return False1166
1167 normalized = _normalize_phone(raw or "")1168 raw_stripped = (raw or "").strip()1169 if re.fullmatch(r"\d{4,8}-\d{2,6}", raw_stripped):1170 return True1171 if normalized.isdigit() and 6 <= len(normalized) <= 14:1172 return True1173 return False1174
1175
1176def _has_strong_phone_context(text: str) -> bool:1177 return bool(_matching_intent_hints(text, PHONE_STRONG_CONTEXT_HINTS))1178
1179
1180def _has_weak_page_signal(soup: BeautifulSoup) -> bool:1181 # Weak match helper used for rejected page accounting.1182 title = _strip_text(soup.title.string) if soup.title and soup.title.string else ""1183 headings = " ".join([_strip_text(h.get_text(" ", strip=True)) for h in soup.select("h1,h2")])1184 text = _strip_text(soup.get_text(" ", strip=True))1185 return _has_contact_phrase(title) or _has_contact_phrase(headings) or _has_contact_phrase(text[:600])1186
1187
1188def _split_txt_record(answer: Any) -> str:1189 if hasattr(answer, "strings"):1190 return "".join(1191 (part.decode("utf-8") if isinstance(part, (bytes, bytearray)) else str(part))1192 for part in getattr(answer, "strings", [])1193 )1194 return str(answer).strip().strip('"')1195
1196
1197def infer_mail_provider(mx_records: List[str]) -> str:1198 lowered = " ".join((mx or "").lower() for mx in mx_records)1199 for provider, markers in MAIL_PROVIDER_HINTS:1200 if any(marker.lower() in lowered for marker in markers):1201 return provider1202 return ""1203
1204
1205def enrich_domain_dns(domain: str, *, dns_timeout: float = 2.5) -> Dict[str, Any]:1206 normalized_domain = (domain or "").strip().lower()1207 payload: Dict[str, Any] = {1208 "domain": normalized_domain,1209 "mx_records": [],1210 "has_mx": False,1211 "spf": None,1212 "dmarc": None,1213 "mail_provider_hint": "",1214 "root_a_records": [],1215 "root_aaaa_records": [],1216 "notes": [],1217 "error": None,1218 }1219 domain = normalized_domain1220 if not payload["domain"]:1221 payload["error"] = "missing domain"1222 return payload1223 if dns is None:1224 payload["notes"].append("dnspython unavailable; DNS enrichment skipped")1225 return payload1226
1227 try:1228 resolver = dns.Resolver(configure=True)1229 resolver.lifetime = float(dns_timeout)1230 # MX records1231 try:1232 mx_answers = resolver.resolve(domain, "MX")1233 mx_records: List[str] = []1234 for ans in mx_answers:1235 try:1236 host = str(ans.exchange).rstrip(".").lower()1237 mx_records.append(host)1238 except Exception:1239 continue1240 payload["mx_records"] = sorted(set(mx_records))1241 payload["has_mx"] = bool(mx_records)1242 payload["mail_provider_hint"] = infer_mail_provider(payload["mx_records"])1243 except Exception as exc:1244 payload["notes"].append(f"MX lookup failed: {exc}")1245
1246 # SPF and DMARC TXT1247 try:1248 txt_answers = resolver.resolve(domain, "TXT")1249 for ans in txt_answers:1250 value = _split_txt_record(ans).strip().strip('"')1251 if value.startswith("v=spf1"):1252 payload["spf"] = value1253 break1254 except Exception as exc:1255 payload["notes"].append(f"SPF lookup failed: {exc}")1256
1257 try:1258 dmarc_answers = resolver.resolve(f"_dmarc.{domain}", "TXT")1259 for ans in dmarc_answers:1260 value = _split_txt_record(ans).strip().strip('"')1261 if value.upper().startswith("V=DMARC1"):1262 payload["dmarc"] = value1263 break1264 except Exception as exc:1265 payload["notes"].append(f"DMARC lookup failed: {exc}")1266
1267 # Root A/AAAA1268 try:1269 a_answers = resolver.resolve(domain, "A")1270 payload["root_a_records"] = sorted({str(ans).rstrip(".") for ans in a_answers})1271 except Exception as exc:1272 payload["notes"].append(f"A lookup failed: {exc}")1273
1274 try:1275 aaaa_answers = resolver.resolve(domain, "AAAA")1276 payload["root_aaaa_records"] = sorted({str(ans).rstrip(".") for ans in aaaa_answers})1277 except Exception as exc:1278 payload["notes"].append(f"AAAA lookup failed: {exc}")1279 except Exception as exc:1280 payload["error"] = str(exc)1281
1282 if payload["error"]:1283 return payload1284 if payload["has_mx"] and not payload["mail_provider_hint"]:1285 payload["mail_provider_hint"] = "Unknown"1286 if not payload["notes"]:1287 payload["notes"] = []1288 return payload1289
1290
1291def _meta_evidence(value: str, page_url: str, source_type: str, evidence: str) -> Dict[str, Any]:1292 return {1293 "value": value,1294 "page_url": page_url,1295 "source_type": source_type,1296 "evidence": evidence,1297 }1298
1299
1300def _extract_jsonld_nodes(node: Any, out: List[Dict[str, Any]]) -> None:1301 if isinstance(node, dict):1302 if "@type" in node:1303 out.append(node)1304 for value in node.values():1305 _extract_jsonld_nodes(value, out)1306 elif isinstance(node, list):1307 for item in node:1308 _extract_jsonld_nodes(item, out)1309
1310
1311def _join_address(node: Dict[str, Any]) -> str:1312 if not isinstance(node, dict):1313 return ""1314 parts = []1315 for key in ("streetAddress", "addressLocality", "addressRegion", "postalCode", "addressCountry"):1316 value = (node.get(key) or "").strip() if isinstance(node.get(key), str) else ""1317 if value:1318 parts.append(value)1319 return ", ".join(parts)1320
1321
1322def _collect_site_metadata_from_jsonld(page_url: str, payload: Any, collector: Dict[str, Any], seen: Set[str]) -> None:1323 metadata = payload1324 nodes: List[Dict[str, Any]] = []1325 _extract_jsonld_nodes(metadata, nodes)1326 for node in nodes:1327 raw_type = node.get("@type")1328 types = set()1329 if isinstance(raw_type, str):1330 types.add(raw_type.lower())1331 elif isinstance(raw_type, list):1332 types.update([str(v).lower() for v in raw_type if isinstance(v, str)])1333
1334 def add_if_new(field: str, value: str, source: str) -> None:1335 if not value:1336 return1337 value_l = value.strip().lower()1338 if not value_l or value_l in seen:1339 return1340 if field == "business_name":1341 if not collector.get("business_name"):1342 collector["business_name"] = _meta_evidence(value, page_url, "jsonld", source)1343 seen.add(value_l)1344 return1345 seen.add(value_l)1346 collector.setdefault(field, []).append(_meta_evidence(value, page_url, "jsonld", source))1347
1348 if any(t in types for t in {"organization", "localbusiness", "professionalservice", "company"}):1349 add_if_new("business_name", (node.get("name") or "").strip() if isinstance(node.get("name"), str) else "", "organization/name")1350 add_if_new("legal_names", (node.get("legalName") or "").strip() if isinstance(node.get("legalName"), str) else "", "organization/legalName")1351 add_if_new("contact_points", (node.get("email") or "").strip() if isinstance(node.get("email"), str) else "", "organization/email")1352 add_if_new("public_emails", (node.get("email") or "").strip() if isinstance(node.get("email"), str) else "", "organization/email")1353 add_if_new("business_phone", (node.get("telephone") or "").strip() if isinstance(node.get("telephone"), str) else "", "organization/telephone")1354 description = (node.get("description") or "").strip() if isinstance(node.get("description"), str) else ""1355 if description and not collector.get("description"):1356 collector["description"] = _meta_evidence(description, page_url, "jsonld", "organization/description")1357
1358 same_as = node.get("sameAs")1359 if isinstance(same_as, str):1360 same_as = [same_as]1361 if isinstance(same_as, list):1362 for profile in same_as:1363 if isinstance(profile, str):1364 add_if_new("same_as", profile.strip(), "organization/sameAs")1365
1366 address = node.get("address")1367 if isinstance(address, dict):1368 add_if_new("addresses", _join_address(address), "organization/address")1369
1370 contact_points = node.get("contactPoint")1371 if isinstance(contact_points, dict):1372 contact_points = [contact_points]1373 if isinstance(contact_points, list):1374 for contact in contact_points:1375 if not isinstance(contact, dict):1376 continue1377 c_email = (contact.get("email") or "").strip() if isinstance(contact.get("email"), str) else ""1378 c_phone = (contact.get("telephone") or "").strip() if isinstance(contact.get("telephone"), str) else ""1379 c_type = (contact.get("contactType") or "").strip() if isinstance(contact.get("contactType"), str) else ""1380 if c_email:1381 add_if_new("contact_points", c_email, "contactPoint/email")1382 add_if_new("public_emails", c_email, "contactPoint/email")1383 if c_phone:1384 add_if_new("business_phone", c_phone, "contactPoint/telephone")1385 if c_type.lower() in {"support", "sales", "sales inquiry", "customer support"}:1386 add_if_new("support_sales_labels", c_type, "contactPoint/contactType")1387
1388 if any(t in types for t in {"contactpoint", "contact-point"}):1389 contact_node_type = str(raw_type or "").lower()1390 add_if_new("contact_points", (node.get("email") or "").strip() if isinstance(node.get("email"), str) else "", f"{contact_node_type}/email")1391 add_if_new("public_emails", (node.get("email") or "").strip() if isinstance(node.get("email"), str) else "", f"{contact_node_type}/public_email")1392 add_if_new("business_phone", (node.get("telephone") or "").strip() if isinstance(node.get("telephone"), str) else "", f"{contact_node_type}/telephone")1393
1394
1395def _classify_email_domain_relation(site_domain: str, email_domain: str) -> str:1396 if not site_domain or not email_domain:1397 return "unknown_relation"1398 s = site_domain.lower().lstrip(".")1399 e = email_domain.lower().lstrip(".")1400 if e == s:1401 return "same_domain"1402 if e == f"www.{s}" or e.endswith(f".{s}"):1403 return "subdomain"1404 return "third_party"1405
1406
1407def enrich_validated_emails(1408 emails: List[Dict[str, Any]],1409 *,1410 site_domain: str,1411 do_dns: bool = False,1412 dns_timeout: float = 2.5,1413) -> List[Dict[str, Any]]:1414 output: List[Dict[str, Any]] = []1415 dns_cache: Dict[str, Dict[str, Any]] = {}1416 for email in emails:1417 value = (email.get("value") or "").strip().lower()1418 if "@" not in value:1419 continue1420 email_domain = value.split("@", 1)[1]1421 enriched = dict(email)1422 enriched["email_domain"] = email_domain1423 enriched["domain_relation"] = _classify_email_domain_relation(site_domain, email_domain)1424
1425 if do_dns:1426 if email_domain not in dns_cache:1427 dns_cache[email_domain] = enrich_domain_dns(email_domain, dns_timeout=dns_timeout)1428 dns_meta = dns_cache[email_domain]1429 enriched["domain_has_mx"] = bool(dns_meta.get("has_mx"))1430 enriched["mail_provider_hint"] = dns_meta.get("mail_provider_hint") or ""1431 enriched["domain_dns"] = dns_meta1432 else:1433 enriched["domain_has_mx"] = None1434 enriched["mail_provider_hint"] = None1435 enriched["domain_dns"] = None1436
1437 output.append(enriched)1438 return output1439
1440
1441def extract_site_metadata(html: str, soup: BeautifulSoup, url: str) -> Dict[str, Any]:1442 metadata: Dict[str, Any] = {1443 "business_name": None,1444 "legal_names": [],1445 "description": None,1446 "addresses": [],1447 "contact_points": [],1448 "same_as": [],1449 "support_sales_labels": [],1450 "public_emails": [],1451 "business_phone": [],1452 "copyright": None,1453 }1454
1455 seen = set()1456
1457 title = _strip_text((soup.title.string if soup.title and soup.title.string else ""))1458 if title and " - " in title:1459 candidate_name = title.split(" - ")[0].strip()1460 if len(candidate_name) > 2:1461 metadata["business_name"] = _meta_evidence(candidate_name, url, "title", "title:first segment before separator")1462 seen.add(candidate_name.lower())1463
1464 meta_desc = soup.find("meta", attrs={"name": "description"})1465 if meta_desc and meta_desc.get("content"):1466 content = _strip_text(meta_desc["content"])1467 if content:1468 metadata["description"] = _meta_evidence(content, url, "meta_description", "meta[name=description]")1469
1470 og_title = soup.find("meta", attrs={"property": "og:title"})1471 og_desc = soup.find("meta", attrs={"property": "og:description"})1472 if og_desc and og_desc.get("content"):1473 content = _strip_text(og_desc["content"])1474 if content and not metadata["description"]:1475 metadata["description"] = _meta_evidence(content, url, "opengraph", "og:description")1476 if og_title and og_title.get("content"):1477 content = _strip_text(og_title["content"])1478 if content and not metadata["business_name"]:1479 metadata["business_name"] = _meta_evidence(content, url, "opengraph", "og:title")1480 seen.add(content.lower())1481
1482 for script in soup.find_all("script", {"type": lambda t: bool(t and "ld+json" in t.lower())}):1483 raw = (script.string or script.get_text(" ", strip=True) or "").strip()1484 if not raw:1485 continue1486 try:1487 payload = json.loads(raw)1488 except Exception:1489 continue1490 _collect_site_metadata_from_jsonld(url, payload, metadata, seen)1491
1492 for section in soup.select("header, footer"):1493 section_text = _strip_text(section.get_text(" ", strip=True))1494 if not section_text:1495 continue1496 low = section_text.lower()1497 if "support" in low:1498 metadata.setdefault("support_sales_labels", []).append(_meta_evidence("support", url, "header_footer_text", section_text[:180]))1499 if "sales" in low:1500 metadata.setdefault("support_sales_labels", []).append(_meta_evidence("sales", url, "header_footer_text", section_text[:180]))1501 if "copyright" in low:1502 snippet = section_text[:220]1503 metadata["copyright"] = _meta_evidence(snippet, url, "header_footer_text", "copyright")1504 if "address" in low:1505 metadata.setdefault("addresses", []).append(_meta_evidence(section_text, url, "header_footer_text", "address label in header/footer"))1506
1507 for tag in soup.find_all("address"):1508 addr = _strip_text(tag.get_text(" ", strip=True))1509 if addr and len(addr) >= 8:1510 if addr.lower() not in seen:1511 metadata.setdefault("addresses", []).append(_meta_evidence(addr, url, "address_tag", "address element"))1512 seen.add(addr.lower())1513
1514 # Keep only deduplicated unique items.1515 for key in ("legal_names", "same_as", "addresses", "contact_points", "support_sales_labels", "public_emails", "business_phone"):1516 unique: List[Dict[str, Any]] = []1517 bucket_seen = set()1518 for item in metadata.get(key, []):1519 if not item:1520 continue1521 v = (item.get("value") or "").strip().lower()1522 if not v or v in bucket_seen:1523 continue1524 bucket_seen.add(v)1525 unique.append(item)1526 metadata[key] = unique1527 return metadata1528
1529
1530def discover_candidate_pages(start_url: str, page_url: str, soup: BeautifulSoup) -> List[Dict[str, Any]]:1531 candidates: Dict[str, Dict[str, Any]] = {}1532 for a in soup.find_all("a", href=True):1533 href = (a.get("href") or "").strip()1534 if not href or href.startswith("#") or href.startswith("javascript:") or href.startswith("mailto:") or href.startswith("tel:"):1535 continue1536
1537 abs_url = normalize_url(href, base=page_url)1538 if not abs_url or not same_site(start_url, abs_url):1539 continue1540
1541 anchor_text = _strip_text(" ".join(a.stripped_strings))1542 parent_chain = [p.name.lower() for p in a.parents if p is not None and hasattr(p, "name") and p.name]1543 context_tokens = " ".join((parent_chain[:5]) + [a.get("class") and " ".join(a.get("class", [])) or "", a.get("id") or "", a.get("aria-label") or "", a.get("title") or ""])1544
1545 score = 01546 reasons: List[str] = []1547 path = _page_host_block(urlparse(abs_url).path)1548
1549 if any(token in parent_chain for token in {"header", "nav", "footer", "aside"}):1550 score += 181551 reasons.append("in site navigation/footer")1552
1553 if _is_contact_url_hint(path):1554 score += 351555 reasons.append("contact-intent path pattern")1556 if _has_contact_phrase(anchor_text + " " + context_tokens):1557 score += 551558 reasons.append("contact phrase in anchor/context")1559 if any(token in (anchor_text or "").lower() for token in ("contact", "support", "sales", "quote", "inquire", "reach")):1560 score += 201561 reasons.append("contact-like anchor text")1562
1563 if abs_url == start_url or path in {"", "/"}:1564 score = max(score, 12)1565
1566 # Do not keep weak matches driven by locations/privacy/etc.1567 if any(w in path for w in WEAK_PATH_HINTS) and "contact" not in path and not _has_contact_phrase(anchor_text):1568 score = 01569
1570 if score < 30:1571 continue1572
1573 existing = candidates.get(abs_url)1574 if not existing or score > existing["score"]:1575 candidates[abs_url] = {1576 "url": abs_url,1577 "score": score,1578 "reason": "; ".join(sorted(set(reasons))),1579 "evidence": _strip_text((anchor_text + " " + context_tokens)[:180]),1580 "source": page_url,1581 }1582
1583 return list(candidates.values())1584
1585
1586def extract_email_candidates(page_url: str, raw_html: str, soup: BeautifulSoup) -> List[Dict[str, Any]]:1587 results: Dict[str, Dict[str, Any]] = {}1588 _ = raw_html1589
1590 # Cloudflare email protection is an explicit site-provided email, just obfuscated in the HTML.1591 for tag in soup.find_all(attrs={"data-cfemail": True}):1592 email = _decode_cfemail(tag.get("data-cfemail") or "")1593 if not email:1594 continue1595 visible_text = _strip_text(tag.get_text(" ", strip=True)) or email1596 snippet = _node_text_snippet(visible_text, visible_text, 0, 0)1597 region, nearby_heading, nearby_labels, nearby_text = _collect_email_context(tag, snippet)1598 _register_email_candidate(1599 results,1600 email,1601 {1602 "value": email,1603 "source_type": "visible_text",1604 "snippet": snippet,1605 "page_url": page_url,1606 "source_attribute": "data-cfemail",1607 "dom_region": region,1608 "nearby_labels": nearby_labels,1609 "nearby_heading": nearby_heading,1610 "nearby_text": nearby_text,1611 "has_contact_context": bool(nearby_labels) or _has_contact_context(nearby_text),1612 "in_contact_block": any(token in (region or "") for token in ("contact", "support", "sales", "footer", "header", "nav")),1613 "source_scope": "cfemail",1614 },1615 )1616
1617 # mailto:1618 for a in soup.find_all("a", href=True):1619 href = (a.get("href") or "").strip()1620 if not href.lower().startswith("mailto:"):1621 continue1622 candidate = href[7:].split("?", 1)[0].strip()1623 if "@" not in candidate:1624 continue1625 candidate = candidate.lower()1626 region, nearby_heading, nearby_labels, nearby_text = _collect_email_context(a, _strip_text(" ".join(a.stripped_strings)) or f"href={href}")1627 _register_email_candidate(1628 results,1629 candidate,1630 {1631 "value": candidate,1632 "source_type": "mailto",1633 "snippet": _strip_text(f"href={href}"),1634 "page_url": page_url,1635 "source_attribute": href,1636 "dom_region": region or "mailto",1637 "nearby_labels": nearby_labels,1638 "nearby_heading": nearby_heading,1639 "nearby_text": nearby_text,1640 "has_contact_context": True,1641 "in_contact_block": any(token in (region or "") for token in ("contact", "support", "sales", "footer", "header", "nav")),1642 "source_scope": "mailto",1643 },1644 )1645
1646 # JSON-LD / structured data1647 for script in soup.find_all("script", {"type": lambda t: bool(t and "ld+json" in t.lower())}):1648 text = (script.string or script.get_text(" ", strip=True) or "").strip()1649 if not text:1650 continue1651 try:1652 payload = json.loads(text)1653 except Exception:1654 continue1655
1656 def walk(node: Any) -> None:1657 if isinstance(node, dict):1658 for key, value in node.items():1659 if isinstance(value, (dict, list)):1660 walk(value)1661 elif isinstance(value, str):1662 if key.lower() in {"email", "emailaddress", "contactemail"} and EMAIL_RE.fullmatch(value.lower()):1663 low = value.lower()1664 _register_email_candidate(1665 results,1666 low,1667 {1668 "value": low,1669 "source_type": "jsonld",1670 "snippet": f"{key}: {value}",1671 "page_url": page_url,1672 "source_attribute": "application/ld+json",1673 "dom_region": "jsonld",1674 "nearby_labels": [],1675 "nearby_heading": "",1676 "nearby_text": "",1677 "has_contact_context": True,1678 "in_contact_block": False,1679 "source_scope": "jsonld",1680 },1681 )1682 elif isinstance(node, list):1683 for item in node:1684 walk(item)1685
1686 walk(payload)1687
1688 # Deliberate script key/value matches.1689 for script in soup.find_all("script"):1690 text = (script.string or script.get_text(" ", strip=True) or "")1691 if not text:1692 continue1693 for match in SCRIPT_EMAIL_RE.finditer(text):1694 email = (match.group(1) or "").strip().lower()1695 if not EMAIL_RE.fullmatch(email):1696 continue1697 key = match.group(0)[:250]1698 _register_email_candidate(1699 results,1700 email,1701 {1702 "value": email,1703 "source_type": "script",1704 "snippet": _strip_text(key),1705 "page_url": page_url,1706 "source_attribute": "inline-script",1707 "dom_region": "script",1708 "nearby_labels": sorted({label for label in CONTACT_KEYWORDS if label in _strip_text(text[:240]).lower()}),1709 "nearby_heading": "",1710 "nearby_text": _strip_text(text[:240]),1711 "has_contact_context": _has_contact_phrase(text[:240]),1712 "context_has_contact_keywords": _has_contact_phrase(text[:240]),1713 "in_contact_block": False,1714 "source_scope": "script",1715 },1716 )1717
1718 def _record_visible(1719 parent_node: Any,1720 node_text: str,1721 *,1722 source_type: str = "visible_text",1723 source_attribute: str = "text",1724 is_contact_scope: bool = False,1725 ) -> None:1726 for match in EMAIL_RE.finditer(node_text):1727 email = match.group(1).lower()1728 idx_start, idx_end = match.span(1)1729 snippet = _node_text_snippet(node_text, match.group(1), idx_start, idx_end)1730 region, nearby_heading, nearby_labels, nearby_text = _collect_email_context(parent_node, snippet)1731 candidate = {1732 "value": email,1733 "source_type": source_type,1734 "snippet": snippet,1735 "page_url": page_url,1736 "source_attribute": source_attribute,1737 "dom_region": region,1738 "nearby_heading": nearby_heading,1739 "nearby_labels": nearby_labels,1740 "nearby_text": nearby_text,1741 "has_contact_context": bool(nearby_labels) or _has_contact_context(nearby_text),1742 "in_contact_block": is_contact_scope or any(1743 token in (region or "") for token in ("contact", "support", "sales", "footer", "header", "nav")1744 ),1745 "source_scope": "text",1746 }1747 _register_email_candidate(results, email, candidate)1748
1749 # Visible text pass over the full DOM.1750 for parent, text in _iter_visible_text_nodes(soup):1751 _record_visible(parent, text, source_type="visible_text", source_attribute="text", is_contact_scope=False)1752
1753 # Anchor text is kept as a separate pass because some sites render this differently from visible text nodes.1754 for a in soup.find_all("a"):1755 anchor_text = _strip_text(" ".join(a.strings or []))1756 if anchor_text:1757 _record_visible(a, anchor_text, source_type="visible_text", source_attribute="anchor_text", is_contact_scope=False)1758
1759 # Human-readable attributes are scanned page-wide; region hints only affect context, not discovery.1760 for tag, attr_name, attr_text in _iter_human_readable_attributes(soup):1761 _record_visible(tag, attr_text, source_type="attribute", source_attribute=attr_name, is_contact_scope=False)1762
1763 return list(results.values())1764
1765
1766def validate_email_candidate(candidate: Dict[str, Any], page_ctx: Dict[str, Any]) -> Optional[Dict[str, Any]]:1767 email = (candidate.get("value") or "").lower()1768 if not EMAIL_RE.fullmatch(email):1769 return None1770 if _looks_like_templated_email(email):1771 candidate["reject_reason"] = "asset_email"1772 return None1773 snippet = candidate.get("snippet", "")1774 source_type = candidate.get("source_type")1775
1776 source_attr = candidate.get("source_attribute")1777 if _in_media_url_context(source_attr, snippet):1778 candidate["reject_reason"] = "asset_email"1779 return None1780
1781 validator_passed = []1782 evidence_parts = [f"source:{source_type}", f"snippet:{snippet}"]1783 nearby_labels = [str(label or "").lower() for label in (candidate.get("nearby_labels") or [])]1784 is_contact_page = bool(page_ctx.get("is_contact_page"))1785 nearby_text = candidate.get("nearby_text") or snippet1786 has_strong_contact_wording = _has_strong_email_contact_wording(nearby_text, nearby_labels)1787
1788 if source_type == "mailto":1789 band = "VERIFIED"1790 validator_passed.append("mailto")1791 elif source_type == "jsonld":1792 band = "VERIFIED"1793 validator_passed.append("jsonld_contact_point")1794 elif source_type == "script":1795 # Script-only detection is only valid when deliberate.1796 if "email" in (snippet or "").lower() and _has_contact_context(snippet):1797 band = "STRONG"1798 validator_passed.append("deliberate_script_kv_pair")1799 else:1800 candidate["reject_reason"] = "asset_email"1801 return None1802 else:1803 dom_region = _normalize_region_hint(candidate.get("dom_region") or "")1804 has_block_context = candidate.get("in_contact_block", False) or any(1805 token in dom_region for token in ("footer", "header", "nav", "main", "article", "contact", "support", "sales", "help")1806 )1807 has_text_context = bool(candidate.get("has_contact_context") or nearby_labels)1808 if has_text_context and has_strong_contact_wording:1809 band = "VERIFIED"1810 validator_passed.append("visible_text_contact_context")1811 validator_passed.append("strong_contact_wording")1812 elif has_block_context or has_text_context:1813 band = "VERIFIED" if has_block_context else "STRONG"1814 validator_passed.append("visible_text_contact_context")1815 elif is_contact_page:1816 band = "STRONG"1817 validator_passed.append("contact_page_context")1818 else:1819 # Visible email without local context is usually noise.1820 candidate["reject_reason"] = "weak_page_match"1821 return None1822 if has_block_context and candidate.get("has_contact_context"):1823 validator_passed.append("contact_region_context")1824
1825 if not validator_passed:1826 candidate["reject_reason"] = "weak_page_match"1827 return None1828
1829 return {1830 "type": "email",1831 "value": email,1832 "source_type": source_type,1833 "page_url": candidate["page_url"],1834 "dom_region": candidate.get("dom_region"),1835 "nearby_labels": nearby_labels,1836 "nearby_text": nearby_text,1837 "evidence": "; ".join(evidence_parts + [f"region:{candidate.get('dom_region', '')}", f"labels:{','.join(nearby_labels)}"]),1838 "confidence_band": band,1839 "validator_passed": validator_passed,1840 }1841
1842
1843def rescue_visible_email_candidate(candidate: Dict[str, Any]) -> Optional[Dict[str, Any]]:1844 if candidate.get("source_type") != "visible_text":1845 return None1846 email = (candidate.get("value") or "").lower()1847 if not EMAIL_RE.fullmatch(email):1848 return None1849 if _looks_like_templated_email(email):1850 return None1851 snippet = candidate.get("snippet", "")1852 if _in_media_url_context(candidate.get("source_attribute"), snippet):1853 return None1854 nearby_labels = [str(label or "").lower() for label in (candidate.get("nearby_labels") or [])]1855 nearby_text = candidate.get("nearby_text") or snippet1856 if not candidate.get("has_contact_context"):1857 return None1858 if not _has_strong_email_contact_wording(nearby_text, nearby_labels):1859 return None1860 return {1861 "type": "email",1862 "value": email,1863 "source_type": "visible_text",1864 "page_url": candidate["page_url"],1865 "dom_region": candidate.get("dom_region"),1866 "nearby_labels": nearby_labels,1867 "nearby_text": nearby_text,1868 "evidence": "; ".join(1869 [1870 "source:visible_text",1871 f"snippet:{snippet}",1872 f"region:{candidate.get('dom_region', '')}",1873 f"labels:{','.join(nearby_labels)}",1874 "rescue:strong_contact_wording",1875 ]1876 ),1877 "confidence_band": "VERIFIED",1878 "validator_passed": ["visible_text_rescue", "strong_contact_wording"],1879 }1880
1881
1882def extract_phone_candidates(page_url: str, raw_html: str, soup: BeautifulSoup) -> List[Dict[str, Any]]:1883 results: Dict[str, Dict[str, Any]] = {}1884 region = _infer_region_from_url(page_url)1885 _ = raw_html1886
1887 def register_phone(candidate: Dict[str, Any]) -> None:1888 key = candidate.get("normalized") or candidate.get("value") or candidate.get("raw") or ""1889 if not key:1890 return1891 existing = results.get(key)1892 if not existing:1893 results[key] = candidate1894 return1895 existing_score = (2 if existing.get("has_context") else 0) + (2 if existing.get("source_type") in {"tel", "callto", "sms", "jsonld"} else 0)1896 candidate_score = (2 if candidate.get("has_context") else 0) + (2 if candidate.get("source_type") in {"tel", "callto", "sms", "jsonld"} else 0)1897 if candidate_score >= existing_score:1898 results[key] = candidate1899
1900 # Link-based phone channels are first-class signals anywhere on the page.1901 for a in soup.find_all("a", href=True):1902 href = (a.get("href") or "").strip()1903 lowered_href = href.lower()1904 if lowered_href.startswith("tel:"):1905 source_type = "tel"1906 raw = href[4:].strip()1907 elif lowered_href.startswith("callto:"):1908 source_type = "callto"1909 raw = href[7:].strip()1910 elif lowered_href.startswith("sms:"):1911 source_type = "sms"1912 raw = href[4:].split("?", 1)[0].strip()1913 else:1914 continue1915 normalized_payload = normalize_phone_number(raw, region=region)1916 number = normalized_payload.get("normalized") or ""1917 if not number:1918 continue1919 dom_region, nearby_heading, nearby_labels, nearby_text = _collect_phone_context(a, _strip_text(" ".join(a.stripped_strings)) or f"href={href}")1920 register_phone({1921 "value": number,1922 "raw": raw,1923 "normalized": number,1924 "source_type": source_type,1925 "snippet": _strip_text(f"href={href}"),1926 "page_url": page_url,1927 "source_attribute": href,1928 "has_context": True,1929 "region": region,1930 "type_hint": classify_phone_context(_strip_text(" ".join([href, nearby_text]))),1931 "dom_region": dom_region,1932 "nearby_heading": nearby_heading,1933 "nearby_labels": nearby_labels,1934 "nearby_text": nearby_text,1935 })1936
1937 def record_phone(parent_node: Any, text: str, source_type: str, source_attribute: str) -> None:1938 for match in PHONE_RE.finditer(text):1939 num = _normalize_phone(match.group(0))1940 if len(num) < 7:1941 continue1942 snippet = _node_text_snippet(text, match.group(0), start := match.start(), end := match.end())1943 raw = match.group(0)1944 normalized_payload = normalize_phone_number(raw, region=region)1945 normalized_number = normalized_payload.get("normalized") or num1946 if not normalized_number:1947 continue1948 dom_region, nearby_heading, nearby_labels, nearby_text = _collect_phone_context(parent_node, snippet)1949 combined_context = " ".join([snippet, nearby_text, nearby_heading])1950 register_phone({1951 "value": normalized_number,1952 "raw": raw,1953 "normalized": normalized_number,1954 "source_type": source_type,1955 "snippet": snippet,1956 "page_url": page_url,1957 "source_attribute": source_attribute,1958 "has_context": _has_strong_phone_context(combined_context),1959 "region": region,1960 "type_hint": classify_phone_context(combined_context),1961 "dom_region": dom_region,1962 "nearby_heading": nearby_heading,1963 "nearby_labels": nearby_labels,1964 "nearby_text": nearby_text,1965 })1966
1967 # Visible text pass over the full DOM.1968 for parent, text in _iter_visible_text_nodes(soup):1969 record_phone(parent, text, "visible_text", "text")1970
1971 for a in soup.find_all("a"):1972 anchor_text = _strip_text(" ".join(a.strings or []))1973 if anchor_text:1974 record_phone(a, anchor_text, "visible_text", "anchor_text")1975
1976 for attr_name in ("aria-label", "title"):1977 for tag in soup.find_all(True, attrs={attr_name: True}):1978 attr_text = _strip_text(tag.get(attr_name) or "")1979 if not attr_text:1980 continue1981 dom_region, nearby_heading, nearby_labels, nearby_text = _collect_phone_context(tag, attr_text)1982 if not _has_strong_phone_context(" ".join([attr_text, nearby_text, nearby_heading])):1983 continue1984 record_phone(tag, attr_text, "attribute", attr_name)1985
1986 # JSON-LD telephones.1987 for script in soup.find_all("script", {"type": lambda t: bool(t and "ld+json" in t.lower())}):1988 text = (script.string or script.get_text(" ", strip=True) or "").strip()1989 if not text:1990 continue1991 try:1992 payload = json.loads(text)1993 except Exception:1994 continue1995
1996 def walk(node: Any) -> None:1997 if isinstance(node, dict):1998 for key, value in node.items():1999 if isinstance(value, (dict, list)):2000 walk(value)2001 elif isinstance(value, str):2002 key_l = str(key).lower()2003 if key_l in {"telephone", "phone", "phonenumber"}:2004 num = _normalize_phone(value)2005 normalized_payload = normalize_phone_number(value, region=region)2006 num = normalized_payload.get("normalized") or num2007 if num:2008 register_phone(2009 {2010 "value": num,2011 "raw": value,2012 "normalized": num,2013 "source_type": "jsonld",2014 "snippet": f"{key}: {value}",2015 "page_url": page_url,2016 "source_attribute": "application/ld+json",2017 "has_context": True,2018 "region": region,2019 "type_hint": classify_phone_context(f"{key}: {value}"),2020 "dom_region": "jsonld",2021 "nearby_heading": "",2022 "nearby_labels": ["phone"],2023 "nearby_text": f"{key}: {value}",2024 },2025 )2026 elif isinstance(node, list):2027 for item in node:2028 walk(item)2029
2030 walk(payload)2031
2032 # Deliberate inline script key/value pairs only.2033 for script in soup.find_all("script"):2034 text = (script.string or script.get_text(" ", strip=True) or "")2035 if not text:2036 continue2037 for match in SCRIPT_PHONE_RE.finditer(text):2038 raw = (match.group(1) or "").strip()2039 normalized_payload = normalize_phone_number(raw, region=region)2040 number = normalized_payload.get("normalized") or ""2041 if not number:2042 continue2043 register_phone(2044 {2045 "value": number,2046 "raw": raw,2047 "normalized": number,2048 "source_type": "script",2049 "snippet": _strip_text(match.group(0)[:250]),2050 "page_url": page_url,2051 "source_attribute": "inline-script",2052 "has_context": True,2053 "region": region,2054 "type_hint": classify_phone_context(match.group(0)),2055 "dom_region": "script",2056 "nearby_heading": "",2057 "nearby_labels": ["phone"],2058 "nearby_text": _strip_text(match.group(0)[:250]),2059 }2060 )2061
2062 return list(results.values())2063
2064
2065def validate_phone(candidate: Dict[str, Any], page_ctx: Dict[str, Any]) -> Optional[Dict[str, Any]]:2066 number = (candidate.get("normalized") or candidate.get("value") or "").replace(" ", "")2067 raw = candidate.get("raw") or number2068 if not number:2069 return None2070
2071 parsed = normalize_phone_number(raw, candidate.get("region"))2072 normalized = parsed.get("normalized")2073 if not normalized:2074 candidate["reject_reason"] = "numeric_noise"2075 return None2076 snippet = candidate.get("snippet", "")2077 source_type = candidate.get("source_type")2078 is_contact_page = bool(page_ctx.get("is_contact_page"))2079 low_snippet = (snippet or "").lower()2080 nearby_labels = [str(label or "").lower() for label in (candidate.get("nearby_labels") or [])]2081 nearby_text = candidate.get("nearby_text") or snippet2082 combined_context = " ".join([snippet, nearby_text, candidate.get("nearby_heading") or ""]).lower()2083
2084 validator_passed = []2085 if any(bad in low_snippet for bad in PHONE_EXCLUDE_CONTEXT_HINTS):2086 candidate["reject_reason"] = "numeric_noise"2087 return None2088 if _looks_like_phone_token_noise(raw, snippet, nearby_text):2089 candidate["reject_reason"] = "numeric_noise"2090 return None2091 if _looks_like_product_identifier_number(raw, snippet, nearby_text):2092 candidate["reject_reason"] = "numeric_noise"2093 return None2094 if len(normalized.lstrip("+")) > 15:2095 candidate["reject_reason"] = "numeric_noise"2096 return None2097 if any(hint in combined_context for hint in PHONE_PRODUCT_NOISE_HINTS) and not any(2098 label in combined_context for label in ("call", "contact", "support", "sales", "tel", "phone")2099 ):2100 candidate["reject_reason"] = "numeric_noise"2101 return None2102
2103 if source_type in {"tel", "callto", "sms"}:2104 validator_passed.append(f"{source_type}_link")2105 band = "VERIFIED"2106 elif source_type == "jsonld":2107 validator_passed.append("jsonld_telephone")2108 band = "VERIFIED"2109 elif source_type == "script":2110 if candidate.get("has_context"):2111 validator_passed.append("deliberate_script_phone")2112 band = "STRONG"2113 else:2114 candidate["reject_reason"] = "numeric_noise"2115 return None2116 elif source_type == "attribute":2117 if candidate.get("source_attribute") not in {"aria-label", "title"}:2118 candidate["reject_reason"] = "numeric_noise"2119 return None2120 if not _has_strong_phone_context(combined_context):2121 candidate["reject_reason"] = "numeric_noise"2122 return None2123 validator_passed.append("attribute_phone_context")2124 band = "STRONG"2125 else:2126 if candidate.get("has_context"):2127 validator_passed.append("visible_text_contact_context")2128 band = "STRONG"2129 else:2130 candidate["reject_reason"] = "numeric_noise"2131 return None2132
2133 return {2134 "type": "phone",2135 "raw": raw,2136 "value": normalized,2137 "normalized": normalized,2138 "source_type": source_type,2139 "type_hint": candidate.get("type_hint") or classify_phone_context(snippet),2140 "page_url": candidate["page_url"],2141 "dom_region": candidate.get("dom_region"),2142 "nearby_labels": nearby_labels,2143 "nearby_text": nearby_text,2144 "evidence": f"source:{source_type}; snippet:{snippet}; region:{candidate.get('dom_region', '')}",2145 "confidence_band": band,2146 "validator_passed": validator_passed,2147 }2148
2149
2150def _field_role(input_node: Any, label_map: Dict[str, str]) -> Optional[str]:2151 text = " ".join(2152 part2153 for part in [2154 (input_node.get("name") or ""),2155 (input_node.get("id") or ""),2156 (input_node.get("placeholder") or ""),2157 (input_node.get("aria-label") or ""),2158 label_map.get((input_node.get("id") or "")) or "",2159 ]2160 if part2161 ).lower()2162
2163 if any(_contains_token(text, hint) for hint in FORM_REVIEW_FIELD_HINTS):2164 return "review"2165 if any(_contains_token(text, hint) for hint in {"business email", "work email", "corporate email"}):2166 return "business_email"2167 if any(_contains_token(text, hint) for hint in {"company size", "team size", "employee count", "number of employees"}):2168 return "company_size"2169 if any(_contains_token(text, hint) for hint in {"company url", "company website", "website", "web site"}):2170 return "company_url"2171 if any(_contains_token(text, hint) for hint in {"number of browser profiles", "browser profiles"}):2172 return "browser_profiles"2173 if any(_contains_token(text, hint) for hint in {"quantity", "qty"}):2174 return "quantity"2175 if any(_contains_token(text, hint) for hint in {"use case", "use-case"}):2176 return "use_case"2177 if any(_contains_token(text, hint) for hint in {"preferred contact method", "best way to contact", "contact method"}):2178 return "preferred_contact_method"2179 for role in FORM_REQUIRED_ROLES | FORM_MESSAGE_ROLES:2180 if role in text:2181 return role2182 if "subject" in text:2183 return "subject"2184 if "company" in text:2185 return "company"2186 if "phone" in text:2187 return "phone"2188 return None2189
2190
2191def _clean_unique_strings(values: Iterable[Any], *, limit: int = 6) -> List[str]:2192 cleaned: List[str] = []2193 seen: Set[str] = set()2194 for value in values:2195 text = _strip_text(str(value or ""))2196 if not text:2197 continue2198 lowered = text.lower()2199 if lowered in seen:2200 continue2201 cleaned.append(text)2202 seen.add(lowered)2203 if len(cleaned) >= limit:2204 break2205 return cleaned2206
2207
2208def _compact_locator_hint(payload: Dict[str, Any]) -> Dict[str, Any]:2209 compact: Dict[str, Any] = {}2210 for key, value in payload.items():2211 if isinstance(value, str):2212 text = _strip_text(value)2213 if text:2214 compact[key] = text2215 elif isinstance(value, list):2216 cleaned = _clean_unique_strings(value, limit=max(len(value), 1))2217 if cleaned:2218 compact[key] = cleaned2219 elif value is not None:2220 compact[key] = value2221 return compact2222
2223
2224def _build_field_locator_hint(field: Dict[str, Any]) -> Dict[str, Any]:2225 placeholder = field.get("placeholder")2226 if _field_placeholder_looks_like_contact_hint(field):2227 placeholder = ""2228 return _compact_locator_hint(2229 {2230 "role": field.get("role"),2231 "tag": field.get("tag"),2232 "input_type": field.get("type"),2233 "label": field.get("label"),2234 "name": field.get("name"),2235 "id": field.get("id"),2236 "placeholder": placeholder,2237 "aria_label": field.get("aria_label"),2238 "autocomplete": field.get("autocomplete"),2239 "class_tokens": field.get("class_tokens") or [],2240 "debug_css_path": field.get("debug_css_path") or "",2241 }2242 )2243
2244
2245def _build_submit_locator_hint(submit_control: Dict[str, Any]) -> Dict[str, Any]:2246 return _compact_locator_hint(2247 {2248 "tag": submit_control.get("tag"),2249 "control_type": submit_control.get("type"),2250 "text": submit_control.get("text"),2251 "name": submit_control.get("name"),2252 "id": submit_control.get("id"),2253 "aria_label": submit_control.get("aria_label"),2254 "class_tokens": submit_control.get("class_tokens") or [],2255 "debug_css_path": submit_control.get("debug_css_path") or "",2256 }2257 )2258
2259
2260def _build_form_locator_hint(candidate: Dict[str, Any]) -> Dict[str, Any]:2261 action = candidate.get("action") or ""2262 action_path = (urlparse(action).path or "").strip() if action else ""2263 return _compact_locator_hint(2264 {2265 "tag": "form",2266 "method": candidate.get("method"),2267 "action": action,2268 "action_path": action_path,2269 "id": candidate.get("form_id"),2270 "class_tokens": (candidate.get("form_class") or "").split(),2271 "text_hint": (candidate.get("form_text") or "")[:160],2272 "dom_signature": candidate.get("dom_signature"),2273 "debug_css_path": candidate.get("form_css_path") or "",2274 }2275 )2276
2277
2278def _pick_primary_submit_control(submit_controls: List[Dict[str, Any]]) -> Dict[str, Any]:2279 for control in submit_controls:2280 label = (control.get("text") or "").lower()2281 if any(_contains_token(label, hint) for hint in FORM_CONTACT_SUBMIT_HINTS | FORM_LEAD_CAPTURE_HINTS):2282 return control2283 return submit_controls[0] if submit_controls else {}2284
2285
2286def _form_has_contact_action_hint(action: str) -> bool:2287 if not action:2288 return False2289 return bool(_matching_intent_hints(action, FORM_CONTACT_ACTION_HINTS)) and not bool(_matching_intent_hints(action, FORM_SALES_FLOW_HINTS))2290
2291
2292def _form_meta_text(candidate: Dict[str, Any], submit_labels: List[str]) -> str:2293 fields = candidate.get("fields", [])2294 submit_controls = candidate.get("submit_controls", [])2295 return " ".join(2296 part2297 for part in [2298 candidate.get("page_url", ""),2299 candidate.get("form_id", ""),2300 candidate.get("form_class", ""),2301 candidate.get("form_css_path", ""),2302 candidate.get("action", ""),2303 " ".join(submit_labels),2304 " ".join((control.get("text") or "") for control in submit_controls),2305 " ".join(" ".join(control.get("class_tokens") or []) for control in submit_controls),2306 candidate.get("form_text", ""),2307 " ".join((field.get("role") or "") for field in fields),2308 " ".join((field.get("meta") or "") for field in fields),2309 ]2310 if part2311 ).lower()2312
2313
2314def _form_has_contact_intent(candidate: Dict[str, Any], submit_labels: List[str]) -> bool:2315 meta_text = _form_meta_text(candidate, submit_labels)2316 if _matching_intent_hints(meta_text, FORM_SALES_FLOW_HINTS):2317 return False2318 if _matching_intent_hints(meta_text, FORM_CONTACT_TEXT_HINTS):2319 return True2320 return _form_has_contact_action_hint(candidate.get("action") or "")2321
2322
2323def _form_has_lead_capture_intent(candidate: Dict[str, Any], submit_labels: List[str]) -> bool:2324 meta_text = _form_meta_text(candidate, submit_labels)2325 return bool(_matching_intent_hints(meta_text, FORM_LEAD_CAPTURE_HINTS))2326
2327
2328def _form_has_hard_reject_intent(candidate: Dict[str, Any], submit_labels: List[str]) -> bool:2329 meta_text = _form_meta_text(candidate, submit_labels)2330 return bool(_matching_intent_hints(meta_text, FORM_HARD_REJECT_HINTS))2331
2332
2333def _form_has_review_intent(candidate: Dict[str, Any], submit_labels: List[str]) -> bool:2334 meta_text = _form_meta_text(candidate, submit_labels)2335 page_path = (urlparse(candidate.get("page_url") or "").path or "").lower()2336 action_path = (urlparse(candidate.get("action") or "").path or "").lower()2337 has_review_path = any(2338 path.endswith("/reviews") or "/reviews/" in path2339 for path in (page_path, action_path)2340 if path2341 )2342 has_review_field = any(2343 field.get("role") == "review"2344 or (field.get("tag") == "textarea" and _contains_token(field.get("meta") or "", "review"))2345 or any(_contains_token(field.get("meta") or "", hint) for hint in FORM_REVIEW_FIELD_HINTS)2346 for field in candidate.get("fields", [])2347 )2348 review_hits = sum(1 for hint in FORM_REVIEW_HINTS if _contains_token(meta_text, hint))2349 return has_review_field or (has_review_path and review_hits > 0) or review_hits > 1 or (review_hits > 0 and _contains_token(meta_text, "terms"))2350
2351
2352def _form_has_sales_flow_intent(candidate: Dict[str, Any], submit_labels: List[str]) -> bool:2353 meta_text = _form_meta_text(candidate, submit_labels)2354 has_sales_signal = bool(_matching_intent_hints(meta_text, FORM_SALES_FLOW_HINTS))2355 has_gate_signal = any(_contains_token(meta_text, hint) for hint in FORM_INTERACTION_GATE_HINTS)2356 return has_sales_signal or (has_gate_signal and _form_has_hard_reject_intent(candidate, submit_labels))2357
2358
2359def _form_has_interaction_gate(candidate: Dict[str, Any]) -> bool:2360 gate_text = " ".join(2361 part2362 for part in [2363 candidate.get("form_id") or "",2364 candidate.get("form_class") or "",2365 candidate.get("form_css_path") or "",2366 ]2367 if part2368 )2369 return bool(_matching_intent_hints(gate_text, FORM_INTERACTION_GATE_HINTS | {"hidden", "overlay", "lightbox"}))2370
2371
2372def _form_is_hidden_modal_outside_contact_page(candidate: Dict[str, Any]) -> bool:2373 if not _form_has_interaction_gate(candidate):2374 return False2375
2376 page_path = (urlparse(candidate.get("page_url") or "").path or "").lower()2377 action_path = (urlparse(candidate.get("action") or "").path or "").lower()2378 return not (_is_contact_url_hint(page_path) or _is_contact_url_hint(action_path))2379
2380
2381def _form_sales_outreach_reasons(candidate: Dict[str, Any], submit_labels: List[str]) -> List[str]:2382 fields = candidate.get("fields", [])2383 role_sequence = [field.get("role") for field in fields if field.get("role")]2384 role_set = set(role_sequence)2385 has_message = bool(role_set & FORM_MESSAGE_ROLES)2386 field_meta = " ".join((field.get("meta") or "") for field in fields)2387
2388 reasons: List[str] = []2389 if _matching_intent_hints(candidate.get("page_url") or "", FORM_SALES_FLOW_HINTS):2390 reasons.append("sales_page_path")2391 if _matching_intent_hints(candidate.get("action") or "", FORM_SALES_FLOW_HINTS):2392 reasons.append("sales_action_path")2393 if _matching_intent_hints(" ".join(submit_labels), FORM_SALES_FLOW_HINTS):2394 reasons.append("sales_submit_cta")2395 if _matching_intent_hints(candidate.get("form_text") or "", FORM_SALES_FLOW_HINTS):2396 reasons.append("sales_form_text")2397 if _matching_intent_hints(field_meta, FORM_SALES_FIELD_HINTS):2398 reasons.append("sales_field_copy")2399
2400 if "business_email" in role_set:2401 reasons.append("business_email_field")2402 if "company_size" in role_set:2403 reasons.append("company_size_field")2404 if {"company", "company_size"} <= role_set:2405 reasons.append("company_qualification_fields")2406 if "company_url" in role_set and role_set & {"company", "business_email", "company_size"}:2407 reasons.append("company_url_qualification")2408 if "use_case" in role_set and (not has_message or role_set & {"company", "business_email", "quantity", "browser_profiles"}):2409 reasons.append("use_case_qualification")2410 if "browser_profiles" in role_set:2411 reasons.append("browser_profile_qualification")2412 if "quantity" in role_set and (not has_message or role_set & {"company", "business_email", "browser_profiles", "use_case"}):2413 reasons.append("quantity_qualification")2414
2415 preferred_contact_idx = next((idx for idx, role in enumerate(role_sequence) if role == "preferred_contact_method"), None)2416 first_message_idx = next((idx for idx, role in enumerate(role_sequence) if role in FORM_MESSAGE_ROLES), None)2417 if preferred_contact_idx is not None and (first_message_idx is None or preferred_contact_idx < first_message_idx):2418 reasons.append("preferred_contact_before_message")2419
2420 if not has_message and len(role_set & FORM_SALES_QUALIFICATION_ROLES) >= 2:2421 reasons.append("lead_qualification_without_message")2422
2423 return sorted(set(reasons))2424
2425
2426def extract_form_candidates(page_url: str, soup: BeautifulSoup) -> List[Dict[str, Any]]:2427 forms: List[Dict[str, Any]] = []2428
2429 label_cache = {}2430 for label in soup.find_all("label"):2431 fid = label.get("for")2432 if fid:2433 label_cache[fid] = _strip_text(label.get_text(" ", strip=True))2434
2435 for form in soup.find_all("form"):2436 action = normalize_url(form.get("action", "") or "", base=page_url) or page_url2437 method = (form.get("method") or "GET").upper()2438 form_id = (form.get("id") or "").lower()2439 form_class = " ".join(form.get("class", [])).lower()2440 form_text = _strip_text(form.get_text(" ", strip=True))2441 fields: List[Dict[str, Any]] = []2442
2443 for field in form.find_all(["input", "textarea", "select"]):2444 f_type = (field.get("type") or field.name or "").lower()2445 if f_type in {"submit", "button", "file", "hidden", "reset"}:2446 continue2447 name = (field.get("name") or "").strip()2448 if not name and not (field.get("id") or field.get("placeholder") or field.get("aria-label")):2449 continue2450 label = label_cache.get((field.get("id") or "").strip(), "")2451 role = _field_role(field, label_cache)2452 meta = " ".join(2453 value2454 for value in (2455 name,2456 field.get("id") or "",2457 field.get("placeholder") or "",2458 field.get("aria-label") or "",2459 label,2460 f_type,2461 )2462 if value2463 ).lower()2464 fields.append(2465 {2466 "tag": field.name.lower(),2467 "name": name,2468 "type": f_type,2469 "id": field.get("id") or "",2470 "label": label,2471 "placeholder": field.get("placeholder") or "",2472 "aria_label": field.get("aria-label") or "",2473 "autocomplete": field.get("autocomplete") or "",2474 "class_tokens": _node_class_tokens(field),2475 "role": role,2476 "meta": meta[:220],2477 "debug_css_path": _css_selector(field),2478 }2479 )2480
2481 if not fields:2482 continue2483
2484 submit_labels: List[str] = []2485 submit_controls: List[Dict[str, Any]] = []2486 for btn in form.find_all(["button", "input"]):2487 btn_type = (btn.get("type") or "").lower()2488 text = _strip_text((btn.get_text(" ", strip=True) or btn.get("value") or ""))2489 if not text:2490 continue2491 if text and (btn_type in {"submit", "button"} or btn.name == "button"):2492 submit_labels.append(text.lower())2493 submit_controls.append(2494 {2495 "tag": btn.name.lower(),2496 "type": btn_type or btn.name.lower(),2497 "text": text,2498 "name": btn.get("name") or "",2499 "id": btn.get("id") or "",2500 "aria_label": btn.get("aria-label") or "",2501 "class_tokens": _node_class_tokens(btn),2502 "debug_css_path": _css_selector(btn),2503 }2504 )2505
2506 forms.append(2507 {2508 "page_url": page_url,2509 "action": action,2510 "method": method,2511 "fields": fields,2512 "submit_labels": submit_labels,2513 "form_id": form_id,2514 "form_class": form_class,2515 "form_text": form_text[:1200],2516 "evidence_nodes": [_strip_text(form.get_text(" ", strip=True))[:240]],2517 "dom_signature": "||".join(2518 sorted(2519 (2520 f"{(f['name'] or f['id'] or 'field')}:{f.get('role') or 'untyped'}"2521 for f in fields2522 )2523 )2524 ),2525 "form_css_path": _css_selector(form),2526 "submit_controls": submit_controls,2527 }2528 )2529
2530 return forms2531
2532
2533def validate_contact_form(candidate: Dict[str, Any]) -> Optional[Dict[str, Any]]:2534 fields = candidate.get("fields", [])2535 submit_labels = [label.lower() for label in candidate.get("submit_labels", [])]2536 submit_controls = candidate.get("submit_controls", [])2537 roles = [f.get("role") for f in fields if f.get("role")]2538 role_set = {r for r in roles if r}2539 if len(fields) <= 1:2540 candidate["reject_reason"] = "non_contact_form"2541 return None2542 if len(role_set) < 2:2543 candidate["reject_reason"] = "non_contact_form"2544 return None2545
2546 meta = " ".join(2547 [2548 (candidate.get("form_id") or ""),2549 (candidate.get("form_class") or ""),2550 candidate.get("action", ""),2551 " ".join(candidate.get("submit_labels", [])),2552 " ".join((f.get("meta") or "") for f in fields),2553 ]2554 ).lower()2555
2556 if any(_contains_token(meta, reject) for reject in FORM_REJECT_HINTS):2557 candidate["reject_reason"] = "non_contact_form"2558 return None2559
2560 if _form_has_review_intent(candidate, submit_labels):2561 candidate["reject_reason"] = "non_contact_form"2562 return None2563
2564 if _form_is_hidden_modal_outside_contact_page(candidate):2565 candidate["reject_reason"] = "non_contact_form"2566 return None2567
2568 has_message = bool(role_set & FORM_MESSAGE_ROLES)2569 has_contact_intent = _form_has_contact_intent(candidate, submit_labels)2570 has_lead_capture_intent = _form_has_lead_capture_intent(candidate, submit_labels)2571 has_hard_reject_intent = _form_has_hard_reject_intent(candidate, submit_labels)2572 has_sales_flow_intent = _form_has_sales_flow_intent(candidate, submit_labels)2573 sales_outreach_reasons = _form_sales_outreach_reasons(candidate, submit_labels)2574 submit_has_contact = any(any(_contains_token(label, hint) for hint in FORM_CONTACT_SUBMIT_HINTS) for label in submit_labels)2575
2576 # Newsletter/demo/download forms often have valid-looking name/email fields.2577 # Treat explicit demo/subscribe flows as non-contact regardless of field shape.2578 if sales_outreach_reasons:2579 candidate["reject_reason"] = "sales_outreach_form"2580 candidate["sales_outreach_signals"] = sales_outreach_reasons2581 return None2582
2583 if has_hard_reject_intent or has_sales_flow_intent:2584 candidate["reject_reason"] = "non_contact_form"2585 return None2586
2587 # Other lead-capture forms are non-contact unless they also have clear inquiry intent.2588 if has_lead_capture_intent and not (has_message and has_contact_intent):2589 candidate["reject_reason"] = "non_contact_form"2590 return None2591
2592 if not has_message and role_set <= {"name", "email"}:2593 candidate["reject_reason"] = "non_contact_form"2594 return None2595
2596 if not has_message and role_set <= FORM_INFO_ONLY_ROLES and not has_contact_intent:2597 candidate["reject_reason"] = "non_contact_form"2598 return None2599
2600 if not has_message and not has_contact_intent:2601 candidate["reject_reason"] = "non_contact_form"2602 return None2603
2604 normalized_roles = sorted(role_set)2605 if not normalized_roles:2606 candidate["reject_reason"] = "non_contact_form"2607 return None2608
2609 validator_passed = ["semantic_field_count"]2610 if has_message:2611 validator_passed.append("message_or_inquiry_or_subject_field")2612 if has_contact_intent:2613 validator_passed.append("contact_intent")2614 if submit_has_contact:2615 validator_passed.append("submit_cta")2616
2617 band = "STRONG"2618 if has_message and (submit_has_contact or has_contact_intent):2619 band = "VERIFIED"2620
2621 action = candidate.get("action", "")2622 if action:2623 validator_passed.append("action_detected")2624
2625 field_locator_hints: Dict[str, Dict[str, Any]] = {}2626 for field in fields:2627 role = field.get("role")2628 if role and role not in field_locator_hints:2629 field_locator_hints[role] = _build_field_locator_hint(field)2630 primary_submit = _pick_primary_submit_control(submit_controls)2631
2632 return {2633 "type": "contact_form",2634 "value": candidate["page_url"],2635 "page_url": candidate["page_url"],2636 "evidence": f"roles={','.join(normalized_roles)}; submit_labels={','.join(submit_labels)[:140]}; action={action}",2637 "confidence_band": band,2638 "validator_passed": validator_passed,2639 "normalized_field_roles": normalized_roles,2640 "action": action,2641 "method": candidate.get("method"),2642 "submit_label": next(2643 (2644 lbl2645 for lbl in candidate.get("submit_labels", [])2646 if any(_contains_token(lbl.lower(), hint) for hint in FORM_CONTACT_SUBMIT_HINTS | FORM_LEAD_CAPTURE_HINTS)2647 ),2648 "",2649 ),2650 "form_signature": f"{candidate.get('method','GET')}|{action}|{candidate.get('dom_signature','')}",2651 "resolution_method": "snapshot_ref",2652 "usage_notes": SNAPSHOT_REF_USAGE_NOTES,2653 "form_locator_hints": _build_form_locator_hint(candidate),2654 "submit_locator_hints": _build_submit_locator_hint(primary_submit),2655 "field_locator_hints": field_locator_hints,2656 "supporting_pages": [],2657 }2658
2659
2660def _run_contact_form_validation_self_checks() -> None:2661 def _candidate(2662 page_url: str,2663 fields: List[Dict[str, Any]],2664 *,2665 form_text: str,2666 submit_labels: List[str],2667 form_class: str = "",2668 action: str = "",2669 ) -> Dict[str, Any]:2670 return {2671 "page_url": page_url,2672 "action": action or page_url,2673 "method": "POST",2674 "fields": fields,2675 "submit_labels": submit_labels,2676 "form_id": "",2677 "form_class": form_class,2678 "form_text": form_text,2679 "evidence_nodes": [form_text[:240]],2680 "dom_signature": "||".join(2681 sorted(f"{(field.get('name') or field.get('id') or 'field')}:{field.get('role') or 'untyped'}" for field in fields)2682 ),2683 "form_css_path": "form",2684 "submit_controls": [],2685 }2686
2687 def _field(role: Optional[str], meta: str, *, tag: str = "input", input_type: str = "text") -> Dict[str, Any]:2688 return {2689 "tag": tag,2690 "name": meta.split()[0],2691 "type": input_type,2692 "id": "",2693 "label": "",2694 "placeholder": "",2695 "aria_label": "",2696 "autocomplete": "",2697 "class_tokens": [],2698 "role": role,2699 "meta": meta,2700 "debug_css_path": "",2701 }2702
2703 assert validate_contact_form(2704 _candidate(2705 "https://example.com/contact-us",2706 [2707 _field("name", "full name"),2708 _field("email", "email address", input_type="email"),2709 _field("message", "message", tag="textarea", input_type="textarea"),2710 ],2711 form_text="Contact us Send us a message",2712 submit_labels=["Send Message"],2713 )2714 ), "direct contact forms with a message should validate"2715
2716 assert not validate_contact_form(2717 _candidate(2718 "https://example.com/reviews/widget",2719 [2720 _field("name", "first name"),2721 _field("email", "email address", input_type="email"),2722 _field("review", "your review", tag="textarea", input_type="textarea"),2723 _field("review", "pros optional"),2724 _field("review", "cons optional"),2725 ],2726 form_text="Rate and Review Your Experience Pros Cons Your LinkedIn Profile URL Terms",2727 submit_labels=["Submit Review"],2728 form_class="submit_review-form",2729 )2730 ), "review submission forms must be rejected"2731
2732 assert not validate_contact_form(2733 _candidate(2734 "https://example.com/contact-sales",2735 [2736 _field("name", "full name"),2737 _field("email", "email address", input_type="email"),2738 _field("company", "company name"),2739 _field("message", "message", tag="textarea", input_type="textarea"),2740 ],2741 form_text="Contact sales to discuss pricing and a demo",2742 submit_labels=["Contact Sales"],2743 )2744 ), "sales contact CTAs must be rejected even when contact wording is present"2745
2746 assert not validate_contact_form(2747 _candidate(2748 "https://example.com/pt/fale-conosco",2749 [2750 _field("name", "nome completo"),2751 _field("email", "email", input_type="email"),2752 _field("company", "empresa"),2753 _field("message", "mensagem", tag="textarea", input_type="textarea"),2754 ],2755 form_text="Fale com o nosso time",2756 submit_labels=["Falar com Vendas"],2757 )2758 ), "localized sales CTAs must be rejected"2759
2760 assert not validate_contact_form(2761 _candidate(2762 "https://example.com/contact-sales",2763 [2764 _field("name", "full name"),2765 _field("email", "email address", input_type="email"),2766 _field("company", "company name"),2767 _field("phone", "phone number", input_type="tel"),2768 ],2769 form_text="Talk to sales Request demo Book a call",2770 submit_labels=["Request Demo"],2771 form_class="provider_popup-form modal",2772 )2773 ), "sales/demo popup flows must be rejected"2774
2775 assert not validate_contact_form(2776 _candidate(2777 "https://example.com/pricing",2778 [2779 _field("name", "full name"),2780 _field("business_email", "business email", input_type="email"),2781 _field("company", "company name"),2782 _field("company_size", "company size"),2783 _field("use_case", "use case"),2784 ],2785 form_text="Book a call with our team",2786 submit_labels=["Book a Call"],2787 )2788 ), "lead qualification fields must be rejected"2789
2790 assert not validate_contact_form(2791 _candidate(2792 "https://example.com/request-quote",2793 [2794 _field("name", "full name"),2795 _field("email", "email address", input_type="email"),2796 _field("company", "company name"),2797 _field("quantity", "quantity"),2798 _field("message", "project details", tag="textarea", input_type="textarea"),2799 ],2800 form_text="Request a quote for your team",2801 submit_labels=["Request Quote"],2802 )2803 ), "request quote flows must be rejected"2804
2805 assert not validate_contact_form(2806 _candidate(2807 "https://example.com/contact-us",2808 [2809 _field("name", "full name"),2810 _field("email", "email address", input_type="email"),2811 ],2812 form_text="Contact us",2813 submit_labels=["Send"],2814 )2815 ), "name and email alone are not enough for a contact form"2816
2817 assert not validate_contact_form(2818 _candidate(2819 "https://example.com/blog/news",2820 [2821 _field("email", "email address", input_type="email"),2822 ],2823 form_text="Subscribe to updates",2824 submit_labels=["Subscribe"],2825 form_class="newsletter-signup",2826 )2827 ), "single-field subscribe forms must be rejected"2828
2829 assert validate_contact_form(2830 _candidate(2831 "https://example.com/contact-us",2832 [2833 _field("name", "full name"),2834 _field("email", "email address", input_type="email"),2835 _field("message", "message", tag="textarea", input_type="textarea"),2836 ],2837 form_text="Contact support Send us a message",2838 submit_labels=["Send Message"],2839 )2840 ), "support contact forms must still validate"2841
2842 assert validate_contact_form(2843 _candidate(2844 "https://example.com/contact",2845 [2846 _field("name", "full name"),2847 _field("email", "email address", input_type="email"),2848 _field("subject", "subject"),2849 _field("message", "message", tag="textarea", input_type="textarea"),2850 ],2851 form_text="General inquiries",2852 submit_labels=["Submit"],2853 )2854 ), "general contact forms with subject and message must still validate"2855
2856 assert not validate_contact_form(2857 {2858 **_candidate(2859 "https://example.com/news/supports-study",2860 [2861 _field("name", "first name"),2862 _field("name", "last name"),2863 _field("phone", "phone number", input_type="tel"),2864 _field("email", "email", input_type="email"),2865 _field("message", "message", tag="textarea", input_type="textarea"),2866 ],2867 form_text="Let's connect We'd love to learn more about your mobility needs and how motiontag can support your goals.",2868 submit_labels=["Abschicken"],2869 action="https://example.com/news/supports-study",2870 ),2871 "form_css_path": "div#modal-formular > div.fallback:nth-of-type(1) > div.ce_form.block > form",2872 }2873 ), "shared hidden modal contact forms must be rejected on non-contact pages"2874
2875 assert validate_contact_form(2876 {2877 **_candidate(2878 "https://example.com/company/contact",2879 [2880 _field("name", "first name"),2881 _field("name", "last name"),2882 _field("phone", "phone number", input_type="tel"),2883 _field("email", "email", input_type="email"),2884 _field("message", "message", tag="textarea", input_type="textarea"),2885 ],2886 form_text="Let's connect We'd love to learn more about your mobility needs and how motiontag can support your goals.",2887 submit_labels=["Send"],2888 action="https://example.com/company/contact",2889 ),2890 "form_css_path": "div#modal-formular > div.fallback:nth-of-type(1) > div.ce_form.block > form",2891 }2892 ), "shared modal contact forms should still validate on contact pages"2893
2894
2895def _run_channel_validation_self_checks() -> None:2896 assert not _has_contact_phrase("motiontag supports study by ETH Zurich"), "supports must not count as contact support intent"2897 assert _has_contact_phrase("contact support"), "explicit support wording must still count as contact intent"2898 assert _looks_like_product_identifier_number(2899 "234710-2001",2900 "Item No. 234710-2001",2901 "Benefits Double layer This product features practical support.",2902 ), "SKU/item numbers must be recognized as product identifiers"2903
2904 page_url = "https://example.com/news/supports-study"2905 article_html = """2906 <html>2907 <body>2908 <article>2909 <h1>Mobility team supports study</h1>2910 <p>Published on 11.04.2021</p>2911 <p>Reference: https://example.com/handle/20.500.11850/500100</p>2912 </article>2913 </body>2914 </html>2915 """2916 article_soup = BeautifulSoup(article_html, "html.parser")2917 phone_candidates = extract_phone_candidates(page_url, article_html, article_soup)2918 assert not any(2919 validate_phone(dict(candidate), {"is_contact_page": False, "url": page_url})2920 for candidate in phone_candidates2921 ), "dates and handle IDs near 'supports' copy must not validate as phones"2922
2923 product_page_url = "https://example.com/products/built-in-support-top"2924 product_html = """2925 <html>2926 <body>2927 <main>2928 <h2>Material & care</h2>2929 <p>Item No. 234710-2001</p>2930 <p>Benefits Double layer This product features practical support.</p>2931 </main>2932 </body>2933 </html>2934 """2935 product_soup = BeautifulSoup(product_html, "html.parser")2936 product_phone_candidates = extract_phone_candidates(product_page_url, product_html, product_soup)2937 assert not any(2938 validate_phone(dict(candidate), {"is_contact_page": False, "url": product_page_url})2939 for candidate in product_phone_candidates2940 ), "SKU-style item numbers must not validate as phones"2941
2942 email_hint = _build_field_locator_hint(2943 {2944 "role": "email",2945 "tag": "input",2946 "type": "email",2947 "label": "Email",2948 "name": "email",2949 "id": "contact-email",2950 "placeholder": "your@email.com",2951 "aria_label": "",2952 "autocomplete": "email",2953 "class_tokens": ["field"],2954 "debug_css_path": "input#contact-email",2955 }2956 )2957 assert "placeholder" not in email_hint, "email example placeholders must not leak into locator hints"2958
2959 phone_hint = _build_field_locator_hint(2960 {2961 "role": "phone",2962 "tag": "input",2963 "type": "tel",2964 "label": "Phone",2965 "name": "phone",2966 "id": "contact-phone",2967 "placeholder": "(xx) xxxxxxxxx",2968 "aria_label": "",2969 "autocomplete": "tel",2970 "class_tokens": ["field"],2971 "debug_css_path": "input#contact-phone",2972 }2973 )2974 assert "placeholder" not in phone_hint, "phone example placeholders must not leak into locator hints"2975
2976 name_hint = _build_field_locator_hint(2977 {2978 "role": "name",2979 "tag": "input",2980 "type": "text",2981 "label": "First name",2982 "name": "first_name",2983 "id": "contact-name",2984 "placeholder": "Enter your first name",2985 "aria_label": "",2986 "autocomplete": "given-name",2987 "class_tokens": ["field"],2988 "debug_css_path": "input#contact-name",2989 }2990 )2991 assert name_hint.get("placeholder") == "Enter your first name", "non-contact placeholders should remain available as hints"2992
2993
2994if __debug__:2995 _run_contact_form_validation_self_checks()2996 _run_channel_validation_self_checks()2997
2998
2999@dataclass3000class RenderedPageSnapshot:3001 final_url: Optional[str]3002 html: Optional[str]3003 script_srcs: List[str]3004 iframe_srcs: List[str]3005 launchers: List[Dict[str, Any]]3006 error: Optional[str] = None3007
3008
3009def _sorted_unique_strings(values: Iterable[Any]) -> List[str]:3010 cleaned = {_strip_text(str(value)) for value in values if _strip_text(str(value))}3011 return sorted(cleaned)3012
3013
3014def _truncate_hint_text(value: str, *, limit: int = 180) -> str:3015 cleaned = _strip_text(value)3016 if len(cleaned) <= limit:3017 return cleaned3018 return cleaned[: limit - 3].rstrip() + "..."3019
3020
3021def _provider_widget_terms(provider: str, cfg: Dict[str, Any]) -> Set[str]:3022 terms = {provider.lower()}3023 terms.update(_strip_text(str(token)).lower() for token in cfg.get("inline", set()) if _strip_text(str(token)))3024 terms.update(_strip_text(str(token)).lower() for token in cfg.get("selector", set()) if _strip_text(str(token)))3025 return {term for term in terms if term}3026
3027
3028def _node_human_readable_attrs(node: Any) -> Dict[str, str]:3029 if not getattr(node, "attrs", None):3030 return {}3031 attrs: Dict[str, str] = {}3032 for name, raw_value in node.attrs.items():3033 text = _stringify_attr_value(raw_value)3034 if _is_human_readable_attr(name, text):3035 attrs[name] = _truncate_hint_text(text)3036 return attrs3037
3038
3039def _node_class_tokens(node: Any) -> List[str]:3040 return _sorted_unique_strings(_strip_text(str(token)) for token in (node.get("class") or []))[:10]3041
3042
3043def _infer_corner_hint(text: str) -> str:3044 lowered = (text or "").lower()3045 pairs = [3046 ("bottom-right", ("bottom", "right")),3047 ("bottom-left", ("bottom", "left")),3048 ("top-right", ("top", "right")),3049 ("top-left", ("top", "left")),3050 ]3051 for label, required in pairs:3052 if all(token in lowered for token in required):3053 return label3054 for single in ("bottom", "top", "right", "left"):3055 if single in lowered:3056 return single3057 return ""3058
3059
3060def _static_widget_position_hints(node: Any) -> Tuple[bool, str]:3061 style = _strip_text(node.get("style") or "")3062 descriptor = " ".join(3063 [3064 style,3065 _strip_text(node.get("id") or ""),3066 " ".join(_node_class_tokens(node)),3067 " ".join(_strip_text(str(value)) for key, value in (node.attrs or {}).items() if str(key).startswith("data-")),3068 ]3069 ).lower()3070 style_compact = style.lower().replace(" ", "")3071 fixed_position = "position:fixed" in style_compact or "position:sticky" in style_compact3072 if not fixed_position:3073 fixed_position = any(token in descriptor for token in WIDGET_FIXED_POSITION_HINTS)3074 corner_hint = _infer_corner_hint(descriptor)3075 return fixed_position, corner_hint3076
3077
3078def _widget_phrase_hits(text: str) -> List[str]:3079 lowered = (text or "").lower()3080 hits = [phrase for phrase in WIDGET_LAUNCHER_PHRASES if phrase in lowered]3081 return sorted(set(hits), key=lambda value: (-len(value), value))3082
3083
3084def _widget_launcher_has_locator_data(launcher: Dict[str, Any]) -> bool:3085 return any(3086 [3087 launcher.get("launcher_text"),3088 launcher.get("aria_label"),3089 launcher.get("title"),3090 launcher.get("id"),3091 launcher.get("class_tokens"),3092 launcher.get("data_attributes"),3093 launcher.get("exact_css_clue"),3094 ]3095 )3096
3097
3098def _is_widget_messaging_href(value: str) -> bool:3099 lowered = (value or "").lower()3100 return any(token in lowered for token in ("wa.me", "api.whatsapp.com", "m.me", "messenger.com", "chat.whatsapp.com"))3101
3102
3103def _widget_launcher_score(launcher: Dict[str, Any], *, page_url: str) -> int:3104 score = 03105 for phrase in launcher.get("phrase_hits", []):3106 if phrase in {"live chat", "send us a message", "message us", "whatsapp"}:3107 score += 33108 elif phrase in {"chat", "need help", "customer support", "support", "talk to us", "help"}:3109 score += 23110 elif phrase == "contact us":3111 score += 23112 else:3113 score += 13114
3115 if launcher.get("tag") == "button":3116 score += 23117 if launcher.get("aria_label"):3118 score += 23119 if launcher.get("title"):3120 score += 13121 if launcher.get("role") == "button":3122 score += 13123 if launcher.get("id"):3124 score += 13125 if launcher.get("class_tokens"):3126 score += 13127 if launcher.get("data_attributes"):3128 score += 13129 if launcher.get("fixed_position"):3130 score += 23131 if launcher.get("corner_hint"):3132 score += 13133 if launcher.get("found_in_rendered_dom"):3134 score += 23135 if launcher.get("visible"):3136 score += 13137 if _is_widget_messaging_href(launcher.get("href") or ""):3138 score += 33139
3140 href = launcher.get("href") or ""3141 if launcher.get("tag") == "a" and href:3142 normalized_href = normalize_url(href, base=page_url)3143 if normalized_href and same_site(page_url, normalized_href):3144 path = (urlparse(normalized_href).path or "").lower()3145 if _is_contact_url_hint(path) and not launcher.get("fixed_position") and not launcher.get("corner_hint"):3146 score -= 33147 return score3148
3149
3150def _widget_launcher_is_actionable(candidate: Dict[str, Any]) -> bool:3151 tag = (candidate.get("tag") or "").lower()3152 role = (candidate.get("role") or "").lower()3153 href = candidate.get("href") or ""3154 text = _strip_text(3155 " ".join(3156 [3157 candidate.get("launcher_text") or "",3158 candidate.get("aria_label") or "",3159 candidate.get("title") or "",3160 ]3161 )3162 )3163 if len(text) > 120 and tag not in {"button", "a"} and role != "button" and not candidate.get("fixed_position"):3164 return False3165 if tag in {"button", "a"} or role == "button":3166 return True3167 if candidate.get("fixed_position") or candidate.get("corner_hint"):3168 return True3169 if _is_widget_messaging_href(href):3170 return True3171 return bool(_matching_intent_hints(text, WIDGET_SUPPORT_INTENT_HINTS | LEAD_GEN_WIDGET_ACTION_HINTS))3172
3173
3174def _is_probable_navigation_link(launcher: Dict[str, Any], *, page_url: str) -> bool:3175 if launcher.get("tag") != "a":3176 return False3177 href = launcher.get("href") or ""3178 if not href or _is_widget_messaging_href(href) or href.lower().startswith(("javascript:", "#")):3179 return False3180 normalized_href = normalize_url(href, base=page_url)3181 if not normalized_href or not same_site(page_url, normalized_href):3182 return False3183 path = (urlparse(normalized_href).path or "").lower()3184 return _is_contact_url_hint(path) or any(token in path for token in ("/contact", "/support", "/help"))3185
3186
3187def _build_widget_launcher_hint(raw_hint: Dict[str, Any], *, page_url: str) -> Optional[Dict[str, Any]]:3188 tag = _strip_text(raw_hint.get("tag") or "").lower()3189 if tag not in WIDGET_LAUNCHER_TAGS:3190 return None3191
3192 data_attributes = {3193 key: _truncate_hint_text(str(value))3194 for key, value in (raw_hint.get("data_attributes") or {}).items()3195 if _is_human_readable_attr(key, _strip_text(str(value)))3196 }3197 hint = {3198 "tag": tag,3199 "launcher_text": _truncate_hint_text(raw_hint.get("launcher_text") or raw_hint.get("text") or ""),3200 "aria_label": _truncate_hint_text(raw_hint.get("aria_label") or ""),3201 "title": _truncate_hint_text(raw_hint.get("title") or ""),3202 "role": _strip_text(raw_hint.get("role") or ""),3203 "id": _strip_text(raw_hint.get("id") or ""),3204 "class_tokens": _sorted_unique_strings(raw_hint.get("class_tokens") or [])[:10],3205 "data_attributes": data_attributes,3206 "exact_css_clue": _strip_text(raw_hint.get("exact_css_clue") or raw_hint.get("css_path") or ""),3207 "href": _strip_text(raw_hint.get("href") or ""),3208 "found_in_rendered_dom": bool(raw_hint.get("found_in_rendered_dom")),3209 "visible": bool(raw_hint.get("visible")) if raw_hint.get("visible") is not None else False,3210 }3211
3212 fixed_position = bool(raw_hint.get("fixed_position"))3213 corner_hint = _strip_text(raw_hint.get("corner_hint") or "")3214 if not fixed_position or not corner_hint:3215 descriptor = " ".join(3216 [3217 hint["launcher_text"],3218 hint["aria_label"],3219 hint["title"],3220 hint["id"],3221 " ".join(hint["class_tokens"]),3222 " ".join(data_attributes.values()),3223 ]3224 )3225 static_fixed, static_corner = _static_widget_position_hints(raw_hint.get("node")) if raw_hint.get("node") is not None else (False, "")3226 fixed_position = fixed_position or static_fixed3227 corner_hint = corner_hint or static_corner or _infer_corner_hint(descriptor)3228 hint["fixed_position"] = fixed_position3229 hint["corner_hint"] = corner_hint3230
3231 combined = " ".join(3232 [3233 hint["launcher_text"],3234 hint["aria_label"],3235 hint["title"],3236 hint["role"],3237 hint["id"],3238 " ".join(hint["class_tokens"]),3239 " ".join(data_attributes.values()),3240 hint["href"],3241 ]3242 )3243 phrase_hits = _widget_phrase_hits(combined)3244 attr_term_hits = sorted({term for term in WIDGET_MATCH_TERMS if term in combined.lower()})3245 hint["phrase_hits"] = phrase_hits3246 hint["attr_term_hits"] = attr_term_hits3247 hint["score"] = _widget_launcher_score(hint, page_url=page_url)3248
3249 if not _widget_launcher_has_locator_data(hint):3250 return None3251 if not phrase_hits and not attr_term_hits and not _is_widget_messaging_href(hint.get("href") or ""):3252 return None3253 if _is_probable_navigation_link(hint, page_url=page_url) and hint["score"] < 5:3254 return None3255 if hint["score"] < 3:3256 return None3257 return hint3258
3259
3260def _launcher_signature(launcher: Dict[str, Any]) -> str:3261 return "|".join(3262 [3263 launcher.get("tag") or "",3264 launcher.get("id") or "",3265 " ".join(launcher.get("class_tokens") or []),3266 launcher.get("aria_label") or "",3267 launcher.get("title") or "",3268 launcher.get("launcher_text") or "",3269 launcher.get("exact_css_clue") or "",3270 ]3271 ).lower()3272
3273
3274def _dedupe_widget_launchers(launchers: List[Dict[str, Any]], *, page_url: str) -> List[Dict[str, Any]]:3275 deduped: Dict[str, Dict[str, Any]] = {}3276 for raw_hint in launchers:3277 hint = _build_widget_launcher_hint(raw_hint, page_url=page_url)3278 if not hint:3279 continue3280 signature = _launcher_signature(hint)3281 existing = deduped.get(signature)3282 if not existing or hint.get("score", 0) > existing.get("score", 0):3283 deduped[signature] = hint3284 return sorted(deduped.values(), key=lambda hint: (-hint.get("score", 0), signature if (signature := _launcher_signature(hint)) else ""))3285
3286
3287def _extract_widget_launcher_hints(soup: BeautifulSoup, page_url: str, *, found_in_rendered_dom: bool = False) -> List[Dict[str, Any]]:3288 launchers: List[Dict[str, Any]] = []3289 for node in soup.find_all(list(WIDGET_LAUNCHER_TAGS)):3290 text = _truncate_hint_text(node.get_text(" ", strip=True))3291 attrs = _node_human_readable_attrs(node)3292 fixed_position, corner_hint = _static_widget_position_hints(node)3293 launchers.append(3294 {3295 "node": node,3296 "tag": (node.name or "").lower(),3297 "launcher_text": text,3298 "aria_label": attrs.get("aria-label") or "",3299 "title": attrs.get("title") or "",3300 "role": _strip_text(node.get("role") or ""),3301 "id": _strip_text(node.get("id") or ""),3302 "class_tokens": _node_class_tokens(node),3303 "data_attributes": {key: value for key, value in attrs.items() if key.startswith("data-")},3304 "exact_css_clue": _css_selector(node),3305 "href": _strip_text(node.get("href") or ""),3306 "fixed_position": fixed_position,3307 "corner_hint": corner_hint,3308 "found_in_rendered_dom": found_in_rendered_dom,3309 "visible": False,3310 }3311 )3312 return _dedupe_widget_launchers(launchers, page_url=page_url)3313
3314
3315def _launcher_matches_provider(provider: str, cfg: Dict[str, Any], launcher: Dict[str, Any]) -> bool:3316 combined = " ".join(3317 [3318 launcher.get("launcher_text") or "",3319 launcher.get("aria_label") or "",3320 launcher.get("title") or "",3321 launcher.get("id") or "",3322 " ".join(launcher.get("class_tokens") or []),3323 " ".join((launcher.get("data_attributes") or {}).values()),3324 ]3325 ).lower()3326 return any(term in combined for term in _provider_widget_terms(provider, cfg))3327
3328
3329def _select_best_widget_launcher(launchers: List[Dict[str, Any]]) -> Optional[Dict[str, Any]]:3330 if not launchers:3331 return None3332 return max(3333 launchers,3334 key=lambda launcher: (3335 launcher.get("score", 0),3336 1 if launcher.get("found_in_rendered_dom") else 0,3337 1 if launcher.get("fixed_position") else 0,3338 len(launcher.get("launcher_text") or ""),3339 ),3340 )3341
3342
3343def _merge_widget_candidates(primary: Dict[str, Any], secondary: Dict[str, Any]) -> Dict[str, Any]:3344 merged = dict(primary)3345 for key in (3346 "matched_host",3347 "matched_script_host",3348 "matched_iframe_host",3349 "matched_inline",3350 "matched_selector",3351 "found_in_rendered_dom",3352 "fixed_position",3353 "has_locatable_launcher",3354 ):3355 merged[key] = bool(primary.get(key)) or bool(secondary.get(key))3356 for key in ("evidence_sources", "exact_selectors", "script_srcs", "iframe_srcs", "matched_by"):3357 merged[key] = _sorted_unique_strings([*(primary.get(key) or []), *(secondary.get(key) or [])])3358 merged["selector_hits"] = max(int(primary.get("selector_hits") or 0), int(secondary.get("selector_hits") or 0))3359
3360 existing_launcher = {3361 field: merged.get(field)3362 for field in (3363 "launcher_text",3364 "aria_label",3365 "title",3366 "role",3367 "id",3368 "class_tokens",3369 "data_attributes",3370 "exact_css_clue",3371 "launcher_score",3372 "found_in_rendered_dom",3373 "fixed_position",3374 "corner_hint",3375 "tag",3376 )3377 }3378 candidate_launcher = {3379 field: secondary.get(field)3380 for field in (3381 "launcher_text",3382 "aria_label",3383 "title",3384 "role",3385 "id",3386 "class_tokens",3387 "data_attributes",3388 "exact_css_clue",3389 "launcher_score",3390 "found_in_rendered_dom",3391 "fixed_position",3392 "corner_hint",3393 "tag",3394 )3395 }3396 if (candidate_launcher.get("launcher_score") or 0) >= (existing_launcher.get("launcher_score") or 0):3397 merged.update({key: value for key, value in candidate_launcher.items() if value not in (None, "", [], {})})3398 return merged3399
3400
3401def _merge_widget_candidate_lists(*groups: List[Dict[str, Any]]) -> List[Dict[str, Any]]:3402 merged: Dict[str, Dict[str, Any]] = {}3403 for group in groups:3404 for candidate in group:3405 key = (candidate.get("provider") or "").lower()3406 if not key:3407 continue3408 existing = merged.get(key)3409 if existing is None:3410 merged[key] = candidate3411 else:3412 merged[key] = _merge_widget_candidates(existing, candidate)3413 return list(merged.values())3414
3415
3416def _url_host_matches(url: str, host_patterns: Iterable[str]) -> bool:3417 try:3418 host = urlparse(url).hostname or ""3419 except Exception:3420 return False3421 host = host.lower().removeprefix("www.")3422 if not host:3423 return False3424 url_lower = url.lower()3425 for pattern in host_patterns:3426 host_pattern = pattern.lower().replace("https://", "").replace("http://", "").lstrip("/")3427 if not host_pattern:3428 continue3429 if "/" in host_pattern:3430 if host_pattern in url_lower:3431 return True3432 host_pattern = host_pattern.split("/", 1)[0]3433 host_pattern = host_pattern.removeprefix("www.")3434 if host == host_pattern or host.endswith(f".{host_pattern}"):3435 return True3436 if "." not in host_pattern and host_pattern in host:3437 return True3438 if host_pattern in url_lower and any(ch == "." for ch in host_pattern):3439 return True3440 return False3441
3442
3443def extract_widget_candidates(3444 page_url: str,3445 raw_html: str,3446 soup: BeautifulSoup,3447 *,3448 rendered_snapshot: Optional[RenderedPageSnapshot] = None,3449) -> List[Dict[str, Any]]:3450 html_lower = raw_html.lower()3451 candidates: Dict[str, Dict[str, Any]] = {}3452 script_nodes = [(s, normalize_url(s.get("src") or "", base=page_url)) for s in soup.find_all("script", src=True)]3453 iframe_nodes = [(i, normalize_url(i.get("src") or "", base=page_url)) for i in soup.find_all("iframe", src=True)]3454 script_srcs = [src for _, src in script_nodes if src]3455 iframe_srcs = [src for _, src in iframe_nodes if src]3456 if rendered_snapshot:3457 page_url = normalize_url(rendered_snapshot.final_url or page_url) or page_url3458 script_srcs = _sorted_unique_strings([*script_srcs, *(rendered_snapshot.script_srcs or [])])3459 iframe_srcs = _sorted_unique_strings([*iframe_srcs, *(rendered_snapshot.iframe_srcs or [])])3460
3461 launchers = _extract_widget_launcher_hints(soup, page_url, found_in_rendered_dom=bool(rendered_snapshot))3462 if rendered_snapshot:3463 launchers = _dedupe_widget_launchers([*launchers, *(rendered_snapshot.launchers or [])], page_url=page_url)3464 generic_launchers = [launcher for launcher in launchers if launcher.get("score", 0) >= 4]3465 generic_best = _select_best_widget_launcher(generic_launchers)3466 provider_detected = False3467
3468 for provider, cfg in WIDGET_PROVIDERS.items():3469 evidences: List[str] = []3470 exact_selectors: List[str] = []3471 matched_by: List[str] = []3472 matched_inline_tokens: List[str] = []3473 matched_selector = False3474 matched_inline = False3475 matched_host = False3476 matched_script_host = False3477 matched_iframe_host = False3478
3479 for node, src in script_nodes:3480 if not src:3481 continue3482 if _url_host_matches(src, cfg["host"]):3483 matched_host = True3484 matched_script_host = True3485 matched_by.append("script_host")3486 evidences.append(f"script host:{src}")3487 selector = _css_selector(node)3488 if selector:3489 exact_selectors.append(selector)3490
3491 for node, src in iframe_nodes:3492 if not src:3493 continue3494 if _url_host_matches(src, cfg["host"]):3495 matched_host = True3496 matched_iframe_host = True3497 matched_by.append("iframe_host")3498 evidences.append(f"iframe host:{src}")3499 selector = _css_selector(node)3500 if selector:3501 exact_selectors.append(selector)3502
3503 for src in script_srcs:3504 if _url_host_matches(src, cfg["host"]) and f"script host:{src}" not in evidences:3505 matched_host = True3506 matched_script_host = True3507 matched_by.append("script_host")3508 evidences.append(f"script host:{src}")3509
3510 for src in iframe_srcs:3511 if _url_host_matches(src, cfg["host"]) and f"iframe host:{src}" not in evidences:3512 matched_host = True3513 matched_iframe_host = True3514 matched_by.append("iframe_host")3515 evidences.append(f"iframe host:{src}")3516
3517 for token in cfg["inline"]:3518 if token.lower() in html_lower:3519 matched_inline = True3520 matched_inline_tokens.append(token)3521 matched_by.append("inline_token")3522 evidences.append(f"inline token:{token}")3523
3524 selector_hits = 03525 for selector in cfg["selector"]:3526 selector_re = re.compile(re.escape(selector), re.I)3527 matches = soup.find_all(3528 lambda tag: bool(3529 getattr(tag, "name", None)3530 and (3531 selector_re.search(_strip_text(tag.get("id") or ""))3532 or any(selector_re.search(_strip_text(str(cls))) for cls in (tag.get("class") or []))3533 )3534 )3535 )3536 if matches:3537 matched_selector = True3538 selector_hits += 13539 matched_by.append("selector")3540 evidences.append(f"selector:{selector}")3541 exact_selectors.extend(_css_selector(match) for match in matches[:3] if _css_selector(match))3542
3543 provider_launchers = [launcher for launcher in launchers if _launcher_matches_provider(provider, cfg, launcher)]3544 selected_launcher = _select_best_widget_launcher(provider_launchers)3545 provider_launcher_confirmed = bool(selected_launcher)3546 if not selected_launcher and (matched_host or matched_inline or matched_selector):3547 if provider not in LEAD_GEN_WIDGET_PROVIDERS:3548 selected_launcher = generic_best3549 elif generic_best and _widget_launcher_is_actionable(generic_best):3550 selected_launcher = generic_best3551 if selected_launcher:3552 matched_by.append("rendered_launcher" if selected_launcher.get("found_in_rendered_dom") else "static_launcher")3553 label = selected_launcher.get("launcher_text") or selected_launcher.get("aria_label") or selected_launcher.get("id") or selected_launcher.get("title")3554 if label:3555 evidences.append(f"launcher:{label}")3556 css_clue = selected_launcher.get("exact_css_clue")3557 if css_clue:3558 exact_selectors.append(css_clue)3559
3560 if not (matched_host or matched_inline or matched_selector or selected_launcher):3561 continue3562 provider_detected = True3563 candidates[provider] = {3564 "provider": provider,3565 "page_url": page_url,3566 "value": provider,3567 "evidence_sources": _sorted_unique_strings(evidences),3568 "exact_selectors": _sorted_unique_strings(exact_selectors),3569 "script_srcs": [src for src in script_srcs if _url_host_matches(src, cfg["host"])],3570 "iframe_srcs": [src for src in iframe_srcs if _url_host_matches(src, cfg["host"])],3571 "matched_by": _sorted_unique_strings(matched_by),3572 "matched_host": matched_host,3573 "matched_script_host": matched_script_host,3574 "matched_iframe_host": matched_iframe_host,3575 "matched_inline": matched_inline,3576 "matched_inline_tokens": _sorted_unique_strings(matched_inline_tokens),3577 "matched_selector": matched_selector,3578 "selector_hits": selector_hits,3579 "found_in_rendered_dom": bool(selected_launcher and selected_launcher.get("found_in_rendered_dom")),3580 "has_locatable_launcher": bool(selected_launcher and _widget_launcher_has_locator_data(selected_launcher)),3581 "provider_launcher_confirmed": provider_launcher_confirmed,3582 "launcher_score": int(selected_launcher.get("score") or 0) if selected_launcher else 0,3583 "launcher_text": selected_launcher.get("launcher_text") if selected_launcher else "",3584 "aria_label": selected_launcher.get("aria_label") if selected_launcher else "",3585 "title": selected_launcher.get("title") if selected_launcher else "",3586 "role": selected_launcher.get("role") if selected_launcher else "",3587 "id": selected_launcher.get("id") if selected_launcher else "",3588 "class_tokens": selected_launcher.get("class_tokens") if selected_launcher else [],3589 "data_attributes": selected_launcher.get("data_attributes") if selected_launcher else {},3590 "fixed_position": bool(selected_launcher and selected_launcher.get("fixed_position")),3591 "corner_hint": selected_launcher.get("corner_hint") if selected_launcher else "",3592 "tag": selected_launcher.get("tag") if selected_launcher else "",3593 "exact_css_clue": selected_launcher.get("exact_css_clue") if selected_launcher else "",3594 }3595
3596 if not provider_detected and generic_best:3597 candidates[GENERIC_WIDGET_PROVIDER] = {3598 "provider": GENERIC_WIDGET_PROVIDER,3599 "page_url": page_url,3600 "value": GENERIC_WIDGET_PROVIDER,3601 "evidence_sources": _sorted_unique_strings(3602 [3603 *(f"launcher phrase:{phrase}" for phrase in generic_best.get("phrase_hits") or []),3604 *(f"launcher attr:{term}" for term in generic_best.get("attr_term_hits") or []),3605 f"launcher:{generic_best.get('launcher_text') or generic_best.get('aria_label') or generic_best.get('id') or generic_best.get('title') or generic_best.get('tag')}",3606 ]3607 ),3608 "exact_selectors": _sorted_unique_strings([generic_best.get("exact_css_clue") or ""]),3609 "script_srcs": [],3610 "iframe_srcs": [],3611 "matched_by": ["rendered_launcher" if generic_best.get("found_in_rendered_dom") else "static_launcher"],3612 "matched_host": False,3613 "matched_script_host": False,3614 "matched_iframe_host": False,3615 "matched_inline": False,3616 "matched_selector": False,3617 "selector_hits": 0,3618 "found_in_rendered_dom": bool(generic_best.get("found_in_rendered_dom")),3619 "has_locatable_launcher": True,3620 "launcher_score": int(generic_best.get("score") or 0),3621 "launcher_text": generic_best.get("launcher_text") or "",3622 "aria_label": generic_best.get("aria_label") or "",3623 "title": generic_best.get("title") or "",3624 "role": generic_best.get("role") or "",3625 "id": generic_best.get("id") or "",3626 "class_tokens": generic_best.get("class_tokens") or [],3627 "data_attributes": generic_best.get("data_attributes") or {},3628 "fixed_position": bool(generic_best.get("fixed_position")),3629 "corner_hint": generic_best.get("corner_hint") or "",3630 "tag": generic_best.get("tag") or "",3631 "exact_css_clue": generic_best.get("exact_css_clue") or "",3632 }3633
3634 merged_candidates: Dict[str, Dict[str, Any]] = {}3635 for candidate in candidates.values():3636 key = (candidate.get("provider") or "").lower()3637 existing = merged_candidates.get(key)3638 if existing is None:3639 merged_candidates[key] = candidate3640 else:3641 merged_candidates[key] = _merge_widget_candidates(existing, candidate)3642 return list(merged_candidates.values())3643
3644
3645def _widget_locator_hints(candidate: Dict[str, Any]) -> Dict[str, Any]:3646 hints = {3647 "provider": candidate.get("provider"),3648 "matched_by": candidate.get("matched_by") or [],3649 "launcher_text": candidate.get("launcher_text"),3650 "aria_label": candidate.get("aria_label"),3651 "title": candidate.get("title"),3652 "role": candidate.get("role"),3653 "tag": candidate.get("tag"),3654 "id": candidate.get("id"),3655 "class_tokens": candidate.get("class_tokens") or [],3656 "data_attributes": candidate.get("data_attributes") or {},3657 "iframe_src": (candidate.get("iframe_srcs") or [None])[0],3658 "iframe_srcs": candidate.get("iframe_srcs") or [],3659 "script_srcs": candidate.get("script_srcs") or [],3660 "css_path_clues": candidate.get("exact_selectors") or [],3661 "found_in_rendered_dom": bool(candidate.get("found_in_rendered_dom")),3662 "fixed_position": bool(candidate.get("fixed_position")),3663 "corner_hint": candidate.get("corner_hint"),3664 }3665 return {key: value for key, value in hints.items() if value not in (None, "", [], {})}3666
3667
3668def _widget_sales_outreach_reasons(candidate: Dict[str, Any]) -> List[str]:3669 provider = _strip_text(candidate.get("provider") or "")3670 launcher_text = " ".join(3671 [3672 candidate.get("launcher_text") or "",3673 candidate.get("aria_label") or "",3674 candidate.get("title") or "",3675 ]3676 )3677 support_text = " ".join(3678 [3679 launcher_text,3680 candidate.get("role") or "",3681 candidate.get("id") or "",3682 " ".join(candidate.get("class_tokens") or []),3683 " ".join(str(value) for value in (candidate.get("data_attributes") or {}).values()),3684 ]3685 )3686 page_url = candidate.get("page_url") or ""3687 provider_context = " ".join(3688 [3689 page_url,3690 launcher_text,3691 " ".join(candidate.get("script_srcs") or []),3692 " ".join(candidate.get("iframe_srcs") or []),3693 ]3694 )3695
3696 reasons: List[str] = []3697 if _matching_intent_hints(launcher_text, FORM_SALES_FLOW_HINTS):3698 reasons.append("sales_launcher_copy")3699 if provider in LEAD_GEN_WIDGET_PROVIDERS and _matching_intent_hints(page_url, FORM_SALES_FLOW_HINTS):3700 reasons.append("lead_gen_provider_on_sales_page")3701 if provider in LEAD_GEN_WIDGET_PROVIDERS and _matching_intent_hints(provider_context, FORM_SALES_FLOW_HINTS | {"book a meeting", "schedule a meeting"}):3702 reasons.append("sales_scheduling_widget")3703 if reasons and not _matching_intent_hints(support_text, WIDGET_SUPPORT_INTENT_HINTS):3704 return sorted(set(reasons))3705 if "sales_launcher_copy" in reasons or "sales_scheduling_widget" in reasons:3706 return sorted(set(reasons))3707 return []3708
3709
3710def validate_widget(candidate: Dict[str, Any]) -> Optional[Dict[str, Any]]:3711 provider = candidate.get("provider")3712 if not provider:3713 candidate["reject_reason"] = "generic_widget"3714 return None3715
3716 evidence_sources = candidate.get("evidence_sources", [])3717 matched_host = bool(candidate.get("matched_host"))3718 matched_script_host = bool(candidate.get("matched_script_host"))3719 matched_iframe_host = bool(candidate.get("matched_iframe_host"))3720 matched_inline = bool(candidate.get("matched_inline"))3721 matched_selector = bool(candidate.get("matched_selector"))3722 has_launcher = bool(candidate.get("has_locatable_launcher"))3723 launcher_score = int(candidate.get("launcher_score") or 0)3724 actionable_launcher = _widget_launcher_is_actionable(candidate)3725 strong_launcher = has_launcher and launcher_score >= 4 and actionable_launcher3726 provider_host_confirmed = matched_script_host or matched_iframe_host3727 provider_dom_confirmed = matched_selector or matched_iframe_host3728 provider_launcher_confirmed = bool(candidate.get("provider_launcher_confirmed"))3729 is_lead_gen_provider = provider in LEAD_GEN_WIDGET_PROVIDERS3730
3731 if not (matched_host or matched_inline or matched_selector or strong_launcher):3732 candidate["reject_reason"] = "generic_widget"3733 return None3734
3735 sales_outreach_reasons = _widget_sales_outreach_reasons(candidate)3736 if sales_outreach_reasons:3737 candidate["reject_reason"] = "sales_outreach_widget"3738 candidate["sales_outreach_signals"] = sales_outreach_reasons3739 return None3740
3741 if is_lead_gen_provider:3742 if provider_host_confirmed and provider_dom_confirmed:3743 band = "VERIFIED"3744 elif provider_host_confirmed or matched_selector:3745 band = "STRONG"3746 elif (matched_inline or provider_launcher_confirmed) and strong_launcher:3747 band = "STRONG"3748 else:3749 candidate["reject_reason"] = "generic_widget"3750 return None3751 else:3752 if provider_host_confirmed and (has_launcher or provider_dom_confirmed):3753 band = "VERIFIED"3754 elif matched_selector and strong_launcher:3755 band = "VERIFIED"3756 elif provider_host_confirmed:3757 band = "STRONG"3758 elif strong_launcher and (matched_inline or matched_selector):3759 band = "STRONG"3760 elif strong_launcher:3761 band = "PLAUSIBLE"3762 else:3763 candidate["reject_reason"] = "generic_widget"3764 return None3765
3766 matched_by = candidate.get("matched_by") or []3767 validator_passed = []3768 if matched_script_host:3769 validator_passed.append("provider_script_host")3770 if matched_iframe_host:3771 validator_passed.append("provider_iframe_host")3772 if matched_inline:3773 validator_passed.append("provider_inline_token")3774 if matched_selector:3775 validator_passed.append("provider_dom_selector")3776 if has_launcher:3777 validator_passed.append("rendered_launcher" if candidate.get("found_in_rendered_dom") else "launcher_locator")3778 if provider_launcher_confirmed:3779 validator_passed.append("provider_launcher_match")3780 if candidate.get("fixed_position"):3781 validator_passed.append("fixed_position_launcher")3782 if candidate.get("corner_hint"):3783 validator_passed.append(f"corner:{candidate.get('corner_hint')}")3784
3785 return {3786 "type": "contact_widget",3787 "value": provider,3788 "page_url": candidate["page_url"],3789 "evidence": "; ".join(evidence_sources),3790 "confidence_band": band,3791 "validator_passed": validator_passed or ["widget_locator"],3792 "provider": provider,3793 "matched_hostname_or_selector": "host" if matched_host else ("inline" if matched_inline else ("selector" if matched_selector else "launcher")),3794 "matched_by": matched_by,3795 "exact_selectors": candidate.get("exact_selectors", []),3796 "widget_locator_hints": _widget_locator_hints(candidate),3797 "launcher_text": candidate.get("launcher_text"),3798 "aria_label": candidate.get("aria_label"),3799 "title": candidate.get("title"),3800 "role": candidate.get("role"),3801 "id": candidate.get("id"),3802 "class_tokens": candidate.get("class_tokens", []),3803 "data_attributes": candidate.get("data_attributes", {}),3804 "script_srcs": candidate.get("script_srcs", []),3805 "iframe_srcs": candidate.get("iframe_srcs", []),3806 "found_in_rendered_dom": bool(candidate.get("found_in_rendered_dom")),3807 "fixed_position": bool(candidate.get("fixed_position")),3808 "corner_hint": candidate.get("corner_hint"),3809 }3810
3811
3812def _run_widget_detection_self_checks() -> None:3813 livechat_static = BeautifulSoup(3814 '<html><body><script src="https://wordpress.livechat.com/api/v2/script/test/widget.js"></script></body></html>',3815 "html.parser",3816 )3817 host_only_candidates = extract_widget_candidates("https://example.com", str(livechat_static), livechat_static)3818 validated_host_only = [validate_widget(candidate) for candidate in host_only_candidates]3819 assert any(item and item.get("provider") == "LiveChat" and item.get("confidence_band") == "STRONG" for item in validated_host_only)3820
3821 intercom_iframe = BeautifulSoup(3822 '<html><body><iframe src="https://widget.intercom.io/widget/demo"></iframe></body></html>',3823 "html.parser",3824 )3825 iframe_candidates = extract_widget_candidates("https://example.com", str(intercom_iframe), intercom_iframe)3826 validated_iframe = [validate_widget(candidate) for candidate in iframe_candidates]3827 assert any(item and item.get("provider") == "Intercom" and item.get("confidence_band") == "VERIFIED" for item in validated_iframe)3828
3829 generic_launcher = BeautifulSoup(3830 '<html><body><button aria-label="Open support chat" class="floating-help-button fixed bottom-right">Need help?</button></body></html>',3831 "html.parser",3832 )3833 generic_candidates = extract_widget_candidates("https://example.com", str(generic_launcher), generic_launcher)3834 validated_generic = [validate_widget(candidate) for candidate in generic_candidates]3835 assert any(item and item.get("provider") == GENERIC_WIDGET_PROVIDER and item.get("confidence_band") == "PLAUSIBLE" for item in validated_generic)3836
3837 sales_launcher = BeautifulSoup(3838 '<html><body><button aria-label="Contact Sales" class="floating-help-button fixed bottom-right">Contact Sales</button></body></html>',3839 "html.parser",3840 )3841 sales_launcher_candidates = extract_widget_candidates("https://example.com/contact-sales", str(sales_launcher), sales_launcher)3842 validated_sales_launcher = [validate_widget(candidate) for candidate in sales_launcher_candidates]3843 assert not any(item for item in validated_sales_launcher), "sales launchers must not validate as contact widgets"3844
3845 calendly_sales = BeautifulSoup(3846 '<html><body><iframe src="https://calendly.com/acme/request-demo"></iframe></body></html>',3847 "html.parser",3848 )3849 calendly_sales_candidates = extract_widget_candidates("https://example.com/contact-sales", str(calendly_sales), calendly_sales)3850 validated_calendly_sales = [validate_widget(candidate) for candidate in calendly_sales_candidates]3851 assert not any(item for item in validated_calendly_sales), "sales scheduling widgets must be rejected"3852
3853 calendly_inline = BeautifulSoup(3854 '<html><body><script>Calendly.initInlineWidget({url:"https://calendly.com/acme/support"});</script><button>Book a support call</button></body></html>',3855 "html.parser",3856 )3857 calendly_inline_candidates = extract_widget_candidates("https://example.com/help", str(calendly_inline), calendly_inline)3858 validated_calendly_inline = [validate_widget(candidate) for candidate in calendly_inline_candidates]3859 assert any(item and item.get("provider") == "Calendly" for item in validated_calendly_inline), "inline Calendly CTA should still validate"3860
3861 calendly_false_positive = BeautifulSoup(3862 '<html><body><script>window.__widgetCache={"provider":"calendly"}</script><div id="details">Tight, body-hugging fit Extra stretchable fabric with built-in support.</div></body></html>',3863 "html.parser",3864 )3865 calendly_false_positive_candidates = extract_widget_candidates(3866 "https://example.com/products/support-top",3867 str(calendly_false_positive),3868 calendly_false_positive,3869 )3870 validated_calendly_false_positive = [validate_widget(candidate) for candidate in calendly_false_positive_candidates]3871 assert not any(item for item in validated_calendly_false_positive), "generic inline mentions must not validate as Calendly widgets"3872
3873
3874if __debug__:3875 _run_widget_detection_self_checks()3876
3877
3878def extract_social_candidates(page_url: str, soup: BeautifulSoup) -> List[Dict[str, Any]]:3879 results: Dict[str, Dict[str, Any]] = {}3880 for a in soup.find_all("a", href=True):3881 raw_href = (a.get("href") or "").strip()3882 if not raw_href or raw_href.startswith("mailto:") or raw_href.startswith("tel:"):3883 continue3884 absolute = normalize_url(raw_href, base=page_url)3885 if not absolute:3886 continue3887 parsed = urlparse(absolute)3888 if not parsed.netloc:3889 continue3890 host = parsed.netloc.lower()3891 host_no_www = host.removeprefix("www.")3892
3893 network = None3894 for name, suffixes in SOCIAL_DOMAINS.items():3895 if any(host_no_www == domain or host_no_www.endswith(domain) or domain in host_no_www for domain in suffixes):3896 if any(token in host_no_www for token in {"assets", "cdn", "static"}):3897 continue3898 network = name3899 break3900 if not network:3901 continue3902
3903 anchor_text = _strip_text(" ".join(a.stripped_strings))3904 aria = (a.get("aria-label") or "").strip()3905 title = (a.get("title") or "").strip()3906 context_nodes = []3907 for parent in list(a.parents)[:5]:3908 if not parent or not hasattr(parent, "name") or not parent.name:3909 continue3910 context_nodes.append(parent.name.lower())3911 classes = " ".join(parent.get("class", [])) if hasattr(parent, "get") else ""3912 context_nodes.append(classes.lower())3913 if parent.get("id"):3914 context_nodes.append(parent.get("id").lower())3915 context = " ".join(context_nodes)3916 snippet = _strip_text(3917 " ".join(3918 token3919 for token in [anchor_text, aria, title, context]3920 if token3921 )[:240]3922 )3923
3924 results[absolute] = {3925 "network": network,3926 "value": absolute,3927 "page_url": page_url,3928 "anchor_text": anchor_text,3929 "aria": aria,3930 "title": title,3931 "snippet": snippet,3932 "context": context.lower(),3933 }3934
3935 return list(results.values())3936
3937
3938def validate_social_profile(candidate: Dict[str, Any], page_ctx: Dict[str, Any]) -> Optional[Dict[str, Any]]:3939 network = candidate.get("network")3940 url = candidate.get("value") or ""3941 if not network or not url:3942 candidate["reject_reason"] = "policy_social_link"3943 return None3944
3945 parsed = urlparse(url)3946 path = (parsed.path or "").lower().strip("/")3947 lowered = url.lower()3948
3949 blocked_social_patterns = (3950 "policy",3951 "/share",3952 "/intent",3953 "/status/",3954 "/watch",3955 "/watch?v=",3956 "/help",3957 "/posts/",3958 "/video",3959 "/reel/",3960 "/p/",3961 "/hashtag/",3962 )3963 if any(tok in lowered for tok in blocked_social_patterns):3964 candidate["reject_reason"] = "policy_social_link"3965 return None3966
3967 if network == "whatsapp":3968 if not any(tok in lowered for tok in ("wa.me", "api.whatsapp.com", "chat.whatsapp.com")):3969 candidate["reject_reason"] = "policy_social_link"3970 return None3971 if not any(w in lowered for w in ("wa.me/", "/phone=")):3972 candidate["reject_reason"] = "weak_page_match"3973 return None3974 band = "VERIFIED"3975 validator = ["direct_whatsapp_target"]3976 elif network == "telegram":3977 if not any(tok in lowered for tok in ("t.me", "telegram.me", "telegram.org")):3978 candidate["reject_reason"] = "policy_social_link"3979 return None3980 band = "VERIFIED"3981 validator = ["direct_telegram_target"]3982 elif network == "messenger":3983 if "m.me/" not in lowered and "messages/" not in lowered and "messenger.com" not in lowered and "dialog/send" not in lowered:3984 candidate["reject_reason"] = "policy_social_link"3985 return None3986 band = "VERIFIED"3987 validator = ["direct_messenger_target"]3988 elif network == "linkedin":3989 if not re.match(r"^/company/|^/in/|^/school/|^/organization/", f"/{path}"):3990 candidate["reject_reason"] = "weak_page_match"3991 return None3992 band = "STRONG"3993 validator = ["linkedin_business_profile"]3994 elif network == "instagram":3995 if any(segment in path for segment in ("p/", "reel/", "reels/", "explore/", "stories/", "tv/")):3996 candidate["reject_reason"] = "weak_page_match"3997 return None3998 if len(path.split("/")) > 2:3999 candidate["reject_reason"] = "weak_page_match"4000 return None4001 band = "STRONG"4002 validator = ["instagram_profile_path"]4003 elif network == "youtube":4004 if any(segment in path for segment in ("watch", "playlist", "shorts", "results")):4005 candidate["reject_reason"] = "policy_social_link"4006 return None4007 if not (path.startswith("channel/") or path.startswith("c/") or path.startswith("user/") or path.startswith("@")):4008 candidate["reject_reason"] = "weak_page_match"4009 return None4010 band = "STRONG"4011 validator = ["youtube_channel"]4012 elif network == "facebook":4013 if any(segment in path for segment in ("policy", "help", "share", "sharer", "watch")):4014 candidate["reject_reason"] = "policy_social_link"4015 return None4016 if not path:4017 candidate["reject_reason"] = "weak_page_match"4018 return None4019 band = "STRONG"4020 validator = ["facebook_profile_or_page"]4021 elif network == "x":4022 if "/intent" in lowered or "/share" in lowered or "/status/" in lowered:4023 candidate["reject_reason"] = "policy_social_link"4024 return None4025 if path and path.split("/")[0] and len(path.split("/")) <= 2 and not path.startswith("i/"):4026 band = "STRONG"4027 validator = ["x_profile"]4028 else:4029 candidate["reject_reason"] = "weak_page_match"4030 return None4031 else:4032 candidate["reject_reason"] = "weak_page_match"4033 return None4034
4035 # require block/label context or contact-like page context.4036 ctx = (candidate.get("snippet") or "").lower() + " " + candidate.get("context", "")4037 if not any(token in ctx for token in ("footer", "header", "nav", "contact", "support", "about", "social", "reach", "get in touch")):4038 if not (page_ctx.get("is_contact_page") or _has_contact_phrase(ctx)):4039 candidate["reject_reason"] = "policy_social_link"4040 return None4041
4042 return {4043 "type": "social_profile",4044 "value": url,4045 "page_url": candidate["page_url"],4046 "evidence": f"network:{network}; profile:{url}; context:{ctx[:180]}",4047 "confidence_band": band,4048 "validator_passed": validator,4049 "network": network,4050 }4051
4052
4053def _canonical_contact_page_score(soup: BeautifulSoup, page_url: str, forms: List[Dict[str, Any]], channels: List[Dict[str, Any]]) -> Tuple[int, List[str]]:4054 score = 04055 reasons: List[str] = []4056
4057 if any(item.get("type") == "contact_form" and item.get("confidence_band") in {"VERIFIED", "STRONG"} for item in forms):4058 score += 454059 reasons.append("validated form")4060
4061 if any(item.get("type") in {"email", "phone"} and item.get("confidence_band") in {"VERIFIED", "STRONG"} for item in channels):4062 score += 354063 reasons.append("direct channel")4064
4065 title = _strip_text(soup.title.string) if soup.title and soup.title.string else ""4066 headings = " ".join(_strip_text(h.get_text(" ", strip=True)) for h in soup.select("h1,h2"))4067 canonical_text = " ".join(_strip_text(x) for x in [page_url, title, headings, candidate_crumbs(soup), soup.get_text(" ", strip=True)[:600]])4068
4069 if _has_contact_phrase(title) or _has_contact_phrase(headings):4070 score += 204071 reasons.append("contact heading/title")4072
4073 if _has_contact_phrase(canonical_text):4074 score += 204075 reasons.append("contact phrase in body/canonical")4076
4077 if "contact" in (urlparse(page_url).path or "").lower():4078 score += 104079 reasons.append("contact path")4080
4081 return score, reasons4082
4083
4084def _build_contact_probe_urls(start_url: str) -> List[str]:4085 parsed = urlparse(start_url)4086 if not (parsed.scheme and parsed.netloc):4087 return []4088
4089 root = f"{parsed.scheme}://{parsed.netloc}"4090 locale_prefixes = ["/"]4091 path_parts = [p for p in (parsed.path or "").strip("/").split("/") if p]4092 if path_parts and re.fullmatch(r"[a-z]{2}(?:-[a-z]{2})?", path_parts[0], flags=re.I):4093 locale_prefixes.append(f"/{path_parts[0]}")4094
4095 seen = set()4096 candidates: List[str] = []4097 for prefix in locale_prefixes:4098 base = root + prefix4099 for path in CONTACT_URL_PROBES:4100 url = normalize_url(urljoin(base + "/", path))4101 if not url:4102 continue4103 if url not in seen:4104 seen.add(url)4105 candidates.append(url)4106 return candidates4107
4108
4109def candidate_crumbs(soup: BeautifulSoup) -> str:4110 # Optional breadcrumb text for E and D checks.4111 crumbs = []4112 for nav in soup.select("nav[aria-label='breadcrumb'], .breadcrumb, .breadcrumbs"):4113 crumbs.append(_strip_text(nav.get_text(" ", strip=True)))4114 return " ".join(crumbs)4115
4116
4117def validate_contact_page(4118 page_url: str,4119 soup: BeautifulSoup,4120 forms: List[Dict[str, Any]],4121 emails: List[Dict[str, Any]],4122 phones: List[Dict[str, Any]],4123 widgets: List[Dict[str, Any]],4124 socials: List[Dict[str, Any]],4125 include_plausible: bool = False,4126) -> Optional[Dict[str, Any]]:4127 score, reasons = _canonical_contact_page_score(soup, page_url, forms + [], emails + phones + widgets + socials)4128 if score >= 80:4129 band = "VERIFIED"4130 elif score >= 55:4131 band = "STRONG"4132 elif include_plausible and score >= 35:4133 band = "PLAUSIBLE"4134 else:4135 return None4136
4137 return {4138 "type": "contact_page",4139 "value": page_url,4140 "page_url": page_url,4141 "evidence": "; ".join(sorted(set(reasons))) if reasons else "contact signals present",4142 "confidence_band": band,4143 "validator_passed": reasons or ["contact_signals"],4144 }4145
4146
4147def dedupe_results(items: List[Dict[str, Any]]) -> List[Dict[str, Any]]:4148 buckets: Dict[Tuple[str, str], Dict[str, Any]] = {}4149
4150 def rank(item: Dict[str, Any]) -> int:4151 if item.get("type") == "email":4152 return _email_dedupe_weight(item)4153 if item.get("type") == "phone":4154 return _phone_dedupe_weight(item)4155 return CONFIDENCE_ORDER.get(item.get("confidence_band", ""), 0)4156
4157 for item in items:4158 item_type = item.get("type", "")4159 value = item.get("value", "")4160 if not item_type or value == "":4161 continue4162 if item_type == "contact_form":4163 value = item.get("form_signature", item.get("value", value))4164 key = (item_type, value)4165 elif item_type == "contact_widget":4166 key = (item_type, item.get("provider", value))4167 else:4168 key = (item_type, value)4169
4170 existing = buckets.get(key)4171 if not existing:4172 buckets[key] = dict(item)4173 buckets[key]["supporting_pages"] = []4174 continue4175
4176 if rank(item) >= rank(existing):4177 existing_pages = set(existing.get("supporting_pages", []))4178 existing_pages.add(item["page_url"])4179 existing["supporting_pages"] = sorted(existing_pages)4180 if item["page_url"] not in existing.get("supporting_pages", []):4181 existing["supporting_pages"].append(item["page_url"])4182 # Keep stronger/ later evidence.4183 if rank(item) > rank(existing) or len(item.get("evidence", "")) > len(existing.get("evidence", "")):4184 existing.update({k: v for k, v in item.items() if k not in {"supporting_pages"}})4185 existing["supporting_pages"] = sorted(set(existing.get("supporting_pages", [])))4186 else:4187 existing_pages = set(existing.get("supporting_pages", []))4188 existing_pages.add(item["page_url"])4189 existing["supporting_pages"] = sorted(existing_pages)4190
4191 return list(buckets.values())4192
4193
4194def _group_item_rank(item: Dict[str, Any]) -> int:4195 if item.get("type") == "email":4196 return _email_dedupe_weight(item)4197 if item.get("type") == "phone":4198 return _phone_dedupe_weight(item)4199 return CONFIDENCE_ORDER.get(item.get("confidence_band", ""), 0) * 1004200
4201
4202def _pick_stronger_group_item(existing: Dict[str, Any], candidate: Dict[str, Any]) -> Dict[str, Any]:4203 existing_rank = _group_item_rank(existing)4204 candidate_rank = _group_item_rank(candidate)4205 if candidate_rank > existing_rank:4206 return candidate4207 if candidate_rank == existing_rank:4208 if len(candidate.get("evidence", "")) > len(existing.get("evidence", "")):4209 return candidate4210 if json.dumps(candidate, sort_keys=True, default=str) < json.dumps(existing, sort_keys=True, default=str):4211 return candidate4212 return existing4213
4214
4215def _group_page_url_key(item: Dict[str, Any]) -> str:4216 raw = (item.get("page_url") or "").strip()4217 return normalize_url(raw) or raw4218
4219
4220def _group_channel_key(item: Dict[str, Any]) -> Optional[Tuple[str, str]]:4221 item_type = item.get("type")4222 if item_type == "email":4223 value = (item.get("value") or "").strip().lower()4224 return (item_type, value) if value else None4225 if item_type == "phone":4226 value = (item.get("normalized") or item.get("value") or "").strip()4227 return (item_type, value) if value else None4228 if item_type == "contact_form":4229 signature = (item.get("form_signature") or "").strip()4230 if signature:4231 return (item_type, signature)4232 action = (item.get("action") or "").strip()4233 method = (item.get("method") or "GET").strip().upper()4234 page_url = _group_page_url_key(item)4235 if not action and not page_url:4236 return None4237 return (item_type, f"{action}|{method}|{page_url}")4238 if item_type == "social_profile":4239 network = (item.get("network") or "").strip().lower()4240 value = (item.get("value") or "").strip()4241 if not network or not value:4242 return None4243 return (item_type, f"{network}|{value}")4244 if item_type == "contact_widget":4245 provider = (item.get("provider") or "").strip().lower()4246 page_url = _group_page_url_key(item)4247 if not provider or not page_url:4248 return None4249 return (item_type, f"{provider}|{page_url}")4250 return None4251
4252
4253def _channel_sort_key(item_type: str, item: Dict[str, Any]) -> Tuple[Any, ...]:4254 base = (-_group_item_rank(item),)4255 if item_type == "email":4256 return base + (4257 (item.get("value") or "").lower(),4258 item.get("source_type") or "",4259 item.get("evidence") or "",4260 )4261 if item_type == "phone":4262 return base + (4263 item.get("normalized") or item.get("value") or "",4264 item.get("type_hint") or "",4265 item.get("source_type") or "",4266 item.get("evidence") or "",4267 )4268 if item_type == "contact_form":4269 return base + (4270 item.get("action") or "",4271 item.get("method") or "",4272 item.get("form_signature") or "",4273 )4274 if item_type == "social_profile":4275 return base + (4276 item.get("network") or "",4277 item.get("value") or "",4278 item.get("evidence") or "",4279 )4280 if item_type == "contact_widget":4281 return base + (4282 item.get("provider") or "",4283 item.get("matched_hostname_or_selector") or "",4284 item.get("evidence") or "",4285 )4286 return base + (json.dumps(item, sort_keys=True, default=str),)4287
4288
4289def _grouped_email_record(item: Dict[str, Any]) -> Dict[str, Any]:4290 return {"value": item.get("value")}4291
4292
4293def _grouped_phone_record(item: Dict[str, Any]) -> Dict[str, Any]:4294 return {"value": item.get("normalized") or item.get("value")}4295
4296
4297def _same_normalized_url(left: Any, right: Any) -> bool:4298 left_raw = (left or "").strip() if isinstance(left, str) else ""4299 right_raw = (right or "").strip() if isinstance(right, str) else ""4300 if not left_raw or not right_raw:4301 return False4302 left_norm = normalize_url(left_raw) or left_raw4303 right_norm = normalize_url(right_raw) or right_raw4304 return left_norm == right_norm4305
4306
4307def _clean_form_locator_hints(4308 locator_hints: Dict[str, Any],4309 *,4310 parent_page_url: str,4311 form_action: Optional[str] = None,4312) -> Dict[str, Any]:4313 cleaned = dict(locator_hints or {})4314 hint_action = cleaned.get("action")4315 if isinstance(hint_action, str) and hint_action:4316 if _same_normalized_url(hint_action, parent_page_url) or (form_action and _same_normalized_url(hint_action, form_action)):4317 cleaned.pop("action", None)4318
4319 action_path = cleaned.get("action_path")4320 if isinstance(action_path, str) and action_path:4321 parent_path = (urlparse(parent_page_url).path or "").strip() if parent_page_url else ""4322 if not cleaned.get("action") and action_path.strip() == parent_path:4323 cleaned.pop("action_path", None)4324 return cleaned4325
4326
4327def _grouped_contact_form_record(item: Dict[str, Any], parent_page_url: str) -> Dict[str, Any]:4328 action = item.get("action")4329 action_matches_parent = bool(action and _same_normalized_url(action, parent_page_url))4330 record = {4331 "method": item.get("method"),4332 "form_signature": item.get("form_signature"),4333 "resolution_method": item.get("resolution_method") or "snapshot_ref",4334 "usage_notes": item.get("usage_notes") or SNAPSHOT_REF_USAGE_NOTES,4335 "form_locator_hints": _clean_form_locator_hints(4336 item.get("form_locator_hints", {}),4337 parent_page_url=parent_page_url,4338 form_action=action,4339 ),4340 "submit_locator_hints": item.get("submit_locator_hints", {}),4341 "field_locator_hints": item.get("field_locator_hints", {}),4342 }4343 if action and not action_matches_parent:4344 record["action"] = action4345 elif action_matches_parent:4346 record["action_is_page_url"] = True4347 return {key: value for key, value in record.items() if value not in (None, "", [], {})}4348
4349
4350def _grouped_social_profile_record(item: Dict[str, Any]) -> Dict[str, Any]:4351 return {4352 "network": item.get("network"),4353 "value": item.get("value"),4354 }4355
4356
4357def _clean_widget_locator_hints(locator_hints: Dict[str, Any], *, parent_page_url: str) -> Dict[str, Any]:4358 cleaned = dict(locator_hints or {})4359 for key in ("page_url", "url", "action"):4360 value = cleaned.get(key)4361 if isinstance(value, str) and _same_normalized_url(value, parent_page_url):4362 cleaned.pop(key, None)4363 for key in ("matched_by", "class_tokens", "script_srcs", "iframe_srcs", "css_path_clues"):4364 if isinstance(cleaned.get(key), list):4365 cleaned[key] = _sorted_unique_strings(cleaned.get(key) or [])4366 if not cleaned[key]:4367 cleaned.pop(key, None)4368 data_attributes = cleaned.get("data_attributes")4369 if isinstance(data_attributes, dict):4370 cleaned["data_attributes"] = {4371 attr_name: _strip_text(str(value))4372 for attr_name, value in data_attributes.items()4373 if _strip_text(str(value))4374 }4375 if not cleaned["data_attributes"]:4376 cleaned.pop("data_attributes", None)4377 return {key: value for key, value in cleaned.items() if value not in (None, "", [], {})}4378
4379
4380def _grouped_contact_widget_record(item: Dict[str, Any], parent_page_url: str) -> Dict[str, Any]:4381 locator_hints = item.get("widget_locator_hints") or {4382 "provider": item.get("provider"),4383 "matched_by": item.get("matched_by") or ([item.get("matched_hostname_or_selector")] if item.get("matched_hostname_or_selector") else []),4384 "launcher_text": item.get("launcher_text"),4385 "aria_label": item.get("aria_label"),4386 "title": item.get("title"),4387 "role": item.get("role"),4388 "id": item.get("id"),4389 "class_tokens": item.get("class_tokens") or [],4390 "data_attributes": item.get("data_attributes") or {},4391 "iframe_src": (item.get("iframe_srcs") or [None])[0],4392 "iframe_srcs": item.get("iframe_srcs") or [],4393 "script_srcs": item.get("script_srcs") or [],4394 "css_path_clues": item.get("exact_selectors", []),4395 "found_in_rendered_dom": bool(item.get("found_in_rendered_dom")),4396 "fixed_position": bool(item.get("fixed_position")),4397 "corner_hint": item.get("corner_hint"),4398 }4399 record = {4400 "provider": item.get("provider"),4401 "resolution_method": item.get("resolution_method") or "snapshot_ref",4402 "usage_notes": item.get("usage_notes")4403 or SNAPSHOT_REF_USAGE_NOTES4404 + ["Use provider, launcher text, aria-label, and any fixed-position or corner clue to match the widget in the snapshot."],4405 "widget_locator_hints": _clean_widget_locator_hints(locator_hints, parent_page_url=parent_page_url),4406 }4407 return {key: value for key, value in record.items() if value not in (None, "", [], {})}4408
4409
4410def _unique_sorted_checks(items: List[Dict[str, Any]]) -> List[str]:4411 checks = {4412 _strip_text(str(check))4413 for item in items4414 for check in (item.get("validator_passed") or [])4415 if _strip_text(str(check))4416 }4417 return sorted(checks, key=lambda value: (value.lower(), value))4418
4419
4420def _short_reason_text(text: str, fallback: str) -> str:4421 cleaned = _strip_text(text)4422 if not cleaned:4423 return fallback4424 parts = []4425 for part in cleaned.split(";"):4426 normalized = _strip_text(part)4427 if normalized and normalized not in parts:4428 parts.append(normalized)4429 if parts:4430 cleaned = "; ".join(parts[:2])4431 if len(cleaned) > 180:4432 return cleaned[:177].rstrip() + "..."4433 return cleaned4434
4435
4436def _confidence_reason_for_item(item: Dict[str, Any]) -> str:4437 item_type = item.get("type")4438 if item_type == "contact_page":4439 return _short_reason_text(item.get("evidence", ""), "validated contact page")4440 if item_type == "contact_form":4441 submit_label = _strip_text(item.get("submit_label") or "")4442 if submit_label:4443 return f"validated contact form ({submit_label})"4444 return "validated contact form"4445 if item_type == "email":4446 source_type = (item.get("source_type") or "").lower()4447 if source_type == "mailto":4448 return "mailto contact email"4449 if source_type == "jsonld":4450 return "email in JSON-LD contact point"4451 return "contact email found on page"4452 if item_type == "phone":4453 source_type = (item.get("source_type") or "").lower()4454 if source_type in {"tel", "callto", "sms"}:4455 return "call link phone number"4456 if source_type == "jsonld":4457 return "phone in JSON-LD contact point"4458 return "phone number found on page"4459 if item_type == "social_profile":4460 network = (item.get("network") or "social").strip()4461 return f"{network} profile linked from page"4462 if item_type == "contact_widget":4463 provider = (item.get("provider") or "contact").strip()4464 return f"{provider} contact widget detected"4465 return _short_reason_text(item.get("evidence", ""), "contact signal found")4466
4467
4468def _confidence_sort_key(item: Dict[str, Any]) -> Tuple[int, int, int, int, int]:4469 return (4470 CONFIDENCE_ORDER.get(item.get("confidence_band", ""), 0),4471 1 if item.get("type") == "contact_page" else 0,4472 _group_item_rank(item),4473 len(item.get("validator_passed") or []),4474 len(item.get("evidence") or ""),4475 )4476
4477
4478def _build_target_confidence(items: List[Dict[str, Any]]) -> Dict[str, Any]:4479 if not items:4480 return {4481 "level": "PLAUSIBLE",4482 "reason": "contact signal found",4483 "checks": [],4484 }4485
4486 best_item = max(items, key=_confidence_sort_key)4487 level = best_item.get("confidence_band")4488 if level not in CONFIDENCE_ORDER:4489 level = "PLAUSIBLE"4490 checks = _unique_sorted_checks(items)4491 return {4492 "level": level,4493 "reason": _confidence_reason_for_item(best_item),4494 "checks": checks,4495 }4496
4497
4498def _collect_page_urls(node: Any) -> Set[str]:4499 collected: Set[str] = set()4500 if isinstance(node, dict):4501 page_url = (node.get("page_url") or "").strip() if isinstance(node.get("page_url"), str) else ""4502 if page_url:4503 normalized = normalize_url(page_url) or page_url4504 collected.add(normalized)4505 for value in node.values():4506 collected.update(_collect_page_urls(value))4507 elif isinstance(node, list):4508 for value in node:4509 collected.update(_collect_page_urls(value))4510 return collected4511
4512
4513def validate_result_schema(result: Dict[str, Any]) -> None:4514 expected_keys = {"input_url", "domain", "targets", "meta"}4515 legacy_keys = {4516 "final_url",4517 "crawl_stats",4518 "rejected_candidates_count",4519 "contact_pages",4520 "items",4521 "automation_targets",4522 "distinct_targets",4523 "site_metadata",4524 "domain_metadata",4525 "compatibility_flags",4526 }4527 assert set(result.keys()) == expected_keys, f"unexpected result keys: {sorted(result.keys())}"4528 assert not (legacy_keys & set(result.keys())), "legacy result keys leaked into output"4529
4530 meta = result.get("meta")4531 assert isinstance(meta, dict), "meta must be an object"4532 assert set(meta.keys()) == {"site", "domain"}, f"unexpected meta keys: {sorted(meta.keys())}"4533
4534 seen_page_urls: Set[str] = set()4535 for target in result.get("targets", []):4536 page_url = _group_page_url_key(target)4537 assert page_url, "targets[].page_url is required"4538 assert target.get("page_url") == page_url, "targets[].page_url must be normalized"4539 assert page_url not in seen_page_urls, f"duplicate target page_url: {page_url}"4540 seen_page_urls.add(page_url)4541
4542 for key, value in result.items():4543 if key == "targets":4544 continue4545 overlap = _collect_page_urls(value) & seen_page_urls4546 assert not overlap, f"overlapping page_url records leaked into top-level field '{key}': {sorted(overlap)}"4547
4548
4549def build_targets(contact_items: List[Dict[str, Any]]) -> List[Dict[str, Any]]:4550 """Build the single canonical actionable target list keyed by normalized page URL."""4551
4552 channel_lists = {4553 "email": "emails",4554 "phone": "phones",4555 "contact_form": "contact_forms",4556 "social_profile": "social_profiles",4557 "contact_widget": "contact_widgets",4558 }4559 page_groups: Dict[str, Dict[str, Any]] = {}4560
4561 for raw_item in contact_items:4562 item_type = raw_item.get("type")4563 if item_type not in channel_lists and item_type != "contact_page":4564 continue4565
4566 page_url = _group_page_url_key(raw_item)4567 if not page_url:4568 continue4569
4570 item = dict(raw_item)4571 item["page_url"] = page_url4572
4573 page_group = page_groups.setdefault(4574 page_url,4575 {4576 "sources": [],4577 **{list_name: {} for list_name in channel_lists.values()},4578 },4579 )4580 page_group["sources"].append(item)4581
4582 if item_type == "contact_page":4583 continue4584
4585 dedupe_key = _group_channel_key(item)4586 if not dedupe_key:4587 continue4588
4589 bucket_name = channel_lists[item_type]4590 existing = page_group[bucket_name].get(dedupe_key)4591 if existing is None:4592 page_group[bucket_name][dedupe_key] = item4593 continue4594 page_group[bucket_name][dedupe_key] = _pick_stronger_group_item(existing, item)4595
4596 targets: List[Dict[str, Any]] = []4597 for page_url in sorted(page_groups):4598 buckets = page_groups[page_url]4599 target = {4600 "page_url": page_url,4601 "confidence": _build_target_confidence(buckets["sources"]),4602 "emails": [4603 _grouped_email_record(item)4604 for item in sorted(buckets["emails"].values(), key=lambda item: _channel_sort_key("email", item))4605 ],4606 "phones": [4607 _grouped_phone_record(item)4608 for item in sorted(buckets["phones"].values(), key=lambda item: _channel_sort_key("phone", item))4609 ],4610 "contact_forms": [4611 _grouped_contact_form_record(item, page_url)4612 for item in sorted(buckets["contact_forms"].values(), key=lambda item: _channel_sort_key("contact_form", item))4613 ],4614 "social_profiles": [4615 _grouped_social_profile_record(item)4616 for item in sorted(buckets["social_profiles"].values(), key=lambda item: _channel_sort_key("social_profile", item))4617 ],4618 "contact_widgets": [4619 _grouped_contact_widget_record(item, page_url)4620 for item in sorted(buckets["contact_widgets"].values(), key=lambda item: _channel_sort_key("contact_widget", item))4621 ],4622 }4623 targets.append(target)4624
4625 seen_page_urls: Set[str] = set()4626 for target in targets:4627 page_url = target.get("page_url") or ""4628 assert page_url and page_url not in seen_page_urls, f"duplicate target page_url: {page_url}"4629 seen_page_urls.add(page_url)4630
4631 return targets4632
4633
4634def parse_sitemap_candidates(start_url: str, session: requests.Session, retries: int, timeout: float) -> List[Dict[str, Any]]:4635 parsed = urlparse(start_url)4636 if not parsed.scheme or not parsed.netloc:4637 return []4638 sitemap_urls = ["/sitemap.xml", "/sitemap_index.xml"]4639 candidates: List[Dict[str, Any]] = []4640 for path in sitemap_urls:4641 candidate = urlunparse((parsed.scheme, parsed.netloc, path, "", "", ""))4642 for attempt in range(retries + 1):4643 try:4644 resp = session.get(candidate, headers=DEFAULT_HEADERS, timeout=timeout, allow_redirects=True)4645 except requests.RequestException:4646 break4647 if resp.status_code in {429, 500, 502, 503, 504} and attempt < retries:4648 time.sleep(0.5 * (attempt + 1))4649 continue4650 if resp.status_code >= 400:4651 break4652 text = resp.text or ""4653 break4654 else:4655 continue4656 try:4657 import xml.etree.ElementTree as ET4658 root = ET.fromstring(text)4659 for node in root.iter():4660 if not node.tag.endswith("loc"):4661 continue4662 url_text = (node.text or "").strip()4663 if not url_text:4664 continue4665 normalized = normalize_url(url_text)4666 if not normalized or not same_site(start_url, normalized):4667 continue4668 path = _page_host_block(urlparse(normalized).path)4669 if _is_contact_url_hint(path):4670 candidates.append(4671 {4672 "url": normalized,4673 "score": 55,4674 "reason": "sitemap contact hint",4675 "evidence": "sitemap entry",4676 "source": start_url,4677 }4678 )4679 except Exception:4680 continue4681 return candidates4682
4683
4684def fetch_page(url: str, session: requests.Session, *, retries: int, timeout: float) -> Tuple[Optional[str], Optional[str], Optional[str]]:4685 for attempt in range(retries + 1):4686 try:4687 resp = session.get(url, headers=DEFAULT_HEADERS, timeout=timeout, allow_redirects=True)4688 if resp.status_code in {429, 500, 502, 503, 504} and attempt < retries:4689 time.sleep(0.5 * (attempt + 1))4690 continue4691 if resp.status_code >= 400:4692 return None, None, f"HTTP {resp.status_code}"4693 ctype = (resp.headers.get("Content-Type") or "").lower()4694 if "text/html" not in ctype and "application/xhtml" not in ctype and "application/json" not in ctype:4695 return None, None, "Non HTML content"4696 return resp.url, resp.text, None4697 except requests.RequestException as exc:4698 if attempt < retries:4699 time.sleep(0.5 * (attempt + 1))4700 continue4701 return None, None, str(exc)4702 return None, None, "unreachable"4703
4704
4705def _playwright_environment_failure(exc: Exception) -> str:4706 message = _strip_text(str(exc))4707 lowered = message.lower()4708 if "executable doesn't exist" in lowered or "playwright install" in lowered:4709 return "Playwright browser executable unavailable"4710 if "no module named" in lowered and "playwright" in lowered:4711 return "Playwright Python package unavailable"4712 return ""4713
4714
4715def _likely_contact_page(page_url: str, soup: BeautifulSoup) -> bool:4716 title = _strip_text(soup.title.string) if soup.title and soup.title.string else ""4717 headings = " ".join(_strip_text(h.get_text(" ", strip=True)) for h in soup.select("h1,h2"))4718 path = (urlparse(page_url).path or "").lower()4719 return _is_contact_url_hint(path) or _has_contact_phrase(" ".join([title, headings, candidate_crumbs(soup)]))4720
4721
4722def _page_has_widget_resource_clues(page_url: str, raw_html: str, soup: BeautifulSoup) -> bool:4723 html_lower = raw_html.lower()4724 inline_terms = {4725 term.lower()4726 for cfg in WIDGET_PROVIDERS.values()4727 for term in cfg.get("inline", set())4728 if len(_strip_text(str(term))) >= 44729 }4730 if any(term in html_lower for term in inline_terms):4731 return True4732
4733 for node in soup.find_all(["script", "iframe"], src=True):4734 src = normalize_url(node.get("src") or "", base=page_url)4735 if not src:4736 continue4737 if any(_url_host_matches(src, cfg["host"]) for cfg in WIDGET_PROVIDERS.values()):4738 return True4739 lowered_src = src.lower()4740 if any(token in lowered_src for token in ("chat", "widget", "messenger", "whatsapp", "livechat")):4741 return True4742 return False4743
4744
4745def _widget_candidates_have_locator_data(candidates: List[Dict[str, Any]]) -> bool:4746 return any(bool(candidate.get("has_locatable_launcher")) or bool(candidate.get("matched_iframe_host")) for candidate in candidates)4747
4748
4749def _should_fetch_rendered_widget_page(4750 page_url: str,4751 raw_html: str,4752 soup: BeautifulSoup,4753 raw_widgets: List[Dict[str, Any]],4754) -> bool:4755 if _widget_candidates_have_locator_data(raw_widgets):4756 return False4757 if any(4758 bool(candidate.get("matched_host")) or bool(candidate.get("matched_inline")) or bool(candidate.get("matched_selector"))4759 for candidate in raw_widgets4760 ):4761 return True4762 return _page_has_widget_resource_clues(page_url, raw_html, soup) and (_likely_contact_page(page_url, soup) or bool(raw_widgets))4763
4764
4765def _playwright_proxy_settings(proxy_url: str) -> Dict[str, str]:4766 parsed = urlparse(proxy_url)4767 if not parsed.scheme or not parsed.hostname:4768 return {"server": proxy_url}4769
4770 server = f"{parsed.scheme}://{parsed.hostname}"4771 if parsed.port:4772 server += f":{parsed.port}"4773
4774 settings = {"server": server}4775 if parsed.username:4776 settings["username"] = unquote(parsed.username)4777 if parsed.password:4778 settings["password"] = unquote(parsed.password)4779 return settings4780
4781
4782def fetch_rendered_page(url: str, *, timeout: float, proxy_url: Optional[str] = None) -> RenderedPageSnapshot:4783 global _PLAYWRIGHT_DISABLED_REASON4784
4785 if _PLAYWRIGHT_DISABLED_REASON:4786 return RenderedPageSnapshot(final_url=url, html=None, script_srcs=[], iframe_srcs=[], launchers=[], error=_PLAYWRIGHT_DISABLED_REASON)4787 if sync_playwright is None:4788 _PLAYWRIGHT_DISABLED_REASON = "Playwright Python package unavailable"4789 return RenderedPageSnapshot(final_url=url, html=None, script_srcs=[], iframe_srcs=[], launchers=[], error=_PLAYWRIGHT_DISABLED_REASON)4790
4791 timeout_ms = max(5000, int(timeout * 1000))4792 try:4793 with sync_playwright() as playwright:4794 launch_kwargs: Dict[str, Any] = {"headless": True}4795 if proxy_url:4796 launch_kwargs["proxy"] = _playwright_proxy_settings(proxy_url)4797 browser = playwright.chromium.launch(**launch_kwargs)4798 try:4799 page = browser.new_page(viewport={"width": 1440, "height": 1024})4800 page.goto(url, wait_until="domcontentloaded", timeout=timeout_ms)4801 try:4802 page.wait_for_load_state("networkidle", timeout=min(timeout_ms, 6000))4803 except PlaywrightTimeoutError:4804 pass4805 page.wait_for_timeout(RENDERED_WIDGET_EXTRA_WAIT_MS)4806
4807 final_url = normalize_url(page.url) or url4808 html = page.content()4809 script_srcs = page.evaluate(4810 "() => Array.from(document.querySelectorAll('script[src]')).map((node) => node.src || '').filter(Boolean)"4811 )4812 iframe_srcs = page.evaluate(4813 "() => Array.from(document.querySelectorAll('iframe[src]')).map((node) => node.src || '').filter(Boolean)"4814 )4815 launchers = page.evaluate(4816 """({ phrases, terms }) => {4817 const clean = (value) => (value || '').replace(/\\s+/g, ' ').trim().slice(0, 180);4818 const cssPath = (element) => {4819 const parts = [];4820 let current = element;4821 let depth = 0;4822 while (current && current.nodeType === 1 && current.tagName && current.tagName.toLowerCase() !== 'html' && depth < 6) {4823 let part = current.tagName.toLowerCase();4824 if (current.id && /^[A-Za-z_][A-Za-z0-9_-]*$/.test(current.id)) {4825 part += `#${current.id}`;4826 parts.push(part);4827 break;4828 }4829 const classes = Array.from(current.classList || []).filter((token) => /^[A-Za-z_][A-Za-z0-9_-]*$/.test(token)).slice(0, 2);4830 if (classes.length) {4831 part += `.${classes.join('.')}`;4832 }4833 if (current.parentElement) {4834 const siblings = Array.from(current.parentElement.children).filter((sibling) => sibling.tagName === current.tagName);4835 if (siblings.length > 1) {4836 part += `:nth-of-type(${siblings.indexOf(current) + 1})`;4837 }4838 }4839 parts.push(part);4840 current = current.parentElement;4841 depth += 1;4842 }4843 return parts.reverse().join(' > ');4844 };4845 const cornerHint = (rect, viewportWidth, viewportHeight) => {4846 const nearTop = rect.top <= 220;4847 const nearBottom = viewportHeight - rect.bottom <= 220;4848 const nearLeft = rect.left <= 220;4849 const nearRight = viewportWidth - rect.right <= 220;4850 if (nearBottom && nearRight) return 'bottom-right';4851 if (nearBottom && nearLeft) return 'bottom-left';4852 if (nearTop && nearRight) return 'top-right';4853 if (nearTop && nearLeft) return 'top-left';4854 if (nearBottom) return 'bottom';4855 if (nearTop) return 'top';4856 if (nearRight) return 'right';4857 if (nearLeft) return 'left';4858 return '';4859 };4860 const viewportWidth = window.innerWidth || document.documentElement.clientWidth || 1440;4861 const viewportHeight = window.innerHeight || document.documentElement.clientHeight || 1024;4862 return Array.from(document.querySelectorAll('button, a, div, span'))4863 .map((element) => {4864 const style = window.getComputedStyle(element);4865 const rect = element.getBoundingClientRect();4866 const visible =4867 style.display !== 'none' &&4868 style.visibility !== 'hidden' &&4869 style.opacity !== '0' &&4870 rect.width >= 16 &&4871 rect.height >= 16 &&4872 rect.bottom >= 0 &&4873 rect.right >= 0;4874 const dataAttributes = {};4875 for (const attr of Array.from(element.attributes || [])) {4876 if (!attr.name.startsWith('data-')) continue;4877 const value = clean(attr.value);4878 if (value) dataAttributes[attr.name] = value;4879 }4880 const item = {4881 tag: (element.tagName || '').toLowerCase(),4882 launcher_text: clean(element.innerText || element.textContent || ''),4883 aria_label: clean(element.getAttribute('aria-label') || ''),4884 title: clean(element.getAttribute('title') || ''),4885 role: clean(element.getAttribute('role') || ''),4886 id: clean(element.id || ''),4887 class_tokens: Array.from(element.classList || []).slice(0, 10),4888 data_attributes: dataAttributes,4889 href: clean(element.getAttribute('href') || ''),4890 exact_css_clue: cssPath(element),4891 fixed_position: ['fixed', 'sticky'].includes(style.position),4892 corner_hint: cornerHint(rect, viewportWidth, viewportHeight),4893 found_in_rendered_dom: true,4894 visible,4895 };4896 const combined = [4897 item.launcher_text,4898 item.aria_label,4899 item.title,4900 item.role,4901 item.id,4902 item.class_tokens.join(' '),4903 Object.values(item.data_attributes).join(' '),4904 item.href,4905 ].join(' ').toLowerCase();4906 item.matches =4907 phrases.some((phrase) => combined.includes(phrase)) ||4908 terms.some((term) => combined.includes(term)) ||4909 /wa\\.me|api\\.whatsapp\\.com|m\\.me|messenger\\.com/.test((item.href || '').toLowerCase()) ||4910 item.fixed_position ||4911 !!item.corner_hint;4912 return item;4913 })4914 .filter((item) => item.visible && item.matches)4915 .slice(0, 80);4916 }""",4917 {"phrases": WIDGET_LAUNCHER_PHRASES, "terms": sorted(WIDGET_MATCH_TERMS)},4918 )4919 return RenderedPageSnapshot(4920 final_url=final_url,4921 html=html,4922 script_srcs=_sorted_unique_strings(normalize_url(src, base=final_url) or src for src in script_srcs or []),4923 iframe_srcs=_sorted_unique_strings(normalize_url(src, base=final_url) or src for src in iframe_srcs or []),4924 launchers=launchers or [],4925 error=None,4926 )4927 finally:4928 browser.close()4929 except Exception as exc:4930 reason = _playwright_environment_failure(exc)4931 if reason:4932 _PLAYWRIGHT_DISABLED_REASON = reason4933 return RenderedPageSnapshot(4934 final_url=url,4935 html=None,4936 script_srcs=[],4937 iframe_srcs=[],4938 launchers=[],4939 error=reason or _truncate_hint_text(str(exc), limit=220),4940 )4941
4942
4943def crawl_domain(4944 start_url: str,4945 *,4946 max_pages: int,4947 timeout: float,4948 retries: int,4949 delay: float,4950 include_plausible: bool = False,4951 enrich_dns: bool = False,4952 include_metadata: bool = False,4953 dns_timeout: float = 2.5,4954 proxy_url: Optional[str] = None,4955 logger: Any = None,4956) -> Dict[str, Any]:4957 with requests.Session() as session:4958 session.headers.update(DEFAULT_HEADERS)4959 if proxy_url:4960 session.proxies.update({"http": proxy_url, "https": proxy_url})4961
4962 queue: deque[tuple[int, str, int]] = deque()4963 seen: Set[str] = set()4964 candidates: Dict[str, int] = {}4965 contact_items: List[Dict[str, Any]] = []4966 visited_site_metadata: List[Dict[str, Any]] = []4967 site_metadata = {4968 "business_name": None,4969 "legal_names": [],4970 "description": None,4971 "addresses": [],4972 "contact_points": [],4973 "same_as": [],4974 "support_sales_labels": [],4975 "public_emails": [],4976 "business_phone": [],4977 "copyright": None,4978 }4979 domain_metadata = None4980 email_domain_metadata_map: Dict[str, Dict[str, Any]] = {}4981
4982 homepage = normalize_url(start_url)4983 if not homepage:4984 raise ValueError(f"invalid start URL: {start_url}")4985 _log_progress(logger, "info", f"Starting crawl for {homepage} (max_pages={max_pages})")4986 queue.append((0, homepage, 100))4987 candidates[homepage] = 1004988
4989 # seed from sitemap only when contact-intent-like.4990 for c in parse_sitemap_candidates(homepage, session, retries=retries, timeout=timeout):4991 if c["url"] not in candidates:4992 queue.append((0, c["url"], c["score"]))4993 candidates[c["url"]] = c["score"]4994
4995 for probe_url in _build_contact_probe_urls(homepage):4996 if not same_site(homepage, probe_url):4997 continue4998 if probe_url not in candidates:4999 queue.append((0, probe_url, 72))5000 candidates[probe_url] = 725001
5002 domain = get_registrable_domain(homepage)5003
5004 while queue and len(seen) < max_pages:5005 _, page_url, _ = queue.popleft()5006 if page_url in seen:5007 continue5008 seen.add(page_url)5009 _log_progress(logger, "info", f"[{len(seen)}/{max_pages}] Fetching {page_url}")5010
5011 final, html, error = fetch_page(page_url, session, retries=retries, timeout=timeout)5012 if error or not final:5013 _log_progress(logger, "warning", f"Skipping {page_url}: {error or 'fetch failed'}")5014 continue5015 final = normalize_url(final)5016 if not final or not same_site(homepage, final):5017 _log_progress(logger, "warning", f"Skipping {page_url}: redirected off-site")5018 continue5019 soup = BeautifulSoup(html, "html.parser")5020 if include_metadata:5021 visited_site_metadata.append(extract_site_metadata(html, soup, final))5022
5023 # stage1: discover candidate pages from this page5024 for c in discover_candidate_pages(homepage, final, soup):5025 url = c["url"]5026 if url in seen:5027 continue5028 if c["score"] > candidates.get(url, 0):5029 candidates[url] = c["score"]5030 queue.append((0, url, c["score"]))5031
5032 # stage2 validation for page candidates5033 page_email_candidates = extract_email_candidates(final, html, soup)5034 page_phone_candidates = extract_phone_candidates(final, html, soup)5035 raw_forms = extract_form_candidates(final, soup)5036 raw_widgets = extract_widget_candidates(final, html, soup)5037 raw_socials = extract_social_candidates(final, soup)5038
5039 rendered_widget_snapshot: Optional[RenderedPageSnapshot] = None5040 if _should_fetch_rendered_widget_page(final, html, soup, raw_widgets):5041 _log_progress(logger, "info", f"Loading rendered fallback for widget detection on {final}")5042 rendered_widget_snapshot = fetch_rendered_page(final, timeout=timeout, proxy_url=proxy_url)5043 if rendered_widget_snapshot.error:5044 _log_progress(logger, "warning", f"Rendered fetch unavailable for {final}: {rendered_widget_snapshot.error}")5045 if rendered_widget_snapshot.html:5046 rendered_widget_soup = BeautifulSoup(rendered_widget_snapshot.html, "html.parser")5047 rendered_widgets = extract_widget_candidates(5048 final,5049 rendered_widget_snapshot.html,5050 rendered_widget_soup,5051 rendered_snapshot=rendered_widget_snapshot,5052 )5053 raw_widgets = _merge_widget_candidate_lists(raw_widgets, rendered_widgets)5054
5055 # Build page context to support stricter validators.5056 page_context = {"is_contact_page": False, "url": final}5057
5058 validated_emails: List[Dict[str, Any]] = []5059 weak_emails: List[Dict[str, Any]] = []5060 accepted_email_ids: Set[int] = set()5061 for candidate in page_email_candidates:5062 validated = validate_email_candidate(candidate, page_context)5063 if validated:5064 validated_emails.append(validated)5065 accepted_email_ids.add(id(candidate))5066 else:5067 if candidate.get("source_type") in {"visible_text", "script"}:5068 weak_emails.append(candidate)5069
5070 validated_phones: List[Dict[str, Any]] = []5071 weak_phones: List[Dict[str, Any]] = []5072 for candidate in page_phone_candidates:5073 validated = validate_phone(candidate, page_context)5074 if validated:5075 validated_phones.append(validated)5076 else:5077 if candidate.get("source_type") in {"visible_text"}:5078 weak_phones.append(candidate)5079
5080 validated_forms: List[Dict[str, Any]] = []5081 for candidate in raw_forms:5082 validated = validate_contact_form(candidate)5083 if validated:5084 validated_forms.append(validated)5085
5086 validated_widgets: List[Dict[str, Any]] = []5087 for candidate in raw_widgets:5088 validated = validate_widget(candidate)5089 if validated:5090 validated_widgets.append(validated)5091
5092 validated_socials: List[Dict[str, Any]] = []5093 for candidate in raw_socials:5094 validated = validate_social_profile(candidate, page_context)5095 if validated:5096 validated_socials.append(validated)5097
5098 # Recompute if this page looks like contact page, then validate with stronger context.5099 title = _strip_text(soup.title.string) if soup.title and soup.title.string else ""5100 headings = " ".join(_strip_text(h.get_text(" ", strip=True)) for h in soup.select("h1,h2"))5101 page_ctx = {5102 "is_contact_page": bool(5103 validated_forms5104 or validated_widgets5105 or validated_socials5106 or validated_emails5107 or validated_phones5108 or _has_contact_phrase(title)5109 or _has_contact_phrase(headings)5110 or _has_contact_phrase(candidate_crumbs(soup))5111 ),5112 }5113 if page_ctx["is_contact_page"]:5114 for candidate in weak_emails:5115 promoted = validate_email_candidate(candidate, page_ctx)5116 if promoted:5117 validated_emails.append(promoted)5118 accepted_email_ids.add(id(candidate))5119 for candidate in weak_phones:5120 promoted = validate_phone(candidate, page_ctx)5121 if promoted:5122 validated_phones.append(promoted)5123
5124 # Final safety net: rescue obvious visible-text support emails even if earlier validation missed them.5125 for candidate in page_email_candidates:5126 if id(candidate) in accepted_email_ids:5127 continue5128 rescued = rescue_visible_email_candidate(candidate)5129 if not rescued:5130 continue5131 validated_emails.append(rescued)5132 accepted_email_ids.add(id(candidate))5133 validated_page = validate_contact_page(5134 final,5135 soup,5136 validated_forms,5137 validated_emails,5138 validated_phones,5139 validated_widgets,5140 validated_socials,5141 include_plausible=include_plausible,5142 )5143 if validated_page:5144 contact_items.append(validated_page)5145
5146 contact_items.extend(validated_emails)5147 contact_items.extend(validated_phones)5148 contact_items.extend(validated_forms)5149 contact_items.extend(validated_widgets)5150 contact_items.extend(validated_socials)5151
5152 # Small polite delay avoids noisy behavior.5153 time.sleep(delay)5154
5155 contact_items = [item for item in contact_items if item.get("confidence_band") in {"VERIFIED", "STRONG", "PLAUSIBLE"}]5156 if not include_plausible:5157 contact_items = [item for item in contact_items if item.get("confidence_band") in {"VERIFIED", "STRONG"}]5158
5159 email_items = [item for item in contact_items if item.get("type") == "email"]5160 non_email_items = [item for item in contact_items if item.get("type") != "email"]5161
5162 if enrich_dns:5163 target_meta = enrich_domain_dns(domain, dns_timeout=dns_timeout)5164 for item in email_items:5165 email_domain = (item.get("email_domain") or "").lower()5166 if not email_domain:5167 value = (item.get("value") or "")5168 if "@" in value:5169 email_domain = value.split("@", 1)[1]5170 if not email_domain:5171 continue5172 if email_domain in email_domain_metadata_map:5173 continue5174 email_domain_metadata_map[email_domain] = enrich_domain_dns(email_domain, dns_timeout=dns_timeout)5175 domain_metadata = {5176 "target_domain": target_meta,5177 "email_domain_metadata": list(email_domain_metadata_map.values()),5178 }5179
5180 contact_items = non_email_items + enrich_validated_emails(5181 email_items,5182 site_domain=domain,5183 do_dns=enrich_dns,5184 dns_timeout=dns_timeout,5185 )5186
5187 # merge site metadata from all visited pages5188 if include_metadata:5189 for page_meta in visited_site_metadata:5190 if not site_metadata["business_name"] and page_meta.get("business_name"):5191 site_metadata["business_name"] = page_meta.get("business_name")5192 for key in ("legal_names", "addresses", "contact_points", "same_as", "support_sales_labels", "public_emails", "business_phone"):5193 collected = site_metadata.get(key, [])5194 seen_local = {x.get("value", "").strip().lower() for x in collected}5195 for item in page_meta.get(key, []):5196 if not item:5197 continue5198 v = (item.get("value") or "").strip().lower()5199 if not v or v in seen_local:5200 continue5201 collected.append(item)5202 seen_local.add(v)5203 site_metadata[key] = collected5204 if not site_metadata["description"] and page_meta.get("description"):5205 site_metadata["description"] = page_meta.get("description")5206 if not site_metadata["copyright"] and page_meta.get("copyright"):5207 site_metadata["copyright"] = page_meta.get("copyright")5208
5209 result: Dict[str, Any] = {5210 "input_url": start_url,5211 "domain": domain,5212 "targets": build_targets(contact_items),5213 "meta": {5214 "site": site_metadata if include_metadata else None,5215 "domain": domain_metadata,5216 },5217 }5218 validate_result_schema(result)5219 _log_progress(logger, "info", f"Finished {homepage}: visited={len(seen)} page(s), targets={len(result['targets'])}")5220 return result5221
5222
5223def _normalize_unique_targets(raw_targets: Iterable[str]) -> List[str]:5224 normalized: List[str] = []5225 seen: Set[str] = set()5226 for target in raw_targets:5227 norm = normalize_url(target)5228 if not norm or norm in seen:5229 continue5230 seen.add(norm)5231 normalized.append(norm)5232 return normalized5233
5234
5235def _extract_start_urls(actor_input: Dict[str, Any]) -> List[str]:5236 raw_targets: List[str] = []5237
5238 single_url = actor_input.get("url")5239 if isinstance(single_url, str):5240 raw_targets.append(single_url)5241
5242 many_urls = actor_input.get("urls")5243 if isinstance(many_urls, list):5244 raw_targets.extend(item for item in many_urls if isinstance(item, str))5245
5246 start_urls = actor_input.get("startUrls")5247 if isinstance(start_urls, list):5248 for item in start_urls:5249 if isinstance(item, str):5250 raw_targets.append(item)5251 elif isinstance(item, dict) and isinstance(item.get("url"), str):5252 raw_targets.append(item["url"])5253
5254 return _normalize_unique_targets(raw_targets)5255
5256
5257def _read_int(actor_input: Dict[str, Any], key: str, default: int, *, minimum: int) -> int:5258 value = actor_input.get(key, default)5259 try:5260 parsed = int(value)5261 except (TypeError, ValueError) as exc:5262 raise ValueError(f"Input '{key}' must be an integer.") from exc5263 if parsed < minimum:5264 raise ValueError(f"Input '{key}' must be at least {minimum}.")5265 return parsed5266
5267
5268def _read_float(actor_input: Dict[str, Any], key: str, default: float, *, minimum: float) -> float:5269 value = actor_input.get(key, default)5270 try:5271 parsed = float(value)5272 except (TypeError, ValueError) as exc:5273 raise ValueError(f"Input '{key}' must be a number.") from exc5274 if parsed < minimum:5275 raise ValueError(f"Input '{key}' must be at least {minimum}.")5276 return parsed5277
5278
5279def _proxy_requested(proxy_settings: Any) -> bool:5280 if not isinstance(proxy_settings, dict):5281 return False5282 if proxy_settings.get("useApifyProxy"):5283 return True5284 proxy_urls = proxy_settings.get("proxyUrls")5285 return isinstance(proxy_urls, list) and any(isinstance(item, str) and item.strip() for item in proxy_urls)5286
5287
5288def _failure_result(url: str, error: str) -> Dict[str, Any]:5289 result = {5290 "input_url": normalize_url(url) or url,5291 "domain": get_registrable_domain(url),5292 "targets": [],5293 "meta": {5294 "site": None,5295 "domain": {5296 "error": _truncate_hint_text(error, limit=300) or "Unknown crawl failure.",5297 },5298 },5299 }5300 validate_result_schema(result)5301 return result5302
5303
5304def _log_progress(logger: Any, level: str, message: str) -> None:5305 if logger is None:5306 return5307 handler = getattr(logger, level, None) or getattr(logger, "info", None)5308 if callable(handler):5309 handler(message)5310
5311
5312def _site_state_key(url: str) -> str:5313 return normalize_url(url) or (url or "").strip()5314
5315
5316def _load_site_state_entries(raw_entries: Any) -> Set[str]:5317 site_keys: Set[str] = set()5318 if not isinstance(raw_entries, list):5319 return site_keys5320 for item in raw_entries:5321 if not isinstance(item, str):5322 continue5323 site_key = _site_state_key(item)5324 if site_key:5325 site_keys.add(site_key)5326 return site_keys5327
5328
5329def _load_site_billing_state(raw_state: Any) -> Tuple[Set[str], Set[str], bool]:5330 state = raw_state if isinstance(raw_state, dict) else {}5331 charged_sites = _load_site_state_entries(state.get("charged_sites"))5332 finalized_sites = _load_site_state_entries(state.get("finalized_sites"))5333 return charged_sites, finalized_sites, bool(state.get("charge_limit_reached", False))5334
5335
5336async def _save_site_billing_state(5337 charged_sites: Set[str],5338 finalized_sites: Set[str],5339 *,5340 charge_limit_reached: bool,5341) -> None:5342 await Actor.set_value(5343 SITE_ANALYZED_STATE_KEY,5344 {5345 "charged_sites": sorted(charged_sites),5346 "finalized_sites": sorted(finalized_sites),5347 "charge_limit_reached": charge_limit_reached,5348 },5349 )5350
5351
5352async def actor_main() -> None:5353 async with Actor:5354 actor_input = await Actor.get_input() or {}5355 targets = _extract_start_urls(actor_input)5356 if not targets:5357 raise ValueError("Provide at least one URL in 'startUrls', 'urls', or 'url'.")5358 Actor.log.info(f"Loaded {len(targets)} start URL(s)")5359
5360 max_pages = _read_int(actor_input, "maxPages", 12, minimum=1)5361 timeout = _read_float(actor_input, "timeoutSecs", 12.0, minimum=1.0)5362 retries = _read_int(actor_input, "retries", 2, minimum=0)5363 delay = _read_float(actor_input, "delaySecs", 0.2, minimum=0.0)5364 dns_timeout = _read_float(actor_input, "dnsTimeoutSecs", 2.5, minimum=0.1)5365 include_plausible = bool(actor_input.get("includePlausible", False))5366 enrich_dns = bool(actor_input.get("enrichDns", False))5367 include_metadata = bool(actor_input.get("includeMetadata", False))5368
5369 proxy_settings = actor_input.get("proxySettings")5370 proxy_configuration = None5371 if _proxy_requested(proxy_settings):5372 proxy_configuration = await Actor.create_proxy_configuration(actor_proxy_input=proxy_settings)5373 if proxy_configuration is None:5374 raise RuntimeError("Proxy configuration could not be created from 'proxySettings'.")5375 Actor.log.info("Proxy configuration enabled")5376
5377 charged_sites, finalized_sites, charge_limit_reached = _load_site_billing_state(5378 await Actor.get_value(SITE_ANALYZED_STATE_KEY)5379 )5380 results: List[Dict[str, Any]] = []5381 for index, url in enumerate(targets, start=1):5382 site_key = _site_state_key(url)5383 if site_key in finalized_sites:5384 Actor.log.info(f"site already finalized, skipping duplicate processing: {url}")5385 continue5386 if charge_limit_reached and site_key not in charged_sites:5387 Actor.log.warning(f"charge limit reached: stopping before site started for {url}")5388 break5389
5390 try:5391 Actor.log.info(f"site started: {url} ({index}/{len(targets)})")5392 proxy_url = None5393 if proxy_configuration is not None:5394 proxy_url = await proxy_configuration.new_url(session_id=get_registrable_domain(url) or None)5395 if not proxy_url:5396 raise RuntimeError("Proxy configuration did not return a proxy URL.")5397
5398 result = crawl_domain(5399 url,5400 max_pages=max_pages,5401 timeout=timeout,5402 retries=retries,5403 delay=delay,5404 include_plausible=include_plausible,5405 enrich_dns=enrich_dns,5406 include_metadata=include_metadata,5407 dns_timeout=dns_timeout,5408 proxy_url=proxy_url,5409 logger=Actor.log,5410 )5411 Actor.log.info(f"site analysis completed: {result.get('input_url') or url}")5412 except Exception as exc: # pragma: no cover - network dependent5413 if site_key in charged_sites:5414 Actor.log.error(f"site failed after prior site-analyzed charge, stopping for retry: {url}: {exc}")5415 raise5416 Actor.log.error(f"site failed: {url}: {exc}")5417 result = _failure_result(url, str(exc))5418 await Actor.push_data(result)5419 results.append(result)5420 continue5421
5422 if site_key in charged_sites:5423 Actor.log.info(f"site-analyzed event already recorded, skipping duplicate charge: {url}")5424 else:5425 charge_result = await Actor.charge(event_name=SITE_ANALYZED_EVENT_NAME)5426 if charge_result.charged_count > 0:5427 charged_sites.add(site_key)5428 await _save_site_billing_state(5429 charged_sites,5430 finalized_sites,5431 charge_limit_reached=charge_limit_reached,5432 )5433 Actor.log.info(f"site-analyzed event recorded: {url}")5434 else:5435 Actor.log.warning(f"site-analyzed event not recorded: {url} (charged_count=0)")5436
5437 if charge_result.event_charge_limit_reached:5438 charge_limit_reached = True5439 await _save_site_billing_state(5440 charged_sites,5441 finalized_sites,5442 charge_limit_reached=charge_limit_reached,5443 )5444 chargeable_within_limit = charge_result.chargeable_within_limit or {}5445 remaining = chargeable_within_limit.get(SITE_ANALYZED_EVENT_NAME)5446 remaining_text = "unknown" if remaining is None else str(remaining)5447 Actor.log.warning(5448 f"charge limit reached: {url} "5449 f"(event={SITE_ANALYZED_EVENT_NAME}, charged_count={charge_result.charged_count}, remaining={remaining_text})"5450 )5451 if charge_result.charged_count == 0:5452 break5453
5454 await Actor.push_data(result)5455 results.append(result)5456 finalized_sites.add(site_key)5457 await _save_site_billing_state(5458 charged_sites,5459 finalized_sites,5460 charge_limit_reached=charge_limit_reached,5461 )5462 if charge_limit_reached:5463 break5464
5465 await Actor.set_value(5466 "OUTPUT",5467 {5468 "processed": len(results),5469 "results": results,5470 },5471 )5472
5473
5474if __name__ == "__main__":5475 asyncio.run(actor_main()){ "title": "Contact Scraper Input", "description": "Provide one or more website URLs. The Actor returns one dataset item per input URL with the grouped contact targets it found.", "type": "object", "schemaVersion": 1, "required": [ "startUrls" ], "properties": { "startUrls": { "title": "Start URLs", "type": "array", "description": "One or more website homepages or contact pages to scan.", "editor": "requestListSources", "prefill": [ { "url": "https://www.hummel.net/" } ] }, "maxPages": { "title": "Max pages per site", "type": "integer", "description": "Maximum number of same-site pages to inspect for each input URL.", "default": 12, "minimum": 1 }, "timeoutSecs": { "title": "Request timeout", "type": "number", "description": "HTTP and rendered-page timeout in seconds.", "default": 12, "minimum": 1, "unit": "s" }, "retries": { "title": "Retries", "type": "integer", "description": "Retry count for transient request failures.", "default": 2, "minimum": 0 }, "delaySecs": { "title": "Delay between pages", "type": "number", "description": "Polite delay between page fetches on the same site.", "default": 0.2, "minimum": 0, "unit": "s" }, "includePlausible": { "title": "Include plausible matches", "type": "boolean", "description": "Include lower-confidence PLAUSIBLE results in addition to VERIFIED and STRONG matches.", "default": false }, "includeMetadata": { "title": "Include site metadata", "type": "boolean", "description": "Extract public business metadata such as name, description, addresses, and sameAs profiles.", "default": false }, "enrichDns": { "title": "Enrich DNS", "type": "boolean", "description": "Resolve MX, SPF, DMARC, and root DNS records for the target and detected email domains.", "default": false }, "dnsTimeoutSecs": { "title": "DNS timeout", "type": "number", "description": "Timeout in seconds for DNS lookups when DNS enrichment is enabled.", "default": 2.5, "minimum": 0.1, "unit": "s" }, "proxySettings": { "title": "Proxy configuration", "type": "object", "description": "Optional. Enable Apify Proxy or supply your own HTTP(S) proxy URLs when needed.", "editor": "proxy", "default": { "useApifyProxy": false } } }}{ "actorSpecification": 1, "name": "contact-scraper", "title": "Contact Scraper", "version": "0.1", "buildTag": "latest", "dockerfile": "../Dockerfile", "input": "./input_schema.json", "output": "./output_schema.json", "storages": { "dataset": "./dataset_schema.json" }}{ "actorOutputSchemaVersion": 1, "title": "Contact Scraper Output", "description": "The Contact Scraper stores one grouped result object per scanned input URL in the default dataset. The linked dataset uses the overview view by default.", "type": "object", "properties": { "results": { "type": "string", "title": "Results", "description": "Default dataset items, with the overview dataset view selected by default in Apify Console.", "template": "{{links.apiDefaultDatasetUrl}}/items?view=overview" } }}{ "actorSpecification": 1, "fields": { "$schema": "http://json-schema.org/draft-07/schema#", "type": "object", "title": "Contact scraper result item", "description": "One dataset item per scanned input URL. Each item groups detected contact channels by normalized page URL and optionally includes site metadata and DNS enrichment metadata.", "required": [ "input_url", "domain", "targets", "meta" ], "additionalProperties": false, "properties": { "input_url": { "type": "string", "title": "Input URL", "description": "Normalized URL that was submitted as the starting point for the scan.", "example": "https://example.com" }, "domain": { "type": [ "string", "null" ], "title": "Registrable domain", "description": "Registrable domain derived from the input URL, or null when a domain could not be derived.", "example": "example.com" }, "targets": { "type": "array", "title": "Grouped contact targets", "description": "Actionable contact findings grouped by normalized page URL. Each target contains page-level confidence and all channels detected on that page.", "items": { "$ref": "#/$defs/target" }, "example": [ { "page_url": "https://example.com/contact", "confidence": { "level": "VERIFIED", "reason": "validated contact form (Send Message)", "checks": [ "contact heading/title", "validated form" ] }, "emails": [ { "value": "contact@example.com" } ], "phones": [ { "value": "+18668323090" } ], "contact_forms": [], "social_profiles": [], "contact_widgets": [] } ] }, "meta": { "$ref": "#/$defs/meta" } }, "$defs": { "confidence": { "type": "object", "title": "Target confidence", "description": "Confidence summary for one grouped page target, derived from the strongest validated contact signal found on that page.", "required": [ "level", "reason", "checks" ], "additionalProperties": false, "properties": { "level": { "type": "string", "title": "Confidence level", "description": "Highest confidence band present among contact signals for this page target.", "enum": [ "VERIFIED", "STRONG", "PLAUSIBLE" ], "example": "VERIFIED" }, "reason": { "type": "string", "title": "Confidence reason", "description": "Human-readable explanation for why the target reached this confidence level.", "example": "validated contact form (Send Message)" }, "checks": { "type": "array", "title": "Validation checks", "description": "Unique validation checks or reasons collected from the signals that make up this target.", "items": { "type": "string" }, "example": [ "contact heading/title", "validated form" ] } } }, "emailRecord": { "type": "object", "title": "Email record", "description": "One validated email address found on the target page.", "required": [ "value" ], "additionalProperties": false, "properties": { "value": { "type": "string", "title": "Email address", "description": "Normalized email address.", "example": "contact@example.com" } } }, "phoneRecord": { "type": "object", "title": "Phone record", "description": "One validated phone number found on the target page.", "required": [ "value" ], "additionalProperties": false, "properties": { "value": { "type": "string", "title": "Phone number", "description": "Normalized phone number, usually in E.164-like format when normalization succeeds.", "example": "+18668323090" } } }, "fieldLocatorHint": { "type": "object", "title": "Field locator hint", "description": "Automation-friendly locator hints for one form field. These hints are meant for matching against a fresh browser snapshot, not for direct CSS automation.", "additionalProperties": false, "properties": { "role": { "type": "string", "title": "Semantic role", "description": "Semantic field role inferred by the scraper, such as name, email, phone, company, subject, or message.", "example": "email" }, "tag": { "type": "string", "title": "Tag", "description": "HTML tag name of the field.", "example": "input" }, "input_type": { "type": "string", "title": "Input type", "description": "Field type attribute or equivalent control type.", "example": "email" }, "label": { "type": "string", "title": "Label", "description": "Visible label text associated with the field.", "example": "Email Address" }, "name": { "type": "string", "title": "Field name", "description": "HTML name attribute.", "example": "email" }, "id": { "type": "string", "title": "Field ID", "description": "HTML id attribute.", "example": "contact_form_Email" }, "placeholder": { "type": "string", "title": "Placeholder", "description": "Placeholder text retained when it is useful for matching and not just an example contact value.", "example": "Enter your first name" }, "aria_label": { "type": "string", "title": "ARIA label", "description": "ARIA label on the field, if present.", "example": "Your message" }, "autocomplete": { "type": "string", "title": "Autocomplete hint", "description": "HTML autocomplete attribute, if present.", "example": "email" }, "class_tokens": { "type": "array", "title": "Class tokens", "description": "Individual CSS class tokens present on the field.", "items": { "type": "string" }, "example": [ "form-control", "contact-email" ] }, "debug_css_path": { "type": "string", "title": "Debug CSS path", "description": "Scraped CSS clue for debugging only. Do not use directly in browser automation commands.", "example": "form#contact-form input[type=\"email\"]" } } }, "submitLocatorHint": { "type": "object", "title": "Submit locator hint", "description": "Locator hints for the primary submit control of a validated contact form.", "additionalProperties": false, "properties": { "tag": { "type": "string", "title": "Tag", "description": "HTML tag name of the submit control.", "example": "button" }, "control_type": { "type": "string", "title": "Control type", "description": "HTML type attribute or equivalent control type.", "example": "submit" }, "text": { "type": "string", "title": "Button text", "description": "Visible text of the submit control.", "example": "Send Message" }, "name": { "type": "string", "title": "Name", "description": "HTML name attribute of the submit control.", "example": "submit" }, "id": { "type": "string", "title": "ID", "description": "HTML id attribute of the submit control.", "example": "contact-submit" }, "aria_label": { "type": "string", "title": "ARIA label", "description": "ARIA label on the submit control, if present.", "example": "Send Message" }, "class_tokens": { "type": "array", "title": "Class tokens", "description": "Individual CSS class tokens present on the submit control.", "items": { "type": "string" }, "example": [ "btn", "btn-primary" ] }, "debug_css_path": { "type": "string", "title": "Debug CSS path", "description": "Scraped CSS clue for debugging only.", "example": "form#contact-form button[type=\"submit\"]" } } }, "formLocatorHint": { "type": "object", "title": "Form locator hint", "description": "Locator hints for the contact form container itself.", "additionalProperties": false, "properties": { "tag": { "type": "string", "title": "Tag", "description": "HTML tag name of the form container.", "example": "form" }, "method": { "type": "string", "title": "Method", "description": "HTTP method declared on the form.", "example": "POST" }, "action": { "type": "string", "title": "Action URL", "description": "Resolved form action URL when it differs from the page URL.", "example": "https://example.com/contact/submit" }, "action_path": { "type": "string", "title": "Action path", "description": "Path portion of the form action URL.", "example": "/contact/submit" }, "id": { "type": "string", "title": "Form ID", "description": "HTML id attribute on the form, if present.", "example": "contact-form" }, "class_tokens": { "type": "array", "title": "Class tokens", "description": "Individual CSS class tokens present on the form.", "items": { "type": "string" }, "example": [ "contact-form", "stacked" ] }, "text_hint": { "type": "string", "title": "Text hint", "description": "Short text snippet from the form that can help identify it in a fresh DOM snapshot.", "example": "Contact us today and we will get back to you." }, "dom_signature": { "type": "string", "title": "DOM signature", "description": "Compact representation of the form field makeup used for deduplication and matching.", "example": "email:email||message:message||name:name" }, "debug_css_path": { "type": "string", "title": "Debug CSS path", "description": "Scraped CSS clue for debugging only.", "example": "main form.contact-form" } } }, "formFieldLocatorMap": { "type": "object", "title": "Field locator hints", "description": "Map of inferred field roles to field locator hints.", "additionalProperties": { "$ref": "#/$defs/fieldLocatorHint" }, "example": { "name": { "role": "name", "tag": "input", "label": "Your Name", "id": "contact_form_Name" }, "email": { "role": "email", "tag": "input", "input_type": "email", "label": "Email Address", "id": "contact_form_Email" }, "message": { "role": "message", "tag": "textarea", "label": "Message", "id": "contact_form_Message" } } }, "contactFormRecord": { "type": "object", "title": "Contact form record", "description": "Validated contact form detected on a target page, with automation-friendly locator hints.", "required": [ "method", "form_signature", "resolution_method", "usage_notes", "form_locator_hints", "field_locator_hints" ], "additionalProperties": false, "properties": { "action": { "type": "string", "title": "Action URL", "description": "Resolved form action URL when it differs from the page URL.", "example": "https://example.com/contact/submit" }, "action_is_page_url": { "type": "boolean", "title": "Action equals page URL", "description": "True when the form action resolves to the same URL as the parent page.", "example": true }, "method": { "type": "string", "title": "Method", "description": "HTTP method declared on the form.", "example": "POST" }, "form_signature": { "type": "string", "title": "Form signature", "description": "Stable signature derived from method, action, and DOM structure for deduplication.", "example": "POST|https://example.com/contact/submit|email:email||message:message||name:name" }, "resolution_method": { "type": "string", "title": "Resolution method", "description": "How automation should re-identify the form later.", "example": "snapshot_ref" }, "usage_notes": { "type": "array", "title": "Usage notes", "description": "Instructions for using the locator hints safely in browser automation.", "items": { "type": "string" }, "example": [ "Open the page_url and run a fresh browser snapshot before interacting.", "Match snapshot elements to these locator hints by role, label, name, id, placeholder, or aria-label." ] }, "form_locator_hints": { "$ref": "#/$defs/formLocatorHint" }, "submit_locator_hints": { "$ref": "#/$defs/submitLocatorHint" }, "field_locator_hints": { "$ref": "#/$defs/formFieldLocatorMap" } } }, "socialProfileRecord": { "type": "object", "title": "Social profile record", "description": "Validated business or direct messaging profile linked from the target page.", "required": [ "network", "value" ], "additionalProperties": false, "properties": { "network": { "type": "string", "title": "Network", "description": "Detected social or messaging network.", "example": "linkedin" }, "value": { "type": "string", "title": "Profile URL", "description": "Direct URL to the validated business profile or messaging endpoint.", "example": "https://www.linkedin.com/company/example/" } } }, "widgetLocatorHint": { "type": "object", "title": "Widget locator hint", "description": "Locator hints for a validated contact widget or chat launcher.", "additionalProperties": false, "properties": { "provider": { "type": "string", "title": "Provider", "description": "Detected widget provider name.", "example": "LiveChat" }, "matched_by": { "type": "array", "title": "Matched by", "description": "Signals that caused the widget to be recognized, such as script host, iframe host, selector, or launcher match.", "items": { "type": "string" }, "example": [ "script_host", "static_launcher" ] }, "launcher_text": { "type": "string", "title": "Launcher text", "description": "Visible text found on the widget launcher.", "example": "Chat with us" }, "aria_label": { "type": "string", "title": "ARIA label", "description": "ARIA label on the widget launcher, if present.", "example": "Open support chat" }, "title": { "type": "string", "title": "Title", "description": "HTML title attribute on the launcher, if present.", "example": "Chat" }, "role": { "type": "string", "title": "Role", "description": "ARIA or inferred role of the launcher.", "example": "button" }, "tag": { "type": "string", "title": "Tag", "description": "HTML tag name of the launcher.", "example": "button" }, "id": { "type": "string", "title": "ID", "description": "HTML id attribute of the launcher, if present.", "example": "livechat-launcher" }, "class_tokens": { "type": "array", "title": "Class tokens", "description": "Individual CSS class tokens present on the launcher.", "items": { "type": "string" }, "example": [ "chat-launcher", "fixed" ] }, "data_attributes": { "type": "object", "title": "Data attributes", "description": "Stringified data-* attributes that can help identify the launcher.", "additionalProperties": { "type": "string" }, "example": { "data-provider": "livechat" } }, "iframe_src": { "type": "string", "title": "Primary iframe source", "description": "First matched widget iframe source, when available.", "example": "https://widget.intercom.io/widget/demo" }, "iframe_srcs": { "type": "array", "title": "Iframe sources", "description": "Matched widget iframe sources.", "items": { "type": "string" }, "example": [ "https://widget.intercom.io/widget/demo" ] }, "script_srcs": { "type": "array", "title": "Script sources", "description": "Matched widget script sources.", "items": { "type": "string" }, "example": [ "https://wordpress.livechat.com/api/v2/script/test/widget.js" ] }, "css_path_clues": { "type": "array", "title": "CSS path clues", "description": "Scraped CSS selector clues for debugging and visual inspection.", "items": { "type": "string" }, "example": [ "button.chat-launcher" ] }, "found_in_rendered_dom": { "type": "boolean", "title": "Found in rendered DOM", "description": "True when the widget launcher was discovered only after a rendered-page fallback.", "example": false }, "fixed_position": { "type": "boolean", "title": "Fixed position", "description": "True when the launcher appears to use a fixed or floating position.", "example": true }, "corner_hint": { "type": "string", "title": "Corner hint", "description": "Position clue such as bottom-right or bottom-left.", "example": "bottom-right" } } }, "contactWidgetRecord": { "type": "object", "title": "Contact widget record", "description": "Validated contact widget or chat launcher with locator hints for automation.", "required": [ "provider", "resolution_method", "usage_notes", "widget_locator_hints" ], "additionalProperties": false, "properties": { "provider": { "type": "string", "title": "Provider", "description": "Detected widget provider, or a generic provider label when only a generic contact launcher could be validated.", "example": "LiveChat" }, "resolution_method": { "type": "string", "title": "Resolution method", "description": "How automation should re-identify the widget later.", "example": "snapshot_ref" }, "usage_notes": { "type": "array", "title": "Usage notes", "description": "Instructions for using widget locator hints safely in browser automation.", "items": { "type": "string" }, "example": [ "Open the page_url and run a fresh browser snapshot before interacting.", "Use provider, launcher text, aria-label, and any fixed-position or corner clue to match the widget in the snapshot." ] }, "widget_locator_hints": { "$ref": "#/$defs/widgetLocatorHint" } } }, "metaEvidence": { "type": "object", "title": "Metadata evidence item", "description": "One extracted metadata value together with the page and extraction source that produced it.", "required": [ "value", "page_url", "source_type", "evidence" ], "additionalProperties": false, "properties": { "value": { "type": "string", "title": "Value", "description": "Extracted metadata value.", "example": "Example Inc." }, "page_url": { "type": "string", "title": "Page URL", "description": "Page where the metadata value was found.", "example": "https://example.com" }, "source_type": { "type": "string", "title": "Source type", "description": "Extraction source such as jsonld, title, opengraph, meta_description, address_tag, or header_footer_text.", "example": "jsonld" }, "evidence": { "type": "string", "title": "Evidence", "description": "Short description of the extraction location or source field.", "example": "organization/name" } } }, "siteMeta": { "type": "object", "title": "Site metadata", "description": "Optional public business metadata aggregated across visited pages when includeMetadata is enabled.", "required": [ "business_name", "legal_names", "description", "addresses", "contact_points", "same_as", "support_sales_labels", "public_emails", "business_phone", "copyright" ], "additionalProperties": false, "properties": { "business_name": { "type": [ "object", "null" ], "title": "Business name", "description": "Best extracted business name, or null when no name could be extracted.", "allOf": [ { "if": { "type": "object" }, "then": { "$ref": "#/$defs/metaEvidence" } } ], "example": { "value": "Example Inc.", "page_url": "https://example.com", "source_type": "jsonld", "evidence": "organization/name" } }, "legal_names": { "type": "array", "title": "Legal names", "description": "Extracted legal entity names.", "items": { "$ref": "#/$defs/metaEvidence" } }, "description": { "type": [ "object", "null" ], "title": "Description", "description": "Best extracted public description, or null when no description could be extracted.", "allOf": [ { "if": { "type": "object" }, "then": { "$ref": "#/$defs/metaEvidence" } } ], "example": { "value": "Example Inc. helps teams reach customers.", "page_url": "https://example.com", "source_type": "meta_description", "evidence": "meta[name=description]" } }, "addresses": { "type": "array", "title": "Addresses", "description": "Extracted postal addresses or address-like snippets.", "items": { "$ref": "#/$defs/metaEvidence" } }, "contact_points": { "type": "array", "title": "Contact points", "description": "Extracted public contact point emails.", "items": { "$ref": "#/$defs/metaEvidence" } }, "same_as": { "type": "array", "title": "SameAs profiles", "description": "Extracted sameAs profile URLs from metadata such as JSON-LD.", "items": { "$ref": "#/$defs/metaEvidence" } }, "support_sales_labels": { "type": "array", "title": "Support or sales labels", "description": "Extracted support or sales labels from metadata or page sections.", "items": { "$ref": "#/$defs/metaEvidence" } }, "public_emails": { "type": "array", "title": "Public emails", "description": "Public email addresses extracted from metadata sources.", "items": { "$ref": "#/$defs/metaEvidence" } }, "business_phone": { "type": "array", "title": "Business phones", "description": "Public phone numbers extracted from metadata sources.", "items": { "$ref": "#/$defs/metaEvidence" } }, "copyright": { "type": [ "object", "null" ], "title": "Copyright snippet", "description": "Detected copyright snippet from page footer or similar content.", "allOf": [ { "if": { "type": "object" }, "then": { "$ref": "#/$defs/metaEvidence" } } ], "example": { "value": "Copyright 2026 Example Inc.", "page_url": "https://example.com", "source_type": "header_footer_text", "evidence": "copyright" } } } }, "dnsDomainMetadata": { "type": "object", "title": "DNS domain metadata", "description": "DNS enrichment payload for a domain when enrichDns is enabled.", "required": [ "domain", "mx_records", "has_mx", "spf", "dmarc", "mail_provider_hint", "root_a_records", "root_aaaa_records", "notes", "error" ], "additionalProperties": false, "properties": { "domain": { "type": "string", "title": "Domain", "description": "Domain that was enriched.", "example": "example.com" }, "mx_records": { "type": "array", "title": "MX records", "description": "Resolved MX record hostnames.", "items": { "type": "string" }, "example": [ "aspmx.l.google.com" ] }, "has_mx": { "type": "boolean", "title": "Has MX", "description": "True when at least one MX record was resolved.", "example": true }, "spf": { "type": [ "string", "null" ], "title": "SPF record", "description": "Resolved SPF TXT record, if present.", "example": "v=spf1 include:_spf.google.com ~all" }, "dmarc": { "type": [ "string", "null" ], "title": "DMARC record", "description": "Resolved DMARC TXT record, if present.", "example": "v=DMARC1; p=none;" }, "mail_provider_hint": { "type": "string", "title": "Mail provider hint", "description": "Best-effort mail provider inference from MX records.", "example": "Google Workspace" }, "root_a_records": { "type": "array", "title": "Root A records", "description": "Resolved IPv4 A records for the domain apex.", "items": { "type": "string" }, "example": [ "93.184.216.34" ] }, "root_aaaa_records": { "type": "array", "title": "Root AAAA records", "description": "Resolved IPv6 AAAA records for the domain apex.", "items": { "type": "string" }, "example": [] }, "notes": { "type": "array", "title": "Notes", "description": "Non-fatal lookup notes, such as partial lookup failures.", "items": { "type": "string" }, "example": [] }, "error": { "type": [ "string", "null" ], "title": "Error", "description": "Fatal DNS enrichment error, if one occurred.", "example": null } } }, "domainMetaFailure": { "type": "object", "title": "Failure metadata", "description": "Domain metadata returned when the crawl failed before enrichment or target extraction completed.", "required": [ "error" ], "additionalProperties": false, "properties": { "error": { "type": "string", "title": "Error", "description": "Short crawl failure description.", "example": "HTTP 403" } } }, "domainMetaEnrichment": { "type": "object", "title": "Domain enrichment metadata", "description": "Domain-level and email-domain DNS enrichment returned when enrichDns is enabled.", "required": [ "target_domain", "email_domain_metadata" ], "additionalProperties": false, "properties": { "target_domain": { "$ref": "#/$defs/dnsDomainMetadata" }, "email_domain_metadata": { "type": "array", "title": "Email domain metadata", "description": "DNS enrichment payloads for the domains of validated email addresses found on the site.", "items": { "$ref": "#/$defs/dnsDomainMetadata" } } } }, "meta": { "type": "object", "title": "Metadata", "description": "Optional site metadata and domain metadata for this scanned input URL.", "required": [ "site", "domain" ], "additionalProperties": false, "properties": { "site": { "type": [ "object", "null" ], "title": "Site metadata", "description": "Aggregated public business metadata when includeMetadata is enabled, otherwise null.", "allOf": [ { "if": { "type": "object" }, "then": { "$ref": "#/$defs/siteMeta" } } ] }, "domain": { "title": "Domain metadata", "description": "Null when DNS enrichment is disabled, a DNS enrichment payload when enrichDns is enabled, or an error object when the crawl failed.", "anyOf": [ { "type": "null" }, { "$ref": "#/$defs/domainMetaFailure" }, { "$ref": "#/$defs/domainMetaEnrichment" } ] } }, "example": { "site": null, "domain": null } }, "target": { "type": "object", "title": "Target page record", "description": "Grouped result for one normalized page URL that contains any validated contact signal.", "required": [ "page_url", "confidence", "emails", "phones", "contact_forms", "social_profiles", "contact_widgets" ], "additionalProperties": false, "properties": { "page_url": { "type": "string", "title": "Page URL", "description": "Normalized page URL used as the grouping key for contact signals.", "example": "https://example.com/contact" }, "confidence": { "$ref": "#/$defs/confidence" }, "emails": { "type": "array", "title": "Emails", "description": "Validated email addresses found on the page.", "items": { "$ref": "#/$defs/emailRecord" } }, "phones": { "type": "array", "title": "Phones", "description": "Validated phone numbers found on the page.", "items": { "$ref": "#/$defs/phoneRecord" } }, "contact_forms": { "type": "array", "title": "Contact forms", "description": "Validated contact forms with locator hints for automation.", "items": { "$ref": "#/$defs/contactFormRecord" } }, "social_profiles": { "type": "array", "title": "Social profiles", "description": "Validated social or messaging profile URLs linked from the page.", "items": { "$ref": "#/$defs/socialProfileRecord" } }, "contact_widgets": { "type": "array", "title": "Contact widgets", "description": "Validated contact widgets or chat launchers with locator hints for automation.", "items": { "$ref": "#/$defs/contactWidgetRecord" } } } } } }, "views": { "overview": { "title": "Overview", "description": "Default table view of grouped scraper outputs per scanned input URL.", "transformation": { "flatten": [ "meta" ], "fields": [ "input_url", "domain", "targets", "meta.site", "meta.domain" ] }, "display": { "component": "table", "properties": { "input_url": { "label": "Input URL", "format": "link" }, "domain": { "label": "Domain", "format": "text" }, "targets": { "label": "Targets", "format": "array" }, "meta.site": { "label": "Site Metadata", "format": "object" }, "meta.domain": { "label": "Domain Metadata", "format": "object" } } } } }}1# Contact Scraper Actor Schema Implementation Plan2
3> **For Claude:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task.4
5**Goal:** Add Apify-native output and dataset schema files that describe the Contact Scraper dataset output and wire them into the Actor definition.6
7**Architecture:** Keep the scraper output code unchanged and describe the existing default dataset shape through `.actor/output_schema.json` and `.actor/dataset_schema.json`. Use the output schema to declare where the results live and use the dataset schema to describe the nested result object and configure the Output tab overview view.8
9**Tech Stack:** Apify Actor definition JSON, Apify output schema, Apify dataset schema, JSON Schema Draft 2020-1210
11---12
13### Task 1: Add Actor output schema14
15**Files:**16- Create: `.actor/output_schema.json`17
18**Step 1: Define the default dataset output**19
20Create an output schema with:21- `actorOutputSchemaVersion: 1`22- a human-readable title and description23- one `results` property pointing to `{{links.apiDefaultDatasetUrl}}/items?view=overview`24
25**Step 2: Keep the output schema minimal**26
27Do not duplicate the full field structure here. Reserve detailed field descriptions for the dataset schema because that is what Apify documents as the right place for describing dataset item structure.28
29### Task 2: Add dataset schema30
31**Files:**32- Create: `.actor/dataset_schema.json`33
34**Step 1: Describe one dataset item**35
36Define a JSON Schema object for:37- `input_url`38- `domain`39- `targets`40- `meta`41
42Include nested descriptions and examples for the grouped target records the scraper emits:43- target page records44- confidence object45- emails46- phones47- contact forms48- social profiles49- contact widgets50- site metadata51- domain metadata52
53**Step 2: Configure the Output tab view**54
55Add an `overview` view that:56- flattens `meta`57- shows `input_url`, `domain`, `targets`, `meta.site`, and `meta.domain`58- formats `input_url` as a link59- formats `targets` as an array60- formats flattened meta fields as objects61
62### Task 3: Wire schemas into actor definition63
64**Files:**65- Modify: `.actor/actor.json`66
67**Step 1: Reference the new schema files**68
69Add:70- `"output": "./output_schema.json"`71- `"storages": { "dataset": "./dataset_schema.json" }`72
73Do not change input schema wiring or other Actor metadata.74
75### Task 4: Verify JSON files76
77**Files:**78- Verify: `.actor/actor.json`79- Verify: `.actor/output_schema.json`80- Verify: `.actor/dataset_schema.json`81
82**Step 1: Run a JSON parse check**83
84Run:85
86```bash87python3 - <<'PY'88import json89for path in [90 '.actor/actor.json',91 '.actor/output_schema.json',92 '.actor/dataset_schema.json',93]:94 with open(path, 'r', encoding='utf-8') as f:95 json.load(f)96 print(f'OK {path}')97PY98```99
100Expected:101- all three files parse successfully102- no JSON syntax errors