MarketingTool/apps/api/app/sources.py

"""Safe, tenant-neutral source adapter contracts.

Network adapters are intentionally capability gated: configuration must explicitly
approve public access, terms, rate limits and credentials (where required).
Adapters never emit or persist credential values.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any, Mapping, Protocol, Sequence
import csv, io, random, time, json
from urllib.parse import urlencode, urlparse
from urllib.request import Request, urlopen

try:
    from .website_scanner import validate_url
except ImportError:
    from website_scanner import validate_url

SECRET_KEYS = {"password", "passwd", "secret", "token", "api_key", "apikey", "authorization", "private_key", "credential"}
NETWORK_KINDS = {"google_places", "bing_local", "approved_directory", "public_website", "permitted_social", "ct_logs", "dns", "rdap"}


def contains_secret(value: Any, path: str = "") -> str | None:
    if isinstance(value, Mapping):
        for key, child in value.items():
            key_text = str(key).lower()
            if key_text in SECRET_KEYS or any(x in key_text for x in ("password", "token", "secret", "api_key", "private_key")):
                return path + str(key)
            found = contains_secret(child, path + str(key) + ".")
            if found: return found
    elif isinstance(value, Sequence) and not isinstance(value, (str, bytes, bytearray)):
        for index, child in enumerate(value):
            found = contains_secret(child, path + str(index) + ".")
            if found: return found
    return None

@dataclass(frozen=True)
class ValidationResult:
    valid: bool
    errors: list[str] = field(default_factory=list)

@dataclass(frozen=True)
class DiscoveryPage:
    records: list[dict[str, str]]
    next_cursor: str | None = None
    metadata: dict[str, Any] = field(default_factory=dict)

@dataclass(frozen=True)
class SourceHealth:
    status: str = "unknown"
    consecutive_failures: int = 0
    circuit_open: bool = False
    last_error: str | None = None

@dataclass(frozen=True)
class NormalizedRecord:
    name: str = ""
    website: str = ""
    email: str = ""
    phone: str = ""
    description: str = ""
    location: str = ""
    source_url: str = ""
    provenance: str = ""
    raw: Mapping[str, Any] = field(default_factory=dict)

class DiscoverySource(Protocol):
    source_code: str
    display_name: str
    def validate_config(self, config: Mapping[str, Any]) -> ValidationResult: ...
    def discover(self, config: Mapping[str, Any], cursor: str | None = None) -> DiscoveryPage: ...
    def health_check(self, config: Mapping[str, Any]) -> SourceHealth: ...

_FIELDS = ("name", "website", "email", "phone", "description", "location")
def normalize_record(row: Mapping[str, Any]) -> dict[str, str]:
    result = {field: str(row.get(field, "") or "").strip() for field in _FIELDS}
    aliases = {"company":"name", "business":"name", "url":"website", "domain":"website", "address":"location", "address_line":"location"}
    for key, target in aliases.items():
        if not result[target] and row.get(key) is not None: result[target] = str(row[key]).strip()
    return result

def normalized_record(row: Mapping[str, Any], *, source_url="", provenance="") -> NormalizedRecord:
    x = normalize_record(row)
    return NormalizedRecord(**x, source_url=source_url, provenance=provenance, raw=dict(row))

def exponential_backoff(attempt: int, base: float = 0.5, maximum: float = 30.0, jitter: float = 0.25) -> float:
    """Bounded exponential delay with symmetric jitter; no sleeping occurs here."""
    delay = min(maximum, base * (2 ** max(0, int(attempt))))
    return max(0.0, delay + random.uniform(-jitter * delay, jitter * delay))

def backoff_delay(attempt: int, base: float = 0.5, maximum: float = 30.0, jitter: float = 0.25) -> float:
    return exponential_backoff(attempt, base, maximum, jitter)

def circuit_is_open(consecutive_failures: int, threshold: int = 3) -> bool:
    return int(consecutive_failures) >= max(1, int(threshold))

def quota_remaining(used: int, limit: int | None) -> int | None:
    """Return remaining quota, clamped so malformed values fail closed."""
    if limit is None: return None
    return max(0, int(limit) - max(0, int(used)))

def quota_allowed(used: int, limit: int | None) -> bool:
    remaining = quota_remaining(used, limit)
    return remaining is None or remaining > 0

def rate_limit_delay(last_request: float | None, min_interval: float) -> float:
    if last_request is None: return 0.0
    return max(0.0, float(min_interval) - (time.monotonic() - last_request))

class _Base:
    kind = ""
    source_code = ""
    display_name = ""
    available = True
    optional = False
    requires_credentials = False
    def validate_config(self, config):
        if not isinstance(config, Mapping): return ValidationResult(False, ["config must be an object"])
        found = contains_secret(config)
        if found: return ValidationResult(False, [f"secret field is not permitted: {found}"])
        return ValidationResult(True)
    validate = validate_config
    def health_check(self, config):
        result = self.validate_config(config)
        return SourceHealth("healthy" if result.valid else "unhealthy", last_error=None if result.valid else "; ".join(result.errors))
    def discover(self, config, cursor=None):
        result = self.validate_config(config)
        if not result.valid: raise ValueError(result.errors[0])
        raise RuntimeError("source_not_configured")

class ManualSource(_Base):
    kind = source_code = "manual"; display_name = "Manual records"
    def validate_config(self, config):
        result = super().validate_config(config)
        if not result.valid: return result
        if not isinstance(config.get("rows"), list): return ValidationResult(False, ["rows must be a list"])
        return ValidationResult(True)
    def discover(self, config, cursor=None):
        result = self.validate_config(config)
        if not result.valid: raise ValueError(result.errors[0])
        return DiscoveryPage([normalize_record(row) for row in config["rows"] if isinstance(row, Mapping)], metadata={"adapter":self.source_code})

class CsvSource(_Base):
    kind = source_code = "csv"; display_name = "CSV import"
    def validate_config(self, config):
        result = super().validate_config(config)
        if not result.valid: return result
        if not isinstance(config.get("csv"), str): return ValidationResult(False, ["csv must be text"])
        try:
            reader = csv.DictReader(io.StringIO(config["csv"]))
            if not reader.fieldnames: return ValidationResult(False, ["CSV header is required"])
        except csv.Error as exc: return ValidationResult(False, [f"invalid CSV: {exc}"])
        return ValidationResult(True)
    def discover(self, config, cursor=None):
        result = self.validate_config(config)
        if not result.valid: raise ValueError(result.errors[0])
        reader = csv.DictReader(io.StringIO(config["csv"].replace("\r\n", "\n")))
        records = []
        for row in reader:
            normalized = {str(k).strip().lower(): v for k, v in row.items()}
            if any(str(v or '').strip() for v in normalized.values()):
                records.append(normalize_record(normalized))
        return DiscoveryPage(records, metadata={"adapter":self.source_code,"columns":reader.fieldnames or [],"record_count":len(records)})


class _HttpJsonSource(_Base):
    """Small, bounded JSON client used only for public standards-based sources."""
    max_bytes = 256 * 1024
    timeout = 8

    def _get_json(self, url):
        safe = validate_url(url)
        request = Request(safe, headers={"User-Agent": "ProspectOS/0.1 (+public-source-research)"})
        with urlopen(request, timeout=self.timeout) as response:
            body = response.read(self.max_bytes + 1)
            if len(body) > self.max_bytes:
                raise ValueError("source_response_too_large")
            return json.loads(body.decode("utf-8", "replace")), safe


class PublicWebsiteSource(_Base):
    kind = source_code = "public_website"
    display_name = "Public website"

    def validate_config(self, config):
        result = super().validate_config(config)
        if not result.valid: return result
        urls = config.get("urls", config.get("url", []))
        if isinstance(urls, str): urls = [urls] if urls.strip() else []
        if not isinstance(urls, list) or not urls or len(urls) > 50:
            return ValidationResult(False, ["urls must contain 1 to 50 public HTTP(S) URLs"])
        for value in urls:
            try: validate_url(str(value))
            except ValueError: return ValidationResult(False, ["unsafe public website URL"])
        return ValidationResult(True)

    def discover(self, config, cursor=None):
        result = self.validate_config(config)
        if not result.valid: raise ValueError(result.errors[0])
        urls = config.get("urls", config.get("url"))
        if isinstance(urls, str): urls = [urls]
        records = []
        for raw_url in urls[:50]:
            safe = validate_url(str(raw_url))
            request = Request(safe, headers={"User-Agent": "ProspectOS/0.1 (+public-source-research)"})
            with urlopen(request, timeout=8) as response:
                body = response.read(128 * 1024).decode("utf-8", "replace")
                final_url = response.geturl()
            from html.parser import HTMLParser
            parser = HTMLParser()
            title = urlparse(final_url).hostname or safe
            records.append(normalize_record({"name": title, "website": final_url, "description": body[:1000]}))
        return DiscoveryPage(records, metadata={"adapter": self.source_code, "record_count": len(records), "bounded": True})


class CtLogsSource(_HttpJsonSource):
    kind = source_code = "ct_logs"
    display_name = "Certificate transparency logs"

    def validate_config(self, config):
        result = super().validate_config(config)
        if not result.valid: return result
        query = str(config.get("domain", config.get("query", ""))).strip()
        if not query or len(query) > 253 or any(ch in query for ch in "\r\n"):
            return ValidationResult(False, ["domain or query is required"])
        return ValidationResult(True)

    def discover(self, config, cursor=None):
        result = self.validate_config(config)
        if not result.valid: raise ValueError(result.errors[0])
        query = str(config.get("domain", config.get("query"))).strip()
        endpoint = "https://crt.sh/?" + urlencode({"q": "%25." + query.lstrip("%.") if not query.startswith("%") else query, "output": "json"})
        payload, source_url = self._get_json(endpoint)
        if not isinstance(payload, list): raise ValueError("invalid_ct_response")
        records, seen = [], set()
        for item in payload[:500]:
            names = str(item.get("name_value", "")) if isinstance(item, dict) else ""
            for name in names.splitlines():
                name = name.strip().lower().lstrip("*.")
                if not name or name in seen or "." not in name: continue
                seen.add(name); records.append(normalize_record({"name": name, "website": "https://" + name}))
        return DiscoveryPage(records[:100], metadata={"adapter": self.source_code, "source_url": source_url, "record_count": len(records), "signal_only": True})


class DnsSource(_Base):
    kind = source_code = "dns"
    display_name = "DNS"

    def validate_config(self, config):
        result = super().validate_config(config)
        if not result.valid: return result
        domains = config.get("domains", config.get("domain", []))
        if isinstance(domains, str): domains = [domains] if domains.strip() else []
        if not isinstance(domains, list) or not domains or len(domains) > 100: return ValidationResult(False, ["domains must contain 1 to 100 names"])
        return ValidationResult(True)

    def discover(self, config, cursor=None):
        result = self.validate_config(config)
        if not result.valid: raise ValueError(result.errors[0])
        import socket
        domains = config.get("domains", config.get("domain")); domains = [domains] if isinstance(domains, str) else domains
        records = []
        for domain in domains[:100]:
            domain = str(domain).strip().lower().rstrip(".")
            if not domain or "." not in domain: continue
            try: addresses = sorted({item[4][0] for item in socket.getaddrinfo(domain, 443, type=socket.SOCK_STREAM)})
            except socket.gaierror: addresses = []
            records.append(normalize_record({"name": domain, "website": "https://" + domain, "description": json.dumps({"a_aaaa": addresses})}))
        return DiscoveryPage(records, metadata={"adapter": self.source_code, "record_count": len(records), "dns_status_only": True})


class RdapSource(_HttpJsonSource):
    kind = source_code = "rdap"
    display_name = "RDAP"

    def validate_config(self, config):
        result = super().validate_config(config)
        if not result.valid: return result
        domain = str(config.get("domain", "")).strip()
        if not domain or "." not in domain: return ValidationResult(False, ["domain is required"])
        return ValidationResult(True)

    def discover(self, config, cursor=None):
        result = self.validate_config(config)
        if not result.valid: raise ValueError(result.errors[0])
        domain = str(config["domain"]).strip().lower().rstrip(".")
        payload, source_url = self._get_json("https://rdap.org/domain/" + domain)
        return DiscoveryPage([normalize_record({"name": domain, "website": "https://" + domain, "description": json.dumps({"rdap": payload}, default=str)[:1000]})], metadata={"adapter": self.source_code, "source_url": source_url, "record_count": 1, "registration_signal_only": True})

class GatedSource(_Base):
    available = False
    optional = True
    requires_credentials = True
    required = "approved"
    def validate_config(self, config):
        result = super().validate_config(config)
        if not result.valid: return result
        if config.get("approved") is not True: return ValidationResult(False, ["source approval is required"])
        if config.get("public_access") is not True: return ValidationResult(False, ["public_access approval is required"])
        if config.get("terms_accepted") is not True: return ValidationResult(False, ["terms_accepted is required"])
        if self.source_code in {"google_places", "bing_local"} and not config.get("credential_ref"):
            return ValidationResult(False, ["approved credential_ref is required"])
        if not isinstance(config.get("rate_limit", 1), (int, float)) or config.get("rate_limit", 1) <= 0:
            return ValidationResult(False, ["positive rate_limit is required"])
        return ValidationResult(True)
    def discover(self, config, cursor=None):
        result = self.validate_config(config)
        if not result.valid: raise ValueError(result.errors[0])
        # Network execution is delegated to an explicitly approved provider; never guess.
        raise RuntimeError("network_adapter_not_configured")


class ApprovedDirectorySource(GatedSource):
    kind = source_code = "approved_directory"
    display_name = "Free public directories"
    available = True
    requires_credentials = False
    optional = False

    def validate_config(self, config):
        result=super().validate_config(config)
        if not result.valid:return result
        provider=str(config.get("provider", "")).strip().lower()
        if provider not in {"openstreetmap", "wikidata", "common_crawl"}: return ValidationResult(False,["provider must be openstreetmap, wikidata, or common_crawl"])
        if not str(config.get("query", "")).strip(): return ValidationResult(False,["query is required"])
        return ValidationResult(True)

    def discover(self, config, cursor=None):
        result=self.validate_config(config)
        if not result.valid: raise ValueError(result.errors[0])
        provider=str(config["provider"]).lower(); query=str(config["query"]).strip(); limit=max(1,min(100,int(config.get("max_records",50))))
        if provider == "openstreetmap":
            area=str(config.get("location", "South Africa")).strip()
            overpass='[out:json][timeout:25];area["name"="%s"]->.a;(nwr["name"](area.a););out center tags;'%(area.replace('"',''))
            req=Request("https://overpass-api.de/api/interpreter",data=overpass.encode(),method="POST",headers={"Content-Type":"application/x-www-form-urlencoded","User-Agent":"ProspectOS/0.1"})
            with urlopen(req,timeout=30) as response: payload=json.loads(response.read(2*1024*1024).decode("utf-8","replace"))
            records=[]
            for element in payload.get("elements",[]):
                tags=element.get("tags",{}); name=tags.get("name","")
                if not name or (query.lower() not in name.lower() and query.lower() not in str(tags).lower()): continue
                records.append(normalize_record({"name":name,"website":tags.get("website") or tags.get("contact:website", ""),"phone":tags.get("phone") or tags.get("contact:phone", ""),"email":tags.get("email") or tags.get("contact:email", ""),"location":", ".join(x for x in (tags.get("addr:street"),tags.get("addr:city"),tags.get("addr:postcode")) if x),"description":"OpenStreetMap public listing"}))
            return DiscoveryPage(records[:limit],metadata={"adapter":self.source_code,"provider":provider,"record_count":len(records)})
        if provider == "wikidata":
            sparql=query if query.lower().startswith("select") else 'SELECT ?item ?itemLabel ?website WHERE {?item rdfs:label ?itemLabel. FILTER(CONTAINS(LCASE(?itemLabel), LCASE("%s"))). OPTIONAL {?item wdt:P856 ?website} FILTER(LANG(?itemLabel)="en")} LIMIT %d'%(query.replace('"',''),limit)
            url="https://query.wikidata.org/sparql?format=json&"+urlencode({"query":sparql})
            payload, _ = _HttpJsonSource()._get_json(url); records=[normalize_record({"name":x.get("itemLabel",{}).get("value",""),"website":x.get("website",{}).get("value","")}) for x in payload.get("results",{}).get("bindings",[])]
            return DiscoveryPage(records[:limit],metadata={"adapter":self.source_code,"provider":provider,"record_count":len(records)})
        index="https://index.commoncrawl.org/CC-MAIN-2026-30-index?url="+query+"&output=json&filter=status:200&collapse=urlkey"
        payload, _ = _HttpJsonSource()._get_json(index); records=[normalize_record({"name":str(x.get("url","")).split('/')[2] if '://' in str(x.get("url","")) else x.get("url", ""),"website":x.get("url","")}) for x in (payload if isinstance(payload,list) else [])]
        return DiscoveryPage(records[:limit],metadata={"adapter":self.source_code,"provider":provider,"record_count":len(records)})

class GooglePlacesSource(GatedSource):
    kind = source_code = "google_places"
    display_name = "Google Places"
    available = True

    def validate_config(self, config):
        result = super().validate_config(config)
        if not result.valid: return result
        if not str(config.get("query", "")).strip(): return ValidationResult(False, ["query is required"])
        return ValidationResult(True)

    def discover(self, config, cursor=None):
        result = self.validate_config(config)
        if not result.valid: raise ValueError(result.errors[0])
        api_key = str(config.get("_api_key", "")).strip()
        if not api_key: raise ValueError("Google Places API key is not configured")
        body = {"textQuery": str(config["query"]).strip(), "pageSize": max(1, min(20, int(config.get("max_records", 20))))}
        if config.get("region_code"): body["regionCode"] = str(config["region_code"]).upper()[:2]
        request = Request("https://places.googleapis.com/v1/places:searchText", data=json.dumps(body).encode(), method="POST", headers={"Content-Type":"application/json", "X-Goog-Api-Key":api_key, "X-Goog-FieldMask":"places.displayName,places.websiteUri,places.nationalPhoneNumber,places.internationalPhoneNumber,places.formattedAddress,places.googleMapsUri"})
        with urlopen(request, timeout=12) as response:
            payload=json.loads(response.read(2*1024*1024).decode("utf-8", "replace"))
        records=[]
        for place in payload.get("places", [])[:20]:
            name=(place.get("displayName") or {}).get("text", "")
            records.append(normalize_record({"name":name, "website":place.get("websiteUri", ""), "phone":place.get("nationalPhoneNumber") or place.get("internationalPhoneNumber", ""), "location":place.get("formattedAddress", ""), "source_url":place.get("googleMapsUri", "") , "description":"Google Places result"}))
        return DiscoveryPage(records, metadata={"adapter":self.source_code,"record_count":len(records),"provider":"google_places"})

class OpenStreetMapSource(ApprovedDirectorySource):
    kind = source_code = "openstreetmap"
    display_name = "OpenStreetMap / Overpass"
    optional = False
    def discover(self, config, cursor=None):
        page = super().discover({**dict(config), "provider": "openstreetmap"}, cursor)
        return DiscoveryPage(page.records, page.next_cursor, {**page.metadata, "adapter": self.source_code})

class WikidataSource(ApprovedDirectorySource):
    kind = source_code = "wikidata"
    display_name = "Wikidata"
    optional = False
    def discover(self, config, cursor=None):
        page = super().discover({**dict(config), "provider": "wikidata"}, cursor)
        return DiscoveryPage(page.records, page.next_cursor, {**page.metadata, "adapter": self.source_code})

class CommonCrawlSource(ApprovedDirectorySource):
    kind = source_code = "common_crawl"
    display_name = "Common Crawl index"
    optional = False
    def discover(self, config, cursor=None):
        page = super().discover({**dict(config), "provider": "common_crawl"}, cursor)
        return DiscoveryPage(page.records, page.next_cursor, {**page.metadata, "adapter": self.source_code})

def _gated(code, name):
    return type(name.replace(" ", ""), (GatedSource,), {"kind":code, "source_code":code, "display_name":name})
BingLocalSource = _gated("bing_local", "Bing / approved local API")
PermittedSocialSource = _gated("permitted_social", "Permitted social")

ADAPTERS = {x.source_code: x for x in (ManualSource, CsvSource, GooglePlacesSource, BingLocalSource, ApprovedDirectorySource, OpenStreetMapSource, WikidataSource, CommonCrawlSource, PublicWebsiteSource, PermittedSocialSource, CtLogsSource, DnsSource, RdapSource)}
# common aliases used by clients
ADAPTER_REGISTRY = ADAPTERS

def adapter_for(kind: str) -> DiscoverySource:
    try: return ADAPTERS[str(kind).strip().lower()]()
    except KeyError: raise ValueError("unsupported source kind")

def available_adapters() -> list[dict[str, object]]:
    return [{"source_code": cls.source_code, "display_name": cls.display_name,
             "available": bool(getattr(cls, "available", False)),
             "optional": bool(getattr(cls, "optional", False)),
             "requires_credentials": bool(getattr(cls, "requires_credentials", False))}
            for cls in ADAPTERS.values()]