"""Optional Browser Use investigator for public website candidates. This module is deliberately isolated and imports browser-use only when called. It performs bounded, headless public research. It does not log in, solve CAPTCHAs, rotate proxies, or continue after an access challenge. """ from __future__ import annotations import asyncio import json import os import re from typing import Any from urllib.parse import urlparse _BLOCKED_HOSTS = { "facebook.com", "instagram.com", "linkedin.com", "youtube.com", "x.com", "twitter.com", "google.com", "google.co.za", "bing.com", "brave.com", "duckduckgo.com", "startpage.com", "yelp.com", "yellowpages.com", } def _normalise_url(value: Any) -> str: text = str(value or "").strip() if not text: return "" if text.startswith("//"): text = "https:" + text if not re.match(r"^https?://", text, re.I): text = "https://" + text parsed = urlparse(text) if parsed.scheme not in {"http", "https"} or not parsed.hostname: return "" host = parsed.hostname.lower().rstrip(".") if host in {"localhost", "127.0.0.1", "::1"} or host.endswith(".local"): return "" if any(host == blocked or host.endswith("." + blocked) for blocked in _BLOCKED_HOSTS): return "" return text[:1000] def _parse_agent_result(raw: Any) -> list[dict[str, Any]]: """Parse and normalise the agent's final JSON candidate list.""" text = str(raw or "").strip() if text.startswith("```"): text = re.sub(r"^```(?:json)?\s*|\s*```$", "", text, flags=re.I | re.S).strip() try: payload = json.loads(text) except (TypeError, ValueError, json.JSONDecodeError): start, end = text.find("{"), text.rfind("}") if start < 0 or end <= start: return [] try: payload = json.loads(text[start:end + 1]) except (TypeError, ValueError, json.JSONDecodeError): return [] if not isinstance(payload, dict): return [] raw_candidates = payload.get("candidates") if not isinstance(raw_candidates, list): return [] output: list[dict[str, Any]] = [] seen: set[str] = set() for item in raw_candidates[:10]: if not isinstance(item, dict): continue url = _normalise_url(item.get("url") or item.get("website") or item.get("domain")) if not url: continue domain = (urlparse(url).hostname or "").lower().removeprefix("www.") if not domain or domain in seen: continue seen.add(domain) try: confidence = max(0, min(100, int(round(float(item.get("confidence", 0)))))) except (TypeError, ValueError): confidence = 0 evidence = item.get("evidence") if not isinstance(evidence, list): evidence = [] output.append({ "url": url, "domain": domain, "confidence": confidence, "evidence": [str(value).strip()[:240] for value in evidence if str(value).strip()][:8], "status": str(item.get("status") or "candidate").strip()[:80], }) return sorted(output, key=lambda item: item["confidence"], reverse=True) def investigate_websites( prospect: dict[str, Any], *, base_url: str, api_key: str, model: str, user_agent: str, ) -> list[dict[str, Any]]: """Run one bounded Browser Use investigation and return candidates.""" if os.environ.get("BROWSER_USE_ENABLED", "1").strip().lower() in {"0", "false", "no", "off"}: raise RuntimeError("browser_use_disabled") if not base_url or not model: raise RuntimeError("browser_use_requires_ai_provider_and_model") if not api_key: raise RuntimeError("browser_use_requires_ai_api_key") try: from browser_use import Agent, BrowserProfile, ChatOpenAI except ImportError as exc: raise RuntimeError("browser_use_not_installed_rebuild_container") from exc name = str(prospect.get("name") or "").strip() address = str(prospect.get("address") or "").strip() category = str(prospect.get("category") or "").strip() phone = str(prospect.get("phone") or "").strip() emails = str(prospect.get("emails") or "").strip() task = f""" You are a bounded public-data website investigator. Find possible official websites for this one business: Business name: {name} Category/services: {category} Address/location: {address} Phone: {phone} Public email(s): {emails} Use only public, unauthenticated pages. Search at most three result pages and inspect at most five candidate domains. Try exact business-name and location/phone searches. Prefer the official business website over social networks, directories, review sites, and map pages. Do not log in, submit forms, solve CAPTCHAs, bypass access challenges, or use a proxy. If a site blocks the browser, record it and move on. Return ONLY JSON in this shape: {{"candidates":[{{"url":"https://example.co.za","confidence":0,"evidence":["name match","phone match"],"status":"candidate"}}]}} Confidence must be an integer from 0 to 100. Include only plausible official website URLs and never invent a URL. """.strip() async def run() -> Any: browser_profile = BrowserProfile( headless=True, executable_path=os.environ.get("PLAYWRIGHT_CHROMIUM_EXECUTABLE_PATH", "/usr/bin/chromium"), minimum_wait_page_load_time=0.5, wait_between_actions=0.7, enable_default_extensions=False, captcha_solver=False, ) llm = ChatOpenAI( model=model, api_key=api_key, base_url=base_url, default_headers={"User-Agent": user_agent}, temperature=0.1, frequency_penalty=None, max_completion_tokens=900, max_retries=1, timeout=45, ) agent = Agent( task=task, llm=llm, browser_profile=browser_profile, use_vision=False, flash_mode=True, use_judge=False, enable_planning=False, enable_signal_handler=False, ) return await agent.run(max_steps=max(3, min(15, int(os.environ.get("BROWSER_USE_MAX_STEPS", "10"))))) timeout = max(30, min(300, int(os.environ.get("BROWSER_USE_TIMEOUT_SECONDS", "120")))) try: history = asyncio.run(asyncio.wait_for(run(), timeout=timeout)) except asyncio.TimeoutError as exc: raise RuntimeError("browser_use_timeout") from exc final = history.final_result() if history is not None and hasattr(history, "final_result") else "" return _parse_agent_result(final)