| 1 | """HTTP utilities for last30days skill (stdlib only).""" |
| 2 | |
| 3 | import json |
| 4 | import os |
| 5 | import re |
| 6 | import socket |
| 7 | import sys |
| 8 | import threading |
| 9 | import time |
| 10 | import urllib.error |
| 11 | import urllib.request |
| 12 | from concurrent.futures import Future |
| 13 | from contextlib import contextmanager |
| 14 | from contextvars import ContextVar, copy_context |
| 15 | from pathlib import Path |
| 16 | from typing import Any, Dict, Optional, Union |
| 17 | from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit, quote |
| 18 | |
| 19 | from . import health |
| 20 | from . import log as _log |
| 21 | |
| 22 | DEFAULT_TIMEOUT = 30 |
| 23 | |
| 24 | |
| 25 | def log(msg: str): |
| 26 | """Log debug message to stderr.""" |
| 27 | _log.debug(msg) |
| 28 | |
| 29 | |
| 30 | MAX_RETRIES = 5 |
| 31 | MAX_429_RETRIES = 2 |
| 32 | RETRY_DELAY = 2.0 |
| 33 | # DNS resolution failures (gaierror) are transient — typically resolved by a |
| 34 | # brief backoff and retry. Use a dedicated minimum attempt count + exponential |
| 35 | # delays (1s, 2s, 4s) so callers that pass a small `retries` value still get a |
| 36 | # meaningful chance to recover from a transient resolution failure. |
| 37 | MIN_DNS_RETRIES = 3 |
| 38 | USER_AGENT = "last30days-skill/3.0 (Assistant Skill)" |
| 39 | |
| 40 | _failure_sink: ContextVar[Optional[list["HTTPError"]]] = ContextVar( |
| 41 | "last30days_http_failure_sink", |
| 42 | default=None, |
| 43 | ) |
| 44 | _expected_miss_statuses: ContextVar[frozenset[int]] = ContextVar( |
| 45 | "last30days_http_expected_miss_statuses", |
| 46 | default=frozenset(), |
| 47 | ) |
| 48 | |
| 49 | _FIXTURE_FORMAT = "last30days-http-fixture/v1" |
| 50 | _FIXTURE_SECRET_KEYS = frozenset( |
| 51 | {"api_key", "apikey", "authorization", "cookie", "key", "secret", "token"} |
| 52 | ) |
| 53 | _fixture_lock = threading.Lock() |
| 54 | _fixture_state: Optional[dict[str, Any]] = None |
| 55 | _NO_FIXTURE = object() |
| 56 | _fixture_module_capture: ContextVar[bool] = ContextVar( |
| 57 | "last30days_fixture_module_capture", |
| 58 | default=False, |
| 59 | ) |
| 60 | |
| 61 | |
| 62 | def _is_secret_key(value: object) -> bool: |
| 63 | key = re.sub(r"[^a-z0-9]+", "_", str(value).lower()).strip("_") |
| 64 | return ( |
| 65 | key in _FIXTURE_SECRET_KEYS |
| 66 | or key.endswith(("_api_key", "_authorization", "_cookie", "_secret", "_token")) |
| 67 | ) |
| 68 | |
| 69 | |
| 70 | def _scrub_fixture_value( |
| 71 | value: Any, |
| 72 | *, |
| 73 | key: str = "", |
| 74 | redactions: frozenset[str] = frozenset(), |
| 75 | ) -> Any: |
| 76 | """Remove credentials before a recorded exchange reaches disk.""" |
| 77 | if key and _is_secret_key(key): |
| 78 | return "<redacted>" |
| 79 | if isinstance(value, dict): |
| 80 | return { |
| 81 | str(child_key): _scrub_fixture_value( |
| 82 | child_value, |
| 83 | key=str(child_key), |
| 84 | redactions=redactions, |
| 85 | ) |
| 86 | for child_key, child_value in value.items() |
| 87 | } |
| 88 | if isinstance(value, list): |
| 89 | return [_scrub_fixture_value(item, redactions=redactions) for item in value] |
| 90 | if isinstance(value, str): |
| 91 | scrubbed = value |
| 92 | for secret in sorted(redactions, key=len, reverse=True): |
| 93 | if len(secret) >= 4: |
| 94 | scrubbed = scrubbed.replace(secret, "<redacted>") |
| 95 | return scrubbed |
| 96 | return value |
| 97 | |
| 98 | |
| 99 | def _collect_secret_values(value: Any, *, key: str = "") -> set[str]: |
| 100 | values: set[str] = set() |
| 101 | if key and _is_secret_key(key) and value not in (None, ""): |
| 102 | values.add(str(value)) |
| 103 | return values |
| 104 | if isinstance(value, dict): |
| 105 | for child_key, child_value in value.items(): |
| 106 | values.update(_collect_secret_values(child_value, key=str(child_key))) |
| 107 | elif isinstance(value, list): |
| 108 | for child in value: |
| 109 | values.update(_collect_secret_values(child)) |
| 110 | return values |
| 111 | |
| 112 | |
| 113 | def _fixture_redactions( |
| 114 | url: str, |
| 115 | headers: dict[str, str], |
| 116 | json_data: Optional[Dict[str, Any]], |
| 117 | ) -> frozenset[str]: |
| 118 | values: set[str] = set() |
| 119 | try: |
| 120 | for key, value in parse_qsl(urlsplit(url).query, keep_blank_values=True): |
| 121 | if _is_secret_key(key) and value: |
| 122 | values.add(value) |
| 123 | except ValueError: |
| 124 | pass |
| 125 | values.update(_collect_secret_values(headers)) |
| 126 | values.update(_collect_secret_values(json_data)) |
| 127 | return frozenset(values) |
| 128 | |
| 129 | |
| 130 | def _scrub_fixture_url(url: str) -> str: |
| 131 | try: |
| 132 | parts = urlsplit(url) |
| 133 | query = urlencode( |
| 134 | [ |
| 135 | (key, "<redacted>" if _is_secret_key(key) else value) |
| 136 | for key, value in parse_qsl(parts.query, keep_blank_values=True) |
| 137 | ] |
| 138 | ) |
| 139 | return urlunsplit((parts.scheme, parts.netloc, parts.path, query, parts.fragment)) |
| 140 | except ValueError: |
| 141 | return url |
| 142 | |
| 143 | |
| 144 | def _fixture_request( |
| 145 | method: str, |
| 146 | url: str, |
| 147 | json_data: Optional[Dict[str, Any]], |
| 148 | raw: bool, |
| 149 | ) -> dict[str, Any]: |
| 150 | request_data: dict[str, Any] = { |
| 151 | "method": method.upper(), |
| 152 | "url": _scrub_fixture_url(url), |
| 153 | "raw": bool(raw), |
| 154 | } |
| 155 | if json_data is not None: |
| 156 | request_data["json"] = _scrub_fixture_value(json_data) |
| 157 | return request_data |
| 158 | |
| 159 | |
| 160 | def _fixture_key(request_data: dict[str, Any]) -> str: |
| 161 | return json.dumps(request_data, sort_keys=True, separators=(",", ":"), ensure_ascii=False) |
| 162 | |
| 163 | |
| 164 | @contextmanager |
| 165 | def recording_requests(path: str | Path): |
| 166 | """Record scrubbed HTTP exchanges to ``path`` for offline eval replay. |
| 167 | |
| 168 | This process-global session is deliberate: source requests run in worker |
| 169 | threads, so a ContextVar would not observe the complete pipeline fan-out. |
| 170 | Nested or concurrent recording/replay sessions are rejected. |
| 171 | """ |
| 172 | global _fixture_state |
| 173 | target = Path(path).expanduser() |
| 174 | if target.suffix.lower() != ".json": |
| 175 | target = target / "http.json" |
| 176 | with _fixture_lock: |
| 177 | if _fixture_state is not None: |
| 178 | raise RuntimeError("An HTTP fixture session is already active") |
| 179 | _fixture_state = { |
| 180 | "mode": "record", |
| 181 | "path": target, |
| 182 | "exchanges": [], |
| 183 | "source_exchanges": [], |
| 184 | # Secret VALUES from the environment, so module-seam recordings |
| 185 | # scrub tokens echoed inside normal string fields (adapter error |
| 186 | # messages, parsed item text), not just secret-named keys. |
| 187 | "redactions": frozenset( |
| 188 | value |
| 189 | for key, value in os.environ.items() |
| 190 | if _is_secret_key(key) and isinstance(value, str) and len(value) >= 4 |
| 191 | ), |
| 192 | } |
| 193 | completed = False |
| 194 | try: |
| 195 | yield target |
| 196 | completed = True |
| 197 | finally: |
| 198 | with _fixture_lock: |
| 199 | state = _fixture_state |
| 200 | _fixture_state = None |
| 201 | if state is not None and completed: |
| 202 | target.parent.mkdir(parents=True, exist_ok=True) |
| 203 | payload = { |
| 204 | "format": _FIXTURE_FORMAT, |
| 205 | "exchanges": state["exchanges"], |
| 206 | "source_exchanges": state["source_exchanges"], |
| 207 | } |
| 208 | temporary = target.with_name(f".{target.name}.tmp") |
| 209 | temporary.write_text( |
| 210 | json.dumps(payload, indent=2, ensure_ascii=False) + "\n", |
| 211 | encoding="utf-8", |
| 212 | ) |
| 213 | if os.name != "nt": |
| 214 | temporary.chmod(0o644) |
| 215 | temporary.replace(target) |
| 216 | |
| 217 | |
| 218 | @contextmanager |
| 219 | def fixture_module_capture(enabled: bool): |
| 220 | """Suppress nested HTTP recording when a whole adapter result is captured.""" |
| 221 | token = _fixture_module_capture.set(enabled) |
| 222 | try: |
| 223 | yield |
| 224 | finally: |
| 225 | _fixture_module_capture.reset(token) |
| 226 | |
| 227 | |
| 228 | @contextmanager |
| 229 | def replaying_requests(path: str | Path): |
| 230 | """Replay recorded exchanges and fail closed on any unrecorded request.""" |
| 231 | global _fixture_state |
| 232 | target = Path(path).expanduser() |
| 233 | if target.is_dir(): |
| 234 | target = target / "http.json" |
| 235 | payload = json.loads(target.read_text(encoding="utf-8")) |
| 236 | if payload.get("format") != _FIXTURE_FORMAT: |
| 237 | raise ValueError(f"Unsupported HTTP fixture format in {target}") |
| 238 | queues: dict[str, list[dict[str, Any]]] = {} |
| 239 | for exchange in payload.get("exchanges") or []: |
| 240 | queues.setdefault(_fixture_key(exchange["request"]), []).append(exchange["response"]) |
| 241 | source_queues: dict[str, list[Any]] = {} |
| 242 | for exchange in payload.get("source_exchanges") or []: |
| 243 | source_queues.setdefault(_fixture_key(exchange["request"]), []).append(exchange) |
| 244 | with _fixture_lock: |
| 245 | if _fixture_state is not None: |
| 246 | raise RuntimeError("An HTTP fixture session is already active") |
| 247 | _fixture_state = { |
| 248 | "mode": "replay", |
| 249 | "path": target, |
| 250 | "queues": queues, |
| 251 | "source_queues": source_queues, |
| 252 | } |
| 253 | try: |
| 254 | yield target |
| 255 | with _fixture_lock: |
| 256 | unused = sum(len(values) for values in queues.values()) + sum( |
| 257 | len(values) for values in source_queues.values() |
| 258 | ) |
| 259 | if unused: |
| 260 | raise AssertionError(f"HTTP fixture replay left {unused} unused exchange(s): {target}") |
| 261 | finally: |
| 262 | with _fixture_lock: |
| 263 | _fixture_state = None |
| 264 | |
| 265 | |
| 266 | def _fixture_replay(request_data: dict[str, Any]) -> Any: |
| 267 | with _fixture_lock: |
| 268 | state = _fixture_state |
| 269 | if state is None or state["mode"] != "replay": |
| 270 | return _NO_FIXTURE |
| 271 | queue = state["queues"].get(_fixture_key(request_data)) |
| 272 | if not queue: |
| 273 | raise AssertionError( |
| 274 | "Unrecorded HTTP request during fixture replay: " |
| 275 | f"{request_data['method']} {request_data['url']}" |
| 276 | ) |
| 277 | response = queue.pop(0) |
| 278 | if response.get("error"): |
| 279 | error = response["error"] |
| 280 | recorded_error = HTTPError( |
| 281 | str(error.get("message") or "Recorded HTTP error"), |
| 282 | status_code=error.get("status_code"), |
| 283 | body=error.get("body"), |
| 284 | outcome_state=error.get("outcome_state"), |
| 285 | ) |
| 286 | _raise(recorded_error) |
| 287 | return response.get("value") |
| 288 | |
| 289 | |
| 290 | def _fixture_record( |
| 291 | request_data: dict[str, Any], |
| 292 | *, |
| 293 | value: Any = None, |
| 294 | error: Optional["HTTPError"] = None, |
| 295 | redactions: frozenset[str] = frozenset(), |
| 296 | ) -> None: |
| 297 | if _fixture_module_capture.get(): |
| 298 | return |
| 299 | with _fixture_lock: |
| 300 | state = _fixture_state |
| 301 | if state is None or state["mode"] != "record": |
| 302 | return |
| 303 | response: dict[str, Any] |
| 304 | if error is None: |
| 305 | response = {"value": _scrub_fixture_value(value, redactions=redactions)} |
| 306 | else: |
| 307 | response = { |
| 308 | "error": _scrub_fixture_value( |
| 309 | { |
| 310 | "message": str(error), |
| 311 | "status_code": error.status_code, |
| 312 | "body": error.body, |
| 313 | "outcome_state": error.outcome_state, |
| 314 | }, |
| 315 | redactions=redactions, |
| 316 | ) |
| 317 | } |
| 318 | state["exchanges"].append({"request": request_data, "response": response}) |
| 319 | |
| 320 | |
| 321 | def fixture_source_replay(request_data: dict[str, Any]) -> tuple[bool, Any]: |
| 322 | """Return a recorded CLI-backed source result when replay is active.""" |
| 323 | scrubbed = _scrub_fixture_value(request_data) |
| 324 | with _fixture_lock: |
| 325 | state = _fixture_state |
| 326 | if state is None or state["mode"] != "replay": |
| 327 | return False, None |
| 328 | queue = state["source_queues"].get(_fixture_key(scrubbed)) |
| 329 | if not queue: |
| 330 | raise AssertionError( |
| 331 | "Unrecorded CLI-backed source request during fixture replay: " |
| 332 | f"{request_data.get('source', 'unknown')}" |
| 333 | ) |
| 334 | exchange = queue.pop(0) |
| 335 | if exchange.get("type") == "error": |
| 336 | error = exchange.get("error") or {} |
| 337 | raise RecordedSourceError( |
| 338 | str(error.get("message") or "Recorded source error"), |
| 339 | exception_type=str(error.get("exception_type") or "Exception"), |
| 340 | outcome_state=error.get("outcome_state"), |
| 341 | ) |
| 342 | return True, exchange.get("value") |
| 343 | |
| 344 | |
| 345 | def fixture_source_record(request_data: dict[str, Any], value: Any) -> None: |
| 346 | """Record the parsed output of a source adapter that bypasses http.py.""" |
| 347 | with _fixture_lock: |
| 348 | state = _fixture_state |
| 349 | if state is None or state["mode"] != "record": |
| 350 | return |
| 351 | session_redactions = state.get("redactions") or frozenset() |
| 352 | state["source_exchanges"].append( |
| 353 | { |
| 354 | "request": _scrub_fixture_value(request_data, redactions=session_redactions), |
| 355 | "value": _scrub_fixture_value(value, redactions=session_redactions), |
| 356 | } |
| 357 | ) |
| 358 | |
| 359 | |
| 360 | def fixture_source_record_error(request_data: dict[str, Any], error: Exception) -> None: |
| 361 | """Record a replayable failure from a source adapter that bypasses http.py.""" |
| 362 | with _fixture_lock: |
| 363 | state = _fixture_state |
| 364 | if state is None or state["mode"] != "record": |
| 365 | return |
| 366 | session_redactions = state.get("redactions") or frozenset() |
| 367 | state["source_exchanges"].append( |
| 368 | { |
| 369 | "request": _scrub_fixture_value(request_data, redactions=session_redactions), |
| 370 | "type": "error", |
| 371 | "error": _scrub_fixture_value( |
| 372 | { |
| 373 | "exception_type": type(error).__name__, |
| 374 | "message": str(error), |
| 375 | "outcome_state": getattr(error, "outcome_state", None), |
| 376 | } |
| 377 | , redactions=session_redactions), |
| 378 | } |
| 379 | ) |
| 380 | |
| 381 | |
| 382 | class RecordedSourceError(RuntimeError): |
| 383 | """Failure restored from a recorded module-backed source exchange.""" |
| 384 | |
| 385 | def __init__( |
| 386 | self, |
| 387 | message: str, |
| 388 | *, |
| 389 | exception_type: str, |
| 390 | outcome_state: Optional[str] = None, |
| 391 | ): |
| 392 | super().__init__(message) |
| 393 | self.exception_type = exception_type |
| 394 | self.outcome_state = outcome_state |
| 395 | |
| 396 | |
| 397 | def _is_dns_failure(err: urllib.error.URLError) -> bool: |
| 398 | """Return True if a URLError was caused by DNS resolution (gaierror).""" |
| 399 | return isinstance(getattr(err, "reason", None), socket.gaierror) |
| 400 | |
| 401 | |
| 402 | class HTTPError(Exception): |
| 403 | """HTTP request error with status code.""" |
| 404 | def __init__( |
| 405 | self, |
| 406 | message: str, |
| 407 | status_code: Optional[int] = None, |
| 408 | body: Optional[str] = None, |
| 409 | outcome_state: Optional[str] = None, |
| 410 | ): |
| 411 | super().__init__(message) |
| 412 | self.status_code = status_code |
| 413 | self.body = body |
| 414 | self.outcome_state = outcome_state or classify_failure( |
| 415 | status_code=status_code, |
| 416 | message=message, |
| 417 | ) |
| 418 | |
| 419 | |
| 420 | @contextmanager |
| 421 | def capture_failures(): |
| 422 | """Capture terminal request failures in the current retrieval context. |
| 423 | |
| 424 | Source modules historically catch ``HTTPError`` and return an empty result. |
| 425 | The context-local sink lets the pipeline retain that failure without shared |
| 426 | mutable state across its worker threads. |
| 427 | """ |
| 428 | failures: list[HTTPError] = [] |
| 429 | token = _failure_sink.set(failures) |
| 430 | try: |
| 431 | yield failures |
| 432 | finally: |
| 433 | _failure_sink.reset(token) |
| 434 | |
| 435 | |
| 436 | @contextmanager |
| 437 | def tee_failures(): |
| 438 | """Observe failures locally WITHOUT hiding them from the enclosing sink. |
| 439 | |
| 440 | ``capture_failures()`` *replaces* the context-local sink, so nesting it |
| 441 | inside a retrieval context swallows the very failure the pipeline needs. |
| 442 | This yields a local list and forwards its contents to the parent sink on |
| 443 | exit, so a swallow site (``get_text`` returns None and drops the status) |
| 444 | can recover what it lost while the pipeline still sees the failure. |
| 445 | """ |
| 446 | parent = _failure_sink.get() |
| 447 | local: list[HTTPError] = [] |
| 448 | token = _failure_sink.set(local) |
| 449 | try: |
| 450 | yield local |
| 451 | finally: |
| 452 | _failure_sink.reset(token) |
| 453 | if parent is not None: |
| 454 | parent.extend(local) |
| 455 | |
| 456 | |
| 457 | @contextmanager |
| 458 | def expected_misses(*status_codes: int): |
| 459 | """Exclude adapter-declared probe misses from captured run failures.""" |
| 460 | token = _expected_miss_statuses.set( |
| 461 | _expected_miss_statuses.get().union(status_codes) |
| 462 | ) |
| 463 | try: |
| 464 | yield |
| 465 | finally: |
| 466 | _expected_miss_statuses.reset(token) |
| 467 | |
| 468 | |
| 469 | def submit_with_context(executor, func, /, *args, **kwargs) -> Future: |
| 470 | """Submit a worker with the caller's failure-capture context.""" |
| 471 | context = copy_context() |
| 472 | return executor.submit(context.run, func, *args, **kwargs) |
| 473 | |
| 474 | |
| 475 | def _record_failure(error: HTTPError) -> None: |
| 476 | if error.status_code in _expected_miss_statuses.get(): |
| 477 | return |
| 478 | sink = _failure_sink.get() |
| 479 | if sink is not None: |
| 480 | sink.append(error) |
| 481 | |
| 482 | |
| 483 | def _raise(error: HTTPError) -> None: |
| 484 | _record_failure(error) |
| 485 | raise error |
| 486 | |
| 487 | |
| 488 | def classify_failure(*, status_code: Optional[int] = None, message: str = "") -> str: |
| 489 | """Map a request failure to the doctor-aligned per-run vocabulary.""" |
| 490 | text = message.lower() |
| 491 | if status_code == 429 or any( |
| 492 | marker in text for marker in ("http 429", "status 429", "rate limit", "too many requests") |
| 493 | ): |
| 494 | return health.RATE_LIMITED |
| 495 | if status_code in (401, 402, 403) or any( |
| 496 | marker in text |
| 497 | for marker in ( |
| 498 | "http 401", |
| 499 | "http 402", |
| 500 | "http 403", |
| 501 | "status 401", |
| 502 | "status 402", |
| 503 | "status 403", |
| 504 | "unauthorized", |
| 505 | "forbidden", |
| 506 | "authentication failed", |
| 507 | "expired token", |
| 508 | ) |
| 509 | ): |
| 510 | return health.AUTH_FAILED |
| 511 | if status_code == 408 or "timed out" in text or "timeout" in text: |
| 512 | return health.TIMEOUT |
| 513 | if any( |
| 514 | marker in text |
| 515 | for marker in ( |
| 516 | "invalid json", |
| 517 | "json decode", |
| 518 | "schema", |
| 519 | "interstitial", |
| 520 | "non-json", |
| 521 | ) |
| 522 | ): |
| 523 | return health.SCHEMA_DRIFT |
| 524 | if any( |
| 525 | marker in text |
| 526 | for marker in ( |
| 527 | "url error", |
| 528 | "connection error", |
| 529 | "connection refused", |
| 530 | "connection reset", |
| 531 | "name or service not known", |
| 532 | "temporary failure in name resolution", |
| 533 | "nodename nor servname", |
| 534 | "dns", |
| 535 | "network is unreachable", |
| 536 | ) |
| 537 | ): |
| 538 | return health.UNREACHABLE |
| 539 | return health.ERROR |
| 540 | |
| 541 | |
| 542 | def request( |
| 543 | method: str, |
| 544 | url: str, |
| 545 | headers: Optional[Dict[str, str]] = None, |
| 546 | json_data: Optional[Dict[str, Any]] = None, |
| 547 | params: Optional[Dict[str, Any]] = None, |
| 548 | timeout: int = DEFAULT_TIMEOUT, |
| 549 | retries: int = MAX_RETRIES, |
| 550 | max_429_retries: int = MAX_429_RETRIES, |
| 551 | raw: bool = False, |
| 552 | ) -> Union[Dict[str, Any], str]: |
| 553 | """Make an HTTP request and return JSON response. |
| 554 | |
| 555 | Args: |
| 556 | method: HTTP method (GET, POST, etc.) |
| 557 | url: Request URL |
| 558 | headers: Optional headers dict |
| 559 | json_data: Optional JSON body (for POST) |
| 560 | params: Optional query-string params. Values are stringified. None values |
| 561 | are dropped. If ``url`` already has a query string, ``params`` is appended. |
| 562 | timeout: Request timeout in seconds |
| 563 | retries: Number of retries on failure |
| 564 | max_429_retries: Maximum 429 retries before giving up (separate cap) |
| 565 | raw: If True, return raw response text instead of parsed JSON |
| 566 | |
| 567 | Returns: |
| 568 | Parsed JSON response as dict, or raw text string if raw=True. |
| 569 | |
| 570 | Raises: |
| 571 | HTTPError: On request failure |
| 572 | """ |
| 573 | headers = headers or {} |
| 574 | headers.setdefault("User-Agent", USER_AGENT) |
| 575 | |
| 576 | if params: |
| 577 | filtered = {k: str(v) for k, v in params.items() if v is not None} |
| 578 | if filtered: |
| 579 | separator = "&" if ("?" in url) else "?" |
| 580 | url = f"{url}{separator}{urlencode(filtered)}" |
| 581 | # Encode any non-ASCII characters to prevent UnicodeEncodeError from |
| 582 | # http.client.HTTPConnection.putrequest (which uses latin-1 internally). |
| 583 | # Only encode path, query, and fragment — not the hostname (netloc), which |
| 584 | # needs IDNA encoding instead of percent-encoding for non-ASCII domains. |
| 585 | parts = urlsplit(url) |
| 586 | safe = '/:@!$&\'()*+,;=-._~%?#[]=+' |
| 587 | url = urlunsplit(( |
| 588 | parts.scheme, |
| 589 | parts.netloc, |
| 590 | quote(parts.path, safe=safe), |
| 591 | quote(parts.query, safe=safe), |
| 592 | quote(parts.fragment, safe=safe), |
| 593 | )) |
| 594 | |
| 595 | fixture_request = _fixture_request(method, url, json_data, raw) |
| 596 | fixture_redactions = _fixture_redactions(url, headers, json_data) |
| 597 | replayed = _fixture_replay(fixture_request) |
| 598 | if replayed is not _NO_FIXTURE: |
| 599 | return replayed |
| 600 | |
| 601 | data = None |
| 602 | if json_data is not None: |
| 603 | data = json.dumps(json_data).encode('utf-8') |
| 604 | headers.setdefault("Content-Type", "application/json") |
| 605 | |
| 606 | req = urllib.request.Request(url, data=data, headers=headers, method=method) |
| 607 | |
| 608 | safe_url = re.sub(r'([?&])(key|api_key|token|secret)=[^&]*', r'\1\2=***', url) |
| 609 | log(f"{method} {safe_url}") |
| 610 | |
| 611 | last_error = None |
| 612 | rate_limit_count = 0 |
| 613 | # DNS failures get a dedicated minimum attempt count + exponential backoff. |
| 614 | # `effective_retries` is the actual loop bound; we expand it on the first |
| 615 | # gaierror if the caller passed a smaller `retries` value than MIN_DNS_RETRIES. |
| 616 | effective_retries = retries |
| 617 | dns_attempts = 0 |
| 618 | attempt = 0 |
| 619 | |
| 620 | def raise_recorded(error: HTTPError) -> None: |
| 621 | _fixture_record(fixture_request, error=error, redactions=fixture_redactions) |
| 622 | _raise(error) |
| 623 | |
| 624 | while attempt < effective_retries: |
| 625 | try: |
| 626 | with urllib.request.urlopen(req, timeout=timeout) as response: |
| 627 | body = response.read().decode('utf-8') |
| 628 | log(f"Response: {response.status} ({len(body)} bytes)") |
| 629 | if raw: |
| 630 | _fixture_record(fixture_request, value=body, redactions=fixture_redactions) |
| 631 | return body |
| 632 | parsed = json.loads(body) if body else {} |
| 633 | _fixture_record(fixture_request, value=parsed, redactions=fixture_redactions) |
| 634 | return parsed |
| 635 | except urllib.error.HTTPError as e: |
| 636 | body = None |
| 637 | try: |
| 638 | body = e.read().decode('utf-8') |
| 639 | except (OSError, UnicodeDecodeError): |
| 640 | pass |
| 641 | log(f"HTTP Error {e.code}: {e.reason}") |
| 642 | if body: |
| 643 | snippet = " ".join(body.split()) |
| 644 | log(f"Error body: {snippet[:200]}") |
| 645 | last_error = HTTPError(f"HTTP {e.code}: {e.reason}", e.code, body) |
| 646 | |
| 647 | # Don't retry client errors (4xx) except rate limits |
| 648 | if 400 <= e.code < 500 and e.code != 429: |
| 649 | raise_recorded(last_error) |
| 650 | |
| 651 | # Cap 429 retries separately to avoid wasting latency |
| 652 | if e.code == 429: |
| 653 | rate_limit_count += 1 |
| 654 | if rate_limit_count >= max_429_retries: |
| 655 | raise_recorded(last_error) |
| 656 | |
| 657 | # HTTP errors respect the caller's original `retries`; only DNS |
| 658 | # failures get the widened `effective_retries` budget. |
| 659 | if attempt < retries - 1: |
| 660 | if e.code == 429: |
| 661 | # Respect Retry-After header, fall back to exponential backoff |
| 662 | retry_after = e.headers.get("Retry-After") if hasattr(e, 'headers') else None |
| 663 | if retry_after: |
| 664 | try: |
| 665 | delay = float(retry_after) |
| 666 | except ValueError: |
| 667 | delay = RETRY_DELAY * (2 ** attempt) + 1 |
| 668 | else: |
| 669 | delay = RETRY_DELAY * (2 ** attempt) + 1 # 3s, 5s, 9s... |
| 670 | log(f"Rate limited (429). Waiting {delay:.1f}s before retry {attempt + 2}/{retries}") |
| 671 | else: |
| 672 | delay = RETRY_DELAY * (2 ** attempt) |
| 673 | time.sleep(delay) |
| 674 | else: |
| 675 | # Caller's original retry budget exhausted; an earlier DNS |
| 676 | # failure may have widened `effective_retries`, but that |
| 677 | # widening is DNS-only — don't grant extra HTTP attempts. |
| 678 | break |
| 679 | except urllib.error.URLError as e: |
| 680 | log(f"URL Error: {e.reason}") |
| 681 | reason = getattr(e, "reason", None) |
| 682 | # urllib commonly wraps socket.timeout (an alias of TimeoutError |
| 683 | # since 3.10) in URLError; classify those as timeouts, not |
| 684 | # unreachable hosts, so the recovery guidance is right. |
| 685 | wrapped_timeout = isinstance(reason, TimeoutError) or "timed out" in str(reason).lower() |
| 686 | last_error = HTTPError( |
| 687 | f"URL Error: {e.reason}", |
| 688 | outcome_state=health.TIMEOUT if wrapped_timeout else health.UNREACHABLE, |
| 689 | ) |
| 690 | if _is_dns_failure(e): |
| 691 | # DNS resolution failures are transient; expand the retry budget |
| 692 | # to MIN_DNS_RETRIES if the caller passed fewer, and use |
| 693 | # exponential backoff (1s, 2s, 4s, ...) instead of the linear |
| 694 | # default. Counts DNS attempts separately so other URLError |
| 695 | # causes don't bypass the regular retry budget. |
| 696 | dns_attempts += 1 |
| 697 | if effective_retries < MIN_DNS_RETRIES: |
| 698 | log( |
| 699 | f"DNS resolution failed; expanding retry budget from " |
| 700 | f"{effective_retries} to {MIN_DNS_RETRIES}" |
| 701 | ) |
| 702 | effective_retries = MIN_DNS_RETRIES |
| 703 | if attempt < effective_retries - 1: |
| 704 | delay = 2 ** (dns_attempts - 1) # 1s, 2s, 4s, 8s, ... |
| 705 | log( |
| 706 | f"DNS resolution failure (attempt {dns_attempts}); " |
| 707 | f"retrying in {delay:.1f}s" |
| 708 | ) |
| 709 | time.sleep(delay) |
| 710 | elif attempt < retries - 1: |
| 711 | # Non-DNS URLError (e.g. ConnectionRefused) respects the |
| 712 | # caller's original retry budget, not the DNS-widened bound. |
| 713 | time.sleep(RETRY_DELAY * (attempt + 1)) |
| 714 | else: |
| 715 | # Caller's original retry budget exhausted; an earlier DNS |
| 716 | # failure widening `effective_retries` does not carry over |
| 717 | # to non-DNS error paths. |
| 718 | break |
| 719 | except json.JSONDecodeError as e: |
| 720 | log(f"JSON decode error: {e}") |
| 721 | last_error = HTTPError( |
| 722 | f"Invalid JSON response: {e}", |
| 723 | outcome_state=health.SCHEMA_DRIFT, |
| 724 | ) |
| 725 | raise_recorded(last_error) |
| 726 | except (OSError, TimeoutError, ConnectionResetError) as e: |
| 727 | # Handle socket-level errors (connection reset, timeout, etc.) |
| 728 | log(f"Connection error: {type(e).__name__}: {e}") |
| 729 | state = health.TIMEOUT if isinstance(e, TimeoutError) else health.UNREACHABLE |
| 730 | last_error = HTTPError( |
| 731 | f"Connection error: {type(e).__name__}: {e}", |
| 732 | outcome_state=state, |
| 733 | ) |
| 734 | if attempt < retries - 1: |
| 735 | # Socket errors respect the caller's original retry budget. |
| 736 | time.sleep(RETRY_DELAY * (attempt + 1)) |
| 737 | else: |
| 738 | # Original budget exhausted; DNS widening doesn't apply here. |
| 739 | break |
| 740 | |
| 741 | attempt += 1 |
| 742 | |
| 743 | if last_error: |
| 744 | raise_recorded(last_error) |
| 745 | error = HTTPError("Request failed with no error details") |
| 746 | raise_recorded(error) |
| 747 | |
| 748 | |
| 749 | def get(url: str, headers: Optional[Dict[str, str]] = None, **kwargs) -> Dict[str, Any]: |
| 750 | """Make a GET request.""" |
| 751 | return request("GET", url, headers=headers, **kwargs) |
| 752 | |
| 753 | |
| 754 | def post(url: str, json_data: Dict[str, Any], headers: Optional[Dict[str, str]] = None, **kwargs) -> Dict[str, Any]: |
| 755 | """Make a POST request with JSON body.""" |
| 756 | return request("POST", url, headers=headers, json_data=json_data, **kwargs) |
| 757 | |
| 758 | |
| 759 | def post_raw(url: str, json_data: Dict[str, Any], headers: Optional[Dict[str, str]] = None, **kwargs) -> str: |
| 760 | """Make a POST request with JSON body and return raw text.""" |
| 761 | return request("POST", url, headers=headers, json_data=json_data, raw=True, **kwargs) |
| 762 | |
| 763 | |
| 764 | BROWSER_USER_AGENT = ( |
| 765 | "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " |
| 766 | "AppleWebKit/537.36 (KHTML, like Gecko) " |
| 767 | "Chrome/124.0.0.0 Safari/537.36" |
| 768 | ) |
| 769 | |
| 770 | |
| 771 | def get_text( |
| 772 | url: str, |
| 773 | timeout: int = DEFAULT_TIMEOUT, |
| 774 | retries: int = 2, |
| 775 | accept: str = "*/*", |
| 776 | headers: Optional[Dict[str, str]] = None, |
| 777 | ) -> Optional[str]: |
| 778 | """Fetch a URL and return decoded text, or None on any failure. |
| 779 | |
| 780 | Keyless helper for Reddit RSS and shreddit HTML endpoints — the free path |
| 781 | that replaced the now-403 ``.json`` endpoints. Sends a browser User-Agent |
| 782 | and never raises: returns None on HTTP error, network failure, or timeout |
| 783 | so tiered callers can fall through to the next source. |
| 784 | |
| 785 | Args: |
| 786 | url: Request URL |
| 787 | timeout: HTTP timeout per attempt in seconds |
| 788 | retries: Number of retries on failure (kept low — these tiers fail fast) |
| 789 | accept: Accept header value (e.g. "application/atom+xml", "text/html") |
| 790 | headers: Optional extra headers merged over the defaults |
| 791 | |
| 792 | Returns: |
| 793 | Decoded response body as text, or None on failure. |
| 794 | """ |
| 795 | merged = { |
| 796 | "User-Agent": BROWSER_USER_AGENT, |
| 797 | "Accept": accept, |
| 798 | "Accept-Language": "en-US,en;q=0.9", |
| 799 | } |
| 800 | if headers: |
| 801 | merged.update(headers) |
| 802 | try: |
| 803 | return request( |
| 804 | "GET", url, headers=merged, timeout=timeout, retries=retries, raw=True |
| 805 | ) |
| 806 | except HTTPError as e: |
| 807 | log(f"get_text failed ({e}): {url}") |
| 808 | return None |
| 809 | |
| 810 | |
| 811 | class RateLimiter: |
| 812 | """Thread-safe token-bucket throttle for an endpoint family. |
| 813 | |
| 814 | The keyless source tiers run under the pipeline's ThreadPoolExecutor, so a |
| 815 | multi-subquery run can fire many requests at the same host at once. A bare |
| 816 | per-request retry budget does not prevent that stampede — it only reacts |
| 817 | after a 429. A token bucket bounds the *sustained* rate while still allowing |
| 818 | a short burst, so legitimate parallelism is preserved (unlike a strict |
| 819 | min-interval gate that would serialize every concurrent caller and could |
| 820 | push later futures past their result timeouts). |
| 821 | |
| 822 | ``rate_per_sec`` tokens refill per second; ``burst`` is the bucket capacity |
| 823 | (max simultaneous calls before throttling kicks in). The lock is released |
| 824 | while sleeping so waiting threads don't serialize on each other. |
| 825 | """ |
| 826 | |
| 827 | def __init__(self, rate_per_sec: float, burst: int | None = None): |
| 828 | self.rate = rate_per_sec |
| 829 | self.capacity = burst if burst is not None else max(1, int(rate_per_sec)) |
| 830 | self._tokens = float(self.capacity) |
| 831 | self._last = time.monotonic() |
| 832 | self._lock = threading.Lock() |
| 833 | |
| 834 | def acquire(self) -> None: |
| 835 | """Consume one token, blocking only when the bucket is empty.""" |
| 836 | while True: |
| 837 | with self._lock: |
| 838 | now = time.monotonic() |
| 839 | # Clamp elapsed to >= 0: a backward clock reading must never |
| 840 | # drive tokens negative (which would spin this loop forever). |
| 841 | elapsed = max(0.0, now - self._last) |
| 842 | self._tokens = min(self.capacity, self._tokens + elapsed * self.rate) |
| 843 | self._last = now |
| 844 | if self._tokens >= 1.0: |
| 845 | self._tokens -= 1.0 |
| 846 | return |
| 847 | wait = (1.0 - self._tokens) / self.rate |
| 848 | time.sleep(wait) |
| 849 | |
| 850 | |
| 851 | # Shared across all keyless Reddit tiers (RSS, listing, shreddit) so their |
| 852 | # combined fan-out is throttled as one family. Burst lets the parallel |
| 853 | # enrichment workers proceed; sustained rate caps the stampede. |
| 854 | REDDIT_KEYLESS_LIMITER = RateLimiter(rate_per_sec=5.0, burst=5) |
| 855 | |
| 856 | |
| 857 | def reddit_keyless_get_text( |
| 858 | url: str, |
| 859 | timeout: int = DEFAULT_TIMEOUT, |
| 860 | retries: int = 2, |
| 861 | accept: str = "*/*", |
| 862 | headers: Optional[Dict[str, str]] = None, |
| 863 | ) -> Optional[str]: |
| 864 | """get_text for the keyless Reddit tiers, throttled by a shared limiter. |
| 865 | |
| 866 | Same contract as :func:`get_text` (returns None on any failure) but spaces |
| 867 | requests via :data:`REDDIT_KEYLESS_LIMITER` so a broad multi-query run does |
| 868 | not stampede Reddit's keyless endpoints and trip blocks. |
| 869 | """ |
| 870 | REDDIT_KEYLESS_LIMITER.acquire() |
| 871 | return get_text(url, timeout=timeout, retries=retries, accept=accept, headers=headers) |
| 872 | |
| 873 | |
| 874 | def scrapecreators_headers(token: str) -> Dict[str, str]: |
| 875 | """Build ScrapeCreators request headers (x-api-key + JSON content type).""" |
| 876 | return { |
| 877 | "x-api-key": token, |
| 878 | "Content-Type": "application/json", |
| 879 | } |
| 880 | |
| 881 | |
| 882 | def get_reddit_json(path: str, timeout: int = DEFAULT_TIMEOUT, retries: int = MAX_RETRIES) -> Dict[str, Any]: |
| 883 | """Fetch Reddit thread JSON. |
| 884 | |
| 885 | Args: |
| 886 | path: Reddit path (e.g., /r/subreddit/comments/id/title) |
| 887 | timeout: HTTP timeout per attempt in seconds |
| 888 | retries: Number of retries on failure |
| 889 | |
| 890 | Returns: |
| 891 | Parsed JSON response |
| 892 | """ |
| 893 | # Ensure path starts with / |
| 894 | if not path.startswith('/'): |
| 895 | path = '/' + path |
| 896 | |
| 897 | # Remove trailing slash and add .json |
| 898 | path = path.rstrip('/') |
| 899 | if not path.endswith('.json'): |
| 900 | path = path + '.json' |
| 901 | |
| 902 | url = f"https://www.reddit.com{path}?raw_json=1" |
| 903 | |
| 904 | headers = { |
| 905 | "User-Agent": USER_AGENT, |
| 906 | "Accept": "application/json", |
| 907 | } |
| 908 | |
| 909 | return get(url, headers=headers, timeout=timeout, retries=retries) |
| 910 |