feat(amazon): route fetches through marketplace geo and language

surfsense_backend/app/proprietary/platforms/amazon/fetch.py

@@ -12,12 +12,12 @@ import logging
 import re
 import time
 from collections.abc import Awaitable, Callable
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from typing import Any
 
 from scrapling.fetchers import AsyncFetcher
 
-from app.utils.proxy import get_proxy_url
+from app.utils.proxy import get_geo_proxy_url, get_sticky_proxy_url
 
 logger = logging.getLogger(__name__)
 
@@ -32,6 +32,10 @@ _BLOCK_MARKERS = (
     "api-services-support@amazon.com",
     "robot check",
     "enter the characters you see below",
+    "token.awswaf.com",
+    "/challenge.js",
+    "awswafintegration",
+    "bm-verify=",
 )
 _CSRF_PATTERNS = (
     re.compile(
@@ -54,6 +58,7 @@ class FetchResult:
     html: str
     url: str
     cookies: dict[str, str]
+    headers: dict[str, str] = field(default_factory=dict)
 
 
 @dataclass
@@ -83,7 +88,9 @@ async def gather_bounded[T](
     return await asyncio.gather(*(run(factory) for factory in factories))
 
 
-def is_blocked(html: str | None, status: int) -> bool:
+def is_blocked(
+    html: str | None, status: int, headers: dict[str, str] | None = None
+) -> bool:
     """Return whether a response is an Amazon anti-bot interstitial.
 
     ``202`` is a soft anti-bot response Amazon serves to some proxy exits (an
@@ -92,10 +99,22 @@ def is_blocked(html: str | None, status: int) -> bool:
     """
     if status in {202, 429, 503}:
         return True
+    if _header_value(headers, "x-amzn-waf-action") == "challenge":
+        return True
     text = (html or "")[:200_000].lower()
     return any(marker in text for marker in _BLOCK_MARKERS)
 
 
+def _header_value(headers: dict[str, str] | None, name: str) -> str | None:
+    if not headers:
+        return None
+    needle = name.lower()
+    for key, value in headers.items():
+        if key.lower() == needle:
+            return str(value).strip().lower()
+    return None
+
+
 def _response_url(page: Any, fallback: str) -> str:
     value = getattr(page, "url", None)
     return str(value) if value else fallback
@@ -106,11 +125,29 @@ def _response_cookies(page: Any) -> dict[str, str]:
     return dict(cookies) if isinstance(cookies, dict) else {}
 
 
+def _response_headers(page: Any) -> dict[str, str]:
+    headers = getattr(page, "headers", None)
+    return dict(headers) if isinstance(headers, dict) else {}
+
+
+def _selected_proxy(
+    proxy: str | None, country: str | None, attempt: int, url: str
+) -> str | None:
+    if proxy is not None:
+        return proxy
+    if country and attempt > 1:
+        session_id = f"amazon-{country}-{attempt}-{abs(hash((url, time.time_ns()))):x}"
+        return get_sticky_proxy_url(session_id, country)
+    return get_geo_proxy_url(country)
+
+
 async def fetch_page(
     url: str,
     *,
     cookies: dict[str, str] | None = None,
     proxy: str | None = None,
+    country: str | None = None,
+    accept_language: str | None = None,
     method: str = "GET",
     data: dict[str, str] | None = None,
     headers: dict[str, str] | None = None,
@@ -119,12 +156,18 @@ async def fetch_page(
     """Fetch a page and retry blocked responses with fresh proxy exits."""
     attempts = _MAX_IP_ATTEMPTS if rotate_on_block else 1
     for attempt in range(1, attempts + 1):
-        selected_proxy = proxy if proxy is not None else get_proxy_url()
+        selected_proxy = _selected_proxy(proxy, country, attempt, url)
         started = time.perf_counter()
         try:
             request = AsyncFetcher.post if method == "POST" else AsyncFetcher.get
+            request_headers = {**_HEADERS}
+            if accept_language:
+                request_headers["Accept-Language"] = (
+                    f"{accept_language},{accept_language.split('-', 1)[0]};q=0.9"
+                )
+            request_headers.update(headers or {})
             kwargs: dict[str, Any] = {
-                "headers": {**_HEADERS, **(headers or {})},
+                "headers": request_headers,
                 "cookies": cookies or {},
                 "proxy": selected_proxy,
                 "stealthy_headers": True,
@@ -141,6 +184,7 @@ async def fetch_page(
 
         status = int(getattr(page, "status", 0) or 0)
         html = getattr(page, "html_content", None) or ""
+        response_headers = _response_headers(page)
         logger.info(
             "[amazon][perf] method=%s status=%s attempt=%s fetch_ms=%.1f url=%s",
             method,
@@ -149,7 +193,7 @@ async def fetch_page(
             (time.perf_counter() - started) * 1000,
             url,
         )
-        if rotate_on_block and is_blocked(html, status):
+        if rotate_on_block and is_blocked(html, status, response_headers):
             logger.info(
                 "Amazon blocked proxy attempt %s/%s for %s", attempt, attempts, url
             )
@@ -159,6 +203,7 @@ async def fetch_page(
             html=html,
             url=_response_url(page, url),
             cookies=_response_cookies(page),
+            headers=response_headers,
         )
     logger.warning("Amazon exhausted %s proxy attempts for %s", attempts, url)
     return None
@@ -169,15 +214,27 @@ async def fetch_html(
     *,
     cookies: dict[str, str] | None = None,
     proxy: str | None = None,
+    country: str | None = None,
+    accept_language: str | None = None,
 ) -> str | None:
     """Return public page HTML, or ``None`` when no usable response is obtained."""
-    result = await fetch_page(url, cookies=cookies, proxy=proxy)
+    result = await fetch_page(
+        url,
+        cookies=cookies,
+        proxy=proxy,
+        country=country,
+        accept_language=accept_language,
+    )
     return result.html if result is not None and result.status == 200 else None
 
 
-async def resolve_shortlink(url: str) -> str | None:
+async def resolve_shortlink(
+    url: str, *, country: str | None = None, accept_language: str | None = None
+) -> str | None:
     """Follow a shortened Amazon URL and return its final destination."""
-    result = await fetch_page(url)
+    result = await fetch_page(
+        url, country=country, accept_language=accept_language
+    )
     return result.url if result is not None and result.status == 200 else None
 
 
@@ -199,6 +256,8 @@ async def fetch_aod_html(
     *,
     cookies: dict[str, str] | None = None,
     proxy: str | None = None,
+    country: str | None = None,
+    accept_language: str | None = None,
 ) -> str | None:
     """Fetch the public all-offers panel for one product.
 
@@ -207,7 +266,13 @@ async def fetch_aod_html(
     expected and the caller falls back to the PDP buy-box winner.
     """
     url = f"{_origin(domain)}/gp/product/ajax/aodAjaxMain/?asin={asin}"
-    return await fetch_html(url, cookies=cookies, proxy=proxy)
+    return await fetch_html(
+        url,
+        cookies=cookies,
+        proxy=proxy,
+        country=country,
+        accept_language=accept_language,
+    )
 
 
 async def fetch_seller_html(
@@ -216,10 +281,16 @@ async def fetch_seller_html(
     *,
     cookies: dict[str, str] | None = None,
     proxy: str | None = None,
+    country: str | None = None,
+    accept_language: str | None = None,
 ) -> str | None:
     """Fetch a public seller profile."""
     return await fetch_html(
-        f"{_origin(domain)}/sp?seller={seller_id}", cookies=cookies, proxy=proxy
+        f"{_origin(domain)}/sp?seller={seller_id}",
+        cookies=cookies,
+        proxy=proxy,
+        country=country,
+        accept_language=accept_language,
     )
 
 
@@ -233,14 +304,12 @@ def should_localize(route: str, deliverable_routes: list[str] | None) -> bool:
     return route.upper() in {value.upper() for value in (deliverable_routes or [])}
 
 
-async def _sticky_proxy(session_id: str) -> str | None:
+async def _sticky_proxy(session_id: str, country: str | None = None) -> str | None:
     """Request a stable proxy exit when the active provider supports it."""
     try:
-        from app.utils.proxy import get_sticky_proxy_url
+        return get_sticky_proxy_url(session_id, country)
-
-        return get_sticky_proxy_url(session_id)
     except (ImportError, NotImplementedError):
-        return get_proxy_url()
+        return get_geo_proxy_url(country)
 
 
 async def get_location_session(
@@ -248,6 +317,8 @@ async def get_location_session(
     *,
     zip_code: str,
     country_code: str | None,
+    country: str | None = None,
+    accept_language: str | None = None,
 ) -> LocationSession | None:
     """Create or reuse an anonymous delivery-location session."""
     key = (domain, zip_code, country_code)
@@ -262,10 +333,12 @@ async def get_location_session(
             return cached
 
         session_id = f"amazon-{abs(hash(key)) & 0xFFFFFFFF:x}"
-        proxy = await _sticky_proxy(session_id)
+        proxy = await _sticky_proxy(session_id, country)
         home = await fetch_page(
             f"{_origin(domain)}/?ref_=nav_logo",
             proxy=proxy,
+            country=country,
+            accept_language=accept_language,
             rotate_on_block=False,
         )
         if home is None or home.status != 200:
@@ -297,6 +370,8 @@ async def get_location_session(
                 "X-Requested-With": "XMLHttpRequest",
                 "Content-Type": "application/x-www-form-urlencoded; charset=UTF-8",
             },
+            country=country,
+            accept_language=accept_language,
             rotate_on_block=False,
         )
         if changed is None or changed.status != 200: