win/lib/shared/browse.py

13.6 KB · 447 lines · python

1#!/usr/bin/env python3
2"""Google search helper for brt -browse."""
3
4from __future__ import annotations
5
6import base64
7import html
8import json
9import os
10import re
11import shutil
12import subprocess
13import sys
14import urllib.parse
15
16USER_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)
20SCRAPE_USER_AGENT = "Mozilla/5.0"
21MAX_RESULTS = 25
22GOOGLE_CSE_URL = "https://www.googleapis.com/customsearch/v1"
23SERPER_URL = "https://google.serper.dev/search"
24
25
26def _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
77def _clean_title(raw: str) -> str:
78 text = re.sub(r"<[^>]+>", "", raw)
79 return html.unescape(text).strip()
80
81
82def _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
108def _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
121def 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
154def 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
173def 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
212def _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
244def 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
269def _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
283def 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
296def 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
309def _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
328def 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
343def _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
364def _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
388def 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
423def 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
445if __name__ == "__main__":
446 sys.exit(main())
447