root/MarketingTool
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add resilient multi-engine browser discovery

Browse Source
This commit is contained in:
Marco0300
2026-09-05 14:59:28 +02:00
parent 6debef6015
commit 521d666ecf
4 changed files with 113 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+17 -5
View File
@@ -1,8 +1,10 @@
# Browser-only ProspectOS
This is the stripped-down replacement for the larger ProspectOS application.
It searches public engines (Google first, then Bing and DuckDuckGo if Google
challenges the server) through a headless Playwright Chromium browser. Each
It searches public engines (Google, Bing, DuckDuckGo, Brave Search, and
Startpage) through a headless Playwright Chromium browser, moving to the next
engine when a server-side challenge, timeout, or empty/changed result page is
encountered. Each
discovery run builds a query from category, keywords, and location, filters
results already present in the local catalogue by root domain, and stores only
new prospects. The explorer supports search and human review states
@@ -19,9 +21,19 @@ docker compose up -d --build
Open `http://SERVER:8000`.
If Google presents a CAPTCHA or bot challenge, the run fails safely. The app
does not bypass challenges or rotate identities. Keep runs small and spaced
out. The implementation uses a fresh headless browser context for each run.
If another service already owns port 8000, choose the host port without
changing the container:
```bash
HOST_PORT=8080 docker compose up -d --build
```
If an engine presents a CAPTCHA or bot challenge, the run moves on to the next
engine. The app does not bypass challenges or rotate identities. Keep runs
small and spaced out. The implementation uses a fresh headless browser
context for each engine and unwraps normal search redirect links before
deduplication. If every public engine blocks the server IP, the run reports a
clear `search_engines_unavailable` error instead of inventing prospects.
## Run locally
api/server.py
+91 -24
View File
@@ -18,7 +18,7 @@ from datetime import datetime, timezone
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from typing import Any
from urllib.parse import parse_qs, quote_plus, urlparse
from urllib.parse import parse_qs, quote_plus, unquote, urlparse
DB_PATH = os.environ.get("PROSPECT_DB", "/data/prospects.db")
MAX_RESULTS = 25
@@ -81,11 +81,79 @@ def blocked(text: str) -> bool:
))
def normalize_result_href(raw_href: str) -> str:
"""Turn a search-result href into the public destination URL.
Engines commonly use protocol-relative links or redirect wrappers (for
example DuckDuckGo's ``uddg`` parameter). We unwrap those links before
applying the catalog/domain filters.
"""
href = (raw_href or "").strip()
if href.startswith("//"):
href = "https:" + href
if href.startswith("/"):
return ""
parsed = urlparse(href)
if parsed.hostname and parsed.hostname.endswith("duckduckgo.com"):
params = parse_qs(parsed.query)
for key in ("uddg", "url", "u"):
target = params.get(key, [""])[0]
if target.startswith(("http://", "https://")):
href = unquote(target)
break
return href
def extract_search_results(page: Any, selector: str, title_selector: str | None,
excluded_hosts: set[str], requested: int) -> list[dict[str, str]]:
"""Extract external result links, with a generic-anchor safety net."""
output: list[dict[str, str]] = []
seen_domains: set[str] = set()
def add_anchor(anchor: Any) -> None:
if len(output) >= requested:
return
try:
title_node = anchor.locator(title_selector) if title_selector else anchor
title = " ".join((title_node.inner_text() or "").split())[:200]
if not title:
title = " ".join(((anchor.get_attribute("aria-label") or anchor.get_attribute("title") or "").split()))[:200]
href = normalize_result_href(anchor.get_attribute("href") or "")
except Exception:
return
domain = clean_domain(href)
if not title or len(title) < 3 or not href.startswith(("http://", "https://")):
return
if not domain or domain in excluded_hosts or domain in seen_domains:
return
seen_domains.add(domain)
output.append({"name": title, "website": href, "domain": domain})
# Prefer stable engine-specific result selectors to avoid collecting nav
# and footer links. If markup changed, fall back to generic anchors.
try:
anchors = page.locator(selector)
for i in range(min(anchors.count(), requested * 4)):
add_anchor(anchors.nth(i))
except Exception:
pass
if not output:
try:
anchors = page.locator("a")
for i in range(min(anchors.count(), 250)):
add_anchor(anchors.nth(i))
if len(output) >= requested:
break
except Exception:
pass
return output
def google_search(query: str, requested: int) -> list[dict[str, str]]:
"""Search public engines through Playwright, with safe engine fallback.
Google frequently challenges datacenter IPs. We do not bypass that
challenge; we close the context and try Bing, then DuckDuckGo, still using
challenge; we close the context and try other public engines, still using
a real headless browser and the same bounded result extraction.
"""
try:
@@ -99,7 +167,9 @@ def google_search(query: str, requested: int) -> list[dict[str, str]]:
engines = [
("google", "https://www.google.com/search?q=" + quote_plus(query) + "&num=" + str(requested), "a:has(h3)", "h3", {"google.com", "google.co.za"}),
("bing", "https://www.bing.com/search?q=" + quote_plus(query) + "&count=" + str(requested), "li.b_algo h2 a", "h2", {"bing.com", "bingj.com"}),
("duckduckgo", "https://html.duckduckgo.com/html/?q=" + quote_plus(query), "a.result__a", None, {"duckduckgo.com"}),
("duckduckgo", "https://html.duckduckgo.com/html/?q=" + quote_plus(query), "a.result__a,a[data-testid='result-title-a']", None, {"duckduckgo.com"}),
("brave", "https://search.brave.com/search?q=" + quote_plus(query), "a.heading-serpresult,a.result-header", None, {"search.brave.com"}),
("startpage", "https://www.startpage.com/sp/search?query=" + quote_plus(query), "a.w-gl__result-title", None, {"startpage.com"}),
]
errors: list[str] = []
with sync_playwright() as p:
@@ -113,35 +183,32 @@ def google_search(query: str, requested: int) -> list[dict[str, str]]:
java_script_enabled=True,
service_workers="block",
viewport={"width": 1280, "height": 900},
user_agent=(
"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36"
),
)
page = context.new_page()
page.set_default_timeout(15000)
page.goto(url, wait_until="domcontentloaded", timeout=15000)
page.wait_for_timeout(250)
page.set_default_timeout(12000)
navigation_timed_out = False
# DOMContentLoaded gives us usable markup without waiting for
# analytics, ads, or other background requests. If an engine
# keeps navigation open, still inspect the partially loaded
# page rather than discarding it as a hard failure.
try:
page.goto(url, wait_until="domcontentloaded", timeout=15000)
except PlaywrightTimeout:
navigation_timed_out = True
page.wait_for_timeout(1000)
body = page.locator("body").inner_text()
html = page.content()
if blocked(body) or blocked(html):
errors.append(engine + "_blocked")
continue
output: list[dict[str, str]] = []
anchors = page.locator(selector)
for i in range(min(anchors.count(), requested * 3)):
anchor = anchors.nth(i)
title_node = anchor.locator(title_selector) if title_selector else anchor
title = " ".join((title_node.inner_text() or "").split())[:200]
href = anchor.get_attribute("href") or ""
domain = clean_domain(href)
if not title or not href or domain in excluded_hosts or not domain:
continue
if not href.startswith(("http://", "https://")):
continue
if domain not in {item["domain"] for item in output}:
output.append({"name": title, "website": href, "domain": domain})
if len(output) >= requested:
break
output = extract_search_results(page, selector, title_selector, excluded_hosts, requested)
if output:
return output
errors.append(engine + "_empty")
errors.append(engine + ("_timeout" if navigation_timed_out else "_empty"))
except PlaywrightTimeout:
errors.append(engine + "_timeout")
except Exception as exc:
@@ -152,7 +219,7 @@ def google_search(query: str, requested: int) -> list[dict[str, str]]:
except Exception: pass
finally:
browser.close()
raise RuntimeError("search_engines_unavailable:" + ",".join(errors[-3:]))
raise RuntimeError("search_engines_unavailable:" + ",".join(errors))
def run_discovery(payload: dict[str, Any]) -> dict[str, Any]:
docker-compose.yml
-1
View File
@@ -4,7 +4,6 @@ services:
context: .
dockerfile: api/Dockerfile
ports:
- "8000:8000"
- "8080:8000"
environment:
PROSPECT_DB: /data/prospects.db
tests/test_app.py
+5
View File
@@ -30,6 +30,11 @@ class BrowserOnlyAppTests(unittest.TestCase):
response = self.conn.getresponse()
return response.status, json.loads(response.read() or b"{}")
def test_search_redirect_links_are_unwrapped(self):
wrapped = "https://duckduckgo.com/l/?uddg=https%3A%2F%2Fwww.example.co.za%2Fservices"
self.assertEqual(server.normalize_result_href(wrapped), "https://www.example.co.za/services")
self.assertEqual(server.normalize_result_href("//example.co.za/about"), "https://example.co.za/about")
@patch("api.server.google_search", return_value=[{"name": "Acme Plumbing", "website": "https://www.acme.co.za", "domain": "acme.co.za"}, {"name": "Other", "website": "https://other.co.za", "domain": "other.co.za"}])
def test_discovery_filters_catalogued_root_domains(self, search):
status, first = self.request("POST", "/api/discover", {"category": "plumbers", "keywords": "plumbing", "location": "Cape Town", "max_results": 5})