build prospect intelligence platform MVP

apps/api/app/domain.py

@@ -0,0 +1,95 @@
+"""Pure, dependency-free prospect domain rules."""
+from __future__ import annotations
+
+import re
+from urllib.parse import urlparse
+
+SCORE_VERSION = "mvp-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:
+    return re.sub(r"[^0-9+]", "", (value or "").strip())
+
+
+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})
+    return result
+
+
+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())