140 lines
7.2 KiB
Python
140 lines
7.2 KiB
Python
"""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"(?<![\w.+-])[A-Z0-9.!#$%&'*+/=?^_`{|}~-]+@[A-Z0-9](?:[A-Z0-9-]{0,61}[A-Z0-9])?(?:\.[A-Z0-9](?:[A-Z0-9-]{0,61}[A-Z0-9])?)+", re.I)
|
|
PHONE_RE = re.compile(r"(?<!\w)(\+?\d[\d ()().-]{6,}\d)(?!\w)")
|
|
SOCIAL_HOSTS = {"facebook.com", "instagram.com", "linkedin.com", "twitter.com", "x.com", "youtube.com", "tiktok.com"}
|
|
FREE_MAIL = {"gmail.com", "googlemail.com", "yahoo.com", "yahoo.co.uk", "hotmail.com", "outlook.com", "live.com", "icloud.com", "aol.com", "proton.me", "protonmail.com", "mail.com"}
|
|
ROLE_NAMES = {"info", "hello", "contact", "sales", "support", "admin", "office", "enquiries", "inquiries", "accounts", "billing", "careers", "hr", "help", "marketing", "bookings", "reception", "service", "customerservice"}
|
|
EXAMPLE_DOMAINS = {"example.com", "example.org", "example.net", "example.test", "invalid", "localhost"}
|
|
|
|
|
|
def _clean(value: str) -> 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
|