| 1 | """Web search retrieval via Brave Search, Exa, Serper, Parallel, or a keyless floor.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | import sys |
| 6 | import urllib.parse |
| 7 | from dataclasses import dataclass |
| 8 | from datetime import datetime |
| 9 | from urllib.parse import urlparse |
| 10 | |
| 11 | from . import dates, env, http, parallel_mcp, schema, web_search_keyless |
| 12 | |
| 13 | |
| 14 | @dataclass(frozen=True) |
| 15 | class GroundedClaimText: |
| 16 | """Candidate text with its exact primary evidence item.""" |
| 17 | |
| 18 | candidate_id: str |
| 19 | title: str |
| 20 | summary: str |
| 21 | item: schema.SourceItem |
| 22 | |
| 23 | |
| 24 | def claim_source_map(report: schema.Report) -> dict[str, GroundedClaimText]: |
| 25 | """Expose only candidate claims that have a clean primary-item trace. |
| 26 | |
| 27 | Freshness verification deliberately starts here instead of scanning all |
| 28 | report prose. A candidate without a primary ``SourceItem`` cannot produce |
| 29 | an auditable per-claim verdict. |
| 30 | """ |
| 31 | grounded: dict[str, GroundedClaimText] = {} |
| 32 | for candidate in report.ranked_candidates: |
| 33 | item = schema.candidate_primary_item(candidate) |
| 34 | if item is None: |
| 35 | continue |
| 36 | grounded[candidate.candidate_id] = GroundedClaimText( |
| 37 | candidate_id=candidate.candidate_id, |
| 38 | title=candidate.title, |
| 39 | summary=candidate.snippet or item.snippet or item.body, |
| 40 | item=item, |
| 41 | ) |
| 42 | return grounded |
| 43 | |
| 44 | |
| 45 | # --------------------------------------------------------------------------- |
| 46 | # Brave Search API |
| 47 | # --------------------------------------------------------------------------- |
| 48 | |
| 49 | def brave_search( |
| 50 | query: str, date_range: tuple[str, str], api_key: str, count: int = 5, |
| 51 | ) -> tuple[list[dict], dict]: |
| 52 | url = ( |
| 53 | "https://api.search.brave.com/res/v1/web/search?" |
| 54 | + urllib.parse.urlencode( |
| 55 | { |
| 56 | "q": query, |
| 57 | "count": count, |
| 58 | "freshness": f"{date_range[0]}to{date_range[1]}", |
| 59 | } |
| 60 | ) |
| 61 | ) |
| 62 | data = http.request("GET", url, headers={"X-Subscription-Token": api_key}, timeout=15) |
| 63 | items = [] |
| 64 | for i, r in enumerate((data.get("web", {}).get("results", []))[:count]): |
| 65 | raw_date = r.get("page_age") or "" |
| 66 | pub_date = _normalize_date(raw_date[:10]) if raw_date else None |
| 67 | if not _in_date_range(pub_date, date_range): |
| 68 | continue |
| 69 | items.append({ |
| 70 | "id": f"WB{i + 1}", |
| 71 | "title": r.get("title", ""), |
| 72 | "url": r.get("url", ""), |
| 73 | "source_domain": _domain(r.get("url", "")), |
| 74 | "snippet": r.get("description", ""), |
| 75 | "date": pub_date, |
| 76 | "relevance": 0.8, |
| 77 | "why_relevant": "Brave web search", |
| 78 | }) |
| 79 | artifact = {"label": "brave", "webSearchQueries": [query], "resultCount": len(items)} |
| 80 | return items, artifact |
| 81 | |
| 82 | |
| 83 | # --------------------------------------------------------------------------- |
| 84 | # Exa AI Search |
| 85 | # --------------------------------------------------------------------------- |
| 86 | |
| 87 | def exa_search( |
| 88 | query: str, date_range: tuple[str, str], api_key: str, count: int = 5, |
| 89 | ) -> tuple[list[dict], dict]: |
| 90 | data = http.request( |
| 91 | "POST", "https://api.exa.ai/search", |
| 92 | headers={"x-api-key": api_key}, |
| 93 | json_data={ |
| 94 | "query": query, |
| 95 | "type": "auto", |
| 96 | "numResults": count, |
| 97 | "startPublishedDate": f"{date_range[0]}T00:00:00.000Z", |
| 98 | "endPublishedDate": f"{date_range[1]}T23:59:59.999Z", |
| 99 | "contents": {"text": {"maxCharacters": 2000}}, |
| 100 | }, |
| 101 | timeout=15, |
| 102 | ) |
| 103 | items = [] |
| 104 | for i, r in enumerate((data.get("results", []))[:count]): |
| 105 | if not isinstance(r, dict): |
| 106 | continue |
| 107 | url = r.get("url", "") |
| 108 | if not url: |
| 109 | continue |
| 110 | raw_date = r.get("publishedDate") or "" |
| 111 | pub_date = _normalize_date(raw_date.split("T")[0] if "T" in raw_date else raw_date[:10]) if raw_date else None |
| 112 | if not _in_date_range(pub_date, date_range): |
| 113 | continue |
| 114 | items.append({ |
| 115 | "id": f"WE{i + 1}", |
| 116 | "title": r.get("title", ""), |
| 117 | "url": url, |
| 118 | "source_domain": _domain(url), |
| 119 | "snippet": (r.get("text") or "")[:500], |
| 120 | "date": pub_date, |
| 121 | "relevance": 0.8, |
| 122 | "why_relevant": "Exa web search", |
| 123 | }) |
| 124 | artifact = {"label": "exa", "webSearchQueries": [query], "resultCount": len(items)} |
| 125 | return items, artifact |
| 126 | |
| 127 | |
| 128 | # --------------------------------------------------------------------------- |
| 129 | # Serper (Google Search wrapper) |
| 130 | # --------------------------------------------------------------------------- |
| 131 | |
| 132 | def serper_search( |
| 133 | query: str, date_range: tuple[str, str], api_key: str, count: int = 5, |
| 134 | ) -> tuple[list[dict], dict]: |
| 135 | data = http.request( |
| 136 | "POST", "https://google.serper.dev/search", |
| 137 | headers={"X-API-KEY": api_key}, |
| 138 | json_data={ |
| 139 | "q": query, |
| 140 | "num": count, |
| 141 | "tbs": f"cdr:1,cd_min:{_serper_date_param(date_range[0])},cd_max:{_serper_date_param(date_range[1])}", |
| 142 | }, |
| 143 | timeout=15, |
| 144 | ) |
| 145 | items = [] |
| 146 | for i, r in enumerate((data.get("organic", []))[:count]): |
| 147 | raw_date = r.get("date") or "" |
| 148 | pub_date = _parse_serper_date(raw_date) |
| 149 | if not _in_date_range(pub_date, date_range): |
| 150 | continue |
| 151 | items.append({ |
| 152 | "id": f"WS{i + 1}", |
| 153 | "title": r.get("title", ""), |
| 154 | "url": r.get("link", ""), |
| 155 | "source_domain": _domain(r.get("link", "")), |
| 156 | "snippet": r.get("snippet", ""), |
| 157 | "date": pub_date, |
| 158 | "relevance": 0.8, |
| 159 | "why_relevant": "Serper web search", |
| 160 | }) |
| 161 | artifact = {"label": "serper", "webSearchQueries": [query], "resultCount": len(items)} |
| 162 | return items, artifact |
| 163 | |
| 164 | |
| 165 | # --------------------------------------------------------------------------- |
| 166 | # Parallel AI Search |
| 167 | # --------------------------------------------------------------------------- |
| 168 | |
| 169 | def parallel_search( |
| 170 | query: str, date_range: tuple[str, str], api_key: str, count: int = 5, |
| 171 | ) -> tuple[list[dict], dict]: |
| 172 | data = http.request( |
| 173 | "POST", "https://api.parallel.ai/v1/search", |
| 174 | headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}, |
| 175 | json_data={ |
| 176 | "search_queries": [query], |
| 177 | "advanced_settings": {"max_results": count}, |
| 178 | }, |
| 179 | timeout=15, |
| 180 | ) |
| 181 | items = [] |
| 182 | for i, r in enumerate((data.get("results", []))[:count]): |
| 183 | if not isinstance(r, dict): |
| 184 | continue |
| 185 | url = r.get("url", "") |
| 186 | if not url: |
| 187 | continue |
| 188 | raw_date = r.get("publish_date") or "" |
| 189 | pub_date = _normalize_date(raw_date[:10]) if raw_date else None |
| 190 | if not _in_date_range(pub_date, date_range): |
| 191 | continue |
| 192 | items.append({ |
| 193 | "id": f"WP{i + 1}", |
| 194 | "title": r.get("title", ""), |
| 195 | "url": url, |
| 196 | "source_domain": _domain(url), |
| 197 | "snippet": ((r.get("excerpts") or [""])[0] or "")[:500], |
| 198 | "date": pub_date, |
| 199 | "relevance": 0.8, |
| 200 | "why_relevant": "Parallel AI web search", |
| 201 | }) |
| 202 | artifact = {"label": "parallel", "webSearchQueries": [query], "resultCount": len(items)} |
| 203 | return items, artifact |
| 204 | |
| 205 | |
| 206 | def _parse_serper_date(raw: str) -> str | None: |
| 207 | if not raw: |
| 208 | return None |
| 209 | normalized = _normalize_date(raw) |
| 210 | if normalized: |
| 211 | return normalized |
| 212 | for fmt in ("%b %d, %Y", "%B %d, %Y", "%Y-%m-%d"): |
| 213 | try: |
| 214 | return datetime.strptime(raw.strip(), fmt).date().isoformat() |
| 215 | except ValueError: |
| 216 | continue |
| 217 | return None |
| 218 | |
| 219 | |
| 220 | |
| 221 | |
| 222 | # --------------------------------------------------------------------------- |
| 223 | # Dispatcher |
| 224 | # --------------------------------------------------------------------------- |
| 225 | |
| 226 | def web_search( |
| 227 | query: str, |
| 228 | date_range: tuple[str, str], |
| 229 | config: dict, |
| 230 | backend: str = "auto", |
| 231 | ) -> tuple[list[dict], dict]: |
| 232 | """Run web search with the specified or auto-detected backend.""" |
| 233 | if backend == "auto": |
| 234 | if config.get("BRAVE_API_KEY"): |
| 235 | backend = "brave" |
| 236 | elif config.get("EXA_API_KEY"): |
| 237 | backend = "exa" |
| 238 | elif config.get("SERPER_API_KEY"): |
| 239 | backend = "serper" |
| 240 | elif config.get("PARALLEL_API_KEY"): |
| 241 | backend = "parallel" |
| 242 | elif env.keyless_web_allowed(config): |
| 243 | # No paid key and the host has no native search -> use the keyless |
| 244 | # floor. On a native-search host this branch is skipped (the model |
| 245 | # supplies web results itself), so the engine returns nothing here. |
| 246 | backend = "keyless" |
| 247 | else: |
| 248 | return [], {} |
| 249 | items: list[dict] = [] |
| 250 | artifact: dict = {} |
| 251 | if backend == "brave": |
| 252 | key = config.get("BRAVE_API_KEY") |
| 253 | if not key: |
| 254 | raise RuntimeError("BRAVE_API_KEY is required when web_backend='brave'") |
| 255 | items, artifact = brave_search(query, date_range, key) |
| 256 | elif backend == "exa": |
| 257 | key = config.get("EXA_API_KEY") |
| 258 | if not key: |
| 259 | raise RuntimeError("EXA_API_KEY is required when web_backend='exa'") |
| 260 | items, artifact = exa_search(query, date_range, key) |
| 261 | elif backend == "serper": |
| 262 | key = config.get("SERPER_API_KEY") |
| 263 | if not key: |
| 264 | raise RuntimeError("SERPER_API_KEY is required when web_backend='serper'") |
| 265 | items, artifact = serper_search(query, date_range, key) |
| 266 | elif backend == "parallel": |
| 267 | key = config.get("PARALLEL_API_KEY") |
| 268 | if not key: |
| 269 | raise RuntimeError("PARALLEL_API_KEY is required when web_backend='parallel'") |
| 270 | items, artifact = parallel_search(query, date_range, key) |
| 271 | elif backend == "parallel-mcp": |
| 272 | items, artifact = parallel_mcp.search( |
| 273 | query, date_range, config.get("PARALLEL_API_KEY") |
| 274 | ) |
| 275 | elif backend == "keyless": |
| 276 | items, artifact = web_search_keyless.keyless_search(query, date_range, config) |
| 277 | elif backend != "none": |
| 278 | raise ValueError(f"Unsupported web backend: {backend!r}") |
| 279 | else: |
| 280 | return [], {} |
| 281 | if items and not _reddit_excluded(config): |
| 282 | # Reddit enrichment is a best-effort secondary fetch on already-retrieved |
| 283 | # web results. Isolate its HTTP failures in a throwaway capture sink so a |
| 284 | # reddit.com fetch failure (e.g. a 403 on a datacenter IP) is not |
| 285 | # attributed to the web/grounding source itself — which would otherwise |
| 286 | # discard the successfully retrieved results and report the source failed. |
| 287 | with http.capture_failures(): |
| 288 | items = _enrich_reddit_items(items) |
| 289 | return items, artifact |
| 290 | |
| 291 | |
| 292 | def _reddit_excluded(config: dict) -> bool: |
| 293 | """Return True when EXCLUDE_SOURCES contains 'reddit'. |
| 294 | |
| 295 | Respects the same suppression knob the pipeline uses for source gating, |
| 296 | so a user who set EXCLUDE_SOURCES=reddit doesn't get Reddit content |
| 297 | smuggled back in via web-search URLs. |
| 298 | """ |
| 299 | raw = (config.get("EXCLUDE_SOURCES") or "").split(",") |
| 300 | return any(s.strip().lower() == "reddit" for s in raw) |
| 301 | |
| 302 | |
| 303 | def _enrich_reddit_items(items: list[dict]) -> list[dict]: |
| 304 | """Enrich web search results that are Reddit URLs with thread body and comments. |
| 305 | |
| 306 | Claude Code's WebFetch blocks reddit.com, so the model can't retrieve |
| 307 | Reddit content from web search results. This fetches it via the public |
| 308 | JSON API (reddit.com/.../.json) which bypasses that restriction. |
| 309 | |
| 310 | Callers should gate this with EXCLUDE_SOURCES=reddit handling (see |
| 311 | `_reddit_excluded`) so a user who explicitly excluded Reddit doesn't |
| 312 | get Reddit content via web-search URLs. |
| 313 | """ |
| 314 | from . import reddit_enrich |
| 315 | from .reddit_enrich import RedditRateLimitError |
| 316 | |
| 317 | for item in items: |
| 318 | url = item.get("url", "") |
| 319 | if "reddit.com" not in url or "/comments/" not in url: |
| 320 | continue |
| 321 | try: |
| 322 | thread_data = reddit_enrich.fetch_thread_data(url, timeout=8) |
| 323 | if not thread_data: |
| 324 | continue |
| 325 | parsed = reddit_enrich.parse_thread_data(thread_data) |
| 326 | # selftext lives under parsed["submission"], not at the top level |
| 327 | selftext = (parsed.get("submission") or {}).get("selftext", "") |
| 328 | if selftext: |
| 329 | item["snippet"] = selftext[:2000] |
| 330 | comments = parsed.get("comments", []) |
| 331 | top = reddit_enrich.get_top_comments(comments) |
| 332 | if top: |
| 333 | item["top_comments"] = [ |
| 334 | {"score": c.get("score", 0), "excerpt": (c.get("body") or "")[:200]} |
| 335 | for c in top[:5] |
| 336 | ] |
| 337 | item["enriched_via"] = "reddit_json_api" |
| 338 | except RedditRateLimitError as exc: |
| 339 | # Stop iterating to avoid flooding more 429s |
| 340 | sys.stderr.write(f"[Web] Reddit rate-limited, halting enrichment: {exc}\n") |
| 341 | break |
| 342 | except Exception as exc: |
| 343 | sys.stderr.write(f"[Web] Reddit enrichment failed for {url}: {exc}\n") |
| 344 | return items |
| 345 | |
| 346 | |
| 347 | # --------------------------------------------------------------------------- |
| 348 | # Helpers |
| 349 | # --------------------------------------------------------------------------- |
| 350 | |
| 351 | def _normalize_date(value: object) -> str | None: |
| 352 | if value is None: |
| 353 | return None |
| 354 | parsed = dates.parse_date(str(value).strip()) |
| 355 | if not parsed: |
| 356 | return None |
| 357 | return parsed.date().isoformat() |
| 358 | |
| 359 | |
| 360 | def _serper_date_param(iso_date: str) -> str: |
| 361 | """Convert YYYY-MM-DD to MM/DD/YYYY for Serper tbs parameter.""" |
| 362 | parts = iso_date.split("-") |
| 363 | return f"{parts[1]}/{parts[2]}/{parts[0]}" |
| 364 | |
| 365 | |
| 366 | def _in_date_range(pub_date: str | None, date_range: tuple[str, str]) -> bool: |
| 367 | if not pub_date: |
| 368 | return False |
| 369 | return date_range[0] <= pub_date <= date_range[1] |
| 370 | |
| 371 | |
| 372 | def _domain(url: str) -> str: |
| 373 | return urlparse(url).netloc.strip().lower() |
| 374 |