MarketingTool/apps/api/app/discovery.py

"""Bounded, public-only prospect discovery with optional criteria search.

There is deliberately no general web search or arbitrary URL input here: callers provide
at most a small set of public seed pages. When seeds are omitted, a configured search
provider supplies bounded seed URLs. Links found on those seeds become candidate sites,
and subsequent crawling is same-origin, bounded, and SSRF-checked by the scanner.
"""
from __future__ import annotations

import re
from html.parser import HTMLParser
from urllib.parse import urljoin, urldefrag, urlparse

from .contact_extractor import extract_contacts
from .website_scanner import MAX_BYTES, MAX_REDIRECTS, _fetch, validate_url
import os
from .ai_research import research as ai_research
from .search_provider import search as search_provider  # deprecated compatibility adapter

MAX_SEEDS = 5
MAX_PAGES = 20
MAX_CANDIDATES = 50
MAX_HTML = MAX_BYTES


class _Links(HTMLParser):
    def __init__(self):
        super().__init__(convert_charrefs=True)
        self.links = []
        self.title = ""
        self.headings = []
        self._tag = ""
        self._buf = []

    def handle_starttag(self, tag, attrs):
        tag = tag.lower(); self._tag = tag
        if tag in {"a", "link"}:
            attrs = {str(k).lower(): str(v or "") for k, v in attrs}
            if attrs.get("href"): self.links.append(attrs["href"])
        if tag in {"title", "h1", "h2", "h3"}: self._buf = []

    def handle_data(self, data):
        if self._tag in {"title", "h1", "h2", "h3"}: self._buf.append(data)

    def handle_endtag(self, tag):
        tag = tag.lower()
        if tag == "title" and self._buf: self.title = " ".join("".join(self._buf).split())[:500]
        if tag in {"h1", "h2", "h3"} and self._buf: self.headings.append(" ".join("".join(self._buf).split())[:300])
        self._tag = ""


def _html(value: bytes) -> str:
    return value[:MAX_HTML].decode("utf-8", "replace")


def _same_site(url, root):
    return (urlparse(url).hostname or "").lower().rstrip(".") == (urlparse(root).hostname or "").lower().rstrip(".")


def _criteria_match(text, criteria):
    keywords = criteria.get("keywords", criteria.get("keyword", []))
    if isinstance(keywords, str): keywords = [keywords]
    if not isinstance(keywords, list) or len(keywords) > 20: raise ValueError("invalid_criteria")
    haystack = text.lower()
    return not keywords or all(str(k).strip().lower() in haystack for k in keywords if str(k).strip())


def discover(criteria, seed_urls=None, *, max_pages=MAX_PAGES, max_candidates=MAX_CANDIDATES):
    if not isinstance(criteria, dict) or len(criteria) > 20: raise ValueError("invalid_criteria")
    try: max_pages = int(max_pages); max_candidates = int(max_candidates)
    except (TypeError, ValueError): raise ValueError("invalid_limits")
    if not 1 <= max_pages <= MAX_PAGES or not 1 <= max_candidates <= MAX_CANDIDATES: raise ValueError("invalid_limits")
    if seed_urls is None:
        try:
            seeds = ai_research(criteria, max_candidates)
            mechanism = "ai_web_research_provider"
        except Exception as exc:
            # SEARCH_PROVIDER_* is retained only as a migration adapter. It is
            # never reported as the primary provider and can be removed later.
            if os.environ.get("SEARCH_PROVIDER_URL", "").strip() and getattr(exc, "args", (None,))[0] == "not_configured":
                seeds = search_provider(criteria, max_candidates)
                mechanism = "criteria_search_provider"  # legacy provenance label
            else:
                raise
    else:
        if not isinstance(seed_urls, list) or not 0 < len(seed_urls) <= MAX_SEEDS: raise ValueError("seed_urls_required")
        seeds = list(seed_urls)
        mechanism = "explicit_seed_allowlist"
    raw_seeds = list(seeds)
    seeds = []
    for raw in raw_seeds:
        try: safe = validate_url(raw)
        except ValueError as exc: raise ValueError("unsafe_seed_url") from exc
        if safe not in seeds: seeds.append(safe)
    candidates = []
    for seed in seeds:
        fetched = _fetch(seed, max_bytes=MAX_HTML, max_redirects=MAX_REDIRECTS)
        if fetched.get("content_type") not in {"text/html", "application/xhtml+xml"}: continue
        parser = _Links(); parser.feed(_html(fetched["body"]))
        seed_text = " ".join([parser.title, *parser.headings])
        linked = []
        for link in parser.links:
            target = urldefrag(urljoin(seed, link))[0]
            if not target or target.startswith(("mailto:", "tel:", "javascript:")): continue
            try: target = validate_url(target)
            except ValueError: continue
            if target not in linked and target not in seeds: linked.append(target)
            if len(linked) >= max_candidates: break
        # A seed acts as a directory/index when it links outward. If it has no
        # usable links, it may itself be the explicitly allowlisted business site.
        if linked:
            candidates.extend(x for x in linked if x not in candidates)
        elif _criteria_match(seed_text, criteria):
            candidates.append(seed)
    results = []
    seen = set()
    for candidate in candidates[:max_candidates]:
        root = candidate; queue = [candidate]; pages = []
        while queue and len(pages) < max_pages:
            page = queue.pop(0)
            if page in {x["url"] for x in pages}: continue
            try: fetched = _fetch(page, max_bytes=MAX_HTML, max_redirects=MAX_REDIRECTS)
            except ValueError: continue
            if fetched.get("content_type") not in {"text/html", "application/xhtml+xml"}: continue
            html = _html(fetched["body"]); parser = _Links(); parser.feed(html)
            pages.append({"url": page, "final_url": fetched.get("final_url") or page, "html": html, "title": parser.title, "headings": parser.headings, "status": fetched.get("status")})
            for link in parser.links:
                target = urldefrag(urljoin(page, link))[0]
                if target and _same_site(target, root):
                    try: target = validate_url(target)
                    except ValueError: continue
                    if target not in {x["url"] for x in pages} and target not in queue: queue.append(target)
            if len(queue) + len(pages) >= max_pages: queue = queue[:max(0, max_pages - len(pages))]
        if not pages: continue
        text = " ".join(x["title"] + " " + " ".join(x["headings"]) for x in pages)
        if not _criteria_match(text, criteria): continue
        domain = (urlparse(root).hostname or "").lower().removeprefix("www.")
        if domain in seen: continue
        seen.add(domain)
        contacts = []
        evidence = []
        for page in pages:
            contacts.extend(extract_contacts(page["html"], page["url"], max_results=100))
            claim = page["title"] or (page["headings"][0] if page["headings"] else "Public business page")
            evidence.append({"kind": "discovery_page", "url": page["url"], "claim": claim, "provenance": "scoped_discovery"})
        deduped = {(x["kind"], x["value"]): x for x in contacts}
        name = next((x["headings"][0] for x in pages if x["headings"]), next((x["title"] for x in pages if x["title"]), domain))
        results.append({"name": name[:200], "website": root, "website_domain": domain, "description": text[:1000], "contacts": list(deduped.values())[:100], "evidence": evidence, "pages": pages, "pages_crawled": len(pages), "provenance": {"mechanism": mechanism, "seed_urls": seeds, "root_url": root}})
    return {"candidates": results, "seeds": seeds, "pages_limit": max_pages, "candidate_limit": max_candidates}