From 82e42728343b61fac25ebde8985184606697c771 Mon Sep 17 00:00:00 2001 From: Anish Sarkar <104695310+AnishSarkar22@users.noreply.github.com> Date: Wed, 15 Jul 2026 15:32:06 +0530 Subject: [PATCH] feat(amazon): route fetches through marketplace geo and language --- .../app/proprietary/platforms/amazon/fetch.py | 109 +++++++++++++++--- 1 file changed, 92 insertions(+), 17 deletions(-) diff --git a/surfsense_backend/app/proprietary/platforms/amazon/fetch.py b/surfsense_backend/app/proprietary/platforms/amazon/fetch.py index c854034b1..9d4b44539 100644 --- a/surfsense_backend/app/proprietary/platforms/amazon/fetch.py +++ b/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) + return get_sticky_proxy_url(session_id, country) 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: