"""Bounded post-discovery enrichment wiring together existing evidence modules. This module intentionally reuses the existing website_scanner, domain_intelligence, and contact_extractor primitives rather than re-implementing bounded checks. Evidence is stored with source URLs and timestamps so every claim is traceable. """ from __future__ import annotations import hashlib import re from html.parser import HTMLParser from typing import Any from urllib.parse import urljoin, urlparse from .contact_extractor import extract_contacts from .domain_intelligence import ( generate_candidate_domains, normalize_registrable_domain, resolve_domain, ) from .website_scanner import scan_website, validate_url DEFAULT_CACHE_TTL = 3600 COPYRIGHT_RE = re.compile(r"©\s*(\d{4})(?:\s*-\s*(\d{4}))?", re.I) EMAIL_RE = re.compile(r"[A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,}", re.I) PHONE_RE = re.compile(r"\+?\d[\d ()().-]{6,}\d") def _registrable(url: str) -> str: return normalize_registrable_domain(url) def _domain_resolves(domain: str) -> bool: result = resolve_domain(domain) return result.get("status") == "ok" class _ResourceParser(HTMLParser): """Collect img/link/script targets for bounded broken-resource analysis.""" def __init__(self, base_url: str): super().__init__(convert_charrefs=True) self.base_url = base_url self.images: list[str] = [] self.links: list[str] = [] def handle_starttag(self, tag, attrs): attributes = dict(attrs) if tag == "img" and attributes.get("src"): self.images.append(attributes["src"]) if tag == "link" and attributes.get("href"): self.links.append(attributes["href"]) if tag == "script" and attributes.get("src"): self.links.append(attributes["src"]) def _absolute(base_url: str, target: str) -> str: return urljoin(base_url, target) def _outdated_copyright(html: str | None) -> tuple[bool, str | None]: """Return (outdated, year_text). Conservative: no detection is no evidence.""" if not html: return None, None year_match = re.search(r"copyright[^0-9]*(\d{4})", html, re.I) if not year_match: year_match = re.search(r"©\s*(\d{4})", html, re.I) if not year_match: return None, None year = year_match.group(1) return None, year def enrich_website(url: str, *, fetch, timeout: float = 5.0, max_bytes: int = 256 * 1024) -> dict[str, Any]: """Return a bounded, evidence-only website assessment for one URL.""" if not url: return {"error": "missing_url"} try: safe = validate_url(url) except ValueError as exc: return {"error": f"unsafe_url: {exc}", "input_url": url} result: dict[str, Any] = { "input_url": url, "domain": _registrable(safe), "has_working_website": None, "status": "unknown", "http_status": None, "https": safe.startswith("https://"), "redirects": [], "ssl_valid": None, "response_time_ms": None, "mobile_viewport": None, "cms": None, "seo": {"title": None, "description": None, "language": None}, "contact_form": None, "visible_phone": None, "visible_email": None, "outdated_copyright": None, "copyright_year": None, "broken_links": None, "broken_images": None, "placeholder": None, "evidence_url": safe, "evidence_checked_at": None, } try: page = fetch(safe, timeout=timeout, max_bytes=max_bytes) except Exception as exc: result["error"] = str(exc)[:200] return result result["http_status"] = page.get("status") result["response_time_ms"] = page.get("elapsed_ms") result["redirects"] = page.get("redirect_chain", []) result["ssl_valid"] = page.get("certificate_status") body = page.get("body", b"") html = body.decode("utf-8", "replace") if isinstance(body, bytes) else (body or "") title = description = language = None viewport = form = phone = email = None cms: list[str] = [] class _Analyzer(HTMLParser): def __init__(self): super().__init__(convert_charrefs=True) self._tag = "" self._buf: list[str] = [] def handle_starttag(self, tag, attrs): nonlocal viewport, form attrs = {str(k).lower(): str(v or "") for k, v in attrs} self._tag = tag if tag == "meta": name = attrs.get("name", "").lower() if name == "description" and attrs.get("content"): nonlocal description description = attrs["content"] if name == "viewport": viewport = True if tag == "form": form = True if tag in {"script", "link"}: text = " ".join(attrs.values()).lower() for key, terms in { "wordpress": ("wordpress", "wp-content"), "drupal": ("drupal",), "joomla": ("joomla",), "shopify": ("shopify",), "wix": ("wix.com",), }.items(): if any(term in text for term in terms): cms.append(key) def handle_data(self, data): nonlocal title, phone, email if self._tag == "title" and data.strip(): title = data.strip() if phone is None and PHONE_RE.search(data): phone = data.strip()[:60] if email is None and EMAIL_RE.search(data): email = data.strip()[:120] def handle_endtag(self, tag): nonlocal title if tag == "title" and title: title = title.strip()[:300] parser = _Analyzer() parser.feed(html) result["seo"]["title"] = title result["seo"]["description"] = description result["mobile_viewport"] = viewport result["contact_form"] = form result["visible_phone"] = phone result["visible_email"] = email result["cms"] = cms[:5] outdated, year = _outdated_copyright(html) result["outdated_copyright"] = outdated result["copyright_year"] = year status = page.get("status") result["has_working_website"] = isinstance(status, int) and 200 <= status < 400 if isinstance(status, int) and (status >= 400 or status < 200): result["status"] = "broken" elif isinstance(status, int): result["status"] = "working" result["html"] = html[:50000] if html else None return result def enrich_domain(domain: str) -> dict[str, Any]: """Bounded domain status: unknown unless resolution evidence is reliable.""" if not domain: return {"error": "missing_domain"} registrable = normalize_registrable_domain(domain) if registrable == "unknown": return {"domain": domain, "status": "unknown", "resolves": False} resolution = resolve_domain(domain) resolves = resolution.get("status") == "ok" has_web = resolves status = "registered" if resolves else ("likely_available" if not resolves else "unknown") return { "domain": registrable, "resolves": resolves, "has_web_service": has_web, "status": status, "resolution": resolution, } def enrich_contacts(html: str, url: str, *, suppressions=None, max_results: int = 50) -> list[dict[str, Any]]: """Public business contact signals with source URL and extraction timestamp.""" return extract_contacts(html, url, suppressions=suppressions, max_results=max_results)