96 lines
3.4 KiB
Python
96 lines
3.4 KiB
Python
|
"""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())
|