Coverage for src/local_deep_research/web_search_engines/wayback_client.py: 98%
107 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +0000
1"""URL-based Wayback availability lookup shared by archive callers."""
3import posixpath
4import re
5from datetime import datetime
6from urllib.parse import unquote, urlsplit
8from requests import Request
10from ..security.safe_requests import safe_get
11from ..utilities.resource_utils import safe_close
12from .rate_limiting import RateLimitError
14AVAILABLE_API = "https://archive.org/wayback/available"
17def _archive_target_key(url: str):
18 """Ignore scheme/www/default ports/fragments, retaining prepared path/query."""
19 prepared = Request("GET", validate_archive_url(url)).prepare().url or ""
20 parsed = urlsplit(prepared)
21 default_port = 443 if parsed.scheme == "https" else 80
22 port = parsed.port if parsed.port != default_port else None
23 return (
24 (parsed.hostname or "").removeprefix("www."),
25 port,
26 parsed.path or "/",
27 parsed.query,
28 )
31def validate_capture_fetch_url(
32 url: str, timestamp: str, expected_url: str | None = None
33) -> None:
34 """Keep page retrieval and every redirect inside the selected capture."""
35 parsed = urlsplit(url)
36 if (
37 parsed.scheme != "https"
38 or parsed.hostname != "web.archive.org"
39 or parsed.port not in {None, 443}
40 or parsed.username is not None
41 or parsed.password is not None
42 or not re.match(
43 rf"^/web/{re.escape(timestamp)}(?:id_|if_)?/https?://", parsed.path
44 )
45 ):
46 raise ValueError("Archive replay left the selected capture")
47 if expected_url is not None: 47 ↛ exitline 47 didn't return from function 'validate_capture_fetch_url' because the condition on line 47 was always true
48 target = url.split(f"/web/{timestamp}", 1)[1].split("/", 1)[1]
49 if _archive_target_key(target) != _archive_target_key(expected_url):
50 raise ValueError("Archive replay changed the captured URL")
53def capture_metadata_from_replay_url(url: str) -> dict | None:
54 """Identify an exact supplied capture without an availability request.
56 A direct replay read needs the same boundary as a discovered capture,
57 even before any citation exists. Unsupported capture forms fail before
58 network access instead of becoming an ordinary live-page fetch.
59 """
60 parsed = urlsplit(url)
61 if parsed.scheme not in {"http", "https"} or not parsed.hostname:
62 return None
63 # Detection includes the spelling Requests will send. These aliases
64 # must not fall through to an unrestricted live-page fetch, but only
65 # the original, canonical capture spelling is accepted below.
66 prepared = urlsplit(Request("GET", url).prepare().url or "")
67 prepared_host = (prepared.hostname or "").rstrip(".")
68 if prepared_host not in {"web.archive.org", "archive.org"}:
69 return None
71 def has_replay_path(path):
72 # Inspect the wrapper, not the embedded target's path. Decode one
73 # layer for detection only; never fetch a normalized substitute.
74 wrapper = unquote(path.split("://", 1)[0]).replace("\\", "/")
75 return any(
76 candidate.lstrip("/").partition("/")[0].lower() == "web"
77 for candidate in (wrapper, posixpath.normpath(wrapper))
78 )
80 if not any(has_replay_path(path) for path in (parsed.path, prepared.path)):
81 return None
82 if prepared_host != "web.archive.org":
83 raise ValueError("Unsupported archive replay capture URL")
84 replay = re.match(r"^/web/([0-9]{14})(?:id_|if_)?/https?://", parsed.path)
85 if replay is None:
86 raise ValueError("Unsupported archive replay capture URL")
87 timestamp = replay.group(1)
88 captured_date = datetime.strptime(timestamp, "%Y%m%d%H%M%S")
89 target = url.split(f"/web/{timestamp}", 1)[1].split("/", 1)[1]
90 validate_archive_url(target)
91 validate_capture_fetch_url(url, timestamp, target)
92 return {
93 "source_type": "archive",
94 "original_url": target,
95 "requested_url": target,
96 "timestamp": timestamp,
97 "formatted_date": captured_date.strftime("%Y-%m-%d %H:%M:%S UTC"),
98 }
101def validate_archive_url(url: str) -> str:
102 """Require one absolute HTTP(S) URL without embedded credentials."""
103 url = url.strip()
104 if not url or any(
105 character.isspace() or ord(character) < 32 for character in url
106 ):
107 raise ValueError(
108 "Supply a single absolute HTTP(S) URL to fetch_archive"
109 )
110 parsed = urlsplit(url)
111 if (
112 parsed.scheme.lower() not in {"http", "https"}
113 or not parsed.hostname
114 or parsed.username is not None
115 or parsed.password is not None
116 ):
117 raise ValueError("Supply an absolute HTTP(S) URL without credentials")
118 # Accessing port validates malformed/non-numeric/out-of-range ports.
119 parsed.port
120 return url
123def normalize_archive_date(value: str | None) -> str | None:
124 """Convert a calendar date to the availability API's YYYYMMDD input."""
125 if value is None:
126 return None
127 if re.fullmatch(r"\d{4}-\d{2}-\d{2}", value):
128 return datetime.strptime(value, "%Y-%m-%d").strftime("%Y%m%d")
129 if re.fullmatch(r"\d{8}", value):
130 return datetime.strptime(value, "%Y%m%d").strftime("%Y%m%d")
131 raise ValueError("Archive date must be a valid YYYY-MM-DD or YYYYMMDD date")
134def get_availability_response(url: str, timestamp=None, *, request_get=None):
135 """Request the fixed archive API without following redirects."""
136 params = {"url": url}
137 if timestamp is not None:
138 params["timestamp"] = timestamp
139 getter = request_get if request_get is not None else safe_get
140 return getter(
141 AVAILABLE_API, params=params, timeout=30, allow_redirects=False
142 )
145def lookup_archive_snapshot(url: str, timestamp=None, *, request_get=None):
146 """Return an accessible snapshot, preserving its actual capture time.
148 With no timestamp the API selects the most recent available capture;
149 otherwise it selects a capture near the requested date. Never substitute
150 the live page when the archive has no capture.
151 """
152 url = validate_archive_url(url)
153 response = get_availability_response(
154 url, timestamp, request_get=request_get
155 )
156 try:
157 if response.status_code == 429:
158 raise RateLimitError(
159 "Wayback archive lookup rate limit exceeded",
160 retry_after=response.headers.get("Retry-After"),
161 )
162 response.raise_for_status()
163 if response.status_code != 200:
164 raise ValueError(
165 "Archive availability API returned an unexpected status"
166 )
167 data = response.json()
168 if not isinstance(data, dict) or not isinstance(
169 data.get("archived_snapshots"), dict
170 ):
171 raise ValueError("Invalid archive availability response")
172 closest = data["archived_snapshots"].get("closest")
173 if closest is None:
174 return None
175 if not isinstance(closest, dict):
176 raise ValueError("Invalid archive snapshot")
177 if (
178 closest.get("available") is not True
179 or str(closest.get("status")) != "200"
180 ):
181 return None
182 captured = closest.get("timestamp")
183 if not isinstance(captured, str) or not re.fullmatch(
184 r"\d{14}", captured
185 ):
186 raise ValueError("Invalid archive capture timestamp")
187 captured_date = datetime.strptime(captured, "%Y%m%d%H%M%S")
188 replay = closest.get("url")
189 if not isinstance(replay, str): 189 ↛ 190line 189 didn't jump to line 190 because the condition on line 189 was never true
190 raise ValueError("Invalid archive replay URL")
191 parsed = urlsplit(replay)
192 prefix = f"/web/{captured}/"
193 if (
194 parsed.scheme.lower() not in {"http", "https"}
195 or parsed.hostname != "web.archive.org"
196 or parsed.port not in {None, 443, 80}
197 or parsed.username is not None
198 or parsed.password is not None
199 or not parsed.path.startswith(prefix)
200 ):
201 raise ValueError("Invalid archive replay URL")
202 # Preserve the embedded target's query, punctuation and fragment.
203 embedded = replay.split(prefix, 1)[1]
204 validate_archive_url(embedded)
205 if _archive_target_key(embedded) != _archive_target_key(url):
206 raise ValueError("Archive capture does not match the requested URL")
207 replay = "https://web.archive.org" + replay[replay.index(prefix) :]
208 return {
209 "original_url": embedded,
210 "requested_url": url,
211 "url": replay,
212 "timestamp": captured,
213 "formatted_date": captured_date.strftime("%Y-%m-%d %H:%M:%S UTC"),
214 "available": True,
215 "status": "200",
216 }
217 finally:
218 safe_close(response, "Wayback availability response")