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

add safe public search-engine fallback

Browse Source
This commit is contained in:
Marco0300
2026-09-05 14:19:56 +02:00
parent 477cbcd539
commit 6debef6015
3 changed files with 43 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/server.py
+36 -20
View File
@@ -82,18 +82,31 @@ def blocked(text: str) -> bool:
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
a real headless browser and the same bounded result extraction.
"""
try:
from playwright.sync_api import sync_playwright, TimeoutError as PlaywrightTimeout
except ImportError as exc:
raise RuntimeError("playwright_not_installed") from exc
url = "https://www.google.com/search?q=" + quote_plus(query) + "&num=" + str(requested)
executable = os.environ.get("PLAYWRIGHT_CHROMIUM_EXECUTABLE_PATH", "").strip()
options: dict[str, Any] = {"headless": True}
if executable:
options["executable_path"] = executable
try:
with sync_playwright() as p:
browser = p.chromium.launch(**options)
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"}),
]
errors: list[str] = []
with sync_playwright() as p:
browser = p.chromium.launch(**options)
try:
for engine, url, selector, title_selector, excluded_hosts in engines:
context = None
try:
context = browser.new_context(
locale="en-ZA",
@@ -108,15 +121,17 @@ def google_search(query: str, requested: int) -> list[dict[str, str]]:
body = page.locator("body").inner_text()
html = page.content()
if blocked(body) or blocked(html):
raise RuntimeError("google_blocked")
errors.append(engine + "_blocked")
continue
output: list[dict[str, str]] = []
anchors = page.locator("a:has(h3)")
anchors = page.locator(selector)
for i in range(min(anchors.count(), requested * 3)):
anchor = anchors.nth(i)
title = " ".join((anchor.locator("h3").inner_text() or "").split())[:200]
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 {"google.com", "google.co.za"} or not domain:
if not title or not href or domain in excluded_hosts or not domain:
continue
if not href.startswith(("http://", "https://")):
continue
@@ -124,19 +139,20 @@ def google_search(query: str, requested: int) -> list[dict[str, str]]:
output.append({"name": title, "website": href, "domain": domain})
if len(output) >= requested:
break
return output
if output:
return output
errors.append(engine + "_empty")
except PlaywrightTimeout:
errors.append(engine + "_timeout")
except Exception as exc:
errors.append(engine + "_error")
finally:
try:
context.close()
except Exception:
pass
browser.close()
except RuntimeError:
raise
except PlaywrightTimeout as exc:
raise RuntimeError("browser_timeout") from exc
except Exception as exc:
raise RuntimeError("browser_error") from exc
if context is not None:
try: context.close()
except Exception: pass
finally:
browser.close()
raise RuntimeError("search_engines_unavailable:" + ",".join(errors[-3:]))
def run_discovery(payload: dict[str, Any]) -> dict[str, Any]: