Coverage for src/local_deep_research/web_search_engines/engines/search_engine_wayback.py: 98%
306 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +0000
1import json
2import re
3from typing import Any, Dict, List, Optional, Tuple
5import requests
6from langchain_core.language_models import BaseLLM
8from ...research_library.downloaders.extraction import extract_content
9from ...security import redact_url_for_log
10from ...security.safe_requests import safe_get
11from ...security.secure_logging import logger
12from ..rate_limiting import RateLimitError
13from ..wayback_client import get_availability_response, validate_archive_url
14from ..search_engine_base import BaseSearchEngine, Exposure, Sensitivity
16# Characters that separate list entries: ASCII and fullwidth commas and
17# semicolons, and the ideographic comma. None is valid in a host name.
18_LIST_SEPARATORS = ",;\uff0c\uff1b\u3001"
20# Characters that end a URL in running text rather than belong to it:
21# sentence stops (ASCII, CJK, fullwidth, ellipsis), Markdown emphasis,
22# straight and curly closing quotes, and closing guillemets.
23_PROSE_URL_ENDINGS = frozenset(
24 ".,;:!?*'\u2019\u201d\u00bb\u203a"
25 "\u3001\u3002\uff0c\uff1b\uff1a\uff01\uff1f\uff0e\u2026"
26)
28# Opening and closing bracket pairs, ASCII, fullwidth and CJK. In running
29# text a closing bracket ends a URL only when the URL holds more of it than
30# of its opener, so a balanced pair in a path stays, as in a Wikipedia title
31# that ends in a bracket.
32_URL_BRACKET_PAIRS: Tuple[Tuple[str, str], ...] = (
33 ("(", ")"),
34 ("[", "]"),
35 ("{", "}"),
36 ("\uff08", "\uff09"),
37 ("\uff3b", "\uff3d"),
38 ("\uff5b", "\uff5d"),
39 ("\u300c", "\u300d"),
40 ("\u300e", "\u300f"),
41 ("\u3010", "\u3011"),
42 ("\u3008", "\u3009"),
43 ("\u300a", "\u300b"),
44 ("\u3014", "\u3015"),
45)
48class WaybackSearchEngine(BaseSearchEngine):
49 """
50 Look up historical versions of supplied URLs or domains in the
51 Internet Archive Wayback Machine. This engine does not discover URLs
52 through another search provider.
53 """
55 # Mark as public search engine
56 is_public = True
57 egress_sensitivity = Sensitivity.NON_SENSITIVE
58 egress_exposure = Exposure.EXPOSING
60 def __init__(
61 self,
62 max_results: int = 10,
63 max_snapshots_per_url: int = 3,
64 llm: Optional[BaseLLM] = None,
65 language: str = "English",
66 max_filtered_results: Optional[int] = None,
67 closest_only: bool = False,
68 settings_snapshot: Optional[Dict[str, Any]] = None,
69 ):
70 """
71 Initialize the Wayback Machine search engine.
73 Args:
74 max_results: Maximum number of search results
75 max_snapshots_per_url: Maximum snapshots to retrieve per URL
76 llm: Language model for relevance filtering
77 language: Language for content processing
78 max_filtered_results: Maximum number of results to keep after filtering
79 closest_only: If True, only retrieves the closest snapshot for each URL
80 """
81 # Initialize the BaseSearchEngine with LLM, max_filtered_results, and max_results
82 super().__init__(
83 llm=llm,
84 max_filtered_results=max_filtered_results,
85 max_results=max_results,
86 settings_snapshot=settings_snapshot,
87 )
88 self.max_snapshots_per_url = max_snapshots_per_url
89 self.language = language
90 self.closest_only = closest_only
92 # API endpoints
93 self.available_api = "https://archive.org/wayback/available"
94 self.cdx_api = "https://web.archive.org/cdx/search/cdx"
96 def _extract_urls_from_query(self, query: str) -> List[str]:
97 """
98 Extract explicit URLs or interpret a bare domain/path as a URL.
99 Topic queries need URL discovery through a configured search
100 provider before they can be looked up in the Wayback Machine.
102 Args:
103 query: The search query or URL
105 Returns:
106 List of URLs to search in the Wayback Machine
107 """
108 # Accept a complete bare domain, optionally with a port, path,
109 # query or fragment, before looking for embedded URLs: query
110 # parameters can themselves contain another URL.
111 target = query.strip()
112 domain_pattern = re.compile(
113 r"(?:[-\w.]|%[\da-fA-F]{2})+\.[\w-]+"
114 r"(?::\d+)?(?:[/?#][^\s<>\"`]*)?"
115 )
116 if domain_pattern.fullmatch(target):
117 logger.info("Query appears to be a domain or partial URL")
118 return [f"http://{target}"]
120 # Inline Markdown link labels are display text, not archive
121 # destinations. Track balanced brackets, including IPv6 labels;
122 # escaped brackets do not open or close a label.
123 label_starts = []
124 label_spans = []
125 for bracket in re.finditer(r"\\.|[\[\]]", query):
126 if bracket.group() == "[":
127 label_starts.append(bracket.start())
128 elif bracket.group() == "]" and label_starts:
129 start = label_starts.pop()
130 if query[bracket.end() : bracket.end() + 1] == "(":
131 label_spans.append((start, bracket.end() + 1))
133 # Labels nest like the brackets that delimit them, and URL starts
134 # only move forward, so a stack of the labels open at each URL
135 # finds its innermost label without rescanning every label.
136 label_spans.sort()
137 open_labels: List[Tuple[int, int]] = []
138 next_label = 0
140 # A URL's greedy run ends at whitespace, a double quote, a
141 # backtick or an angle bracket, and every URL of a compact list
142 # shares one run. No URL slices or rescans the rest of its run:
143 # searches that look past it are cached for the run's later URLs,
144 # so a long pasted list stays linear.
145 url_start = re.compile(r"https?://(?=[^\s<>\"`])", re.IGNORECASE)
146 run_boundary = re.compile(r"[\s<>\"`]")
147 quote_separator = re.compile(
148 rf"'[{_LIST_SEPARATORS}]*'(?=https?://)", re.IGNORECASE
149 )
150 path_separator = re.compile(
151 rf"[{_LIST_SEPARATORS}](?=https?://)", re.IGNORECASE
152 )
153 path_mark = re.compile(r"[?#]")
154 authority_mark = re.compile(r"[/?#]")
155 host_separator = re.compile(rf"[{_LIST_SEPARATORS}]")
156 bracket_pairs = {"(": re.compile(r"[()]"), "[": re.compile(r"[\[\]]")}
157 run_end = 0
158 closing_quote = None
159 first_hits: Dict[re.Pattern, Optional[int]] = {}
161 def first_in_run(pattern: re.Pattern, position: int) -> Optional[int]:
162 # A run's URLs are visited in order. An earlier search that
163 # found nothing, or found a match at or after this position,
164 # already answers the question.
165 hit = first_hits.get(pattern, -1)
166 if hit is not None and hit < position:
167 found = pattern.search(query, position, run_end)
168 hit = found.start() if found else None
169 first_hits[pattern] = hit
170 return hit
172 def host_separator_in(
173 scheme_end: int, authority_end: int
174 ) -> Optional[int]:
175 # The host follows the authority's last "@".
176 host_start = query.rfind("@", scheme_end, authority_end) + 1
177 cut = host_separator.search(
178 query, max(host_start, scheme_end), authority_end
179 )
180 return cut.start() if cut else None
182 urls = []
183 cursor = 0
184 while match := url_start.search(query, cursor):
185 start = match.start()
186 while (
187 next_label < len(label_spans)
188 and label_spans[next_label][0] < start
189 ):
190 label = label_spans[next_label]
191 next_label += 1
192 while open_labels and open_labels[-1][1] <= label[0]: 192 ↛ 193line 192 didn't jump to line 193 because the condition on line 192 was never true
193 open_labels.pop()
194 open_labels.append(label)
195 while open_labels and open_labels[-1][1] <= start:
196 open_labels.pop()
197 if open_labels:
198 cursor = open_labels[-1][1]
199 continue
201 if start >= run_end:
202 boundary = run_boundary.search(query, start)
203 run_end = boundary.start() if boundary else len(query)
204 first_hits.clear()
205 # An unmatched opening prose quote must not turn an
206 # internal apostrophe into a closing delimiter: the run's
207 # last apostrophe closes a quote only when nothing but
208 # closing punctuation follows it. The backward scan stops
209 # at this URL's scheme at the latest.
210 tail = run_end
211 while query[tail - 1] in ".,;:!?)]}":
212 tail -= 1
213 closing_quote = tail - 1 if query[tail - 1] == "'" else None
215 # List separators after the host are prose/list separators,
216 # including a compact list with no whitespace. Keep these
217 # characters in userinfo, paths and query values.
218 # The authority read stops at the next "/", "?" or "#", and a
219 # following URL's scheme supplies one, so it never spans the
220 # rest of a compact list.
221 scheme_end = match.end()
222 authority_match = authority_mark.search(query, scheme_end, run_end)
223 authority_end = (
224 authority_match.start() if authority_match else run_end
225 )
226 host_separator_at = host_separator_in(scheme_end, authority_end)
228 url_end = run_end
229 preceding = query[start - 1] if start else ""
230 quoted = False
231 if preceding == "'":
232 # Strip the enclosing quote, retaining apostrophes in
233 # the target. A following quoted URL can start a list.
234 separator = first_in_run(quote_separator, start)
235 quote_end = closing_quote if separator is None else separator
236 if quote_end is not None:
237 url_end = quote_end
238 quoted = True
240 # Markdown links, parentheses and square brackets add a
241 # closing delimiter. Preserve balanced pairs inside the URL,
242 # including IPv6 hosts, and literal closers in raw URLs. A
243 # closer after a separator in the host cannot change the
244 # target, so the scan stops there.
245 wrapped = preceding in bracket_pairs
246 if wrapped:
247 depth = 0
248 for bracket in bracket_pairs[preceding].finditer(
249 query,
250 start,
251 run_end if host_separator_at is None else host_separator_at,
252 ):
253 if bracket.group() == preceding:
254 depth += 1
255 elif depth:
256 depth -= 1
257 else:
258 url_end = bracket.start()
259 break
261 delimited = quoted or (bool(preceding) and preceding in '"<`')
262 if not wrapped and not delimited:
263 # A compact list may separate complete path URLs too.
264 # Query/fragment values can contain literal nested URLs;
265 # explicitly delimited targets also retain literal paths.
266 path_end = first_in_run(path_mark, start)
267 separator = first_in_run(path_separator, start)
268 if separator is not None and (
269 path_end is None or separator < path_end
270 ):
271 url_end = separator
273 if host_separator_at is not None and url_end >= host_separator_at:
274 url_end = host_separator_at
275 elif url_end < authority_end:
276 # A cut inside the authority shortens it: an "@" after
277 # the cut no longer marks userinfo, so re-read the host.
278 separator = host_separator_in(scheme_end, url_end)
279 if separator is not None: 279 ↛ 280line 279 didn't jump to line 280 because the condition on line 279 was never true
280 url_end = separator
281 url = query[start:url_end]
283 # A complete raw URL or an explicitly delimited destination
284 # keeps literal punctuation. Inside a parenthesis or square
285 # bracket, a list puts its separator before the closer or the
286 # next space. In prose, peel sentence endings, quotes, emphasis,
287 # CJK punctuation and a closer (ASCII, fullwidth or CJK) without
288 # its opener, without damaging balanced pairs in URL paths.
289 if wrapped:
290 url = url.rstrip(_LIST_SEPARATORS)
291 elif not delimited and url != target:
292 unmatched = {
293 closer: max(0, url.count(closer) - url.count(opener))
294 for opener, closer in _URL_BRACKET_PAIRS
295 }
296 end = len(url)
297 while end: 297 ↛ 306line 297 didn't jump to line 306 because the condition on line 297 was always true
298 character = url[end - 1]
299 if character in _PROSE_URL_ENDINGS:
300 end -= 1
301 elif unmatched.get(character):
302 end -= 1
303 unmatched[character] -= 1
304 else:
305 break
306 url = url[:end]
308 urls.append(url)
309 # Resume after the actual target, not the rest of its run:
310 # the run may also hold the next URL or Markdown link.
311 cursor = start + len(url)
312 if urls:
313 logger.info(f"Found {len(urls)} URLs in query")
314 return urls
316 # Return empty list if nothing worked
317 logger.warning(
318 "Wayback needs a URL or domain; discover URLs with a configured "
319 "search provider first"
320 )
321 return []
323 def _format_timestamp(self, timestamp: str) -> str:
324 """Format Wayback Machine timestamp into readable date"""
325 if len(timestamp) < 14:
326 return timestamp
328 try:
329 year = timestamp[0:4]
330 month = timestamp[4:6]
331 day = timestamp[6:8]
332 hour = timestamp[8:10]
333 minute = timestamp[10:12]
334 second = timestamp[12:14]
335 return f"{year}-{month}-{day} {hour}:{minute}:{second}"
336 except Exception:
337 logger.debug("Timestamp formatting failed, returning original")
338 return timestamp
340 def _get_wayback_snapshots(self, url: str) -> List[Dict[str, Any]]:
341 """
342 Get snapshots from the Wayback Machine for a specific URL.
344 Args:
345 url: URL to get snapshots for
347 Returns:
348 List of snapshot dictionaries
349 """
350 snapshots = []
352 try:
353 if self.closest_only:
354 # Get only the closest snapshot
355 response = get_availability_response(url, request_get=safe_get)
357 # Check for rate limit
358 if response.status_code == 429:
359 raise RateLimitError( # noqa: TRY301 — re-raised for base retry handling
360 "Wayback Machine rate limit exceeded",
361 retry_after=response.headers.get("Retry-After"),
362 )
364 data = response.json()
366 if (
367 "archived_snapshots" in data
368 and "closest" in data["archived_snapshots"]
369 ):
370 snapshot = data["archived_snapshots"]["closest"]
371 snapshot_url = snapshot["url"]
372 timestamp = snapshot["timestamp"]
374 snapshots.append(
375 {
376 "timestamp": timestamp,
377 "formatted_date": self._format_timestamp(timestamp),
378 "url": snapshot_url,
379 "original_url": url,
380 "available": snapshot.get("available", True),
381 "status": snapshot.get("status", "200"),
382 }
383 )
384 else:
385 # Get multiple snapshots using CDX API
386 response = safe_get(
387 self.cdx_api,
388 params={
389 "url": url,
390 "output": "json",
391 "fl": "timestamp,original,statuscode,mimetype",
392 "collapse": "timestamp:4", # Group by year
393 "limit": self.max_snapshots_per_url,
394 },
395 )
397 # Check for rate limit
398 if response.status_code == 429:
399 raise RateLimitError( # noqa: TRY301 — re-raised by except RateLimitError for base class retry
400 "Wayback Machine CDX API rate limit exceeded",
401 retry_after=response.headers.get("Retry-After"),
402 )
404 # Check if response is valid JSON
405 data = response.json()
407 # First item is the header
408 if len(data) > 1:
409 headers = data[0]
410 for item in data[1:]:
411 snapshot = dict(zip(headers, item, strict=False))
412 timestamp = snapshot.get("timestamp", "")
414 wayback_url = (
415 f"https://web.archive.org/web/{timestamp}/{url}"
416 )
418 snapshots.append(
419 {
420 "timestamp": timestamp,
421 "formatted_date": self._format_timestamp(
422 timestamp
423 ),
424 "url": wayback_url,
425 "original_url": url,
426 "available": True,
427 "status": snapshot.get("statuscode", "200"),
428 }
429 )
431 # Limit to max snapshots per URL
432 snapshots = snapshots[: self.max_snapshots_per_url]
434 except RateLimitError:
435 # Re-raise rate limit errors for base class retry handling
436 raise
437 except Exception as e:
438 if not isinstance(
439 e, (json.JSONDecodeError, requests.exceptions.JSONDecodeError)
440 ):
441 # A non-JSON reply is per URL (e.g. the archive's
442 # "blocked site" page), not a Wayback outage.
443 self._record_search_failure(e)
444 safe_msg = self._scrub_error(e)
445 logger.exception(
446 f"Error getting Wayback snapshots for {redact_url_for_log(url)} ({type(e).__name__}): {safe_msg}"
447 )
449 return snapshots
451 def _get_previews(self, query: str) -> List[Dict[str, Any]]:
452 """
453 Get preview information for Wayback Machine snapshots.
455 Args:
456 query: The search query
458 Returns:
459 List of preview dictionaries
460 """
461 logger.info("Getting Wayback Machine previews")
463 urls = []
464 for url in self._extract_urls_from_query(query):
465 # fetch_archive's URL check: a user name or password before the
466 # host never reaches the archive or the results. A credential in
467 # a query value is sent as written, as with fetch_archive, and
468 # unlike fetch_archive no per-target egress check runs here.
469 try:
470 urls.append(validate_archive_url(url))
471 except ValueError:
472 logger.warning(
473 f"Skipping Wayback target {redact_url_for_log(url)}: not "
474 "one absolute HTTP(S) URL without credentials"
475 )
476 # Each target costs an archive request, so look up no more targets
477 # than this search returns results.
478 if len(urls) > self.max_results:
479 logger.info(
480 f"Looking up the first {self.max_results} of {len(urls)} URLs"
481 )
482 urls = urls[: self.max_results]
484 if not urls:
485 logger.warning("No URLs found in the query")
486 return []
488 # Get snapshots for each URL
489 all_snapshots = []
490 for url in urls:
491 snapshots = self._get_wayback_snapshots(url)
492 all_snapshots.extend(snapshots)
494 # Apply rate limiting between requests
495 if len(urls) > 1:
496 self.rate_tracker.apply_rate_limit(self.engine_type)
498 # Format as previews
499 previews = []
500 for snapshot in all_snapshots:
501 preview = {
502 "id": f"{snapshot['timestamp']}_{snapshot['original_url']}",
503 "title": f"Archive of {snapshot['original_url']} ({snapshot['formatted_date']})",
504 "link": snapshot["url"],
505 "snippet": f"Archived version from {snapshot['formatted_date']}",
506 "original_url": snapshot["original_url"],
507 "timestamp": snapshot["timestamp"],
508 "formatted_date": snapshot["formatted_date"],
509 }
510 previews.append(preview)
512 logger.info(f"Found {len(previews)} Wayback Machine snapshots")
513 return previews
515 def _remove_boilerplate(self, html: str) -> str:
516 """Remove boilerplate using the shared extraction pipeline."""
517 if not html or not html.strip():
518 return ""
519 try:
520 return extract_content(html, language=self.language) or ""
521 except Exception as e:
522 safe_msg = self._scrub_error(e)
523 logger.exception(
524 f"Error removing boilerplate ({type(e).__name__}): {safe_msg}"
525 )
526 return html
528 def _get_wayback_content(self, url: str) -> Tuple[str, str]:
529 """
530 Retrieve content from a Wayback Machine URL.
532 Args:
533 url: Wayback Machine URL
535 Returns:
536 Tuple of (raw_html, cleaned_text)
537 """
538 try:
539 headers = {
540 "User-Agent": "Mozilla/5.0 (Local Deep Research Bot; research project)"
541 }
542 response = safe_get(url, headers=headers, timeout=10)
543 raw_html = response.text
545 # Clean the HTML
546 cleaned_text = self._remove_boilerplate(raw_html)
548 return raw_html, cleaned_text
549 except Exception as e:
550 safe_msg = self._scrub_error(e)
551 logger.exception(
552 f"Error retrieving content from {redact_url_for_log(url)} ({type(e).__name__}): {safe_msg}"
553 )
554 return "", f"Error retrieving content: {safe_msg}"
556 def _get_full_content(
557 self, relevant_items: List[Dict[str, Any]]
558 ) -> List[Dict[str, Any]]:
559 """
560 Get full content for the relevant Wayback Machine snapshots.
562 Args:
563 relevant_items: List of relevant preview dictionaries
565 Returns:
566 List of result dictionaries with full content
567 """
568 logger.info(
569 f"Getting full content for {len(relevant_items)} Wayback Machine snapshots"
570 )
572 results = []
573 for item in relevant_items:
574 wayback_url = item.get("link")
575 if not wayback_url:
576 results.append(item)
577 continue
579 logger.info(
580 f"Retrieving content from {redact_url_for_log(wayback_url)}"
581 )
583 try:
584 # Retrieve content
585 raw_html, full_content = self._get_wayback_content(wayback_url)
587 # Add full content to the result
588 result = item.copy()
589 result["raw_html"] = raw_html
590 result["full_content"] = full_content
591 result["full_content_is_complete"] = bool(
592 raw_html and full_content
593 )
595 results.append(result)
597 # Apply rate limiting
598 self.rate_tracker.apply_rate_limit(self.engine_type)
599 except Exception as e:
600 safe_msg = self._scrub_error(e)
601 logger.exception(
602 f"Error processing {redact_url_for_log(wayback_url)} ({type(e).__name__}): {safe_msg}"
603 )
604 results.append(item)
606 return results
608 def search_by_url(
609 self, url: str, max_snapshots: int | None = None
610 ) -> List[Dict[str, Any]]:
611 """
612 Search for archived versions of a specific URL.
614 Args:
615 url: The URL to search for archives
616 max_snapshots: Maximum number of snapshots to return
618 Returns:
619 List of snapshot dictionaries
620 """
621 max_snapshots = max_snapshots or self.max_snapshots_per_url
623 snapshots = self._get_wayback_snapshots(url)
624 previews = []
626 for snapshot in snapshots[:max_snapshots]:
627 preview = {
628 "id": f"{snapshot['timestamp']}_{snapshot['original_url']}",
629 "title": f"Archive of {snapshot['original_url']} ({snapshot['formatted_date']})",
630 "link": snapshot["url"],
631 "snippet": f"Archived version from {snapshot['formatted_date']}",
632 "original_url": snapshot["original_url"],
633 "timestamp": snapshot["timestamp"],
634 "formatted_date": snapshot["formatted_date"],
635 }
636 previews.append(preview)
638 return self._get_full_content_cached(previews)
640 def search_by_date_range(
641 self, url: str, start_date: str, end_date: str
642 ) -> List[Dict[str, Any]]:
643 """
644 Search for archived versions of a URL within a date range.
646 Args:
647 url: The URL to search for archives
648 start_date: Start date in format YYYYMMDD
649 end_date: End date in format YYYYMMDD
651 Returns:
652 List of snapshot dictionaries
653 """
654 try:
655 # Use CDX API with date range
656 response = safe_get(
657 self.cdx_api,
658 params={
659 "url": url,
660 "output": "json",
661 "fl": "timestamp,original,statuscode,mimetype",
662 "from": start_date,
663 "to": end_date,
664 "limit": self.max_snapshots_per_url,
665 },
666 )
668 # Process response
669 data = response.json()
671 # First item is the header
672 if len(data) <= 1:
673 return []
675 headers = data[0]
676 snapshots = []
678 for item in data[1:]:
679 snapshot = dict(zip(headers, item, strict=False))
680 timestamp = snapshot.get("timestamp", "")
682 wayback_url = f"https://web.archive.org/web/{timestamp}/{url}"
684 snapshots.append(
685 {
686 "id": f"{timestamp}_{url}",
687 "title": f"Archive of {url} ({self._format_timestamp(timestamp)})",
688 "link": wayback_url,
689 "snippet": f"Archived version from {self._format_timestamp(timestamp)}",
690 "original_url": url,
691 "timestamp": timestamp,
692 "formatted_date": self._format_timestamp(timestamp),
693 }
694 )
696 return self._get_full_content_cached(snapshots)
698 except Exception as e:
699 safe_msg = self._scrub_error(e)
700 logger.exception(
701 f"Error searching date range for {redact_url_for_log(url)} ({type(e).__name__}): {safe_msg}"
702 )
703 return []
705 def get_latest_snapshot(self, url: str) -> Optional[Dict[str, Any]]:
706 """
707 Get the most recent snapshot of a URL.
709 Args:
710 url: The URL to get the latest snapshot for
712 Returns:
713 Dictionary with snapshot information or None if not found
714 """
715 try:
716 response = safe_get(self.available_api, params={"url": url})
717 data = response.json()
719 if (
720 "archived_snapshots" in data
721 and "closest" in data["archived_snapshots"]
722 ):
723 snapshot = data["archived_snapshots"]["closest"]
724 timestamp = snapshot["timestamp"]
725 wayback_url = snapshot["url"]
727 result = {
728 "id": f"{timestamp}_{url}",
729 "title": f"Latest archive of {url} ({self._format_timestamp(timestamp)})",
730 "link": wayback_url,
731 "snippet": f"Archived version from {self._format_timestamp(timestamp)}",
732 "original_url": url,
733 "timestamp": timestamp,
734 "formatted_date": self._format_timestamp(timestamp),
735 }
737 raw_html, full_content = self._get_wayback_content(wayback_url)
738 result["raw_html"] = raw_html
739 result["full_content"] = full_content
740 result["full_content_is_complete"] = bool(
741 raw_html and full_content
742 )
744 return result
746 return None
748 except Exception as e:
749 safe_msg = self._scrub_error(e)
750 logger.exception(
751 f"Error getting latest snapshot for {redact_url_for_log(url)} ({type(e).__name__}): {safe_msg}"
752 )
753 return None