| 1 | #!/usr/bin/env python3 |
| 2 | """Google search helper for brt -browse.""" |
| 3 | |
| 4 | from __future__ import annotations |
| 5 | |
| 6 | import base64 |
| 7 | import html |
| 8 | import json |
| 9 | import os |
| 10 | import re |
| 11 | import shutil |
| 12 | import subprocess |
| 13 | import sys |
| 14 | import urllib.parse |
| 15 | |
| 16 | USER_AGENT = ( |
| 17 | "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " |
| 18 | "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36" |
| 19 | ) |
| 20 | SCRAPE_USER_AGENT = "Mozilla/5.0" |
| 21 | MAX_RESULTS = 25 |
| 22 | GOOGLE_CSE_URL = "https://www.googleapis.com/customsearch/v1" |
| 23 | SERPER_URL = "https://google.serper.dev/search" |
| 24 | |
| 25 | |
| 26 | def _fetch( |
| 27 | url: str, |
| 28 | headers: dict | None = None, |
| 29 | data: bytes | None = None, |
| 30 | *, |
| 31 | user_agent: str | None = None, |
| 32 | ) -> str: |
| 33 | curl = shutil.which("curl") |
| 34 | if not curl: |
| 35 | raise RuntimeError("curl is required for brt -browse") |
| 36 | |
| 37 | agent = user_agent or (headers or {}).get("User-Agent") or USER_AGENT |
| 38 | cmd = [ |
| 39 | curl, |
| 40 | "-fsSL", |
| 41 | "--max-time", |
| 42 | "15", |
| 43 | "-A", |
| 44 | agent, |
| 45 | ] |
| 46 | for key, value in (headers or {}).items(): |
| 47 | if key.lower() in ("user-agent", "cookie", "accept-language", "content-type", "x-api-key"): |
| 48 | continue |
| 49 | cmd.extend(["-H", f"{key}: {value}"]) |
| 50 | for key, value in (headers or {}).items(): |
| 51 | if key.lower() == "cookie": |
| 52 | cmd.extend(["-H", f"Cookie: {value}"]) |
| 53 | elif key.lower() == "accept-language": |
| 54 | cmd.extend(["-H", f"Accept-Language: {value}"]) |
| 55 | elif key.lower() == "content-type": |
| 56 | cmd.extend(["-H", f"Content-Type: {value}"]) |
| 57 | elif key.lower() == "x-api-key": |
| 58 | cmd.extend(["-H", f"X-API-KEY: {value}"]) |
| 59 | |
| 60 | if data is not None: |
| 61 | cmd.extend(["-X", "POST", "--data-binary", "@-"]) |
| 62 | proc = subprocess.run( |
| 63 | cmd + [url], |
| 64 | input=data, |
| 65 | capture_output=True, |
| 66 | check=False, |
| 67 | ) |
| 68 | else: |
| 69 | proc = subprocess.run(cmd + [url], capture_output=True, check=False) |
| 70 | |
| 71 | if proc.returncode != 0: |
| 72 | err = proc.stderr.decode("utf-8", errors="replace").strip() |
| 73 | raise RuntimeError(err or f"curl failed ({proc.returncode})") |
| 74 | return proc.stdout.decode("utf-8", errors="replace") |
| 75 | |
| 76 | |
| 77 | def _clean_title(raw: str) -> str: |
| 78 | text = re.sub(r"<[^>]+>", "", raw) |
| 79 | return html.unescape(text).strip() |
| 80 | |
| 81 | |
| 82 | def _normalize_url(raw: str) -> str | None: |
| 83 | url = html.unescape(raw).strip() |
| 84 | if url.startswith("//"): |
| 85 | url = "https:" + url |
| 86 | if url.startswith("/url?"): |
| 87 | parsed = urllib.parse.urlparse(url) |
| 88 | qs = urllib.parse.parse_qs(parsed.query) |
| 89 | url = qs.get("q", qs.get("url", [""]))[0] |
| 90 | url = urllib.parse.unquote(url) |
| 91 | if not url.startswith(("http://", "https://")): |
| 92 | return None |
| 93 | if any( |
| 94 | skip in url |
| 95 | for skip in ( |
| 96 | "google.com/search", |
| 97 | "google.com/url", |
| 98 | "webcache.googleusercontent.com", |
| 99 | "accounts.google.com", |
| 100 | "support.google.com", |
| 101 | "policies.google.com", |
| 102 | ) |
| 103 | ): |
| 104 | return None |
| 105 | return url |
| 106 | |
| 107 | |
| 108 | def _dedupe_results(items: list[dict]) -> list[dict]: |
| 109 | seen: set[str] = set() |
| 110 | out: list[dict] = [] |
| 111 | for item in items: |
| 112 | url = item.get("url", "") |
| 113 | if not url or url in seen: |
| 114 | continue |
| 115 | seen.add(url) |
| 116 | title = item.get("title", "").strip() or url |
| 117 | out.append({"title": title, "url": url}) |
| 118 | return out |
| 119 | |
| 120 | |
| 121 | def search_google_cse(query: str, offset: int, limit: int) -> list[dict]: |
| 122 | api_key = os.environ.get("BRT_GOOGLE_API_KEY", "").strip() |
| 123 | cx = os.environ.get("BRT_GOOGLE_CX", "").strip() |
| 124 | if not api_key or not cx: |
| 125 | return [] |
| 126 | |
| 127 | want = min(MAX_RESULTS, offset + limit) |
| 128 | results: list[dict] = [] |
| 129 | start = 1 # Google CSE is 1-indexed |
| 130 | while len(results) < want: |
| 131 | batch = min(10, want - len(results)) |
| 132 | params = urllib.parse.urlencode( |
| 133 | { |
| 134 | "key": api_key, |
| 135 | "cx": cx, |
| 136 | "q": query, |
| 137 | "num": batch, |
| 138 | "start": start, |
| 139 | } |
| 140 | ) |
| 141 | payload = json.loads(_fetch(f"{GOOGLE_CSE_URL}?{params}")) |
| 142 | items = payload.get("items") or [] |
| 143 | if not items: |
| 144 | break |
| 145 | for item in items: |
| 146 | results.append({"title": item.get("title", ""), "url": item.get("link", "")}) |
| 147 | start += len(items) |
| 148 | if len(items) < batch: |
| 149 | break |
| 150 | |
| 151 | return results[offset : offset + limit] |
| 152 | |
| 153 | |
| 154 | def search_serper(query: str, offset: int, limit: int) -> list[dict]: |
| 155 | api_key = os.environ.get("BRT_SERPER_API_KEY", "").strip() |
| 156 | if not api_key: |
| 157 | return [] |
| 158 | |
| 159 | page = offset // 10 + 1 |
| 160 | body = json.dumps({"q": query, "num": min(10, limit), "page": page}).encode() |
| 161 | headers = { |
| 162 | "X-API-KEY": api_key, |
| 163 | "Content-Type": "application/json", |
| 164 | "User-Agent": USER_AGENT, |
| 165 | } |
| 166 | payload = json.loads(_fetch(SERPER_URL, headers=headers, data=body)) |
| 167 | organic = payload.get("organic") or [] |
| 168 | results = [{"title": item.get("title", ""), "url": item.get("link", "")} for item in organic] |
| 169 | inner_offset = offset % 10 |
| 170 | return results[inner_offset : inner_offset + limit] |
| 171 | |
| 172 | |
| 173 | def search_google_html(query: str, offset: int, limit: int) -> list[dict]: |
| 174 | want = min(MAX_RESULTS, offset + limit) |
| 175 | results: list[dict] = [] |
| 176 | start = 0 |
| 177 | |
| 178 | while len(results) < want: |
| 179 | params = urllib.parse.urlencode( |
| 180 | { |
| 181 | "q": query, |
| 182 | "num": 10, |
| 183 | "hl": "en", |
| 184 | "pws": "0", |
| 185 | "start": start, |
| 186 | "gbv": "1", |
| 187 | "filter": "0", |
| 188 | } |
| 189 | ) |
| 190 | url = f"https://www.google.com/search?{params}" |
| 191 | headers = { |
| 192 | "User-Agent": SCRAPE_USER_AGENT, |
| 193 | "Accept-Language": "en-US,en;q=0.9", |
| 194 | "Cookie": "CONSENT=YES+; SOCS=CAI", |
| 195 | } |
| 196 | try: |
| 197 | page = _fetch(url, headers=headers) |
| 198 | except (RuntimeError, OSError): |
| 199 | break |
| 200 | |
| 201 | batch = _parse_google_html(page) |
| 202 | if not batch: |
| 203 | break |
| 204 | results.extend(batch) |
| 205 | start += 10 |
| 206 | if len(batch) < 5 or start >= MAX_RESULTS: |
| 207 | break |
| 208 | |
| 209 | return _dedupe_results(results)[offset : offset + limit] |
| 210 | |
| 211 | |
| 212 | def _parse_google_html(page: str) -> list[dict]: |
| 213 | results: list[dict] = [] |
| 214 | |
| 215 | for block in re.findall(r"<div[^>]+class=\"[^\"]*\bg\b[^\"]*\"[^>]*>(.*?)</div>\s*</div>", page, re.S | re.I): |
| 216 | title_match = re.search(r"<h3[^>]*>(.*?)</h3>", block, re.S | re.I) |
| 217 | if not title_match: |
| 218 | continue |
| 219 | link_match = re.search(r"<a[^>]+href=\"([^\"]+)\"[^>]*>", block, re.I) |
| 220 | if not link_match: |
| 221 | continue |
| 222 | url = _normalize_url(link_match.group(1)) |
| 223 | if not url: |
| 224 | continue |
| 225 | results.append({"title": _clean_title(title_match.group(1)), "url": url}) |
| 226 | |
| 227 | if results: |
| 228 | return results |
| 229 | |
| 230 | titles = re.findall(r"<h3[^>]*>(.*?)</h3>", page, re.S | re.I) |
| 231 | links = re.findall(r"<a[^>]+href=\"(/url\?q=[^\"]+|https?://[^\"]+)\"[^>]*>", page, re.I) |
| 232 | for link in links: |
| 233 | url = _normalize_url(link) |
| 234 | if not url: |
| 235 | continue |
| 236 | title = _clean_title(titles[len(results)]) if len(results) < len(titles) else url |
| 237 | results.append({"title": title, "url": url}) |
| 238 | if len(results) >= 10: |
| 239 | break |
| 240 | |
| 241 | return results |
| 242 | |
| 243 | |
| 244 | def search_ddg_html(query: str, offset: int, limit: int) -> list[dict]: |
| 245 | want = min(MAX_RESULTS, offset + limit) |
| 246 | results: list[dict] = [] |
| 247 | page = offset // 10 |
| 248 | |
| 249 | while len(results) < want: |
| 250 | params = urllib.parse.urlencode({"q": query, "s": page * 10}) |
| 251 | url = f"https://html.duckduckgo.com/html/?{params}" |
| 252 | headers = {"User-Agent": SCRAPE_USER_AGENT} |
| 253 | try: |
| 254 | page_html = _fetch(url, headers=headers) |
| 255 | except (RuntimeError, OSError): |
| 256 | break |
| 257 | |
| 258 | batch = _parse_ddg_html(page_html) |
| 259 | if not batch: |
| 260 | break |
| 261 | results.extend(batch) |
| 262 | page += 1 |
| 263 | if len(batch) < 5 or page * 10 >= MAX_RESULTS: |
| 264 | break |
| 265 | |
| 266 | return _dedupe_results(results)[offset : offset + limit] |
| 267 | |
| 268 | |
| 269 | def _decode_bing_redirect(href: str) -> str | None: |
| 270 | href = html.unescape(href).strip() |
| 271 | parsed = urllib.parse.urlparse(href) |
| 272 | encoded = urllib.parse.parse_qs(parsed.query).get("u", [""])[0] |
| 273 | if not encoded: |
| 274 | return href if href.startswith(("http://", "https://")) else None |
| 275 | payload = encoded[2:] if len(encoded) > 2 else encoded |
| 276 | try: |
| 277 | pad = "=" * (-len(payload) % 4) |
| 278 | return base64.b64decode(payload + pad).decode("utf-8", errors="replace") |
| 279 | except (ValueError, UnicodeDecodeError): |
| 280 | return None |
| 281 | |
| 282 | |
| 283 | def search_bing_html(query: str, offset: int, limit: int) -> list[dict]: |
| 284 | want = min(MAX_RESULTS, offset + limit) |
| 285 | params = urllib.parse.urlencode({"q": query, "count": 10, "first": 1}) |
| 286 | url = f"https://www.bing.com/search?{params}" |
| 287 | headers = {"User-Agent": SCRAPE_USER_AGENT} |
| 288 | try: |
| 289 | page = _fetch(url, headers=headers) |
| 290 | except (RuntimeError, OSError): |
| 291 | return [] |
| 292 | |
| 293 | return _dedupe_results(_parse_bing_html(page))[offset : offset + limit] |
| 294 | |
| 295 | |
| 296 | def search_brave_html(query: str, offset: int, limit: int) -> list[dict]: |
| 297 | want = min(MAX_RESULTS, offset + limit) |
| 298 | params = urllib.parse.urlencode({"q": query}) |
| 299 | url = f"https://search.brave.com/search?{params}" |
| 300 | headers = {"User-Agent": SCRAPE_USER_AGENT} |
| 301 | try: |
| 302 | page = _fetch(url, headers=headers) |
| 303 | except (RuntimeError, OSError): |
| 304 | return [] |
| 305 | |
| 306 | return _dedupe_results(_parse_brave_html(page))[:want][offset : offset + limit] |
| 307 | |
| 308 | |
| 309 | def _parse_brave_html(page: str) -> list[dict]: |
| 310 | results: list[dict] = [] |
| 311 | for part in page.split('data-type="web"')[1:]: |
| 312 | link_match = re.search(r'<a[^>]+href="(https?://[^"]+)"', part) |
| 313 | if not link_match: |
| 314 | continue |
| 315 | url = link_match.group(1) |
| 316 | if "search.brave.com" in url: |
| 317 | continue |
| 318 | title_match = re.search(r'class="title[^"]*"[^>]*>([\s\S]*?)</', part) |
| 319 | if title_match: |
| 320 | title = _clean_title(title_match.group(1)) |
| 321 | else: |
| 322 | title = url |
| 323 | if title: |
| 324 | results.append({"title": title, "url": url}) |
| 325 | return results |
| 326 | |
| 327 | |
| 328 | def search_web_scrape(query: str, offset: int, limit: int) -> list[dict]: |
| 329 | want = min(MAX_RESULTS, offset + limit) |
| 330 | merged: list[dict] = [] |
| 331 | for search_fn in (search_brave_html, search_bing_html): |
| 332 | try: |
| 333 | batch = search_fn(query, 0, want) |
| 334 | except Exception: |
| 335 | batch = [] |
| 336 | merged.extend(batch) |
| 337 | merged = _dedupe_results(merged) |
| 338 | if len(merged) >= want: |
| 339 | break |
| 340 | return merged[offset : offset + limit] |
| 341 | |
| 342 | |
| 343 | def _parse_bing_html(page: str) -> list[dict]: |
| 344 | results: list[dict] = [] |
| 345 | for match in re.finditer(r"<h2[^>]*>([\s\S]*?)</h2>", page, re.I): |
| 346 | link_match = re.search( |
| 347 | r'<a[^>]+href="([^"]+)"[^>]*>([\s\S]*?)</a>', |
| 348 | match.group(1), |
| 349 | re.I, |
| 350 | ) |
| 351 | if not link_match: |
| 352 | continue |
| 353 | title = _clean_title(link_match.group(2)) |
| 354 | if not title: |
| 355 | continue |
| 356 | url = _decode_bing_redirect(link_match.group(1)) |
| 357 | if not url: |
| 358 | continue |
| 359 | url = _normalize_url(url) or url |
| 360 | results.append({"title": title, "url": url}) |
| 361 | return results |
| 362 | |
| 363 | |
| 364 | def _parse_ddg_html(page: str) -> list[dict]: |
| 365 | results: list[dict] = [] |
| 366 | for match in re.finditer( |
| 367 | r'class="result__a"[^>]*href="([^"]+)"[^>]*>(.*?)</a>', |
| 368 | page, |
| 369 | re.S | re.I, |
| 370 | ): |
| 371 | href = match.group(1) |
| 372 | title = _clean_title(match.group(2)) |
| 373 | if "duckduckgo.com/y.js" in href or not title: |
| 374 | continue |
| 375 | if href.startswith("//"): |
| 376 | href = "https:" + href |
| 377 | if "uddg=" in href: |
| 378 | parsed = urllib.parse.urlparse(href) |
| 379 | href = urllib.parse.unquote( |
| 380 | urllib.parse.parse_qs(parsed.query).get("uddg", [""])[0] |
| 381 | ) |
| 382 | url = _normalize_url(href) |
| 383 | if url: |
| 384 | results.append({"title": title, "url": url}) |
| 385 | return results |
| 386 | |
| 387 | |
| 388 | def search_all(query: str, max_results: int = MAX_RESULTS) -> list[dict]: |
| 389 | query = query.strip() |
| 390 | if not query: |
| 391 | return [] |
| 392 | |
| 393 | max_results = max(1, min(MAX_RESULTS, max_results)) |
| 394 | for backend in (search_google_cse, search_serper, search_google_html): |
| 395 | try: |
| 396 | batch = backend(query, 0, max_results) |
| 397 | except Exception: |
| 398 | batch = [] |
| 399 | batch = _dedupe_results(batch) |
| 400 | if batch: |
| 401 | return batch[:max_results] |
| 402 | |
| 403 | for fallback, search_fn in ( |
| 404 | ("web", search_web_scrape), |
| 405 | ("DuckDuckGo", search_ddg_html), |
| 406 | ): |
| 407 | try: |
| 408 | batch = _dedupe_results(search_fn(query, 0, max_results)) |
| 409 | except Exception: |
| 410 | batch = [] |
| 411 | if batch: |
| 412 | label = "open web" if fallback == "web" else fallback |
| 413 | print( |
| 414 | f"note: Google unavailable; showing {label} results (set " |
| 415 | "BRT_GOOGLE_API_KEY + BRT_GOOGLE_CX or BRT_SERPER_API_KEY for " |
| 416 | "official Google results)", |
| 417 | file=sys.stderr, |
| 418 | ) |
| 419 | return batch[:max_results] |
| 420 | return [] |
| 421 | |
| 422 | |
| 423 | def main() -> int: |
| 424 | if len(sys.argv) < 3 or sys.argv[1] != "search": |
| 425 | print( |
| 426 | "usage: browse.py search <query> [max_results]", |
| 427 | file=sys.stderr, |
| 428 | ) |
| 429 | return 1 |
| 430 | |
| 431 | query = sys.argv[2] |
| 432 | max_results = int(sys.argv[3]) if len(sys.argv) > 3 else MAX_RESULTS |
| 433 | results = search_all(query, max_results) |
| 434 | if not results: |
| 435 | print( |
| 436 | "error: no search results (set keys in ~/.brt/.env or run brt -config)", |
| 437 | file=sys.stderr, |
| 438 | ) |
| 439 | return 1 |
| 440 | |
| 441 | json.dump(results, sys.stdout, ensure_ascii=False) |
| 442 | return 0 |
| 443 | |
| 444 | |
| 445 | if __name__ == "__main__": |
| 446 | sys.exit(main()) |
| 447 | |