215 lines
7.5 KiB
Python
215 lines
7.5 KiB
Python
"""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)
|