Coverage for src/local_deep_research/web_search_engines/engines/search_engine

1import re

2from typing import Any, Dict, List, Optional, Tuple

4from langchain_core.language_models import BaseLLM

6from ...research_library.downloaders.extraction import extract_content

7from ...security import redact_url_for_log

8from ...security.safe_requests import safe_get

9from ...security.secure_logging import logger

10from ..rate_limiting import RateLimitError

11from ..search_engine_base import BaseSearchEngine, Exposure, Sensitivity

14class WaybackSearchEngine(BaseSearchEngine):

15 """

16 Internet Archive Wayback Machine search engine implementation

17 Provides access to historical versions of web pages

18 """

20 # Mark as public search engine

21 is_public = True

22 egress_sensitivity = Sensitivity.NON_SENSITIVE

23 egress_exposure = Exposure.EXPOSING

25 def __init__(

26 self,

27 max_results: int = 10,

28 max_snapshots_per_url: int = 3,

29 llm: Optional[BaseLLM] = None,

30 language: str = "English",

31 max_filtered_results: Optional[int] = None,

32 closest_only: bool = False,

33 settings_snapshot: Optional[Dict[str, Any]] = None,

34 ):

35 """

36 Initialize the Wayback Machine search engine.

38 Args:

39 max_results: Maximum number of search results

40 max_snapshots_per_url: Maximum snapshots to retrieve per URL

41 llm: Language model for relevance filtering

42 language: Language for content processing

43 max_filtered_results: Maximum number of results to keep after filtering

44 closest_only: If True, only retrieves the closest snapshot for each URL

45 """

46 # Initialize the BaseSearchEngine with LLM, max_filtered_results, and max_results

47 super().__init__(

48 llm=llm,

49 max_filtered_results=max_filtered_results,

50 max_results=max_results,

51 settings_snapshot=settings_snapshot,

52 )

53 self.max_snapshots_per_url = max_snapshots_per_url

54 self.language = language

55 self.closest_only = closest_only

57 # API endpoints

58 self.available_api = "https://archive.org/wayback/available"

59 self.cdx_api = "https://web.archive.org/cdx/search/cdx"

61 def _extract_urls_from_query(self, query: str) -> List[str]:

62 """

63 Extract URLs from a query string or interpret as an URL if possible.

64 For non-URL queries, use a DuckDuckGo search to find relevant URLs.

66 Args:

67 query: The search query or URL

69 Returns:

70 List of URLs to search in the Wayback Machine

71 """

72 # Check if the query is already a URL

73 url_pattern = re.compile(r"https?://(?:[-\w.]|(?:%[\da-fA-F]{2}))+")

74 urls = url_pattern.findall(query)

76 if urls:

77 logger.info(f"Found {len(urls)} URLs in query")

78 return urls

80 # Check if query is a domain without http prefix

81 domain_pattern = re.compile(r"^(?:[-\w.]|(?:%[\da-fA-F]{2}))+\.\w+$")

82 if domain_pattern.match(query):

83 logger.info(f"Query appears to be a domain: {query}")

84 return [f"http://{query}"]

86 # For non-URL queries, use DuckDuckGo to find relevant URLs

87 logger.info(

88 "Query is not a URL, using DuckDuckGo to find relevant URLs"

89 )

90 try:

91 # Import DuckDuckGo search engine

92 from langchain_community.utilities import DuckDuckGoSearchAPIWrapper

94 # Use max_results from parent class, but limit to 5 for URL discovery

95 url_search_limit = min(5, self.max_results)

96 ddg = DuckDuckGoSearchAPIWrapper(max_results=url_search_limit)

97 # Pass max_results as a positional argument

98 results = ddg.results(query, url_search_limit)

100 # Extract URLs from results

101 ddg_urls = [

102 str(result.get("link"))

103 for result in results

104 if result.get("link")

105 ]

106 if ddg_urls:

107 logger.info(

108 f"Found {len(ddg_urls)} URLs from DuckDuckGo search"

109 )

110 return ddg_urls

111 except Exception as e:

112 safe_msg = self._scrub_error(e)

113 logger.exception(

114 f"Error using DuckDuckGo for URL discovery ({type(e).__name__}): {safe_msg}"

115 )

116

117 # Fallback: treat the query as a potential domain or path

118 if "/" in query and "." in query:

119 logger.info(f"Treating query as a partial URL: {query}")

120 return [f"http://{query}"]

121 if "." in query: 121 ↛ 122line 121 didn't jump to line 122 because the condition on line 121 was never true

122 logger.info(f"Treating query as a domain: {query}")

123 return [f"http://{query}"]

124

125 # Return empty list if nothing worked

126 logger.warning(f"Could not extract any URLs from query: {query}")

127 return []

128

129 def _format_timestamp(self, timestamp: str) -> str:

130 """Format Wayback Machine timestamp into readable date"""

131 if len(timestamp) < 14:

132 return timestamp

133

134 try:

135 year = timestamp[0:4]

136 month = timestamp[4:6]

137 day = timestamp[6:8]

138 hour = timestamp[8:10]

139 minute = timestamp[10:12]

140 second = timestamp[12:14]

141 return f"{year}-{month}-{day} {hour}:{minute}:{second}"

142 except Exception:

143 logger.debug("Timestamp formatting failed, returning original")

144 return timestamp

145

146 def _get_wayback_snapshots(self, url: str) -> List[Dict[str, Any]]:

147 """

148 Get snapshots from the Wayback Machine for a specific URL.

149

150 Args:

151 url: URL to get snapshots for

152

153 Returns:

154 List of snapshot dictionaries

155 """

156 snapshots = []

157

158 try:

159 if self.closest_only:

160 # Get only the closest snapshot

161 response = safe_get(self.available_api, params={"url": url})

162

163 # Check for rate limit

164 if response.status_code == 429:

165 raise RateLimitError("Wayback Machine rate limit exceeded") # noqa: TRY301 — re-raised by except RateLimitError for base class retry

166

167 data = response.json()

168

169 if (

170 "archived_snapshots" in data

171 and "closest" in data["archived_snapshots"]

172 ):

173 snapshot = data["archived_snapshots"]["closest"]

174 snapshot_url = snapshot["url"]

175 timestamp = snapshot["timestamp"]

176

177 snapshots.append(

178 {

179 "timestamp": timestamp,

180 "formatted_date": self._format_timestamp(timestamp),

181 "url": snapshot_url,

182 "original_url": url,

183 "available": snapshot.get("available", True),

184 "status": snapshot.get("status", "200"),

185 }

186 )

187 else:

188 # Get multiple snapshots using CDX API

189 response = safe_get(

190 self.cdx_api,

191 params={

192 "url": url,

193 "output": "json",

194 "fl": "timestamp,original,statuscode,mimetype",

195 "collapse": "timestamp:4", # Group by year

196 "limit": self.max_snapshots_per_url,

197 },

198 )

199

200 # Check for rate limit

201 if response.status_code == 429:

202 raise RateLimitError( # noqa: TRY301 — re-raised by except RateLimitError for base class retry

203 "Wayback Machine CDX API rate limit exceeded"

204 )

205

206 # Check if response is valid JSON

207 data = response.json()

208

209 # First item is the header

210 if len(data) > 1:

211 headers = data[0]

212 for item in data[1:]:

213 snapshot = dict(zip(headers, item, strict=False))

214 timestamp = snapshot.get("timestamp", "")

215

216 wayback_url = (

217 f"https://web.archive.org/web/{timestamp}/{url}"

218 )

219

220 snapshots.append(

221 {

222 "timestamp": timestamp,

223 "formatted_date": self._format_timestamp(

224 timestamp

225 ),

226 "url": wayback_url,

227 "original_url": url,

228 "available": True,

229 "status": snapshot.get("statuscode", "200"),

230 }

231 )

232

233 # Limit to max snapshots per URL

234 snapshots = snapshots[: self.max_snapshots_per_url]

235

236 except RateLimitError:

237 # Re-raise rate limit errors for base class retry handling

238 raise

239 except Exception as e:

240 safe_msg = self._scrub_error(e)

241 logger.exception(

242 f"Error getting Wayback snapshots for {redact_url_for_log(url)} ({type(e).__name__}): {safe_msg}"

243 )

244

245 return snapshots

246

247 def _get_previews(self, query: str) -> List[Dict[str, Any]]:

248 """

249 Get preview information for Wayback Machine snapshots.

250

251 Args:

252 query: The search query

253

254 Returns:

255 List of preview dictionaries

256 """

257 logger.info(f"Getting Wayback Machine previews for query: {query}")

258

259 # Extract URLs from query

260 urls = self._extract_urls_from_query(query)

261

262 if not urls:

263 logger.warning(f"No URLs found in query: {query}")

264 return []

265

266 # Get snapshots for each URL

267 all_snapshots = []

268 for url in urls:

269 snapshots = self._get_wayback_snapshots(url)

270 all_snapshots.extend(snapshots)

271

272 # Apply rate limiting between requests

273 if len(urls) > 1: 273 ↛ 274line 273 didn't jump to line 274 because the condition on line 273 was never true

274 self.rate_tracker.apply_rate_limit(self.engine_type)

275

276 # Format as previews

277 previews = []

278 for snapshot in all_snapshots:

279 preview = {

280 "id": f"{snapshot['timestamp']}_{snapshot['original_url']}",

281 "title": f"Archive of {snapshot['original_url']} ({snapshot['formatted_date']})",

282 "link": snapshot["url"],

283 "snippet": f"Archived version from {snapshot['formatted_date']}",

284 "original_url": snapshot["original_url"],

285 "timestamp": snapshot["timestamp"],

286 "formatted_date": snapshot["formatted_date"],

287 }

288 previews.append(preview)

289

290 logger.info(f"Found {len(previews)} Wayback Machine snapshots")

291 return previews

292

293 def _remove_boilerplate(self, html: str) -> str:

294 """Remove boilerplate using the shared extraction pipeline."""

295 if not html or not html.strip():

296 return ""

297 try:

298 return extract_content(html, language=self.language) or ""

299 except Exception as e:

300 safe_msg = self._scrub_error(e)

301 logger.exception(

302 f"Error removing boilerplate ({type(e).__name__}): {safe_msg}"

303 )

304 return html

305

306 def _get_wayback_content(self, url: str) -> Tuple[str, str]:

307 """

308 Retrieve content from a Wayback Machine URL.

309

310 Args:

311 url: Wayback Machine URL

312

313 Returns:

314 Tuple of (raw_html, cleaned_text)

315 """

316 try:

317 headers = {

318 "User-Agent": "Mozilla/5.0 (Local Deep Research Bot; research project)"

319 }

320 response = safe_get(url, headers=headers, timeout=10)

321 raw_html = response.text

322

323 # Clean the HTML

324 cleaned_text = self._remove_boilerplate(raw_html)

325

326 return raw_html, cleaned_text

327 except Exception as e:

328 safe_msg = self._scrub_error(e)

329 logger.exception(

330 f"Error retrieving content from {redact_url_for_log(url)} ({type(e).__name__}): {safe_msg}"

331 )

332 return "", f"Error retrieving content: {safe_msg}"

333

334 def _get_full_content(

335 self, relevant_items: List[Dict[str, Any]]

336 ) -> List[Dict[str, Any]]:

337 """

338 Get full content for the relevant Wayback Machine snapshots.

339

340 Args:

341 relevant_items: List of relevant preview dictionaries

342

343 Returns:

344 List of result dictionaries with full content

345 """

346 logger.info(

347 f"Getting full content for {len(relevant_items)} Wayback Machine snapshots"

348 )

349

350 results = []

351 for item in relevant_items:

352 wayback_url = item.get("link")

353 if not wayback_url:

354 results.append(item)

355 continue

356

357 logger.info(

358 f"Retrieving content from {redact_url_for_log(wayback_url)}"

359 )

360

361 try:

362 # Retrieve content

363 raw_html, full_content = self._get_wayback_content(wayback_url)

364

365 # Add full content to the result

366 result = item.copy()

367 result["raw_html"] = raw_html

368 result["full_content"] = full_content

369

370 results.append(result)

371

372 # Apply rate limiting

373 self.rate_tracker.apply_rate_limit(self.engine_type)

374 except Exception as e:

375 safe_msg = self._scrub_error(e)

376 logger.exception(

377 f"Error processing {redact_url_for_log(wayback_url)} ({type(e).__name__}): {safe_msg}"

378 )

379 results.append(item)

380

381 return results

382

383 def search_by_url(

384 self, url: str, max_snapshots: int | None = None

385 ) -> List[Dict[str, Any]]:

386 """

387 Search for archived versions of a specific URL.

388

389 Args:

390 url: The URL to search for archives

391 max_snapshots: Maximum number of snapshots to return

392

393 Returns:

394 List of snapshot dictionaries

395 """

396 max_snapshots = max_snapshots or self.max_snapshots_per_url

397

398 snapshots = self._get_wayback_snapshots(url)

399 previews = []

400

401 for snapshot in snapshots[:max_snapshots]:

402 preview = {

403 "id": f"{snapshot['timestamp']}_{snapshot['original_url']}",

404 "title": f"Archive of {snapshot['original_url']} ({snapshot['formatted_date']})",

405 "link": snapshot["url"],

406 "snippet": f"Archived version from {snapshot['formatted_date']}",

407 "original_url": snapshot["original_url"],

408 "timestamp": snapshot["timestamp"],

409 "formatted_date": snapshot["formatted_date"],

410 }

411 previews.append(preview)

412

413 return self._get_full_content(previews)

414

415 def search_by_date_range(

416 self, url: str, start_date: str, end_date: str

417 ) -> List[Dict[str, Any]]:

418 """

419 Search for archived versions of a URL within a date range.

420

421 Args:

422 url: The URL to search for archives

423 start_date: Start date in format YYYYMMDD

424 end_date: End date in format YYYYMMDD

425

426 Returns:

427 List of snapshot dictionaries

428 """

429 try:

430 # Use CDX API with date range

431 response = safe_get(

432 self.cdx_api,

433 params={

434 "url": url,

435 "output": "json",

436 "fl": "timestamp,original,statuscode,mimetype",

437 "from": start_date,

438 "to": end_date,

439 "limit": self.max_snapshots_per_url,

440 },

441 )

442

443 # Process response

444 data = response.json()

445

446 # First item is the header

447 if len(data) <= 1:

448 return []

449

450 headers = data[0]

451 snapshots = []

452

453 for item in data[1:]:

454 snapshot = dict(zip(headers, item, strict=False))

455 timestamp = snapshot.get("timestamp", "")

456

457 wayback_url = f"https://web.archive.org/web/{timestamp}/{url}"

458

459 snapshots.append(

460 {

461 "id": f"{timestamp}_{url}",

462 "title": f"Archive of {url} ({self._format_timestamp(timestamp)})",

463 "link": wayback_url,

464 "snippet": f"Archived version from {self._format_timestamp(timestamp)}",

465 "original_url": url,

466 "timestamp": timestamp,

467 "formatted_date": self._format_timestamp(timestamp),

468 }

469 )

470

471 return self._get_full_content(snapshots)

472

473 except Exception as e:

474 safe_msg = self._scrub_error(e)

475 logger.exception(

476 f"Error searching date range for {redact_url_for_log(url)} ({type(e).__name__}): {safe_msg}"

477 )

478 return []

479

480 def get_latest_snapshot(self, url: str) -> Optional[Dict[str, Any]]:

481 """

482 Get the most recent snapshot of a URL.

483

484 Args:

485 url: The URL to get the latest snapshot for

486

487 Returns:

488 Dictionary with snapshot information or None if not found

489 """

490 try:

491 response = safe_get(self.available_api, params={"url": url})

492 data = response.json()

493

494 if (

495 "archived_snapshots" in data

496 and "closest" in data["archived_snapshots"]

497 ):

498 snapshot = data["archived_snapshots"]["closest"]

499 timestamp = snapshot["timestamp"]

500 wayback_url = snapshot["url"]

501

502 result = {

503 "id": f"{timestamp}_{url}",

504 "title": f"Latest archive of {url} ({self._format_timestamp(timestamp)})",

505 "link": wayback_url,

506 "snippet": f"Archived version from {self._format_timestamp(timestamp)}",

507 "original_url": url,

508 "timestamp": timestamp,

509 "formatted_date": self._format_timestamp(timestamp),

510 }

511

512 raw_html, full_content = self._get_wayback_content(wayback_url)

513 result["raw_html"] = raw_html

514 result["full_content"] = full_content

515

516 return result

517

518 return None

519

520 except Exception as e:

521 safe_msg = self._scrub_error(e)

522 logger.exception(

523 f"Error getting latest snapshot for {redact_url_for_log(url)} ({type(e).__name__}): {safe_msg}"

524 )

525 return None

Coverage for src/local_deep_research/web_search_engines/engines/search_engine_wayback.py: 97%

199 statements