MarketingTool/apps/api/app/domain_intelligence.py

"""Conservative, dependency-free domain intelligence helpers.

This module deliberately reports uncertainty rather than inferring DNS or registrar facts.
"""
from __future__ import annotations

import ipaddress
import re
import socket
import threading
import time
import unicodedata
from datetime import datetime, timezone
from difflib import SequenceMatcher
from urllib.parse import urlparse

# Small explicitly maintained PSL subset. Unknown suffixes are never guessed.
PUBLIC_SUFFIXES = frozenset({
    "com", "org", "net", "za", "co.za", "org.za", "net.za", "ac.za", "gov.za", "edu.za",
})
DEFAULT_TTL_SECONDS = 300
MAX_DOMAIN_LENGTH = 253
MAX_CANDIDATES = 20


def _host(value: str | None) -> str:
    raw = str(value or "").strip().lower()
    if not raw:
        return ""
    parsed = urlparse(raw if "://" in raw else "//" + raw)
    host = (parsed.hostname or "").rstrip(".").lower()
    if host.startswith("www."):
        host = host[4:]
    return host


def _valid_hostname(host: str) -> bool:
    if not host or len(host) > MAX_DOMAIN_LENGTH or "." not in host:
        return False
    try:
        ipaddress.ip_address(host)
        return False
    except ValueError:
        pass
    if any(len(label) > 63 or not re.fullmatch(r"[a-z0-9](?:[a-z0-9-]*[a-z0-9])?", label) for label in host.split(".")):
        return False
    return True


def normalize_registrable_domain(value: str | None) -> str:
    """Return the registrable domain, or ``unknown`` for unsupported/unsafe input."""
    host = _host(value)
    if not _valid_hostname(host):
        return "unknown"
    labels = host.split(".")
    suffix = next((".".join(labels[-n:]) for n in (3, 2, 1) if ".".join(labels[-n:]) in PUBLIC_SUFFIXES), None)
    if not suffix or len(labels) <= suffix.count(".") + 1:
        return "unknown"
    return ".".join(labels[-(suffix.count(".") + 2):])


def domain_info(value: str | None) -> dict:
    host = _host(value)
    registered = normalize_registrable_domain(value)
    return {"input": str(value or ""), "hostname": host, "registrable_domain": registered,
            "status": "ok" if registered != "unknown" else "unknown"}


def _metadata(ttl_seconds: int = DEFAULT_TTL_SECONDS) -> dict:
    now = time.time()
    checked = datetime.fromtimestamp(now, timezone.utc).replace(microsecond=0).isoformat()
    return {"checked_at": checked, "ttl_seconds": ttl_seconds,
            "cache_expires_at": datetime.fromtimestamp(now + ttl_seconds, timezone.utc).replace(microsecond=0).isoformat()}


def resolve_domain(domain: str, *, timeout: float = 3.0, resolver=None, ttl_seconds: int = DEFAULT_TTL_SECONDS) -> dict:
    """Resolve only a validated hostname using stdlib socket; never fetches a URL."""
    normalized = normalize_registrable_domain(domain)
    host = _host(domain)
    result = {"domain": normalized, "hostname": host, "status": "unknown", "addresses": [], **_metadata(ttl_seconds)}
    if normalized == "unknown" or not _valid_hostname(host):
        return result
    resolver = resolver or socket.getaddrinfo
    try:
        # getaddrinfo has no per-call timeout. Run it in a daemon worker so a
        # resolver stall cannot block an HTTP handler indefinitely.
        outcome = {}
        def lookup():
            try: outcome["records"] = resolver(host, None, 0, socket.SOCK_STREAM)
            except BaseException as exc: outcome["exception"] = exc
        worker = threading.Thread(target=lookup, daemon=True)
        worker.start(); worker.join(max(0.01, float(timeout)))
        if worker.is_alive():
            result["status"] = "timeout"
            return result
        if "exception" in outcome: raise outcome["exception"]
        records = outcome.get("records", [])
        addresses = sorted({str(item[4][0]) for item in records if item and len(item) > 4 and item[0] in (socket.AF_INET, socket.AF_INET6)})
        result["addresses"] = addresses
        result["status"] = "ok" if addresses else "unknown"
    except socket.gaierror as exc:
        code = getattr(exc, "errno", None)
        if code is None and exc.args: code = exc.args[0]
        result["status"] = "nxdomain" if code in (socket.EAI_NONAME, socket.EAI_NODATA, -2, -5) else "error"
        result["error_code"] = code
    except (TimeoutError, socket.timeout):
        result["status"] = "timeout"
    except OSError as exc:
        result["status"] = "error"; result["error_code"] = exc.__class__.__name__
    return result


def capability_hook(domain: str, capability: str, *, resolver=None, ttl_seconds: int = DEFAULT_TTL_SECONDS) -> dict:
    """MX/NS/TXT hook. No optional resolver means not_configured, not empty."""
    normalized = normalize_registrable_domain(domain)
    base = {"domain": normalized, "capability": capability, "status": "unknown", "records": [], **_metadata(ttl_seconds)}
    if normalized == "unknown": return base
    if capability not in {"mx", "ns", "txt"}: base.update(status="error", error_code="unsupported_capability"); return base
    if resolver is None:
        base.update(status="not_configured", reason="resolver_not_configured")
        return base
    try:
        records = resolver(normalized, capability)
        base.update(status="ok" if records else "unknown", records=list(records or []))
    except TimeoutError: base["status"] = "timeout"
    except socket.gaierror: base["status"] = "nxdomain"
    except Exception as exc: base.update(status="error", error_code=exc.__class__.__name__)
    return base


def resolve_mx(domain: str, *, resolver=None, ttl_seconds: int = DEFAULT_TTL_SECONDS) -> dict:
    return capability_hook(domain, "mx", resolver=resolver, ttl_seconds=ttl_seconds)


def resolve_ns(domain: str, *, resolver=None, ttl_seconds: int = DEFAULT_TTL_SECONDS) -> dict:
    return capability_hook(domain, "ns", resolver=resolver, ttl_seconds=ttl_seconds)


def resolve_txt(domain: str, *, resolver=None, ttl_seconds: int = DEFAULT_TTL_SECONDS) -> dict:
    return capability_hook(domain, "txt", resolver=resolver, ttl_seconds=ttl_seconds)


registrable_domain = normalize_registrable_domain


def association_confidence(candidate_domain: str, observed_domain: str, *, business_name: str = "") -> dict:
    candidate = normalize_registrable_domain(candidate_domain); observed = normalize_registrable_domain(observed_domain)
    if candidate == "unknown" or observed == "unknown": return {"level":"unknown", "score":0.0, "reasons":["unsupported_domain"]}
    if candidate == observed: return {"level":"high", "score":1.0, "reasons":["same_registrable_domain"]}
    token = re.sub(r"[^a-z0-9]", "", unicodedata.normalize("NFKD", str(business_name).lower()))
    score = SequenceMatcher(None, token, candidate.split(".")[0]).ratio() if token else 0.0
    return {"level":"medium" if score >= .8 else "low", "score":round(score, 4), "reasons":["name_domain_similarity"] if score >= .8 else ["different_registrable_domain"]}


def _slug(value: object) -> str:
    text = unicodedata.normalize("NFKD", str(value or "")).encode("ascii", "ignore").decode().lower()
    words = re.findall(r"[a-z0-9]+", text)
    return "-".join(words)[:40].strip("-")


def generate_candidate_domains(business_name: str, service: str = "", location: str = "", *, suffixes=("co.za", "com"), limit: int = MAX_CANDIDATES) -> list[str]:
    limit = max(0, min(int(limit), MAX_CANDIDATES)); name, svc, loc = _slug(business_name), _slug(service), _slug(location)
    if not name: return []
    parts = [name, f"{name}-{svc}" if svc else "", f"{name}-{loc}" if loc else "", f"{svc}-{loc}" if svc and loc else ""]
    if svc: parts += [f"{name}{svc}"]
    out=[]
    for base in parts:
        if not base or len(base) > 63: continue
        for suffix in suffixes:
            if suffix not in PUBLIC_SUFFIXES: continue
            domain = f"{base}.{suffix}"
            if normalize_registrable_domain(domain) != "unknown" and domain not in out: out.append(domain)
            if len(out) >= limit: return out
    return out