130 lines
6.6 KiB
Python
130 lines
6.6 KiB
Python
"""Bounded, public-only prospect discovery using an explicit seed allowlist.
|
|
|
|
There is deliberately no general web search or arbitrary URL input here: callers provide
|
|
at most a small set of public seed pages. Only 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
|
|
|
|
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, *, max_pages=MAX_PAGES, max_candidates=MAX_CANDIDATES):
|
|
if not isinstance(criteria, dict) or len(criteria) > 20: raise ValueError("invalid_criteria")
|
|
if not isinstance(seed_urls, list) or not 0 < len(seed_urls) <= MAX_SEEDS: raise ValueError("seed_urls_required")
|
|
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")
|
|
seeds = []
|
|
for raw in seed_urls:
|
|
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": "explicit_seed_allowlist", "seed_urls": seeds, "root_url": root}})
|
|
return {"candidates": results, "seeds": seeds, "pages_limit": max_pages, "candidate_limit": max_candidates}
|