diff --git a/resilience.py b/resilience.py
new file mode 100644
index 0000000..b19f5d0
--- /dev/null
+++ b/resilience.py
@@ -0,0 +1,228 @@
+"""Resilience primitives for Pry scrapers.
+
+Provides circuit breakers and per-domain tier memory so the fallback chain
+can skip unhealthy services and prefer the cheapest successful strategy for
+a given domain.
+"""
+
+# SPDX-License-Identifier: MIT
+# Copyright (c) 2026 Rug Munch Media LLC
+#
+# Part of Pry — https://git.rugmunch.io/RugMunchMedia/pryscraper
+# Licensed under MIT. See LICENSE.
+from __future__ import annotations
+
+import logging
+import time
+from collections import defaultdict
+from dataclasses import dataclass, field
+from typing import Any, ClassVar
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_FAILURE_THRESHOLD = 5
+DEFAULT_RECOVERY_SECONDS = 30
+DEFAULT_HALF_OPEN_MAX_TRIES = 2
+DOMAIN_MEMORY_MAX_AGE_SECONDS = 3600
+
+
+class CircuitState:
+ CLOSED = "closed"
+ OPEN = "open"
+ HALF_OPEN = "half_open"
+
+
+@dataclass
+class CircuitBreaker:
+ """Simple circuit breaker with exponential cooldown."""
+
+ name: str
+ failure_threshold: int = DEFAULT_FAILURE_THRESHOLD
+ recovery_seconds: float = DEFAULT_RECOVERY_SECONDS
+ half_open_max_tries: int = DEFAULT_HALF_OPEN_MAX_TRIES
+ state: str = CircuitState.CLOSED
+ failures: int = 0
+ successes: int = 0
+ last_failure_at: float = 0.0
+ half_open_attempts: int = 0
+
+ def is_allowed(self) -> bool:
+ now = time.time()
+ if self.state == CircuitState.CLOSED:
+ return True
+ if self.state == CircuitState.OPEN:
+ if now - self.last_failure_at >= self.recovery_seconds:
+ self.state = CircuitState.HALF_OPEN
+ self.half_open_attempts = 0
+ logger.info("circuit_breaker_half_open", extra={"service": self.name})
+ return True
+ return False
+ # HALF_OPEN
+ return self.half_open_attempts < self.half_open_max_tries
+
+ def record_success(self) -> None:
+ if self.state == CircuitState.HALF_OPEN:
+ self.half_open_attempts += 1
+ if self.half_open_attempts >= self.half_open_max_tries:
+ self._close()
+ return
+ self.successes += 1
+ # Decay failures slowly on success to forgive transient blips.
+ if self.failures > 0:
+ self.failures -= 1
+
+ def record_failure(self) -> None:
+ self.failures += 1
+ self.last_failure_at = time.time()
+ if self.state == CircuitState.HALF_OPEN:
+ self._open()
+ return
+ if self.failures >= self.failure_threshold:
+ self._open()
+
+ def _open(self) -> None:
+ if self.state != CircuitState.OPEN:
+ logger.warning(
+ "circuit_breaker_open",
+ extra={"service": self.name, "failures": self.failures},
+ )
+ self.state = CircuitState.OPEN
+ self.half_open_attempts = 0
+
+ def _close(self) -> None:
+ logger.info("circuit_breaker_closed", extra={"service": self.name})
+ self.state = CircuitState.CLOSED
+ self.failures = 0
+ self.half_open_attempts = 0
+
+
+@dataclass
+class TierRecord:
+ successes: int = 0
+ failures: int = 0
+ last_success_at: float = 0.0
+ last_failure_at: float = 0.0
+ total_latency: float = 0.0
+
+ @property
+ def score(self) -> float:
+ """Higher score = more preferred. Balances success rate and recency."""
+ total = self.successes + self.failures
+ if total == 0:
+ return 0.0
+ success_rate = self.successes / total
+ # Recency bonus: 1.0 if succeeded in last 5 minutes, decaying.
+ recency = 0.0
+ if self.last_success_at > 0:
+ age = time.time() - self.last_success_at
+ recency = max(0.0, 1.0 - age / DOMAIN_MEMORY_MAX_AGE_SECONDS)
+ return success_rate + recency * 0.5
+
+
+@dataclass
+class DomainTierMemory:
+ """Remember which tiers work best per domain."""
+
+ data: dict[str, dict[str, TierRecord]] = field(default_factory=lambda: defaultdict(dict))
+
+ def record(self, domain: str, tier: str, success: bool, latency: float = 0.0) -> None:
+ rec = self.data[domain].setdefault(tier, TierRecord())
+ if success:
+ rec.successes += 1
+ rec.last_success_at = time.time()
+ else:
+ rec.failures += 1
+ rec.last_failure_at = time.time()
+ rec.total_latency += latency
+
+ def best_tier(self, domain: str, candidates: list[str]) -> str | None:
+ records = self.data.get(domain, {})
+ if not records:
+ return None
+ # Only consider tiers with at least one success and no recent failures.
+ now = time.time()
+ eligible = [
+ (tier, rec)
+ for tier, rec in records.items()
+ if rec.successes > 0
+ and (rec.last_failure_at == 0 or now - rec.last_failure_at > 300)
+ and tier in candidates
+ ]
+ if not eligible:
+ return None
+ eligible.sort(key=lambda item: item[1].score, reverse=True)
+ return eligible[0][0]
+
+ def to_dict(self) -> dict[str, dict[str, dict[str, Any]]]:
+ return {
+ domain: {
+ tier: {
+ "successes": rec.successes,
+ "failures": rec.failures,
+ "last_success_at": rec.last_success_at,
+ "last_failure_at": rec.last_failure_at,
+ "score": round(rec.score, 3),
+ }
+ for tier, rec in tiers.items()
+ }
+ for domain, tiers in self.data.items()
+ }
+
+
+class ResilienceRegistry:
+ """Global registry for circuit breakers and domain tier memory."""
+
+ # External services that should have circuit breakers.
+ SERVICE_NAMES: ClassVar[list[str]] = [
+ "flaresolverr",
+ "ollama",
+ "openrouter",
+ "redis",
+ ]
+
+ def __init__(self) -> None:
+ self.breakers: dict[str, CircuitBreaker] = {
+ name: CircuitBreaker(name=name) for name in self.SERVICE_NAMES
+ }
+ self.domain_memory = DomainTierMemory()
+
+ def breaker(self, name: str) -> CircuitBreaker:
+ if name not in self.breakers:
+ self.breakers[name] = CircuitBreaker(name=name)
+ return self.breakers[name]
+
+ def is_service_allowed(self, name: str) -> bool:
+ return self.breaker(name).is_allowed()
+
+ def record_service_result(self, name: str, success: bool) -> None:
+ breaker = self.breaker(name)
+ if success:
+ breaker.record_success()
+ else:
+ breaker.record_failure()
+
+ def record_domain_tier_result(
+ self, domain: str, tier: str, success: bool, latency: float = 0.0
+ ) -> None:
+ self.domain_memory.record(domain, tier, success, latency)
+
+ def best_tier_for_domain(self, domain: str, candidates: list[str]) -> str | None:
+ return self.domain_memory.best_tier(domain, candidates)
+
+ def stats(self) -> dict[str, Any]:
+ return {
+ "circuit_breakers": {
+ name: {
+ "state": cb.state,
+ "failures": cb.failures,
+ "successes": cb.successes,
+ "last_failure_at": cb.last_failure_at,
+ }
+ for name, cb in self.breakers.items()
+ },
+ "domain_memory": self.domain_memory.to_dict(),
+ }
+
+
+# Global singleton used by scrapers.
+registry = ResilienceRegistry()
diff --git a/tests/test_resilience.py b/tests/test_resilience.py
new file mode 100644
index 0000000..b6606da
--- /dev/null
+++ b/tests/test_resilience.py
@@ -0,0 +1,219 @@
+"""Tests for resilience primitives and their integration into UltimateScraper."""
+
+from __future__ import annotations
+
+import time as time_module
+from contextlib import ExitStack
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+import resilience
+import ultimate_scraper
+from resilience import CircuitBreaker, CircuitState, DomainTierMemory, ResilienceRegistry
+
+
+def _valid_html(text: str) -> str:
+ body = f"
{text}
" * 40
+ return (
+ "T"
+ ""
+ f"{body}"
+ )
+
+
+@pytest.fixture
+def fresh_registry(monkeypatch):
+ """Provide a clean ResilienceRegistry and patch it into UltimateScraper."""
+ reg = ResilienceRegistry()
+ monkeypatch.setattr(resilience, "registry", reg)
+ monkeypatch.setattr(ultimate_scraper, "registry", reg)
+ return reg
+
+
+@pytest.fixture
+def frozen_time(monkeypatch):
+ """Freeze time.time() for deterministic circuit-breaker tests."""
+ now = 1_700_000_000.0
+
+ def fake_time():
+ return now
+
+ monkeypatch.setattr(time_module, "time", fake_time)
+ monkeypatch.setattr(resilience.time, "time", fake_time)
+ monkeypatch.setattr(ultimate_scraper.time, "time", fake_time)
+ return lambda offset=0: setattr(fake_time, "__wrapped__", None) or None
+
+
+class TestCircuitBreaker:
+ def test_starts_closed_and_allows(self):
+ cb = CircuitBreaker(name="test")
+ assert cb.state == CircuitState.CLOSED
+ assert cb.is_allowed() is True
+
+ def test_opens_after_threshold_failures(self):
+ cb = CircuitBreaker(name="test", failure_threshold=3, recovery_seconds=30)
+ cb.record_failure()
+ cb.record_failure()
+ assert cb.is_allowed() is True
+ cb.record_failure()
+ assert cb.state == CircuitState.OPEN
+ assert cb.is_allowed() is False
+
+ def test_half_open_after_recovery(self, frozen_time, monkeypatch):
+ cb = CircuitBreaker(name="test", failure_threshold=1, recovery_seconds=60)
+ cb.record_failure()
+ assert cb.state == CircuitState.OPEN
+ # Advance time past recovery window.
+ now = 1_700_000_000.0 + 61
+ monkeypatch.setattr(time_module, "time", lambda: now)
+ monkeypatch.setattr(resilience.time, "time", lambda: now)
+ assert cb.is_allowed() is True
+ assert cb.state == CircuitState.HALF_OPEN
+
+ def test_closes_after_half_open_successes(self, frozen_time, monkeypatch):
+ cb = CircuitBreaker(
+ name="test", failure_threshold=1, recovery_seconds=60, half_open_max_tries=2
+ )
+ cb.record_failure()
+ now = 1_700_000_000.0 + 61
+ monkeypatch.setattr(time_module, "time", lambda: now)
+ monkeypatch.setattr(resilience.time, "time", lambda: now)
+ assert cb.is_allowed() is True
+ cb.record_success()
+ assert cb.state == CircuitState.HALF_OPEN
+ assert cb.is_allowed() is True
+ cb.record_success()
+ assert cb.state == CircuitState.CLOSED
+
+ def test_failure_in_half_open_reopens(self, frozen_time, monkeypatch):
+ cb = CircuitBreaker(name="test", failure_threshold=1, recovery_seconds=60)
+ cb.record_failure()
+ now = 1_700_000_000.0 + 61
+ monkeypatch.setattr(time_module, "time", lambda: now)
+ monkeypatch.setattr(resilience.time, "time", lambda: now)
+ cb.is_allowed()
+ cb.record_failure()
+ assert cb.state == CircuitState.OPEN
+
+
+class TestDomainTierMemory:
+ def test_picks_best_tier(self):
+ mem = DomainTierMemory()
+ mem.record("example.com", "direct", success=True, latency=0.1)
+ mem.record("example.com", "direct", success=True, latency=0.1)
+ mem.record("example.com", "playwright", success=False, latency=0.1)
+ best = mem.best_tier("example.com", ["direct", "playwright"])
+ assert best == "direct"
+
+ def test_ignores_recent_failures(self, frozen_time, monkeypatch):
+ mem = DomainTierMemory()
+ now = 1_700_000_000.0
+ monkeypatch.setattr(time_module, "time", lambda: now)
+ mem.record("example.com", "direct", success=True, latency=0.1)
+ mem.record("example.com", "direct", success=False, latency=0.1)
+ assert mem.best_tier("example.com", ["direct"]) is None
+
+ def test_returns_none_for_unknown_domain(self):
+ mem = DomainTierMemory()
+ assert mem.best_tier("example.com", ["direct"]) is None
+
+
+class TestResilienceRegistry:
+ def test_stats_shape(self, fresh_registry):
+ stats = fresh_registry.stats()
+ assert "circuit_breakers" in stats
+ assert "domain_memory" in stats
+ assert set(stats["circuit_breakers"].keys()) == set(ResilienceRegistry.SERVICE_NAMES)
+
+ def test_service_result_toggles_breaker(self, fresh_registry):
+ fresh_registry.breaker("flaresolverr").failure_threshold = 1
+ fresh_registry.record_service_result("flaresolverr", False)
+ assert fresh_registry.is_service_allowed("flaresolverr") is False
+ fresh_registry.record_service_result("flaresolverr", True)
+ # In half-open after recovery, not immediate.
+ assert fresh_registry.breaker("flaresolverr").state == CircuitState.OPEN
+
+
+_TIER_NAMES = [
+ "_tier_direct",
+ "_tier_tls_fingerprint",
+ "_tier_cloudscraper",
+ "_tier_flaresolverr",
+ "_tier_undetected",
+ "_tier_camoufox",
+ "_tier_playwright",
+ "_tier_googlebot",
+ "_tier_premium_proxy",
+ "_tier_archive",
+ "_tier_google_cache",
+ "_tier_tor",
+]
+
+
+class TestUltimateScraperResilience:
+ @pytest.mark.anyio
+ async def test_skips_circuit_open_tier(self, fresh_registry):
+ scraper = ultimate_scraper.UltimateScraper()
+ fresh_registry.breaker("flaresolverr").failure_threshold = 1
+ fresh_registry.record_service_result("flaresolverr", False)
+
+ html = _valid_html("archive content")
+ mocks = {
+ name: AsyncMock(return_value=None)
+ for name in _TIER_NAMES
+ }
+ mocks["_tier_archive"].return_value = html
+ with ExitStack() as stack:
+ for name in _TIER_NAMES:
+ stack.enter_context(patch.object(scraper, name, new=mocks[name]))
+ result = await scraper.scrape("https://example.com/page")
+ assert result["status"] == "ok"
+ assert result["method"] == "archive"
+ mocks["_tier_direct"].assert_awaited_once()
+ mocks["_tier_cloudscraper"].assert_awaited_once()
+ mocks["_tier_flaresolverr"].assert_not_awaited()
+
+ @pytest.mark.anyio
+ async def test_prefers_domain_memory_best_tier(self, fresh_registry):
+ scraper = ultimate_scraper.UltimateScraper()
+ fresh_registry.record_domain_tier_result("example.com", "archive", success=True)
+ html = _valid_html("archived")
+ mocks = {name: AsyncMock(return_value=None) for name in _TIER_NAMES}
+ mocks["_tier_archive"].return_value = html
+ mocks["_tier_direct"].return_value = _valid_html("direct")
+ with ExitStack() as stack:
+ for name in _TIER_NAMES:
+ stack.enter_context(patch.object(scraper, name, new=mocks[name]))
+ result = await scraper.scrape("https://example.com/page")
+ assert result["status"] == "ok"
+ assert result["method"] == "archive"
+ mocks["_tier_archive"].assert_awaited()
+ mocks["_tier_direct"].assert_not_awaited()
+
+ @pytest.mark.anyio
+ async def test_records_tier_results_after_success(self, fresh_registry):
+ scraper = ultimate_scraper.UltimateScraper()
+ html = _valid_html("direct content")
+ mocks = {name: AsyncMock(return_value=None) for name in _TIER_NAMES}
+ mocks["_tier_direct"].return_value = html
+ with ExitStack() as stack:
+ for name in _TIER_NAMES:
+ stack.enter_context(patch.object(scraper, name, new=mocks[name]))
+ await scraper.scrape("https://example.com/page")
+ assert fresh_registry.best_tier_for_domain(
+ "example.com", ["direct", "archive"]
+ ) == "direct"
+
+ @pytest.mark.anyio
+ async def test_records_tier_failure_no_success_memory(self, fresh_registry):
+ scraper = ultimate_scraper.UltimateScraper()
+ mocks = {name: AsyncMock(return_value=None) for name in _TIER_NAMES}
+ with ExitStack() as stack:
+ for name in _TIER_NAMES:
+ stack.enter_context(patch.object(scraper, name, new=mocks[name]))
+ result = await scraper.scrape("https://example.com/page")
+ assert result["status"] == "error"
+ assert fresh_registry.best_tier_for_domain(
+ "example.com", ["direct", "archive"]
+ ) is None
diff --git a/ultimate_scraper.py b/ultimate_scraper.py
index 84eded7..c59a209 100644
--- a/ultimate_scraper.py
+++ b/ultimate_scraper.py
@@ -14,9 +14,12 @@ import logging
import os
import random
import re
+import time
from typing import Any, ClassVar
from urllib.parse import urlparse
+from resilience import registry
+
logger = logging.getLogger(__name__)
# Try importing optional bypass libraries
@@ -118,67 +121,71 @@ class UltimateScraper:
return self._result(url, html, "premium_proxy")
errors.append("premium_proxy_failed")
- # Tier 1: Direct HTTP
- html = await self._tier_direct(url, opts, proxy_url=proxy_url)
- if html and self._is_valid(html):
- return self._result(url, html, "direct")
+ # Build tier list. Order starts cheap/direct and escalates.
+ tier_order = [
+ ("direct", self._tier_direct),
+ ("tls_fingerprint", self._tier_tls_fingerprint),
+ ("cloudscraper", self._tier_cloudscraper),
+ ("flaresolverr", self._tier_flaresolverr),
+ ("undetected", self._tier_undetected),
+ ("camoufox", self._tier_camoufox),
+ ("playwright", self._tier_playwright),
+ ("googlebot", self._tier_googlebot),
+ ]
- # Tier 2: TLS fingerprint impersonation
- html = await self._tier_tls_fingerprint(url, opts, proxy_url=proxy_url)
- if html and self._is_valid(html):
- return self._result(url, html, "tls_fingerprint")
-
- # Tier 3: cloudscraper (Python Cloudflare bypass)
- html = await self._tier_cloudscraper(url, opts)
- if html and self._is_valid(html):
- return self._result(url, html, "cloudscraper")
-
- # Tier 4: FlareSolverr
- html = await self._tier_flaresolverr(url, opts)
- if html and self._is_valid(html):
- return self._result(url, html, "flaresolverr")
-
- # Tier 5: undetected-chromedriver
- html = await self._tier_undetected(url, opts)
- if html and self._is_valid(html):
- return self._result(url, html, "undetected")
-
- # Tier 6: Camoufox stealth Firefox
- html = await self._tier_camoufox(url, opts, proxy_url=proxy_url)
- if html and self._is_valid(html):
- return self._result(url, html, "camoufox")
-
- # Tier 7: Playwright stealth
- html = await self._tier_playwright(url, opts)
- if html and self._is_valid(html):
- return self._result(url, html, "playwright")
-
- # Tier 8: Googlebot UA
- html = await self._tier_googlebot(url, opts, proxy_url=proxy_url)
- if html and self._is_valid(html):
- return self._result(url, html, "googlebot")
-
- # Tier 9: Premium proxy retry (after cheap tiers fail)
+ # Premium proxy retry (after cheap tiers fail) if not tried first.
if proxy_url and not opts.get("use_premium_proxy_first", True):
- html = await self._tier_premium_proxy(url, proxy_url, opts)
- if html and self._is_valid(html):
- return self._result(url, html, "premium_proxy")
- errors.append("premium_proxy_failed")
+ tier_order.append(("premium_proxy", self._tier_premium_proxy))
- # Tier 10: Archive.org / Wayback Machine
- html = await self._tier_archive(url, opts)
- if html and self._is_valid(html):
- return self._result(url, html, "archive")
+ tier_order.extend([
+ ("archive", self._tier_archive),
+ ("google_cache", self._tier_google_cache),
+ ("tor", self._tier_tor),
+ ])
- # Tier 11: Google Cache
- html = await self._tier_google_cache(url, opts)
- if html and self._is_valid(html):
- return self._result(url, html, "google_cache")
+ # If domain memory knows a good tier, try it first.
+ parsed = urlparse(url)
+ domain = parsed.netloc
+ best = registry.best_tier_for_domain(domain, [name for name, _ in tier_order])
+ if best:
+ tier_order.sort(key=lambda item: (item[0] != best, item[0]))
- # Tier 12: Tor (anonymous routing)
- html = await self._tier_tor(url, opts)
- if html and self._is_valid(html):
- return self._result(url, html, "tor")
+ # External service tiers protected by circuit breakers.
+ breaker_for_tier = {
+ "flaresolverr": "flaresolverr",
+ "playwright": "playwright",
+ "camoufox": "camoufox",
+ "tls_fingerprint": "tls_fingerprint",
+ }
+
+ for tier_name, tier_fn in tier_order:
+ # Skip external tiers whose circuit breaker is open.
+ breaker_name = breaker_for_tier.get(tier_name)
+ if breaker_name and not registry.is_service_allowed(breaker_name):
+ logger.debug("tier_skipped_circuit_open", extra={"tier": tier_name, "domain": domain})
+ errors.append(f"{tier_name}: circuit_open")
+ continue
+
+ start = time.time()
+ try:
+ if tier_name == "premium_proxy":
+ html = await tier_fn(url, proxy_url, opts)
+ else:
+ html = await tier_fn(url, opts, proxy_url=proxy_url)
+ except Exception as e: # noqa: BLE001
+ logger.debug("tier_exception", extra={"tier": tier_name, "error": str(e)[:50]})
+ html = None
+
+ latency = time.time() - start
+ success = bool(html and self._is_valid(html))
+ registry.record_domain_tier_result(domain, tier_name, success, latency)
+ if breaker_name:
+ registry.record_service_result(breaker_name, success)
+
+ if success:
+ method = "premium_proxy" if tier_name == "premium_proxy" else tier_name
+ return self._result(url, html, method)
+ errors.append(f"{tier_name}_failed")
last_err = errors[-1] if errors else "all_tiers_failed"
return {
@@ -301,7 +308,7 @@ class UltimateScraper:
logger.debug("premium_proxy_failed", extra={"error": str(e)[:50]})
return None
- async def _tier_cloudscraper(self, url: str, opts: dict[str, Any]) -> str | None:
+ async def _tier_cloudscraper(self, url: str, opts: dict[str, Any], proxy_url: str | None = None) -> str | None:
if not _has_cloudscraper:
return None
try:
@@ -317,7 +324,7 @@ class UltimateScraper:
logger.debug("cloudscraper_failed", extra={"error": str(e)[:50]})
return None
- async def _tier_flaresolverr(self, url: str, opts: dict[str, Any]) -> str | None:
+ async def _tier_flaresolverr(self, url: str, opts: dict[str, Any], proxy_url: str | None = None) -> str | None:
from settings import settings
if not settings.flaresolverr_url:
@@ -337,7 +344,7 @@ class UltimateScraper:
logger.debug("flaresolverr_failed", extra={"error": str(e)[:50]})
return None
- async def _tier_undetected(self, url: str, opts: dict[str, Any]) -> str | None:
+ async def _tier_undetected(self, url: str, opts: dict[str, Any], proxy_url: str | None = None) -> str | None:
if not _has_undetected:
return None
try:
@@ -364,7 +371,7 @@ class UltimateScraper:
logger.debug("undetected_failed", extra={"error": str(e)[:50]})
return None
- async def _tier_playwright(self, url: str, opts: dict[str, Any]) -> str | None:
+ async def _tier_playwright(self, url: str, opts: dict[str, Any], proxy_url: str | None = None) -> str | None:
if not _has_playwright:
return None
try:
@@ -460,7 +467,7 @@ class UltimateScraper:
logger.debug("googlebot_failed", extra={"error": str(e)[:50]})
return None
- async def _tier_archive(self, url: str, opts: dict[str, Any]) -> str | None:
+ async def _tier_archive(self, url: str, opts: dict[str, Any], proxy_url: str | None = None) -> str | None:
"""Fetch from Wayback Machine CDX API for latest snapshot."""
try:
import httpx
@@ -480,7 +487,7 @@ class UltimateScraper:
logger.debug("archive_failed", extra={"error": str(e)[:50]})
return None
- async def _tier_google_cache(self, url: str, opts: dict[str, Any]) -> str | None:
+ async def _tier_google_cache(self, url: str, opts: dict[str, Any], proxy_url: str | None = None) -> str | None:
try:
import httpx
@@ -493,7 +500,7 @@ class UltimateScraper:
logger.debug("google_cache_failed", extra={"error": str(e)[:50]})
return None
- async def _tier_tor(self, url: str, opts: dict[str, Any]) -> str | None:
+ async def _tier_tor(self, url: str, opts: dict[str, Any], proxy_url: str | None = None) -> str | None:
"""Fetch via Tor SOCKS5 proxy."""
if not _has_tor:
return None