MarketingTool/apps/api/app/domain.py

"""Pure, dependency-free prospect domain rules."""
from __future__ import annotations

import re
import unicodedata
from difflib import SequenceMatcher
from urllib.parse import urlparse

SCORE_VERSION = "mvp-1"
MATCH_SCORE_VERSION = "phase6-1"
_SOCIAL = {"facebook.com", "instagram.com", "linkedin.com", "twitter.com", "x.com", "youtube.com", "tiktok.com"}


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


def normalize_phone(value: str | None) -> str:
    raw = str(value or "").strip()
    if not raw:
        return ""
    # Keep a leading international plus and digits only; never invent a country
    # code for an unknown number. South African local and 00 prefixes are safe
    # canonicalization cases because their numbering plan is unambiguous.
    compact = re.sub(r"[^0-9+]", "", raw)
    if compact.startswith("00"):
        compact = "+" + compact[2:]
    if compact.startswith("+27"):
        rest = compact[3:]
        if rest.startswith("0"):
            rest = rest[1:]
        return "+27" + rest
    if compact.startswith("0") and len(compact) == 10:
        return "+27" + compact[1:]
    if compact.startswith("+"):
        return "+" + re.sub(r"\D", "", compact[1:])
    return re.sub(r"\D", "", compact)


def _location_part(value: object) -> str:
    text = " ".join(str(value or "").split()).strip().lower()
    return "".join(c for c in unicodedata.normalize("NFKD", text) if not unicodedata.combining(c))


def normalize_location(value: object = None, *, province=None, city=None, suburb=None) -> dict:
    if isinstance(value, dict):
        province, city, suburb = value.get("province", province), value.get("city", city), value.get("suburb", suburb)
    elif value is not None and not any(x is not None for x in (province, city, suburb)):
        province = value
    return {"province": _location_part(province), "city": _location_part(city), "suburb": _location_part(suburb)}


def normalize_business(raw: dict) -> dict:
    name = " ".join(str(raw.get("name", "")).split())
    email = str(raw.get("email", "")).strip().lower()
    website = str(raw.get("website", "")).strip()
    domain = normalize_domain(raw.get("website_domain") or website)
    phone = normalize_phone(raw.get("phone"))
    result = dict(raw)
    result.update({"name": name, "email": email, "website": website, "website_domain": domain, "phone": phone})
    result.update(normalize_location(raw.get("location", raw)))
    return result


def match_businesses(source: dict, candidates: list[dict], threshold: float = 0.72) -> list[dict]:
    """Return deterministic, explainable suggestions; this function never merges."""
    left = normalize_business(source)
    output = []
    for raw in candidates:
        right = normalize_business(raw)
        signals = []
        if left["website_domain"] and left["website_domain"] == right["website_domain"]:
            signals.append((1.0, "exact_website_domain"))
        if left["email"] and left["email"] == right["email"]:
            signals.append((1.0, "exact_email"))
        if left["phone"] and left["phone"] == right["phone"]:
            signals.append((1.0, "exact_phone"))
        if left["name"] and right["name"]:
            similarity = SequenceMatcher(None, re.sub(r"[^a-z0-9]", "", left["name"].lower()), re.sub(r"[^a-z0-9]", "", right["name"].lower())).ratio()
            if similarity >= 0.65: signals.append((similarity, "similar_name"))
        for field, reason in (("province", "same_province"), ("city", "same_city"), ("suburb", "same_suburb")):
            if left[field] and left[field] == right[field]: signals.append((0.08, reason))
        if not signals: continue
        exact = [s for s, r in signals if r.startswith("exact_")]
        name = next((s for s, r in signals if r == "similar_name"), 0.0)
        confidence = max(exact or [0.0]) if exact else min(0.99, 0.65 * name + sum(s for s, r in signals if r.startswith("same_")))
        if confidence >= threshold:
            output.append({"id": raw.get("id"), "confidence": round(confidence, 4), "reasons": [r for _, r in signals], "score_version": MATCH_SCORE_VERSION})
    return sorted(output, key=lambda x: (-x["confidence"], x["id"] if isinstance(x["id"], int) else str(x["id"])))


def classify_website(website_or_domain: str | None) -> str:
    domain = normalize_domain(website_or_domain)
    if not domain:
        return "missing"
    if any(domain == item or domain.endswith("." + item) for item in _SOCIAL):
        return "social_profile"
    return "business_site"


def score_business(business: dict) -> dict:
    b = normalize_business(business)
    score = 0
    factors = []
    if b.get("name"):
        score += 20; factors.append("named_business")
    site_class = classify_website(b.get("website_domain") or b.get("website"))
    if site_class == "business_site":
        score += 30; factors.append("business_site")
    if b.get("email"):
        score += 25; factors.append("email")
    if b.get("phone"):
        score += 15; factors.append("phone")
    if b.get("description"):
        score += 10; factors.append("description")
    return {"score": min(score, 100), "score_version": SCORE_VERSION, "factors": factors, "website_class": site_class}


def suppression_values(business: dict) -> set[str]:
    b = normalize_business(business)
    return {x for x in (b.get("email"), b.get("website_domain"), b.get("phone")) if x}


def is_suppressed(business: dict, suppressions: list[dict]) -> bool:
    values = suppression_values(business)
    for item in suppressions:
        kind, value = item.get("kind", ""), str(item.get("value", "")).strip().lower()
        if kind == "domain": value = normalize_domain(value)
        elif kind == "phone": value = normalize_phone(value)
        if value and value in values:
            return True
    return False


def deduplication_key(business: dict) -> tuple[str, str]:
    b = normalize_business(business)
    if b["website_domain"]: return ("domain", b["website_domain"])
    if b["email"]: return ("email", b["email"])
    if b["phone"]: return ("phone", b["phone"])
    return ("name", re.sub(r"[^a-z0-9]", "", b["name"].lower()))


def deduplicate_businesses(rows: list[dict]) -> list[dict]:
    chosen: dict[tuple[str, str], dict] = {}
    for raw in rows:
        item = normalize_business(raw)
        key = deduplication_key(item)
        if key not in chosen or sum(bool(v) for v in item.values()) > sum(bool(v) for v in chosen[key].values()):
            chosen[key] = item
    return list(chosen.values())