"""Conservative extraction of public business contact signals from approved HTML. This module never fetches URLs. Callers must provide HTML obtained from the approved website scanner and a source URL already associated with the business. """ from __future__ import annotations import html as html_lib import re from html.parser import HTMLParser from urllib.parse import unquote, urlparse MAX_HTML_BYTES = 512 * 1024 MAX_RESULTS = 100 EMAIL_RE = re.compile(r"(? str: return re.sub(r"\s+", " ", html_lib.unescape(value or "")).strip() def _decode_obfuscation(value: str) -> str: value = html_lib.unescape(unquote(value or "")) value = re.sub(r"\s*(?:\[|\(|\{|\s)at(?:\]|\)|\}|\s)\s*", "@", value, flags=re.I) value = re.sub(r"\s*(?:\[|\(|\{|\s)dot(?:\]|\)|\}|\s)*", ".", value, flags=re.I) return value def valid_email(value: str) -> bool: value = value.strip().lower() if len(value) > 254 or value.count("@") != 1 or value.split("@", 1)[1] in EXAMPLE_DOMAINS: return False return bool(EMAIL_RE.fullmatch(value)) and not any(x in value for x in ("password", "token", "secret", "credential", "apikey")) def normalize_phone(value: str) -> str: value = value.strip() digits = re.sub(r"\D", "", value) if value.startswith("+") and digits: return "+" + digits return digits class _ContactParser(HTMLParser): def __init__(self): super().__init__(convert_charrefs=True) self.text: list[str] = [] self.links: list[tuple[str, str]] = [] self.form_fields: list[str] = [] self._ignore = 0 self._anchor = "" self._in_form = False def handle_starttag(self, tag, attrs): attrs = {str(k).lower(): str(v or "") for k, v in attrs} tag = tag.lower() if tag in {"script", "style", "noscript", "template", "svg"}: self._ignore += 1 if tag == "a": self._anchor = attrs.get("href", "") if tag == "form": self._in_form = True if self._in_form and tag in {"input", "textarea", "select"}: name = attrs.get("name", "") or attrs.get("id", "") typ = attrs.get("type", "") self.form_fields.append(_clean(" ".join((name, typ, attrs.get("placeholder", ""))))) def handle_endtag(self, tag): tag = tag.lower() if tag in {"script", "style", "noscript", "template", "svg"} and self._ignore: self._ignore -= 1 if tag == "a": self._anchor = "" if tag == "form": self._in_form = False def handle_data(self, data): if self._ignore: return if data.strip(): self.text.append(data) if self._anchor: self.links.append((self._anchor, data)) def _record(kind, value, label, source_url, confidence, classification="unknown", *, suppressed=False, provenance="visible_text"): return {"kind": kind, "value": value, "label": label[:200], "classification": classification, "confidence": round(max(0.0, min(1.0, confidence)), 2), "source_url": source_url, "public_business": True, "mx_status": "unknown", "suppressed": bool(suppressed), "do_not_contact": bool(suppressed), "provenance": provenance} def extract_contacts(source_html: str, source_url: str, *, suppressions=None, max_results=MAX_RESULTS) -> list[dict]: if not isinstance(source_html, str) or len(source_html.encode("utf-8")) > MAX_HTML_BYTES: raise ValueError("html_too_large") try: max_results = int(max_results) except (ValueError, TypeError): raise ValueError("invalid_limits") if max_results < 1 or max_results > MAX_RESULTS: raise ValueError("invalid_limits") parser = _ContactParser(); parser.feed(source_html) visible = _decode_obfuscation(_clean(" ".join(parser.text))) suppression = {(str(x.get("kind", "")), str(x.get("value", "")).lower()) for x in (suppressions or [])} out, seen = [], set() def add(kind, value, label, confidence, classification="unknown", provenance="visible_text"): value = value.strip().lower() if kind == "email" else value.strip() if kind == "email": if not valid_email(value): return local, domain = value.rsplit("@", 1) classification = "free_mail" if domain in FREE_MAIL else ("role" if local in ROLE_NAMES else "named") key = (kind, value); blocked = ("email", value) in suppression or ("domain", domain) in suppression else: if kind in {"phone", "whatsapp"}: value = normalize_phone(value) if len(re.sub(r"\D", "", value)) < 7: return key = (kind, value); blocked = (kind, value.lower()) in suppression if key in seen or len(out) >= max_results: return seen.add(key); out.append(_record(kind, value, label or kind.title(), source_url, confidence, classification, suppressed=blocked, provenance=provenance)) for href, anchor_text in parser.links: raw = _decode_obfuscation(href) if raw.lower().startswith("mailto:"): address = raw[7:].split("?", 1)[0] add("email", address, _clean(anchor_text), 0.98, provenance="mailto") elif raw.lower().startswith("tel:"): add("phone", raw[4:].split("?", 1)[0], _clean(anchor_text), 0.98, provenance="tel") else: parsed = urlparse(raw if "://" in raw else "https://" + raw) host = (parsed.hostname or "").lower().removeprefix("www.") if host == "wa.me" or "whatsapp" in host: number = re.sub(r"\D", "", parsed.path) if number: add("whatsapp", "+" + number, _clean(anchor_text) or "WhatsApp", 0.98, provenance="whatsapp_link") elif any(host == d or host.endswith("." + d) for d in SOCIAL_HOSTS): add("social", raw, _clean(anchor_text) or host, 0.95, provenance="social_link") for match in EMAIL_RE.finditer(visible): context = visible[max(0, match.start() - 40):match.start()] if re.search(r"(?:password|passwd|token|secret|credential|api[_ -]?key|authorization)\s*[:=]?\s*$", context, re.I): continue add("email", match.group(0), "Email", 0.88) for value in PHONE_RE.findall(visible): add("phone", value, "Phone", 0.82) # A form is provenance, not proof of a destination address. if parser.form_fields: for field in parser.form_fields[:5]: if re.search(r"email|contact|phone|whatsapp", field, re.I): add("form", field, "Contact form", 0.7, provenance="form_field") return out