diff --git a/docker/.env.example b/docker/.env.example index 2aa0806a8..cd9789fae 100644 --- a/docker/.env.example +++ b/docker/.env.example @@ -441,7 +441,7 @@ SURFSENSE_ENABLE_DOOM_LOOP=true # WEB_CRAWL_CAPTCHA_MICROS_PER_SOLVE=3000 # Debit the credit wallet per *item returned* by the platform-native scrapers -# (Reddit, Google Search, Google Maps, YouTube). Default FALSE keeps scraping +# (Reddit, Google Search, Google Maps, Amazon, YouTube). Default FALSE keeps scraping # effectively free for self-hosted installs. Each rate is micro-USD per item, # config-driven: = round(USD_per_1000_items * 1_000). Defaults sit # at/above Apify's first-party actor rates (we charge no subscription tiers, @@ -452,6 +452,7 @@ SURFSENSE_ENABLE_DOOM_LOOP=true # GOOGLE_SEARCH_MICROS_PER_SERP=5500 # GOOGLE_MAPS_MICROS_PER_PLACE=3500 # GOOGLE_MAPS_MICROS_PER_REVIEW=1500 +# AMAZON_MICROS_PER_PRODUCT=3500 # YOUTUBE_MICROS_PER_VIDEO=2500 # YOUTUBE_MICROS_PER_COMMENT=1500 # TIKTOK_MICROS_PER_VIDEO=3500 diff --git a/surfsense_backend/.env.example b/surfsense_backend/.env.example index 3d2355460..58e579e05 100644 --- a/surfsense_backend/.env.example +++ b/surfsense_backend/.env.example @@ -277,8 +277,8 @@ MICROS_PER_PAGE=1000 # WEB_CRAWL_CAPTCHA_MICROS_PER_SOLVE=3000 # Debit the credit wallet per *item returned* by the platform-native scrapers -# (Reddit, Google Search, Google Maps, YouTube). Default FALSE keeps scraping -# effectively free for self-hosted/OSS installs; hosted deployments set TRUE. +# (Reddit, Google Search, Google Maps, Amazon, YouTube). Default FALSE keeps +# scraping effectively free for self-hosted/OSS installs; hosted set TRUE. # Each rate is micro-USD per item, fully config-driven (no hardcoded rate): # = round(USD_per_1000_items * 1_000) # 3500 == $3.50/1000 | 5000 == $5/1000 | 2000 == $2/1000 @@ -290,6 +290,7 @@ MICROS_PER_PAGE=1000 # GOOGLE_SEARCH_MICROS_PER_SERP=5500 # GOOGLE_MAPS_MICROS_PER_PLACE=3500 # GOOGLE_MAPS_MICROS_PER_REVIEW=1500 +# AMAZON_MICROS_PER_PRODUCT=3500 # YOUTUBE_MICROS_PER_VIDEO=2500 # YOUTUBE_MICROS_PER_COMMENT=1500 # INSTAGRAM_SCRAPE_MICROS_PER_ITEM=3500 diff --git a/surfsense_backend/app/agents/chat/multi_agent_chat/constants.py b/surfsense_backend/app/agents/chat/multi_agent_chat/constants.py index 6bbd808fb..333d4d274 100644 --- a/surfsense_backend/app/agents/chat/multi_agent_chat/constants.py +++ b/surfsense_backend/app/agents/chat/multi_agent_chat/constants.py @@ -29,6 +29,7 @@ CONNECTOR_TYPE_TO_CONNECTOR_AGENT_MAPS: dict[str, str] = { # connected app. Tokens are searchable-type strings (Composio Gmail/Calendar # map to the GOOGLE_* tokens in connector_searchable_types). SUBAGENT_TO_REQUIRED_CONNECTOR_MAP: dict[str, frozenset[str]] = { + "amazon": frozenset(), "deliverables": frozenset(), "knowledge_base": frozenset(), "web_crawler": frozenset(), diff --git a/surfsense_backend/app/agents/chat/multi_agent_chat/subagents/builtins/amazon/__init__.py b/surfsense_backend/app/agents/chat/multi_agent_chat/subagents/builtins/amazon/__init__.py new file mode 100644 index 000000000..24f27b429 --- /dev/null +++ b/surfsense_backend/app/agents/chat/multi_agent_chat/subagents/builtins/amazon/__init__.py @@ -0,0 +1 @@ +"""``amazon`` builtin subagent: structured public Amazon product data.""" diff --git a/surfsense_backend/app/agents/chat/multi_agent_chat/subagents/builtins/amazon/agent.py b/surfsense_backend/app/agents/chat/multi_agent_chat/subagents/builtins/amazon/agent.py new file mode 100644 index 000000000..cc2e261b4 --- /dev/null +++ b/surfsense_backend/app/agents/chat/multi_agent_chat/subagents/builtins/amazon/agent.py @@ -0,0 +1,43 @@ +"""``amazon`` route: ``SurfSenseSubagentSpec`` builder for deepagents.""" + +from __future__ import annotations + +from typing import Any + +from langchain_core.language_models import BaseChatModel +from langchain_core.tools import BaseTool + +from app.agents.chat.multi_agent_chat.subagents.shared.md_file_reader import ( + read_md_file, +) +from app.agents.chat.multi_agent_chat.subagents.shared.spec import SurfSenseSubagentSpec +from app.agents.chat.multi_agent_chat.subagents.shared.subagent_builder import ( + pack_subagent, +) + +from .tools.index import NAME, RULESET, load_tools + + +def build_subagent( + *, + dependencies: dict[str, Any], + model: BaseChatModel | None = None, + middleware_stack: dict[str, Any] | None = None, + mcp_tools: list[BaseTool] | None = None, +) -> SurfSenseSubagentSpec: + tools = [*load_tools(dependencies=dependencies), *(mcp_tools or [])] + description = ( + read_md_file(__package__, "description").strip() + or "Scrapes public Amazon product data for a URL or search term." + ) + system_prompt = read_md_file(__package__, "system_prompt").strip() + return pack_subagent( + name=NAME, + description=description, + system_prompt=system_prompt, + tools=tools, + ruleset=RULESET, + dependencies=dependencies, + model=model, + middleware_stack=middleware_stack, + ) diff --git a/surfsense_backend/app/agents/chat/multi_agent_chat/subagents/builtins/amazon/description.md b/surfsense_backend/app/agents/chat/multi_agent_chat/subagents/builtins/amazon/description.md new file mode 100644 index 000000000..c66baab39 --- /dev/null +++ b/surfsense_backend/app/agents/chat/multi_agent_chat/subagents/builtins/amazon/description.md @@ -0,0 +1,2 @@ +Amazon product specialist: scrapes public Amazon listings and returns structured product data — title, ASIN, brand, price and list price, star rating and review breakdown, availability, images, features, best-seller ranks, marketplace offers, third-party sellers, product variants, and on-page customer reviews. Works from a search term (e.g. "mechanical keyboard") or from Amazon product, search, category, best-seller, or short (a.co / amzn.to) URLs, on any Amazon marketplace domain. Only public, anonymous data — no login or seller account. Can enrich results with extra offers and seller profiles, expand product variants, and localize pricing/availability by country and ZIP. +Use it for product research, price and buy-box tracking, review mining, catalog enrichment by ASIN, and best-seller/category monitoring. Triggers include "find X on Amazon", "Amazon price of X", "reviews for this Amazon product", "compare these Amazon products", "top-selling X on Amazon", and "look up this ASIN/Amazon URL". Not for general web search (use the Google Search specialist), reading an arbitrary non-Amazon page (use the web crawling specialist), or other marketplaces. diff --git a/surfsense_backend/app/agents/chat/multi_agent_chat/subagents/builtins/amazon/system_prompt.md b/surfsense_backend/app/agents/chat/multi_agent_chat/subagents/builtins/amazon/system_prompt.md new file mode 100644 index 000000000..b18d90f73 --- /dev/null +++ b/surfsense_backend/app/agents/chat/multi_agent_chat/subagents/builtins/amazon/system_prompt.md @@ -0,0 +1,67 @@ +You are the SurfSense Amazon sub-agent. +You receive delegated instructions from a supervisor agent and return structured results for supervisor synthesis. + + +Answer the delegated question from public Amazon product data gathered with your verb, comparing against earlier results already in this conversation when the task calls for it. + + + +- `amazon_scrape` +- `read_run` / `search_run` (free readers for stored scrape output) + + + +- Discovering products: call `amazon_scrape` with `search_terms` (e.g. ["mechanical keyboard"]), setting `domain` when a non-US marketplace matters (e.g. "www.amazon.co.uk"). +- Specific products: pass Amazon product URLs (or search / category / best-seller / short a.co URLs) in `urls`. +- Faster listings: set `include_details=false` to return card-only results without opening each product page. +- Pricing and buy box: raise `max_offers` to pull additional marketplace offers; set `include_sellers=true` to attach seller profiles. +- Variants: raise `max_variants` to return variants as separate results; set `include_variant_prices=true` for per-variant prices. +- Localized pricing/availability: set `country_code` and `zip_code`. +- Batch multiple URLs or search terms into one call rather than many single-source calls. + +- Comparison requests: pull the current products, compare against prior values already in this conversation's earlier tool results, and report concrete deltas (price up/down, rating change, rank moves, stock changes). + + + +- Use only tools in ``. +- Report only results present in the tool output. Never invent titles, ASINs, prices, ratings, or rankings. +- Provide at least one of `urls` or `search_terms`; they cannot be combined arbitrarily beyond the source cap. + + + +- Do not perform general web search — that is the Google Search specialist's job. +- Do not read or extract an arbitrary non-Amazon page — return the URL for the web crawling specialist. +- Do not generate deliverables or perform connector mutations; return findings for the supervisor to act on. +- Only public, anonymous Amazon data — never anything behind a login or seller account. + + + +- Report uncertainty explicitly when evidence is incomplete or conflicting. +- Never present unverified claims as facts. + + + +- Underspecified request — no usable search term or URL — return `status=blocked` with the missing fields. +- Tool failure: return `status=error` with a concise recovery `next_step`. +- No useful evidence: return `status=blocked` with a narrower query or the scope you still need. + + + +Return **only** one JSON object (no markdown/prose): +{ + "status": "success" | "partial" | "blocked" | "error", + "action_summary": string, + "evidence": { + "findings": string[], + "sources": string[], + "confidence": "high" | "medium" | "low" + }, + "next_step": string | null, + "missing_fields": string[] | null, + "assumptions": string[] | null +} + +Route-specific rules: +- `evidence.findings`: max 10 entries, each a single sentence stating one distinct product or delta. Do not paste raw payloads. +- `evidence.sources`: max 10 URLs, one per finding when applicable. List each URL once. + diff --git a/surfsense_backend/app/agents/chat/multi_agent_chat/subagents/builtins/amazon/tools/__init__.py b/surfsense_backend/app/agents/chat/multi_agent_chat/subagents/builtins/amazon/tools/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/surfsense_backend/app/agents/chat/multi_agent_chat/subagents/builtins/amazon/tools/index.py b/surfsense_backend/app/agents/chat/multi_agent_chat/subagents/builtins/amazon/tools/index.py new file mode 100644 index 000000000..70162f8d4 --- /dev/null +++ b/surfsense_backend/app/agents/chat/multi_agent_chat/subagents/builtins/amazon/tools/index.py @@ -0,0 +1,27 @@ +"""``amazon`` sub-agent tools: the Amazon product scrape capability verb.""" + +from __future__ import annotations + +from typing import Any + +from langchain_core.tools import BaseTool + +from app.agents.chat.multi_agent_chat.shared.permissions import Ruleset +from app.capabilities.amazon.scrape.definition import AMAZON_SCRAPE +from app.capabilities.core.access.agent import build_capability_tools + +NAME = "amazon" + +RULESET = Ruleset(origin=NAME, rules=[]) + +_CI_VERBS = [AMAZON_SCRAPE] + + +def load_tools( + *, dependencies: dict[str, Any] | None = None, **kwargs: Any +) -> list[BaseTool]: + d = {**(dependencies or {}), **kwargs} + return build_capability_tools( + workspace_id=d.get("workspace_id"), + capabilities=_CI_VERBS, + ) diff --git a/surfsense_backend/app/agents/chat/multi_agent_chat/subagents/registry.py b/surfsense_backend/app/agents/chat/multi_agent_chat/subagents/registry.py index de37edfb9..7b8b8a6f1 100644 --- a/surfsense_backend/app/agents/chat/multi_agent_chat/subagents/registry.py +++ b/surfsense_backend/app/agents/chat/multi_agent_chat/subagents/registry.py @@ -12,6 +12,9 @@ from langchain_core.tools import BaseTool from app.agents.chat.multi_agent_chat.constants import ( SUBAGENT_TO_REQUIRED_CONNECTOR_MAP, ) +from app.agents.chat.multi_agent_chat.subagents.builtins.amazon.agent import ( + build_subagent as build_amazon_subagent, +) from app.agents.chat.multi_agent_chat.subagents.builtins.deliverables.agent import ( build_subagent as build_deliverables_subagent, ) @@ -80,6 +83,7 @@ class SubagentBuilder(Protocol): SUBAGENT_BUILDERS_BY_NAME: dict[str, SubagentBuilder] = { + "amazon": build_amazon_subagent, "deliverables": build_deliverables_subagent, "dropbox": build_dropbox_subagent, "google_drive": build_google_drive_subagent, diff --git a/surfsense_backend/app/capabilities/__init__.py b/surfsense_backend/app/capabilities/__init__.py index d159536fb..9c97ef59a 100644 --- a/surfsense_backend/app/capabilities/__init__.py +++ b/surfsense_backend/app/capabilities/__init__.py @@ -2,4 +2,6 @@ from __future__ import annotations +from app.capabilities import amazon as _amazon # noqa: F401 + __all__: list[str] = [] diff --git a/surfsense_backend/app/capabilities/amazon/__init__.py b/surfsense_backend/app/capabilities/amazon/__init__.py new file mode 100644 index 000000000..76addfa68 --- /dev/null +++ b/surfsense_backend/app/capabilities/amazon/__init__.py @@ -0,0 +1,5 @@ +"""Amazon capability namespace.""" + +from __future__ import annotations + +from app.capabilities.amazon.scrape import definition as _scrape # noqa: F401 diff --git a/surfsense_backend/app/capabilities/amazon/scrape/__init__.py b/surfsense_backend/app/capabilities/amazon/scrape/__init__.py new file mode 100644 index 000000000..a360208df --- /dev/null +++ b/surfsense_backend/app/capabilities/amazon/scrape/__init__.py @@ -0,0 +1,3 @@ +"""Amazon product scraping capability.""" + +from __future__ import annotations diff --git a/surfsense_backend/app/capabilities/amazon/scrape/definition.py b/surfsense_backend/app/capabilities/amazon/scrape/definition.py new file mode 100644 index 000000000..012d74911 --- /dev/null +++ b/surfsense_backend/app/capabilities/amazon/scrape/definition.py @@ -0,0 +1,22 @@ +"""Registration for the ``amazon.scrape`` capability.""" + +from __future__ import annotations + +from app.capabilities.amazon.scrape.executor import build_scrape_executor +from app.capabilities.amazon.scrape.schemas import ScrapeInput, ScrapeOutput +from app.capabilities.core import BillingUnit, Capability, register_capability + +AMAZON_SCRAPE = Capability( + name="amazon.scrape", + description=( + "Scrape public Amazon product details, search results, offers, sellers, " + "best-seller rankings, and on-page reviews." + ), + input_schema=ScrapeInput, + output_schema=ScrapeOutput, + executor=build_scrape_executor(), + billing_unit=BillingUnit.AMAZON_PRODUCT, + docs_url="/docs/connectors/native/amazon", +) + +register_capability(AMAZON_SCRAPE) diff --git a/surfsense_backend/app/capabilities/amazon/scrape/executor.py b/surfsense_backend/app/capabilities/amazon/scrape/executor.py new file mode 100644 index 000000000..e353c17e4 --- /dev/null +++ b/surfsense_backend/app/capabilities/amazon/scrape/executor.py @@ -0,0 +1,59 @@ +"""Executor for the ``amazon.scrape`` capability.""" + +from __future__ import annotations + +from collections.abc import Awaitable, Callable +from urllib.parse import quote_plus + +from app.capabilities.amazon.scrape.schemas import ( + MAX_AMAZON_RESULTS, + ScrapeInput, + ScrapeOutput, +) +from app.capabilities.core import Executor +from app.capabilities.core.progress import emit_progress +from app.proprietary.platforms.amazon import AmazonScrapeInput, scrape_products + +ScrapeFn = Callable[..., Awaitable[list[dict]]] + + +def build_scrape_executor(scrape_fn: ScrapeFn | None = None) -> Executor: + """Bind the capability input mapping to a replaceable scraper function.""" + scrape_fn = scrape_fn or scrape_products + + async def execute(payload: ScrapeInput) -> ScrapeOutput: + search_urls = [ + f"https://{payload.domain}/s?k={quote_plus(term)}" + for term in payload.search_terms + ] + input_model = AmazonScrapeInput( + categoryOrProductUrls=[ + {"url": url} for url in [*payload.urls, *search_urls] + ], + maxItemsPerStartUrl=payload.max_items, + language=payload.language, + countryCode=payload.country_code, + zipCode=payload.zip_code, + scrapeProductDetails=payload.include_details, + maxOffers=payload.max_offers, + scrapeSellers=payload.include_sellers, + maxProductVariantsAsSeparateResults=payload.max_variants, + scrapeProductVariantPrices=payload.include_variant_prices, + ) + emit_progress( + "starting", + "Scraping Amazon products", + total=payload.estimated_units, + unit="product", + ) + items = await scrape_fn(input_model, limit=MAX_AMAZON_RESULTS) + emit_progress( + "done", + f"Scraped {sum('error' not in item for item in items)} product(s)", + current=len(items), + total=payload.estimated_units, + unit="product", + ) + return ScrapeOutput(items=items) + + return execute diff --git a/surfsense_backend/app/capabilities/amazon/scrape/schemas.py b/surfsense_backend/app/capabilities/amazon/scrape/schemas.py new file mode 100644 index 000000000..e510e2eea --- /dev/null +++ b/surfsense_backend/app/capabilities/amazon/scrape/schemas.py @@ -0,0 +1,61 @@ +"""Input and output contracts for ``amazon.scrape``.""" + +from __future__ import annotations + +from pydantic import BaseModel, Field, model_validator + +from app.proprietary.platforms.amazon import ProductItem + +MAX_AMAZON_SOURCES = 20 +MAX_AMAZON_RESULTS = 1000 + + +class ScrapeInput(BaseModel): + """Agent-facing controls for public product discovery and enrichment.""" + + urls: list[str] = Field(default_factory=list, max_length=MAX_AMAZON_SOURCES) + search_terms: list[str] = Field(default_factory=list, max_length=MAX_AMAZON_SOURCES) + max_items: int = Field(default=10, ge=1, le=100) + domain: str = Field( + default="www.amazon.com", pattern=r"^(?:www\.)?amazon\.[a-z.]+$" + ) + language: str | None = None + country_code: str | None = Field(default=None, min_length=2, max_length=2) + zip_code: str | None = Field(default=None, min_length=1, max_length=20) + include_details: bool = True + max_offers: int = Field(default=0, ge=0, le=100) + include_sellers: bool = False + max_variants: int = Field(default=0, ge=0, le=100) + include_variant_prices: bool = False + + @model_validator(mode="after") + def _require_source(self) -> ScrapeInput: + if not (self.urls or self.search_terms): + raise ValueError("Provide at least one URL or search term.") + if len(self.urls) + len(self.search_terms) > MAX_AMAZON_SOURCES: + raise ValueError( + f"Provide no more than {MAX_AMAZON_SOURCES} combined sources." + ) + return self + + @property + def estimated_units(self) -> int: + """Worst-case returned products within the hard per-run ceiling.""" + search_products = len(self.search_terms) * self.max_items + direct_products = len(self.urls) * (1 + self.max_variants) + return min(search_products + direct_products, MAX_AMAZON_RESULTS) + + +class ScrapeOutput(BaseModel): + """Products and structured per-input errors in emission order.""" + + items: list[ProductItem] = Field(default_factory=list) + + @property + def billable_units(self) -> int: + """Count successful products; error items are never billed.""" + return sum( + 1 + for item in self.items + if not (item.model_extra and item.model_extra.get("error")) + ) diff --git a/surfsense_backend/app/capabilities/core/billing.py b/surfsense_backend/app/capabilities/core/billing.py index 67abf4164..c2c7f6b5b 100644 --- a/surfsense_backend/app/capabilities/core/billing.py +++ b/surfsense_backend/app/capabilities/core/billing.py @@ -33,6 +33,7 @@ _PLATFORM_RATE_KEYS: dict[BillingUnit, str] = { BillingUnit.GOOGLE_SEARCH_SERP: "GOOGLE_SEARCH_MICROS_PER_SERP", BillingUnit.GOOGLE_MAPS_PLACE: "GOOGLE_MAPS_MICROS_PER_PLACE", BillingUnit.GOOGLE_MAPS_REVIEW: "GOOGLE_MAPS_MICROS_PER_REVIEW", + BillingUnit.AMAZON_PRODUCT: "AMAZON_MICROS_PER_PRODUCT", BillingUnit.YOUTUBE_VIDEO: "YOUTUBE_MICROS_PER_VIDEO", BillingUnit.YOUTUBE_COMMENT: "YOUTUBE_MICROS_PER_COMMENT", BillingUnit.INSTAGRAM_ITEM: "INSTAGRAM_SCRAPE_MICROS_PER_ITEM", @@ -54,6 +55,7 @@ _UNIT_NOUNS: dict[BillingUnit, str] = { BillingUnit.GOOGLE_SEARCH_SERP: "SERP", BillingUnit.GOOGLE_MAPS_PLACE: "place", BillingUnit.GOOGLE_MAPS_REVIEW: "review", + BillingUnit.AMAZON_PRODUCT: "product", BillingUnit.YOUTUBE_VIDEO: "video", BillingUnit.YOUTUBE_COMMENT: "comment", BillingUnit.INSTAGRAM_ITEM: "item", diff --git a/surfsense_backend/app/capabilities/core/types.py b/surfsense_backend/app/capabilities/core/types.py index 67b9175fc..2e8d3b6cc 100644 --- a/surfsense_backend/app/capabilities/core/types.py +++ b/surfsense_backend/app/capabilities/core/types.py @@ -23,6 +23,7 @@ class BillingUnit(StrEnum): GOOGLE_SEARCH_SERP = "google_search_serp" GOOGLE_MAPS_PLACE = "google_maps_place" GOOGLE_MAPS_REVIEW = "google_maps_review" + AMAZON_PRODUCT = "amazon_product" YOUTUBE_VIDEO = "youtube_video" YOUTUBE_COMMENT = "youtube_comment" INSTAGRAM_ITEM = "instagram_item" diff --git a/surfsense_backend/app/config/__init__.py b/surfsense_backend/app/config/__init__.py index 92e448d9a..41b94a7e1 100644 --- a/surfsense_backend/app/config/__init__.py +++ b/surfsense_backend/app/config/__init__.py @@ -697,9 +697,8 @@ class Config: # per item; retune with an env change + restart (no code/migration): # = round(USD_per_1000_items * 1_000) # $3.50/1000 -> 3500 | $5.00/1000 -> 5000 | $2.00/1000 -> 2000 - # Defaults sit at/above Apify's first-party actor rates (Jul 2026), which - # is justified because SurfSense charges no subscription tiers, no - # per-run actor-start fees, and no separate proxy/compute/storage billing. + # Defaults include margin for proxy, compute, and storage costs while + # remaining independently adjustable for each platform. PLATFORM_SCRAPE_BILLING_ENABLED = ( os.getenv("PLATFORM_SCRAPE_BILLING_ENABLED", "FALSE").upper() == "TRUE" ) @@ -715,6 +714,7 @@ class Config: GOOGLE_MAPS_MICROS_PER_REVIEW = int( os.getenv("GOOGLE_MAPS_MICROS_PER_REVIEW", "1500") ) + AMAZON_MICROS_PER_PRODUCT = int(os.getenv("AMAZON_MICROS_PER_PRODUCT", "3500")) YOUTUBE_MICROS_PER_VIDEO = int(os.getenv("YOUTUBE_MICROS_PER_VIDEO", "2500")) # Kept separate from the video rate so comments can be re-tuned toward the # cheaper per-comment market ($0.40-2.00/1k) without touching video pricing. diff --git a/surfsense_backend/app/proprietary/platforms/amazon/README.md b/surfsense_backend/app/proprietary/platforms/amazon/README.md new file mode 100644 index 000000000..83cb210ac --- /dev/null +++ b/surfsense_backend/app/proprietary/platforms/amazon/README.md @@ -0,0 +1,50 @@ +# Amazon Product Scraper + +Current status: product details, search and category discovery, offers, sellers, +best-seller rankings, on-page reviews, localized delivery sessions, and product +variants are implemented. The `amazon.scrape` capability exposes the scraper +with per-product billing and a hard per-run result ceiling. + +## Scope + +The scraper reads public pages available to anonymous visitors. It does not log +in, use account cookies, or retrieve account-gated content. Delivery +localization uses an anonymous session cookie pinned to the same proxy exit. + +Deep review pagination currently requires an account on Amazon and is therefore +out of scope. Reviews embedded in a public product page are returned in +`productPageReviews` and `productPageReviewsFromOtherCountries`. + +## Architecture + +- `schemas.py` defines the stable input, product, and error models. +- `url_resolver.py` classifies product, search, best-sellers, and shortened URLs. +- `fetch.py` owns proxy-aware HTTP access, block detection, retries, and + anonymous location sessions. +- `parsers.py` contains pure, defensive HTML parsers. +- `scraper.py` coordinates discovery, enrichment, concurrency, limits, and + in-stream errors. + +## Implementation progress + +- Done: public product detail parsing and shortened-link dispatch. +- Done: search/category paging with card-only and detailed modes. +- Done: public offers and seller enrichment. +- Done: best-seller rankings and product-page reviews. +- Done: anonymous localized delivery sessions on sticky proxy exits. +- Done: variant expansion, variant prices, capability registration, and billing. + +## Verification + +Offline fixtures cover every parser and flow: + +```bash +cd surfsense_backend +uv run pytest tests/unit/platforms/amazon tests/unit/capabilities/amazon +``` + +The manual live check requires proxy credentials: + +```bash +uv run python scripts/e2e_amazon_scraper.py +``` diff --git a/surfsense_backend/app/proprietary/platforms/amazon/__init__.py b/surfsense_backend/app/proprietary/platforms/amazon/__init__.py new file mode 100644 index 000000000..e2cf8cb7d --- /dev/null +++ b/surfsense_backend/app/proprietary/platforms/amazon/__init__.py @@ -0,0 +1,12 @@ +"""Platform-native Amazon Product Scraper.""" + +from .schemas import AmazonScrapeInput, ErrorItem, ProductItem +from .scraper import iter_products, scrape_products + +__all__ = [ + "AmazonScrapeInput", + "ErrorItem", + "ProductItem", + "iter_products", + "scrape_products", +] diff --git a/surfsense_backend/app/proprietary/platforms/amazon/fetch.py b/surfsense_backend/app/proprietary/platforms/amazon/fetch.py new file mode 100644 index 000000000..2593fcd48 --- /dev/null +++ b/surfsense_backend/app/proprietary/platforms/amazon/fetch.py @@ -0,0 +1,388 @@ +"""Network access for public Amazon pages. + +All requests use the configured residential proxy. Responses that contain an +anti-bot interstitial are retried with a fresh proxy exit, while ordinary HTTP +failures are returned to the caller for domain-specific error handling. +""" + +from __future__ import annotations + +import asyncio +import logging +import re +import time +from collections.abc import Awaitable, Callable +from dataclasses import dataclass, field +from typing import Any + +from scrapling.fetchers import AsyncFetcher + +from app.utils.proxy import get_geo_proxy_url, get_sticky_proxy_url + +logger = logging.getLogger(__name__) + +_MAX_IP_ATTEMPTS = 8 +_REQUEST_TIMEOUT_S = 30 +_HEADERS = { + "Accept-Language": "en-US,en;q=0.9", + "Accept": "text/html,application/xhtml+xml,application/json;q=0.9,*/*;q=0.8", +} +_BLOCK_MARKERS = ( + "/errors/validatecaptcha", + "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( + r"""(?:name|id)=["'](?:anti-csrftoken-a2z|csrf-token)["'][^>]*""" + r"""(?:value|content)=["']([^"']+)""", + re.IGNORECASE, + ), + re.compile( + r"""["'](?:anti-csrftoken-a2z|csrf-token)["']\s*[:=]\s*["']([^"']+)""", + re.IGNORECASE, + ), +) + + +@dataclass(frozen=True) +class FetchResult: + """The response details needed by scraper flows.""" + + status: int + html: str + url: str + cookies: dict[str, str] + headers: dict[str, str] = field(default_factory=dict) + + +@dataclass +class LocationSession: + """Anonymous delivery-location state pinned to one proxy exit.""" + + proxy: str | None + cookies: dict[str, str] + country_code: str | None + zip_code: str + location_text: str | None + created_at: float + + +async def gather_bounded[T]( + factories: list[Callable[[], Awaitable[T]]], *, concurrency: int +) -> list[T]: + """Run async factories concurrently while preserving input order.""" + if not factories: + return [] + semaphore = asyncio.Semaphore(max(1, concurrency)) + + async def run(factory: Callable[[], Awaitable[T]]) -> T: + async with semaphore: + return await factory() + + return await asyncio.gather(*(run(factory) for factory in factories)) + + +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 + empty/accepted body rather than the page), so it is treated as blocked and + retried on a fresh exit rather than parsed as a real page. + """ + 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 + + +def _response_cookies(page: Any) -> dict[str, str]: + cookies = getattr(page, "cookies", None) + 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, + rotate_on_block: bool = True, +) -> FetchResult | None: + """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 = _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": request_headers, + "cookies": cookies or {}, + "proxy": selected_proxy, + "stealthy_headers": True, + "timeout": _REQUEST_TIMEOUT_S, + } + if method == "POST": + kwargs["data"] = data or {} + page = await request(url, **kwargs) + except Exception as exc: + logger.warning("Amazon request failed for %s: %s", url, exc) + if proxy is not None: + return None + continue + + 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, + status, + attempt, + (time.perf_counter() - started) * 1000, + url, + ) + if rotate_on_block and is_blocked(html, status, response_headers): + logger.info( + "Amazon blocked proxy attempt %s/%s for %s", attempt, attempts, url + ) + continue + return FetchResult( + status=status, + 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 + + +async def fetch_html( + url: str, + *, + 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, + 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, *, 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, country=country, accept_language=accept_language) + return result.url if result is not None and result.status == 200 else None + + +def _origin(domain: str) -> str: + return f"https://{domain}" + + +def _csrf_token(html: str) -> str | None: + for pattern in _CSRF_PATTERNS: + match = pattern.search(html) + if match: + return match.group(1) + return None + + +async def fetch_aod_html( + asin: str, + domain: str, + *, + 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. + + ``aodAjaxMain`` is a path segment, not a query param. Amazon also serves this + panel as a JS-only modal for many (especially US) ASINs, so a 404 here is + 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, + country=country, + accept_language=accept_language, + ) + + +async def fetch_seller_html( + seller_id: str, + domain: str, + *, + 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, + country=country, + accept_language=accept_language, + ) + + +_LOCATION_SESSION_TTL_S = 30 * 60 +_location_sessions: dict[tuple[str, str, str | None], LocationSession] = {} +_location_locks: dict[tuple[str, str, str | None], asyncio.Lock] = {} + + +def should_localize(route: str, deliverable_routes: list[str] | None) -> bool: + """Return whether a request route should use delivery-location state.""" + return route.upper() in {value.upper() for value in (deliverable_routes or [])} + + +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: + return get_sticky_proxy_url(session_id, country) + except (ImportError, NotImplementedError): + return get_geo_proxy_url(country) + + +async def get_location_session( + domain: str, + *, + 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) + cached = _location_sessions.get(key) + if cached and time.time() - cached.created_at < _LOCATION_SESSION_TTL_S: + return cached + + lock = _location_locks.setdefault(key, asyncio.Lock()) + async with lock: + cached = _location_sessions.get(key) + if cached and time.time() - cached.created_at < _LOCATION_SESSION_TTL_S: + return cached + + session_id = f"amazon-{abs(hash(key)) & 0xFFFFFFFF:x}" + 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: + return None + csrf = _csrf_token(home.html) + if csrf is None: + logger.warning("Amazon location session did not expose a CSRF token") + return None + + endpoint = f"{_origin(domain)}/gp/delivery/ajax/address-change.html" + payload = { + "locationType": "LOCATION_INPUT", + "zipCode": zip_code, + "storeContext": "generic", + "deviceType": "web", + "pageType": "Gateway", + "actionSource": "glow", + } + cookies = dict(home.cookies) + cookies["anti-csrftoken-a2z"] = csrf + changed = await fetch_page( + endpoint, + cookies=cookies, + proxy=proxy, + method="POST", + data=payload, + headers={ + "anti-csrftoken-a2z": csrf, + "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: + return None + cookies.update(changed.cookies) + location_text = zip_code + session = LocationSession( + proxy=proxy, + cookies=cookies, + country_code=country_code.upper() if country_code else None, + zip_code=zip_code, + location_text=location_text, + created_at=time.time(), + ) + _location_sessions[key] = session + return session diff --git a/surfsense_backend/app/proprietary/platforms/amazon/locale.py b/surfsense_backend/app/proprietary/platforms/amazon/locale.py new file mode 100644 index 000000000..1e8b7b880 --- /dev/null +++ b/surfsense_backend/app/proprietary/platforms/amazon/locale.py @@ -0,0 +1,33 @@ +"""Marketplace locale helpers for Amazon fetch routing. + +Only proxy countries verified against the configured DataImpulse plan are mapped +here. Unmapped marketplaces fall back to the default proxy exit. +""" + +_MARKETPLACE_COUNTRY = { + "com": "us", + "co.uk": "gb", + "de": "de", + "fr": "fr", + "it": "it", + "es": "es", +} + +_MARKETPLACE_LANGUAGE = { + "com": "en-US", + "co.uk": "en-GB", + "de": "de-DE", + "fr": "fr-FR", + "it": "it-IT", + "es": "es-ES", +} + + +def proxy_country_for(marketplace: str | None) -> str | None: + """Return a confirmed proxy exit country for one Amazon marketplace.""" + return _MARKETPLACE_COUNTRY.get((marketplace or "").lower()) + + +def accept_language_for(marketplace: str | None) -> str: + """Return the browser language header that matches the marketplace UI.""" + return _MARKETPLACE_LANGUAGE.get((marketplace or "").lower(), "en-US") diff --git a/surfsense_backend/app/proprietary/platforms/amazon/parsers.py b/surfsense_backend/app/proprietary/platforms/amazon/parsers.py new file mode 100644 index 000000000..4ee722637 --- /dev/null +++ b/surfsense_backend/app/proprietary/platforms/amazon/parsers.py @@ -0,0 +1,586 @@ +"""Pure parsers for public Amazon HTML. + +Selectors cover stable element IDs first and use small, explicit fallbacks for +layout variants. Missing sections return empty or nullable values so isolated +markup changes do not discard an otherwise usable product. +""" + +from __future__ import annotations + +import json +import re +from html import unescape +from typing import Any +from urllib.parse import parse_qs, urljoin, urlparse + +from scrapling.parser import Adaptor + +_NUMBER_RE = re.compile(r"[\d,.]+") +_ASIN_RE = re.compile(r"^[A-Z0-9]{10}$") +_ASIN_IN_TEXT_RE = re.compile(r"\b[A-Z0-9]{10}\b") +_CURRENCY_BY_SYMBOL = {"$": "USD", "€": "EUR", "£": "GBP", "¥": "JPY", "₹": "INR"} +_JSON_ASSIGNMENT_RE = re.compile( + r"(?:dimensionValuesDisplayData|variationValues|dimensionToAsinMap)" + r"""\s*["']?\s*:\s*(\{.*?\})(?:,\s*["']|\s*;)""", + re.DOTALL, +) + + +def _one(node: Any, selector: str) -> Any | None: + found = node.css(selector) + return found[0] if found else None + + +def _text(node: Any | None) -> str | None: + if node is None: + return None + raw = node.get_all_text(strip=True) + return re.sub(r"\s+", " ", raw).strip() or None + + +def _texts(node: Any, selector: str) -> list[str]: + values: list[str] = [] + for match in node.css(selector): + value = _text(match) + if value and value not in values: + values.append(value) + return values + + +def _first_text(node: Any, *selectors: str) -> str | None: + for selector in selectors: + value = _text(_one(node, selector)) + if value: + return value + return None + + +def _attr(node: Any | None, name: str) -> str | None: + if node is None: + return None + value = node.attrib.get(name) + return str(value).strip() if value else None + + +def _first_attr(node: Any, name: str, *selectors: str) -> str | None: + for selector in selectors: + value = _attr(_one(node, selector), name) + if value: + return value + return None + + +def _integer(value: str | None) -> int | None: + if not value: + return None + match = _NUMBER_RE.search(value) + if not match: + return None + digits = re.sub(r"\D", "", match.group(0)) + return int(digits) if digits else None + + +def _float(value: str | None) -> float | None: + if not value: + return None + match = _NUMBER_RE.search(value) + if not match: + return None + token = match.group(0) + if "," in token and "." in token: + decimal = "," if token.rfind(",") > token.rfind(".") else "." + grouping = "." if decimal == "," else "," + token = token.replace(grouping, "").replace(decimal, ".") + elif ( + token.count(",") == 1 and "." not in token and len(token.rsplit(",", 1)[1]) <= 2 + ): + token = token.replace(",", ".") + else: + token = token.replace(",", "") + try: + return float(token) + except ValueError: + return None + + +def _price(value: str | None) -> dict[str, Any] | None: + amount = _float(value) + if amount is None: + return None + currency = next( + ( + code + for symbol, code in _CURRENCY_BY_SYMBOL.items() + if symbol in (value or "") + ), + None, + ) + if currency is None and value: + match = re.search(r"\b(USD|EUR|GBP|JPY|INR|CAD|AUD)\b", value) + currency = match.group(1) if match else None + return {"value": amount, "currency": currency} + + +def _absolute(domain: str, href: str | None) -> str | None: + return urljoin(f"https://{domain}", href) if href else None + + +def _asin_from_href(href: str | None) -> str | None: + if not href: + return None + match = re.search(r"/(?:dp|gp/product)/([A-Z0-9]{10})(?:[/?]|$)", href) + return match.group(1) if match else None + + +def _embedded_json(html: str, key: str) -> Any | None: + """Decode a JSON value assigned to ``key`` inside an inline script.""" + marker = re.search(rf"""["']?{re.escape(key)}["']?\s*:\s*""", html) + if marker is None: + return None + start = marker.end() + while start < len(html) and html[start].isspace(): + start += 1 + if start >= len(html) or html[start] not in "[{": + return None + opening = html[start] + closing = "}" if opening == "{" else "]" + depth = 0 + quoted = False + escaped = False + for index in range(start, len(html)): + char = html[index] + if quoted: + if escaped: + escaped = False + elif char == "\\": + escaped = True + elif char == '"': + quoted = False + continue + if char == '"': + quoted = True + elif char == opening: + depth += 1 + elif char == closing: + depth -= 1 + if depth == 0: + try: + return json.loads(html[start : index + 1]) + except json.JSONDecodeError: + return None + return None + + +def _stars_breakdown(doc: Adaptor) -> dict[str, float | None] | None: + breakdown: dict[str, float | None] = {} + for row in doc.css("#histogramTable tr, [data-hook='rating-histogram'] tr"): + label = _first_text(row, ".a-text-left", "td:first-child", "span") + percent = _first_text(row, ".a-text-right", "td:last-child") + star = _integer(label) + value = _float(percent) + if star and 1 <= star <= 5 and value is not None: + breakdown[f"{star}star"] = value / 100 if value > 1 else value + return breakdown or None + + +def _variant_data( + html: str, +) -> tuple[list[str], list[dict[str, Any]], list[dict[str, Any]]]: + display = _embedded_json(html, "dimensionValuesDisplayData") + variants = _embedded_json(html, "variationValues") + asins: list[str] = [] + details: list[dict[str, Any]] = [] + attributes: list[dict[str, Any]] = [] + + if isinstance(display, dict): + for asin, values in display.items(): + if _ASIN_RE.match(str(asin)): + asins.append(str(asin)) + details.append({"asin": str(asin), "values": values}) + if isinstance(variants, dict): + for name, values in variants.items(): + attributes.append({"name": name, "values": values}) + for value in variants.values(): + if isinstance(value, list): + for candidate in value: + if isinstance(candidate, str) and _ASIN_RE.match(candidate): + asins.append(candidate) + return list(dict.fromkeys(asins)), details, attributes + + +def _attributes(doc: Adaptor) -> tuple[list[dict[str, str]], dict[str, str]]: + pairs: list[dict[str, str]] = [] + mapped: dict[str, str] = {} + for row in doc.css( + "#productDetails_techSpec_section_1 tr, " + "#productDetails_detailBullets_sections1 tr, " + "#detailBullets_feature_div li" + ): + name = _first_text(row, "th", ".a-text-bold") + value = _first_text(row, "td", "span:not(.a-text-bold)") + if name and value: + name = name.rstrip(": \u200e") + pairs.append({"name": name, "value": value}) + mapped[name] = value + return pairs, mapped + + +def _reviews( + doc: Adaptor, domain: str +) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: + local: list[dict[str, Any]] = [] + foreign: list[dict[str, Any]] = [] + for node in doc.css("[data-hook='review']"): + review = { + "id": _attr(node, "id"), + "title": _first_text(node, "[data-hook='review-title']"), + "stars": _float(_first_text(node, "[data-hook='review-star-rating']")), + "text": _first_text(node, "[data-hook='review-body']"), + "author": _first_text(node, ".a-profile-name"), + "date": _first_text(node, "[data-hook='review-date']"), + "verified": _one(node, "[data-hook='avp-badge']") is not None, + "helpfulVotes": _integer( + _first_text(node, "[data-hook='helpful-vote-statement']") + ), + } + if any(value is not None for value in review.values()): + target = ( + foreign + if "other countries" in (_text(node) or "").lower() + or "cm_cr_arp_d_rvw_rvwer" in (_attr(node, "class") or "") + else local + ) + target.append(review) + return local, foreign + + +def parse_product( + html: str, *, asin: str, url: str, domain: str | None = None +) -> dict[str, Any]: + """Parse one product detail page into output-model fields.""" + doc = Adaptor(html) + domain = domain or (urlparse(url).hostname or "www.amazon.com") + title = _first_text(doc, "#productTitle", "#title") + price_text = _first_text( + doc, + "#corePrice_feature_div .a-offscreen", + "#corePriceDisplay_desktop_feature_div .a-offscreen", + "#apex_desktop .a-price .a-offscreen", + "#priceblock_ourprice", + "#priceblock_dealprice", + ".priceToPay .a-offscreen", + "#price_inside_buybox", + "#centerCol .a-price .a-offscreen", + ) + list_price_text = _first_text( + doc, + "#corePrice_feature_div .basisPrice .a-offscreen", + "#corePriceDisplay_desktop_feature_div .basisPrice .a-offscreen", + ".priceBlockStrikePriceString", + "#centerCol .a-price.a-text-price .a-offscreen", + ) + availability = _first_text(doc, "#availability", "#outOfStock") + availability_lower = (availability or "").lower() + in_stock = ( + None + if availability is None + else not any( + word in availability_lower for word in ("unavailable", "out of stock") + ) + ) + brand_link = _one(doc, "#bylineInfo") + brand_text = _text(brand_link) + brand = ( + re.sub(r"^(visit the |brand:\s*)| store$", "", brand_text, flags=re.IGNORECASE) + if brand_text + else None + ) + variant_asins, variant_details, variant_attributes = _variant_data(html) + attrs, attrs_mapped = _attributes(doc) + local_reviews, foreign_reviews = _reviews(doc, domain) + + high_res: list[str] = [] + gallery: list[str] = [] + for image in doc.css("#altImages img, #landingImage, #imgTagWrapperId img"): + thumb = _attr(image, "src") + if thumb and thumb not in gallery: + gallery.append(thumb) + dynamic = _attr(image, "data-a-dynamic-image") + if dynamic: + try: + decoded = json.loads(unescape(dynamic)) + for candidate in decoded: + if candidate not in high_res: + high_res.append(candidate) + except (json.JSONDecodeError, TypeError): + pass + zoom = _attr(image, "data-old-hires") + if zoom and zoom not in high_res: + high_res.append(zoom) + + ranks: list[dict[str, Any]] = [] + for node in doc.css( + "#detailBulletsWrapper_feature_div li, #productDetails_detailBullets_sections1 tr" + ): + text = _text(node) + if not text or "best sellers rank" not in text.lower(): + continue + for rank, category in re.findall(r"#([\d,]+)\s+in\s+([^#(]+)", text): + ranks.append( + {"rank": int(rank.replace(",", "")), "category": category.strip()} + ) + + aplus = _one(doc, "#aplus, #aplus_feature_div") + brand_story = _one(doc, "#brandStory_feature_div, .apm-brand-story") + store_href = _attr(brand_link, "href") + reviews_link = _first_attr( + doc, "href", "#acrCustomerReviewLink", "[data-hook='see-all-reviews-link-foot']" + ) + seller_link = _one( + doc, + "#sellerProfileTriggerId, #merchant-info a[href*='seller='], a[href*='/sp?seller=']", + ) + seller_href = _attr(seller_link, "href") + seller_id = (parse_qs(urlparse(seller_href or "").query).get("seller") or [None])[0] + # Fall back to the buy-box "Sold by" name when there is no seller profile link + # (Amazon-direct items and some link-less third-party merchants). + seller_name = _text(seller_link) or _first_text( + doc, + "#tabular-buybox [tabular-attribute-name='Sold by'] .tabular-buybox-text", + "#merchantInfoFeature_feature_div .offer-display-feature-text-message", + ) + return { + "title": title, + "url": url, + "asin": asin, + "originalAsin": asin, + "brand": brand, + "author": _first_text( + doc, ".author", "#bylineInfo_feature_div .contributorNameID" + ), + "price": _price(price_text), + "listPrice": _price(list_price_text), + "shippingPrice": _price( + _first_text(doc, "#deliveryBlockMessage .a-color-secondary") + ), + "inStock": in_stock, + "inStockText": availability, + "delivery": _first_text( + doc, "#mir-layout-DELIVERY_BLOCK-slot-PRIMARY_DELIVERY_MESSAGE_LARGE" + ), + "fastestDelivery": _first_text( + doc, "#mir-layout-DELIVERY_BLOCK-slot-SECONDARY_DELIVERY_MESSAGE_LARGE" + ), + "condition": _first_text( + doc, "#buyNew_noncbb .a-color-price", "#usedAccordionRow" + ), + "stars": _float( + _first_text(doc, "#acrPopover", "[data-hook='rating-out-of-text']") + ), + "starsBreakdown": _stars_breakdown(doc), + "reviewsCount": _integer(_first_text(doc, "#acrCustomerReviewText")), + "answeredQuestions": _integer(_first_text(doc, "#askATFLink")), + "aiReviewsSummary": ( + { + "text": _first_text( + doc, "#product-summary, [data-hook='cr-insights-widget']" + ) + } + if _first_text(doc, "#product-summary, [data-hook='cr-insights-widget']") + else None + ), + "monthlyPurchaseVolume": _first_text( + doc, "#social-proofing-faceout-title-tk_bought" + ), + "breadCrumbs": " > ".join( + _texts(doc, "#wayfinding-breadcrumbs_feature_div li a") + ) + or None, + "description": _first_text( + doc, "#productDescription", "#bookDescription_feature_div" + ), + "features": _texts(doc, "#feature-bullets li span.a-list-item"), + "sustainabilityFeatures": [ + {"text": value} + for value in _texts(doc, "#sustainability_feature_div .a-list-item") + ], + "videosCount": _integer(_first_text(doc, "#videoCount")), + "visitStoreLink": ( + {"text": brand_text, "url": _absolute(domain, store_href)} + if store_href + else None + ), + "thumbnailImage": _first_attr(doc, "src", "#landingImage", "#imgBlkFront"), + "galleryThumbnails": gallery, + "highResolutionImages": high_res, + "aPlusContent": ( + { + "text": _text(aplus), + "images": [ + _attr(img, "src") for img in aplus.css("img") if _attr(img, "src") + ], + } + if aplus is not None + else None + ), + "brandStory": {"text": _text(brand_story)} if brand_story is not None else None, + "returnPolicy": _first_text( + doc, "#returnsInfoFeature_feature_div", "#RETURNS_POLICY" + ), + "support": _first_text(doc, "#support_feature_div"), + "variantAsins": variant_asins, + "variantDetails": variant_details, + "variantAttributes": variant_attributes, + "attributes": attrs, + "attributesMapped": attrs_mapped or None, + "productOverview": [ + { + "name": _first_text(row, "td:first-child"), + "value": _first_text(row, "td:last-child"), + } + for row in doc.css("#productOverview_feature_div tr") + if _first_text(row, "td:first-child") + ], + "seller": ( + { + "id": seller_id, + "name": seller_name, + "url": _absolute(domain, seller_href), + } + if (seller_link is not None or seller_name) + else None + ), + "bestsellerRanks": ranks, + "isAmazonChoice": _one(doc, "#acBadge_feature_div, .ac-badge-wrapper") + is not None, + "amazonChoiceText": _first_text(doc, "#acBadge_feature_div, .ac-badge-wrapper"), + "reviewsLink": _absolute(domain, reviews_link), + "hasReviews": bool(local_reviews or foreign_reviews), + "productPageReviews": local_reviews, + "productPageReviewsFromOtherCountries": foreign_reviews, + } + + +def parse_search_page( + html: str, *, page: int = 1, domain: str = "www.amazon.com" +) -> list[dict[str, Any]]: + """Parse product cards from a search or category page.""" + doc = Adaptor(html) + cards: list[dict[str, Any]] = [] + for position, node in enumerate( + doc.css( + "[data-component-type='s-search-result'][data-asin], [data-asin].s-result-item" + ), + start=1, + ): + asin = (_attr(node, "data-asin") or "").upper() + if not _ASIN_RE.match(asin): + continue + href = _first_attr(node, "href", "h2 a", "a.a-link-normal") + title = _first_text(node, "h2", "h2 span", ".a-size-base-plus") + cards.append( + { + "asin": asin, + "originalAsin": asin, + "title": title, + "url": _absolute(domain, href) or f"https://{domain}/dp/{asin}", + "price": _price(_first_text(node, ".a-price .a-offscreen")), + "stars": _float(_first_text(node, ".a-icon-alt")), + "reviewsCount": _integer( + _first_text(node, "[aria-label$='ratings']", ".s-underline-text") + ), + "thumbnailImage": _first_attr(node, "src", ".s-image"), + "categoryPageData": { + "position": position, + "page": page, + "isSponsored": "sponsored" in (_text(node) or "").lower(), + "isBestSeller": _one(node, ".a-badge-text, .s-badge-text") + is not None, + }, + } + ) + return cards + + +def parse_aod_offers( + html: str, *, domain: str = "www.amazon.com" +) -> list[dict[str, Any]]: + """Parse rows from the public all-offers panel.""" + doc = Adaptor(html) + offers: list[dict[str, Any]] = [] + for position, node in enumerate( + doc.css("#aod-offer, .aod-information-block"), start=1 + ): + seller_link = _one( + node, "#aod-offer-soldBy a[href*='seller='], a[href*='/sp?']" + ) + href = _attr(seller_link, "href") + query = parse_qs(urlparse(href or "").query) + seller_id = (query.get("seller") or [None])[0] + offers.append( + { + "position": position, + "price": _price(_first_text(node, ".a-price .a-offscreen")), + "condition": _first_text( + node, "#aod-offer-heading", ".aod-information-block" + ), + "delivery": _first_text(node, "#mir-layout-DELIVERY_BLOCK"), + "seller": { + "id": seller_id, + "name": _text(seller_link) + or _first_text(node, "#aod-offer-soldBy .a-color-base"), + "url": _absolute(domain, href), + }, + "isPinnedOffer": _one(node, "#aod-pinned-offer, .aod-pinned-offer") + is not None, + } + ) + return offers + + +def parse_seller( + html: str, *, seller_id: str | None = None, domain: str = "www.amazon.com" +) -> dict[str, Any]: + """Parse the public summary of a seller profile.""" + doc = Adaptor(html) + return { + "id": seller_id, + "name": _first_text(doc, "#sellerName", "#seller-profile-container h1", "h1"), + "url": f"https://{domain}/sp?seller={seller_id}" if seller_id else None, + "reviewsCount": _integer( + _first_text(doc, "#seller-feedback-summary", "#feedback-summary-table") + ), + "averageRating": _float( + _first_text(doc, "#feedback-summary-table .a-icon-alt", ".a-icon-star") + ), + } + + +def parse_bestsellers_page( + html: str, *, page: int = 1, domain: str = "www.amazon.com" +) -> list[dict[str, Any]]: + """Parse ranked products from a best-sellers page.""" + doc = Adaptor(html) + items: list[dict[str, Any]] = [] + nodes = doc.css("#gridItemRoot, .zg-grid-general-faceout, [id^='p13n-asin-index-']") + for fallback_rank, node in enumerate(nodes, start=1): + href = _first_attr(node, "href", "a[href*='/dp/']", "a[href*='/gp/product/']") + asin = _asin_from_href(href) or (_attr(node, "data-asin") or None) + if not asin or not _ASIN_RE.match(asin): + continue + rank = _integer(_first_text(node, ".zg-bdg-text")) or fallback_rank + items.append( + { + "asin": asin, + "originalAsin": asin, + "title": _first_text( + node, "._cDEzb_p13n-sc-css-line-clamp-3_g3dy1", "img" + ), + "url": _absolute(domain, href) or f"https://{domain}/dp/{asin}", + "price": _price(_first_text(node, ".a-price .a-offscreen")), + "stars": _float(_first_text(node, ".a-icon-alt")), + "thumbnailImage": _first_attr(node, "src", "img"), + "bestsellerPageData": {"rank": rank, "page": page}, + } + ) + return items diff --git a/surfsense_backend/app/proprietary/platforms/amazon/schemas.py b/surfsense_backend/app/proprietary/platforms/amazon/schemas.py new file mode 100644 index 000000000..4ba0b239f --- /dev/null +++ b/surfsense_backend/app/proprietary/platforms/amazon/schemas.py @@ -0,0 +1,245 @@ +# ruff: noqa: N815 +"""Input/output models for the Amazon Product Scraper. + +The skeleton accepts the full input surface; output fields the implementation +does not source yet are emitted as ``None``/``[]`` so parity is additive — +exactly like the YouTube / Maps / Google Search models. + +Outputs use ``extra="allow"`` on purpose: it lets the output shape grow without +breaking existing consumers. Only a small set of stable, always-populated nested +objects (``price``, ``starsBreakdown``, ``seller``, ``visitStoreLink``) are +typed; the sprawling, layout-volatile sections (A+ content, brand story, +attribute tables, offers, on-page reviews, ...) stay loose ``dict``/``list`` and +lean on ``extra="allow"`` rather than pinning a shape the parsers haven't +verified against live HTML yet. + +Public, anonymous data only: there is no auth/login field on the input surface. +Deep review pagination is login-gated on today's Amazon and is out of scope; the +on-page ``productPageReviews`` are the only reviews modeled here. +""" + +from __future__ import annotations + +from typing import Any, Literal + +from pydantic import BaseModel, ConfigDict, Field + +# --------------------------------------------------------------------------- # +# Typed nested objects (the stable, always-populated ones). # +# --------------------------------------------------------------------------- # + + +class Price(BaseModel): + """A monetary amount as Amazon renders it (``price``, ``listPrice``, + ``shippingPrice``, and the ``min``/``max`` of ``priceRange``).""" + + model_config = ConfigDict(extra="allow") + + value: float | None = None + currency: str | None = None + + +class StarsBreakdown(BaseModel): + """Per-star rating distribution (fractions summing to ~1.0). + + Amazon's keys start with a digit (``5star`` ... ``1star``), which is not a + valid Python identifier, so each field is aliased. ``populate_by_name`` lets + tests construct it either way; ``to_output`` serializes ``by_alias`` so the + wire shape retains the digit-prefixed keys. + """ + + model_config = ConfigDict(extra="allow", populate_by_name=True) + + five_star: float | None = Field(default=None, alias="5star") + four_star: float | None = Field(default=None, alias="4star") + three_star: float | None = Field(default=None, alias="3star") + two_star: float | None = Field(default=None, alias="2star") + one_star: float | None = Field(default=None, alias="1star") + + +class Seller(BaseModel): + """The featured-offer seller stamped on the product item.""" + + model_config = ConfigDict(extra="allow") + + id: str | None = None + name: str | None = None + url: str | None = None + reviewsCount: int | None = None + averageRating: float | None = None + + +class VisitStoreLink(BaseModel): + """The brand-store link (``visitStoreLink``).""" + + model_config = ConfigDict(extra="allow") + + text: str | None = None + url: str | None = None + + +# --------------------------------------------------------------------------- # +# Input surface. # +# --------------------------------------------------------------------------- # + + +class AmazonScrapeInput(BaseModel): + """Full input surface for the Amazon Product Scraper. + + ``categoryOrProductUrls`` mixes category/search, product, bestsellers, and + shortened URLs; ``proxyCountry`` defaults to ``AUTO_SELECT_PROXY_COUNTRY`` + (derived from each URL's domain); ``scrapeProductDetails`` defaults on (deep + scrape). ``extra="allow"`` keeps a verbatim payload valid even for add-ons + this scraper does not model. + """ + + model_config = ConfigDict(extra="allow") + + # Discovery (required) + categoryOrProductUrls: list[dict] = Field(min_length=1) + + # Result caps + maxItemsPerStartUrl: int | None = Field(default=None, ge=0) + maxSearchPagesPerStartUrl: int = Field(default=9999, ge=1) + maxProductVariantsAsSeparateResults: int = Field(default=0, ge=0) + maxOffers: int = Field(default=0, ge=0) + + # Localization + language: str | None = None + proxyCountry: str = "AUTO_SELECT_PROXY_COUNTRY" + countryCode: str | None = None + zipCode: str | None = None + locationDeliverableRoutes: list[str] | None = Field( + default_factory=lambda: ["PRODUCT", "SEARCH", "OFFERS"] + ) + + # Feature toggles + scrapeSellers: bool = False + useCaptchaSolver: bool = False + scrapeProductVariantPrices: bool = False + scrapeProductDetails: bool | None = True + + +# --------------------------------------------------------------------------- # +# Output item. # +# --------------------------------------------------------------------------- # + + +class ProductItem(BaseModel): + """Amazon Product Scraper output item (one per product). + + Unsourced fields default to ``None``/``[]``; ``extra="allow"`` keeps the + contract open so later milestones add fields without breaking consumers. + """ + + model_config = ConfigDict(extra="allow", populate_by_name=True) + + # Identity / core + title: str | None = None + url: str | None = None + asin: str | None = None + originalAsin: str | None = None + brand: str | None = None + author: str | None = None + price: Price | None = None + listPrice: Price | None = None + shippingPrice: Price | None = None + inStock: bool | None = None + inStockText: str | None = None + delivery: str | None = None + fastestDelivery: str | None = None + condition: str | None = None + + # Ratings / social + stars: float | None = None + starsBreakdown: StarsBreakdown | None = None + reviewsCount: int | None = None + answeredQuestions: int | None = None + aiReviewsSummary: dict | None = None + monthlyPurchaseVolume: str | None = None + + # Content + breadCrumbs: str | None = None + description: str | None = None + features: list[str] = Field(default_factory=list) + sustainabilityFeatures: list[dict] = Field(default_factory=list) + videosCount: int | None = None + visitStoreLink: VisitStoreLink | None = None + thumbnailImage: str | None = None + galleryThumbnails: list[str] = Field(default_factory=list) + highResolutionImages: list[str] = Field(default_factory=list) + importantInformation: dict | None = None + bookDescription: str | None = None + aPlusContent: dict | None = None + brandStory: dict | None = None + productComparison: dict | None = None + + # Commerce + returnPolicy: str | None = None + support: str | None = None + priceRange: dict | None = None + variantAsins: list[str] = Field(default_factory=list) + variantDetails: list[dict] = Field(default_factory=list) + variantAttributes: list[dict] = Field(default_factory=list) + attributes: list[dict] = Field(default_factory=list) + attributesMapped: dict | None = None + productOverview: list[dict] = Field(default_factory=list) + manufacturerAttributes: list[dict] = Field(default_factory=list) + seller: Seller | None = None + bestsellerRanks: list[dict] = Field(default_factory=list) + isAmazonChoice: bool | None = None + amazonChoiceText: str | None = None + offers: list[dict] = Field(default_factory=list) + + # Reviews (public, on-page only) + reviewsLink: str | None = None + hasReviews: bool | None = None + productPageReviews: list[dict] = Field(default_factory=list) + productPageReviewsFromOtherCountries: list[dict] = Field(default_factory=list) + + # Provenance / meta + locationText: str | None = None + unNormalizedProductUrl: str | None = None + loadedCountryCode: str | None = None + categoryPageData: dict | None = None + bestsellerPageData: dict | None = None + input: str | None = None + + def to_output(self) -> dict[str, Any]: + """Serialize to the flat dict output shape (keeps extras, aliases). + + ``by_alias`` restores digit-prefixed keys like ``starsBreakdown.5star``; + ``exclude_none=False`` keeps unsourced keys present so consumers never + break on a missing field. + """ + return self.model_dump(by_alias=True, exclude_none=False) + + +# --------------------------------------------------------------------------- # +# Error item (the failure model — pushed into the stream, not raised). # +# --------------------------------------------------------------------------- # + +ErrorCode = Literal[ + "invalid_url", + "invalid_input", + "product_not_found", + "shortened_url_invalid", + "bestsellers_category_not_found", + "no_results_found", +] + + +class ErrorItem(BaseModel): + """A per-input failure emitted into the dataset instead of a normal item. + + Consumers tell error items apart from products by the presence of ``error``. + ``invalid_input`` additionally terminates the run (enforced in later + milestones); all other codes are per-input and non-fatal. + """ + + model_config = ConfigDict(extra="allow") + + error: ErrorCode + errorDescription: str | None = None + input: str | None = None + url: str | None = None diff --git a/surfsense_backend/app/proprietary/platforms/amazon/scraper.py b/surfsense_backend/app/proprietary/platforms/amazon/scraper.py new file mode 100644 index 000000000..57a52c077 --- /dev/null +++ b/surfsense_backend/app/proprietary/platforms/amazon/scraper.py @@ -0,0 +1,541 @@ +"""Orchestrate public Amazon product discovery and enrichment. + +The streaming core dispatches each start URL to a product, search, best-sellers, +or shortened-link flow. Network and parsing concerns remain isolated in their +own modules so markup changes and retry policy can be tested independently. +""" + +from __future__ import annotations + +import asyncio +from collections.abc import AsyncIterator +from typing import Any +from urllib.parse import parse_qsl, urlencode, urlparse, urlunparse + +from .fetch import ( + fetch_aod_html, + fetch_page, + fetch_seller_html, + gather_bounded, + get_location_session, + resolve_shortlink, + should_localize, +) +from .locale import accept_language_for, proxy_country_for +from .parsers import ( + parse_aod_offers, + parse_bestsellers_page, + parse_product, + parse_search_page, + parse_seller, +) +from .schemas import AmazonScrapeInput, ErrorItem, ProductItem +from .url_resolver import ResolvedUrl, resolve_url + +__all__ = ["iter_products", "scrape_products"] + +_DETAIL_CONCURRENCY = 8 +_SEARCH_PAGE_LIMIT = 7 +_DEFAULT_ITEMS_PER_START_URL = 100 + + +def _error( + code: str, description: str, *, input_url: str | None, url: str | None = None +) -> dict[str, Any]: + return ErrorItem( + error=code, + errorDescription=description, + input=input_url, + url=url or input_url, + ).model_dump() + + +def _buybox_offer(fields: dict[str, Any]) -> dict[str, Any] | None: + """Synthesize a single offer from the PDP buy box. + + Amazon serves the All-Offers-Display panel as a JS-only modal for many + (especially US) ASINs, so the AOD ajax endpoint 404s. Fall back to the + featured buy-box winner already parsed from the product page. + """ + price = fields.get("price") + seller = fields.get("seller") + seller = seller if isinstance(seller, dict) else None + if not price and not (seller and seller.get("name")): + return None + return { + "position": 1, + "price": price, + "condition": fields.get("condition") or "New", + "delivery": fields.get("delivery"), + "seller": seller, + "isPinnedOffer": True, + } + + +def _page_url(url: str, page: int) -> str: + parsed = urlparse(url) + query = dict(parse_qsl(parsed.query, keep_blank_values=True)) + query["page"] = str(page) + return urlunparse(parsed._replace(query=urlencode(query))) + + +async def _location_context( + resolved: ResolvedUrl, input_model: AmazonScrapeInput, route: str +) -> tuple[dict[str, str] | None, str | None, str | None, str | None]: + if ( + not input_model.zipCode + or not resolved.domain + or not should_localize(route, input_model.locationDeliverableRoutes) + ): + return None, None, None, None + session = await get_location_session( + resolved.domain, + zip_code=input_model.zipCode, + country_code=input_model.countryCode, + country=proxy_country_for(resolved.marketplace), + accept_language=accept_language_for(resolved.marketplace), + ) + if session is None: + return None, None, None, None + return ( + session.cookies, + session.proxy, + session.location_text, + session.country_code, + ) + + +async def _product_flow( + resolved: ResolvedUrl, + input_model: AmazonScrapeInput, + *, + provenance: dict[str, Any] | None = None, + allow_variants: bool = True, +) -> AsyncIterator[dict[str, Any]]: + """Fetch and enrich one product detail page.""" + asin = resolved.asin + if not asin or not resolved.domain: + yield _error( + "product_not_found", + "The product URL did not contain a valid ASIN.", + input_url=resolved.url, + ) + return + + cookies, proxy, location_text, loaded_country = await _location_context( + resolved, input_model, "PRODUCT" + ) + country = proxy_country_for(resolved.marketplace) + accept_language = accept_language_for(resolved.marketplace) + response = await fetch_page( + resolved.url, + cookies=cookies, + proxy=proxy, + country=country, + accept_language=accept_language, + ) + if response is None: + yield _error( + "product_not_found", + "The product page could not be loaded after retrying available proxy exits.", + input_url=resolved.url, + ) + return + if response.status in {404, 410}: + yield _error( + "product_not_found", + "The product page was not found.", + input_url=resolved.url, + ) + return + if response.status != 200: + yield _error( + "product_not_found", + f"The product page returned HTTP {response.status}.", + input_url=resolved.url, + ) + return + + fields = await asyncio.to_thread( + parse_product, + response.html, + asin=asin, + url=response.url, + domain=resolved.domain, + ) + if not fields.get("title"): + yield _error( + "product_not_found", + "The response did not contain a recognizable product.", + input_url=resolved.url, + url=response.url, + ) + return + fields.update(provenance or {}) + fields["input"] = resolved.url + fields["unNormalizedProductUrl"] = resolved.url + fields["locationText"] = location_text + fields["loadedCountryCode"] = loaded_country + + if input_model.maxOffers > 0: + offer_cookies, offer_proxy, _, _ = await _location_context( + resolved, input_model, "OFFERS" + ) + offers_html = await fetch_aod_html( + asin, + resolved.domain, + cookies=offer_cookies or cookies, + proxy=offer_proxy or proxy, + country=country, + accept_language=accept_language, + ) + offers = ( + parse_aod_offers(offers_html, domain=resolved.domain) if offers_html else [] + ) + if not offers: + buybox = _buybox_offer(fields) + offers = [buybox] if buybox else [] + if offers: + fields["offers"] = offers[: input_model.maxOffers] + + if input_model.scrapeSellers: + seller_ids: list[str] = [] + featured = fields.get("seller") + if isinstance(featured, dict) and featured.get("id"): + seller_ids.append(featured["id"]) + for offer in fields.get("offers") or []: + seller = offer.get("seller") if isinstance(offer, dict) else None + if isinstance(seller, dict) and seller.get("id"): + seller_ids.append(seller["id"]) + seller_ids = list(dict.fromkeys(seller_ids)) + + async def load_seller(seller_id: str) -> dict[str, Any] | None: + html = await fetch_seller_html( + seller_id, + resolved.domain, + cookies=cookies, + proxy=proxy, + country=country, + accept_language=accept_language, + ) + return ( + parse_seller(html, seller_id=seller_id, domain=resolved.domain) + if html + else None + ) + + sellers = await gather_bounded( + [ + lambda seller_id=seller_id: load_seller(seller_id) + for seller_id in seller_ids + ], + concurrency=_DETAIL_CONCURRENCY, + ) + by_id = { + seller["id"]: seller for seller in sellers if seller and seller.get("id") + } + if featured and featured.get("id") in by_id: + fields["seller"] = by_id[featured["id"]] + for offer in fields.get("offers") or []: + seller = offer.get("seller") + if isinstance(seller, dict) and seller.get("id") in by_id: + offer["seller"] = by_id[seller["id"]] + + variant_asins = [ + value for value in fields.get("variantAsins") or [] if value != asin + ] + variant_cap = ( + min(len(variant_asins), input_model.maxProductVariantsAsSeparateResults) + if allow_variants + else 0 + ) + price_cap = ( + len(variant_asins) + if allow_variants and input_model.scrapeProductVariantPrices + else 0 + ) + fetch_count = max(variant_cap, price_cap) + variant_items: list[dict[str, Any]] = [] + if fetch_count: + + async def load_variant(variant_asin: str) -> list[dict[str, Any]]: + variant_url = f"https://{resolved.domain}/dp/{variant_asin}" + variant_resolved = resolve_url(variant_url) + if variant_resolved is None: + return [] + return [ + item + async for item in _product_flow( + variant_resolved, input_model, allow_variants=False + ) + ] + + batches = await gather_bounded( + [ + lambda variant_asin=variant_asin: load_variant(variant_asin) + for variant_asin in variant_asins[:fetch_count] + ], + concurrency=_DETAIL_CONCURRENCY, + ) + variant_items = [ + item for batch in batches for item in batch if "error" not in item + ] + if input_model.scrapeProductVariantPrices: + prices = { + item.get("asin"): item.get("price") + for item in variant_items + if item.get("asin") + } + details = fields.get("variantDetails") or [ + {"asin": variant_asin} for variant_asin in variant_asins + ] + for detail in details: + if detail.get("asin") in prices: + detail["price"] = prices[detail["asin"]] + fields["variantDetails"] = details + + yield ProductItem(**fields).to_output() + for item in variant_items[:variant_cap]: + item["originalAsin"] = asin + yield item + + +async def _search_flow( + resolved: ResolvedUrl, input_model: AmazonScrapeInput +) -> AsyncIterator[dict[str, Any]]: + """Page through search results and optionally fetch product details.""" + if not resolved.domain: + yield _error( + "no_results_found", "The search domain was invalid.", input_url=resolved.url + ) + return + cap = ( + input_model.maxItemsPerStartUrl + if input_model.maxItemsPerStartUrl is not None + else _DEFAULT_ITEMS_PER_START_URL + ) + max_pages = min(input_model.maxSearchPagesPerStartUrl, _SEARCH_PAGE_LIMIT) + cookies, proxy, _, _ = await _location_context(resolved, input_model, "SEARCH") + country = proxy_country_for(resolved.marketplace) + accept_language = accept_language_for(resolved.marketplace) + seen: set[str] = set() + emitted = 0 + for page in range(1, max_pages + 1): + html_response = await fetch_page( + _page_url(resolved.url, page), + cookies=cookies, + proxy=proxy, + country=country, + accept_language=accept_language, + ) + cards = ( + await asyncio.to_thread( + parse_search_page, + html_response.html, + page=page, + domain=resolved.domain, + ) + if html_response is not None and html_response.status == 200 + else [] + ) + cards = [ + card for card in cards if card.get("asin") and card["asin"] not in seen + ] + for card in cards: + seen.add(card["asin"]) + if not cards: + if page == 1: + yield _error( + "no_results_found", + "The search page did not contain any products.", + input_url=resolved.url, + ) + return + cards = cards[: max(0, cap - emitted)] + if input_model.scrapeProductDetails is False: + for card in cards: + card["input"] = resolved.url + yield ProductItem(**card).to_output() + emitted += 1 + else: + + async def load_card(card: dict[str, Any]) -> list[dict[str, Any]]: + product = resolve_url(card["url"]) + if product is None: + return [] + provenance = {"categoryPageData": card["categoryPageData"]} + return [ + item + async for item in _product_flow( + product, input_model, provenance=provenance + ) + ] + + batches = await gather_bounded( + [lambda card=card: load_card(card) for card in cards], + concurrency=_DETAIL_CONCURRENCY, + ) + for batch in batches: + for item in batch: + if "error" not in item and emitted >= cap: + return + yield item + if "error" not in item: + emitted += 1 + if emitted >= cap: + return + + +async def _bestsellers_flow( + resolved: ResolvedUrl, input_model: AmazonScrapeInput +) -> AsyncIterator[dict[str, Any]]: + """Page through a best-sellers category and emit ranked products.""" + if not resolved.domain: + yield _error( + "bestsellers_category_not_found", + "The best-sellers domain was invalid.", + input_url=resolved.url, + ) + return + cap = ( + input_model.maxItemsPerStartUrl + if input_model.maxItemsPerStartUrl is not None + else _DEFAULT_ITEMS_PER_START_URL + ) + country = proxy_country_for(resolved.marketplace) + accept_language = accept_language_for(resolved.marketplace) + seen: set[str] = set() + emitted = 0 + for page in range(1, min(input_model.maxSearchPagesPerStartUrl, 2) + 1): + response = await fetch_page( + _page_url(resolved.url, page), + country=country, + accept_language=accept_language, + ) + cards = ( + await asyncio.to_thread( + parse_bestsellers_page, + response.html, + page=page, + domain=resolved.domain, + ) + if response is not None and response.status == 200 + else [] + ) + cards = [ + card for card in cards if card.get("asin") and card["asin"] not in seen + ] + for card in cards: + seen.add(card["asin"]) + if not cards: + if page == 1: + yield _error( + "bestsellers_category_not_found", + "The page did not contain a recognizable best-sellers category.", + input_url=resolved.url, + ) + return + cards = cards[: max(0, cap - emitted)] + if input_model.scrapeProductDetails is False: + for card in cards: + card["input"] = resolved.url + yield ProductItem(**card).to_output() + emitted += 1 + else: + + async def load_card(card: dict[str, Any]) -> list[dict[str, Any]]: + product = resolve_url(card["url"]) + if product is None: + return [] + return [ + item + async for item in _product_flow( + product, + input_model, + provenance={"bestsellerPageData": card["bestsellerPageData"]}, + ) + ] + + for batch in await gather_bounded( + [lambda card=card: load_card(card) for card in cards], + concurrency=_DETAIL_CONCURRENCY, + ): + for item in batch: + if "error" not in item and emitted >= cap: + return + yield item + if "error" not in item: + emitted += 1 + if emitted >= cap: + return + + +async def _shortened_flow( + resolved: ResolvedUrl, input_model: AmazonScrapeInput +) -> AsyncIterator[dict[str, Any]]: + """Resolve a shortened link and dispatch its final Amazon URL.""" + target = await resolve_shortlink(resolved.url) + final = resolve_url(target) if target else None + if final is None or final.kind == "shortened": + yield _error( + "shortened_url_invalid", + "The shortened link did not resolve to a supported Amazon page.", + input_url=resolved.url, + url=target, + ) + return + async for item in _FLOWS[final.kind](final, input_model): + if "input" in item: + item["input"] = resolved.url + yield item + + +_FLOWS = { + "product": _product_flow, + "search": _search_flow, + "bestsellers": _bestsellers_flow, + "shortened": _shortened_flow, +} + + +async def iter_products( + input_model: AmazonScrapeInput, +) -> AsyncIterator[dict[str, Any]]: + """Yield product items for every start URL. + + Each ``categoryOrProductUrls`` entry is classified and dispatched to its + per-kind flow. A URL that cannot be recognized as an Amazon product / + search / bestsellers / shortened link yields an ``invalid_url`` error item + (the failure model is in-stream error items, not exceptions). + """ + for entry in input_model.categoryOrProductUrls: + url = entry.get("url") if isinstance(entry, dict) else None + resolved = resolve_url(url) if url else None + if resolved is None: + yield _error( + "invalid_url", + "Start URL was malformed or not a recognized Amazon URL.", + input_url=url, + ) + continue + async for item in _FLOWS[resolved.kind](resolved, input_model): + yield item + + +async def scrape_products( + input_model: AmazonScrapeInput, *, limit: int | None = None +) -> list[dict[str, Any]]: + """Collect :func:`iter_products` into a list, honoring an optional ``limit``. + + ``limit`` is a request-time policy guard (used by the route/capability), NOT + a ceiling in the streaming core. + """ + from app.capabilities.core.progress import emit_progress + + results: list[dict[str, Any]] = [] + async for item in iter_products(input_model): + results.append(item) + emit_progress("scraping", current=len(results), total=limit, unit="product") + if limit is not None and len(results) >= limit: + break + return results diff --git a/surfsense_backend/app/proprietary/platforms/amazon/url_resolver.py b/surfsense_backend/app/proprietary/platforms/amazon/url_resolver.py new file mode 100644 index 000000000..bc2102d6c --- /dev/null +++ b/surfsense_backend/app/proprietary/platforms/amazon/url_resolver.py @@ -0,0 +1,98 @@ +"""Classify an Amazon start URL and extract its identifiers. + +Covers the ``categoryOrProductUrls`` shapes accepted on input: product +pages (``/dp/ASIN``, ``/gp/product/ASIN``), search / category pages (``/s?k=``, +``/s?bbn=``, ``/s?rh=``), bestsellers pages (``/zgbs/`` / ``/gp/bestsellers/``), +and shortened links (``a.co`` / ``amzn.to`` / ``amzn.eu``, which need a network +redirect to resolve — classified here, resolved later in the fetch layer). + +Pure, no I/O. The ``marketplace`` suffix (``com``, ``de``, ``co.uk``, ...) is +derived from the host so the fetch layer can default the proxy country and UI +language without a separate geocoding step. +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass +from typing import Literal +from urllib.parse import parse_qs, urlparse + +ResolvedKind = Literal["product", "search", "bestsellers", "shortened"] + +_SHORTLINK_HOSTS = ("a.co", "amzn.to", "amzn.eu") + +# ASIN: a 10-char uppercase alphanumeric id, carried in the canonical product +# path forms (``/dp/``, ``/gp/product/``, ``/product/``) or a ``?asin=`` query. +_ASIN_PATH_RE = re.compile(r"/(?:dp|gp/product|product|d)/([A-Z0-9]{10})(?:[/?]|$)") +_ASIN_QUERY_RE = re.compile(r"[?&]asin=([A-Z0-9]{10})", re.IGNORECASE) + +# Search/category pages carry their query in these params even without a /s path. +_SEARCH_QUERY_KEYS = ("k", "bbn", "rh") + + +@dataclass(frozen=True) +class ResolvedUrl: + kind: ResolvedKind + url: str + asin: str | None = None + domain: str | None = None + marketplace: str | None = None # TLD suffix, e.g. "com", "de", "co.uk" + + +def extract_asin(url: str) -> str | None: + """Pull the 10-char ASIN out of an Amazon product URL, if present.""" + match = _ASIN_PATH_RE.search(url) + if match: + return match.group(1) + match = _ASIN_QUERY_RE.search(url) + return match.group(1).upper() if match else None + + +def _marketplace_from_host(host: str) -> str | None: + """The Amazon TLD suffix (``com``, ``de``, ``co.uk``, ...), or ``None``. + + Strips common subdomain prefixes and the ``amazon`` label, returning the + remainder (e.g. ``www.amazon.co.jp`` -> ``co.jp``). + """ + host = host.lower() + for prefix in ("www.", "smile."): + if host.startswith(prefix): + host = host[len(prefix) :] + if not host.startswith("amazon."): + return None + return host[len("amazon.") :] or None + + +def resolve_url(url: str) -> ResolvedUrl | None: + """Classify an Amazon start URL into a scrape job, or ``None`` if unrecognized.""" + parsed = urlparse(url) + host = (parsed.hostname or "").lower() + path = parsed.path or "" + + # Shortened links resolve to their real URL later, in the fetch layer. + if host in _SHORTLINK_HOSTS: + return ResolvedUrl("shortened", url) + + if "amazon." not in host: + return None + + marketplace = _marketplace_from_host(host) + + # Bestsellers before search: a /zgbs page has no /s path but is its own kind. + if "/zgbs/" in path or path.startswith("/gp/bestsellers"): + return ResolvedUrl("bestsellers", url, domain=host, marketplace=marketplace) + + # Product: canonical /dp/ (etc.) path with an ASIN. + asin = extract_asin(url) + if asin is not None: + return ResolvedUrl( + "product", url, asin=asin, domain=host, marketplace=marketplace + ) + + # Search / category: the /s path, or any search-carrying query param. + query = parse_qs(parsed.query) + if path.startswith("/s") or any(k in query for k in _SEARCH_QUERY_KEYS): + return ResolvedUrl("search", url, domain=host, marketplace=marketplace) + + return None diff --git a/surfsense_backend/app/routes/__init__.py b/surfsense_backend/app/routes/__init__.py index 85450d4b7..509233e15 100644 --- a/surfsense_backend/app/routes/__init__.py +++ b/surfsense_backend/app/routes/__init__.py @@ -1,6 +1,7 @@ from fastapi import APIRouter, Depends # Import verb namespaces for their registration side effects before the door builds. +import app.capabilities.amazon import app.capabilities.google_maps import app.capabilities.google_search import app.capabilities.instagram diff --git a/surfsense_backend/app/utils/proxy/__init__.py b/surfsense_backend/app/utils/proxy/__init__.py index 3e4158587..48fb222c5 100644 --- a/surfsense_backend/app/utils/proxy/__init__.py +++ b/surfsense_backend/app/utils/proxy/__init__.py @@ -15,6 +15,16 @@ def get_proxy_url() -> str | None: return get_active_provider().get_proxy_url() +def get_geo_proxy_url(country: str | None = None) -> str | None: + """Proxy URL pinned to an exit country when the provider supports it.""" + return get_active_provider().get_geo_proxy_url(country) + + +def get_sticky_proxy_url(session_id: str, country: str | None = None) -> str | None: + """Proxy URL pinned to a stable vendor session when supported.""" + return get_active_provider().get_sticky_proxy_url(session_id, country) + + def get_playwright_proxy() -> dict[str, str] | None: """Playwright-style proxy dict, or ``None`` when not configured.""" return get_active_provider().get_playwright_proxy() @@ -41,9 +51,11 @@ def get_residential_proxy_url() -> str | None: __all__ = [ "ProxyProvider", "get_active_provider", + "get_geo_proxy_url", "get_playwright_proxy", "get_proxy_url", "get_requests_proxies", "get_residential_proxy_url", + "get_sticky_proxy_url", "is_pool_backed", ] diff --git a/surfsense_backend/app/utils/proxy/base.py b/surfsense_backend/app/utils/proxy/base.py index ed67cd384..f225dcfb1 100644 --- a/surfsense_backend/app/utils/proxy/base.py +++ b/surfsense_backend/app/utils/proxy/base.py @@ -67,6 +67,24 @@ class ProxyProvider(ABC): return None return {"http": proxy_url, "https": proxy_url} + def get_geo_proxy_url(self, country: str | None = None) -> str | None: + """Return a proxy URL pinned to ``country`` when supported. + + Providers without vendor-specific country routing safely fall back to + their ordinary proxy URL. + """ + return self.get_proxy_url() + + def get_sticky_proxy_url( + self, session_id: str, country: str | None = None + ) -> str | None: + """Return a proxy URL pinned to ``session_id`` when supported. + + Providers without vendor-specific session routing safely fall back to + their ordinary proxy URL. + """ + return self.get_geo_proxy_url(country) + def get_location(self) -> str: """Return the proxy's configured exit region (e.g. ``"us"``), or ``""``. diff --git a/surfsense_backend/app/utils/proxy/providers/dataimpulse.py b/surfsense_backend/app/utils/proxy/providers/dataimpulse.py index 17d09325f..b2a151e03 100644 --- a/surfsense_backend/app/utils/proxy/providers/dataimpulse.py +++ b/surfsense_backend/app/utils/proxy/providers/dataimpulse.py @@ -16,15 +16,13 @@ Example URL:: http://__cr.us:@gw.dataimpulse.com:823 -ponytail: sticky sessions (a stable exit IP across requests) are another -username suffix (``__sid.``) — the lever the Reddit scraper's README flags as -a TODO for its ``loid``-per-IP flow. Not built yet: Reddit isn't wired to a -route, so there's no caller to thread a session id through. Add a -``get_sticky_proxy_url(session_id)`` here (rewriting the username) when it lands. +Sticky sessions use the vendor's ``__sid.`` username suffix, allowing a +caller to keep cookies and requests on the same exit IP. """ import logging -from urllib.parse import urlsplit +import re +from urllib.parse import quote, urlsplit, urlunsplit from app.config import Config from app.utils.proxy.base import ProxyProvider @@ -34,6 +32,22 @@ logger = logging.getLogger(__name__) # DataImpulse encodes country routing as a "__cr." username suffix; the # country token runs until the next "__" param (e.g. "__sid") or the end. _COUNTRY_MARKER = "__cr." +_COUNTRY_RE = re.compile(r"__cr\.[A-Za-z]{2,}") +_SESSION_RE = re.compile(r"__sid\.[A-Za-z0-9_-]+") + + +def _safe_session_id(session_id: str) -> str: + safe_id = re.sub(r"[^A-Za-z0-9_-]", "-", session_id).strip("-") + if not safe_id: + raise ValueError("session_id must contain at least one letter or digit") + return safe_id + + +def _safe_country(country: str | None) -> str | None: + if country is None: + return None + safe_country = re.sub(r"[^a-z]", "", country.lower()) + return safe_country or None class DataImpulseProvider(ProxyProvider): @@ -54,3 +68,41 @@ class DataImpulseProvider(ProxyProvider): if _COUNTRY_MARKER not in username: return "" return username.split(_COUNTRY_MARKER, 1)[1].split("__", 1)[0].lower() + + def _rewrite_proxy_url( + self, *, country: str | None = None, session_id: str | None = None + ) -> str | None: + url = self.get_proxy_url() + if not url: + return None + parts = urlsplit(url) + username = _SESSION_RE.sub("", _COUNTRY_RE.sub("", parts.username or "")) + safe_country = _safe_country(country) + if safe_country is not None: + username += f"__cr.{safe_country}" + elif _COUNTRY_MARKER in (parts.username or ""): + username += f"__cr.{self.get_location()}" + if session_id is not None: + username += f"__sid.{_safe_session_id(session_id)}" + userinfo = quote(username, safe="%") + if parts.password is not None: + userinfo += f":{quote(parts.password, safe='%')}" + host = parts.hostname or "" + netloc = f"{userinfo}@{host}" + if parts.port: + netloc += f":{parts.port}" + return urlunsplit( + (parts.scheme, netloc, parts.path, parts.query, parts.fragment) + ) + + def get_geo_proxy_url(self, country: str | None = None) -> str | None: + """Return the configured URL with a country-routing suffix when requested.""" + if not _safe_country(country): + return self.get_proxy_url() + return self._rewrite_proxy_url(country=country) + + def get_sticky_proxy_url( + self, session_id: str, country: str | None = None + ) -> str | None: + """Return the configured URL with deterministic country/session suffixes.""" + return self._rewrite_proxy_url(country=country, session_id=session_id) diff --git a/surfsense_backend/scripts/e2e_amazon_scraper.py b/surfsense_backend/scripts/e2e_amazon_scraper.py new file mode 100644 index 000000000..a784c8786 --- /dev/null +++ b/surfsense_backend/scripts/e2e_amazon_scraper.py @@ -0,0 +1,84 @@ +"""Manual end-to-end check for the public Amazon scraper. + +Run from the backend directory: + + uv run python scripts/e2e_amazon_scraper.py + uv run python scripts/e2e_amazon_scraper.py --refresh-fixtures + +The script requires live network access and the configured residential proxy. +The optional flag replaces the product and search parser fixtures with current +live responses. The script is intentionally excluded from pytest. +""" + +from __future__ import annotations + +import asyncio +import json +import sys +from pathlib import Path + +from dotenv import load_dotenv + +_BACKEND_ROOT = Path(__file__).resolve().parent.parent +sys.path.insert(0, str(_BACKEND_ROOT)) +for _candidate in (_BACKEND_ROOT / ".env", _BACKEND_ROOT.parent / ".env"): + if _candidate.exists(): + load_dotenv(_candidate) + break + +from app.proprietary.platforms.amazon import ( # noqa: E402 + AmazonScrapeInput, + scrape_products, +) +from app.proprietary.platforms.amazon.fetch import fetch_page # noqa: E402 + +_PRODUCT_URL = "https://www.amazon.com/dp/B09V3KXJPB" +_SEARCH_URL = "https://www.amazon.com/s?k=wireless+mouse" +_FIXTURE_DIR = _BACKEND_ROOT / "tests" / "unit" / "platforms" / "amazon" / "fixtures" + + +def _check(label: str, passed: bool) -> bool: + print(f"[{'PASS' if passed else 'FAIL'}] {label}") + return passed + + +async def main() -> int: + product_items = await scrape_products( + AmazonScrapeInput(categoryOrProductUrls=[{"url": _PRODUCT_URL}]), + limit=1, + ) + product = product_items[0] if product_items else {} + print(json.dumps(product, indent=2, ensure_ascii=False)[:3000]) + product_ok = _check( + "product detail has identity and title", + bool(product.get("asin") and product.get("title")), + ) + + search_items = await scrape_products( + AmazonScrapeInput( + categoryOrProductUrls=[{"url": _SEARCH_URL}], + maxItemsPerStartUrl=3, + scrapeProductDetails=False, + ) + ) + search_ok = _check( + "search returns product cards", + bool(search_items) + and all( + item.get("asin") and item.get("categoryPageData") for item in search_items + ), + ) + fixture_ok = True + if "--refresh-fixtures" in sys.argv: + _FIXTURE_DIR.mkdir(parents=True, exist_ok=True) + for name, url in (("product.html", _PRODUCT_URL), ("search.html", _SEARCH_URL)): + response = await fetch_page(url) + saved = response is not None and response.status == 200 + if saved: + (_FIXTURE_DIR / name).write_text(response.html, encoding="utf-8") + fixture_ok &= _check(f"refreshed {name}", saved) + return 0 if product_ok and search_ok and fixture_ok else 1 + + +if __name__ == "__main__": + raise SystemExit(asyncio.run(main())) diff --git a/surfsense_backend/tests/unit/agents/multi_agent_chat/test_subagent_composition.py b/surfsense_backend/tests/unit/agents/multi_agent_chat/test_subagent_composition.py index ee889cd17..f578f3457 100644 --- a/surfsense_backend/tests/unit/agents/multi_agent_chat/test_subagent_composition.py +++ b/surfsense_backend/tests/unit/agents/multi_agent_chat/test_subagent_composition.py @@ -27,6 +27,7 @@ pytestmark = pytest.mark.unit # specialist is a deliberate product change and must be reflected here. _EXPECTED_SUBAGENTS = frozenset( { + "amazon", "deliverables", "dropbox", "google_drive", diff --git a/surfsense_backend/tests/unit/capabilities/amazon/__init__.py b/surfsense_backend/tests/unit/capabilities/amazon/__init__.py new file mode 100644 index 000000000..8b1378917 --- /dev/null +++ b/surfsense_backend/tests/unit/capabilities/amazon/__init__.py @@ -0,0 +1 @@ + diff --git a/surfsense_backend/tests/unit/capabilities/amazon/scrape/__init__.py b/surfsense_backend/tests/unit/capabilities/amazon/scrape/__init__.py new file mode 100644 index 000000000..8b1378917 --- /dev/null +++ b/surfsense_backend/tests/unit/capabilities/amazon/scrape/__init__.py @@ -0,0 +1 @@ + diff --git a/surfsense_backend/tests/unit/capabilities/amazon/scrape/test_executor.py b/surfsense_backend/tests/unit/capabilities/amazon/scrape/test_executor.py new file mode 100644 index 000000000..806828f31 --- /dev/null +++ b/surfsense_backend/tests/unit/capabilities/amazon/scrape/test_executor.py @@ -0,0 +1,45 @@ +from __future__ import annotations + +from app.capabilities.amazon.scrape.executor import build_scrape_executor +from app.capabilities.amazon.scrape.schemas import MAX_AMAZON_RESULTS, ScrapeInput +from app.proprietary.platforms.amazon import AmazonScrapeInput + + +class _FakeScraper: + def __init__(self) -> None: + self.calls: list[tuple[AmazonScrapeInput, int | None]] = [] + + async def __call__( + self, input_model: AmazonScrapeInput, *, limit: int | None = None + ) -> list[dict]: + self.calls.append((input_model, limit)) + return [{"asin": "B09V3KXJPB", "title": "Product"}] + + +async def test_executor_maps_agent_input_to_scraper_input(): + scraper = _FakeScraper() + execute = build_scrape_executor(scraper) + + output = await execute( + ScrapeInput( + search_terms=["wireless mouse"], + urls=["https://www.amazon.com/dp/B09V3KXJPB"], + max_items=5, + max_offers=2, + include_sellers=True, + zip_code="10001", + country_code="US", + ) + ) + + assert output.items[0].asin == "B09V3KXJPB" + input_model, limit = scraper.calls[0] + assert input_model.categoryOrProductUrls == [ + {"url": "https://www.amazon.com/dp/B09V3KXJPB"}, + {"url": "https://www.amazon.com/s?k=wireless+mouse"}, + ] + assert input_model.maxItemsPerStartUrl == 5 + assert input_model.maxOffers == 2 + assert input_model.scrapeSellers is True + assert input_model.zipCode == "10001" + assert limit == MAX_AMAZON_RESULTS diff --git a/surfsense_backend/tests/unit/capabilities/amazon/scrape/test_schemas.py b/surfsense_backend/tests/unit/capabilities/amazon/scrape/test_schemas.py new file mode 100644 index 000000000..34487ea2b --- /dev/null +++ b/surfsense_backend/tests/unit/capabilities/amazon/scrape/test_schemas.py @@ -0,0 +1,42 @@ +from __future__ import annotations + +import pytest +from pydantic import ValidationError + +from app.capabilities.amazon.scrape.schemas import ( + MAX_AMAZON_RESULTS, + ScrapeInput, + ScrapeOutput, +) + + +def test_estimated_units_cover_search_and_direct_variant_fanout(): + payload = ScrapeInput( + search_terms=["mouse", "keyboard"], + urls=["https://www.amazon.com/dp/B09V3KXJPB"], + max_items=20, + max_variants=3, + ) + + assert payload.estimated_units == 44 + + +def test_estimated_units_respect_hard_run_ceiling(): + payload = ScrapeInput(search_terms=["x"] * 20, max_items=100) + assert payload.estimated_units == MAX_AMAZON_RESULTS + + +def test_at_least_one_source_is_required(): + with pytest.raises(ValidationError): + ScrapeInput() + + +def test_error_items_are_not_billable(): + output = ScrapeOutput( + items=[ + {"asin": "B09V3KXJPB", "title": "Product"}, + {"error": "product_not_found", "errorDescription": "Missing"}, + ] + ) + + assert output.billable_units == 1 diff --git a/surfsense_backend/tests/unit/capabilities/amazon/test_registry.py b/surfsense_backend/tests/unit/capabilities/amazon/test_registry.py new file mode 100644 index 000000000..bcebc1c60 --- /dev/null +++ b/surfsense_backend/tests/unit/capabilities/amazon/test_registry.py @@ -0,0 +1,13 @@ +from __future__ import annotations + +from app.capabilities.amazon.scrape.schemas import ScrapeInput, ScrapeOutput +from app.capabilities.core import BillingUnit +from app.capabilities.core.store import get_capability + + +def test_amazon_scrape_is_registered_and_metered(): + capability = get_capability("amazon.scrape") + + assert capability.input_schema is ScrapeInput + assert capability.output_schema is ScrapeOutput + assert capability.billing_unit is BillingUnit.AMAZON_PRODUCT diff --git a/surfsense_backend/tests/unit/capabilities/google_maps/test_registry.py b/surfsense_backend/tests/unit/capabilities/google_maps/test_registry.py index c6a007645..409d0f78a 100644 --- a/surfsense_backend/tests/unit/capabilities/google_maps/test_registry.py +++ b/surfsense_backend/tests/unit/capabilities/google_maps/test_registry.py @@ -7,6 +7,7 @@ import pytest from app.capabilities import ( google_maps, # noqa: F401 — importing the namespace registers its verbs ) +from app.capabilities.core import BillingUnit from app.capabilities.core.store import get_capability from app.capabilities.google_maps.reviews.schemas import ReviewsInput, ReviewsOutput from app.capabilities.google_maps.scrape.schemas import ScrapeInput, ScrapeOutput @@ -14,19 +15,19 @@ from app.capabilities.google_maps.scrape.schemas import ScrapeInput, ScrapeOutpu pytestmark = pytest.mark.unit -def test_google_maps_scrape_is_registered_and_free(): +def test_google_maps_scrape_is_registered_and_billable(): cap = get_capability("google_maps.scrape") assert cap.name == "google_maps.scrape" assert cap.input_schema is ScrapeInput assert cap.output_schema is ScrapeOutput - assert cap.billing_unit is None + assert cap.billing_unit is BillingUnit.GOOGLE_MAPS_PLACE -def test_google_maps_reviews_is_registered_and_free(): +def test_google_maps_reviews_is_registered_and_billable(): cap = get_capability("google_maps.reviews") assert cap.name == "google_maps.reviews" assert cap.input_schema is ReviewsInput assert cap.output_schema is ReviewsOutput - assert cap.billing_unit is None + assert cap.billing_unit is BillingUnit.GOOGLE_MAPS_REVIEW diff --git a/surfsense_backend/tests/unit/capabilities/reddit/test_registry.py b/surfsense_backend/tests/unit/capabilities/reddit/test_registry.py index e0bca5f26..a6452470a 100644 --- a/surfsense_backend/tests/unit/capabilities/reddit/test_registry.py +++ b/surfsense_backend/tests/unit/capabilities/reddit/test_registry.py @@ -7,16 +7,17 @@ import pytest from app.capabilities import ( reddit, # noqa: F401 — importing the namespace registers its verbs ) +from app.capabilities.core import BillingUnit from app.capabilities.core.store import get_capability from app.capabilities.reddit.scrape.schemas import ScrapeInput, ScrapeOutput pytestmark = pytest.mark.unit -def test_reddit_scrape_is_registered_and_free(): +def test_reddit_scrape_is_registered_and_billable(): cap = get_capability("reddit.scrape") assert cap.name == "reddit.scrape" assert cap.input_schema is ScrapeInput assert cap.output_schema is ScrapeOutput - assert cap.billing_unit is None + assert cap.billing_unit is BillingUnit.REDDIT_ITEM diff --git a/surfsense_backend/tests/unit/capabilities/youtube/test_registry.py b/surfsense_backend/tests/unit/capabilities/youtube/test_registry.py index 756b65176..7fc98c697 100644 --- a/surfsense_backend/tests/unit/capabilities/youtube/test_registry.py +++ b/surfsense_backend/tests/unit/capabilities/youtube/test_registry.py @@ -7,6 +7,7 @@ import pytest from app.capabilities import ( youtube, # noqa: F401 — importing the namespace registers its verbs ) +from app.capabilities.core import BillingUnit from app.capabilities.core.store import get_capability from app.capabilities.youtube.comments.schemas import CommentsInput, CommentsOutput from app.capabilities.youtube.scrape.schemas import ScrapeInput, ScrapeOutput @@ -14,19 +15,19 @@ from app.capabilities.youtube.scrape.schemas import ScrapeInput, ScrapeOutput pytestmark = pytest.mark.unit -def test_youtube_scrape_is_registered_and_free(): +def test_youtube_scrape_is_registered_and_billable(): cap = get_capability("youtube.scrape") assert cap.name == "youtube.scrape" assert cap.input_schema is ScrapeInput assert cap.output_schema is ScrapeOutput - assert cap.billing_unit is None + assert cap.billing_unit is BillingUnit.YOUTUBE_VIDEO -def test_youtube_comments_is_registered_and_free(): +def test_youtube_comments_is_registered_and_billable(): cap = get_capability("youtube.comments") assert cap.name == "youtube.comments" assert cap.input_schema is CommentsInput assert cap.output_schema is CommentsOutput - assert cap.billing_unit is None + assert cap.billing_unit is BillingUnit.YOUTUBE_COMMENT diff --git a/surfsense_backend/tests/unit/platforms/amazon/__init__.py b/surfsense_backend/tests/unit/platforms/amazon/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/surfsense_backend/tests/unit/platforms/amazon/fixtures/bestsellers.html b/surfsense_backend/tests/unit/platforms/amazon/fixtures/bestsellers.html new file mode 100644 index 000000000..d6323e9c2 --- /dev/null +++ b/surfsense_backend/tests/unit/platforms/amazon/fixtures/bestsellers.html @@ -0,0 +1,17 @@ + + + +
+ #1 + Example Wireless Headphones + $49.99 + 4.6 out of 5 stars + Example Wireless Headphones +
+
+ #2 + Example Earbuds + $29.99 +
+ + diff --git a/surfsense_backend/tests/unit/platforms/amazon/fixtures/blocked.html b/surfsense_backend/tests/unit/platforms/amazon/fixtures/blocked.html new file mode 100644 index 000000000..3d4517b61 --- /dev/null +++ b/surfsense_backend/tests/unit/platforms/amazon/fixtures/blocked.html @@ -0,0 +1,9 @@ + + + Robot Check + +
+

Enter the characters you see below

+
+ + diff --git a/surfsense_backend/tests/unit/platforms/amazon/fixtures/offers.html b/surfsense_backend/tests/unit/platforms/amazon/fixtures/offers.html new file mode 100644 index 000000000..d3ff57ccc --- /dev/null +++ b/surfsense_backend/tests/unit/platforms/amazon/fixtures/offers.html @@ -0,0 +1,17 @@ + + + +
+
+
New
+ $47.50 + +
FREE delivery tomorrow
+
+
+
Used - Very Good
+ $39.00 + +
+ + diff --git a/surfsense_backend/tests/unit/platforms/amazon/fixtures/product.html b/surfsense_backend/tests/unit/platforms/amazon/fixtures/product.html new file mode 100644 index 000000000..450cda040 --- /dev/null +++ b/surfsense_backend/tests/unit/platforms/amazon/fixtures/product.html @@ -0,0 +1,51 @@ + + + + + Example Wireless Headphones + Visit the Example Store + Example Retailer +
+ $49.99 + $79.99 +
+
+ 4.6 out of 5 stars + 1,234 ratings + Reviews +
+ + + +
5 star80%
4 star12%
+
In Stock
+
2K+ bought in past month
+
  • Long battery life
  • Noise cancelling
+
Detailed product description.
+ Example Wireless Headphones +

Premium materials and comfort.

Premium materials and comfort
+
Designed for everyday listening.
+
  • Best Sellers Rank: #12 in Electronics #3 in Headphones
+
ModelWH-100
+
ColorBlack
+
Amazon's Choice
+ +
+ Alex + Excellent sound + 5.0 out of 5 stars + Reviewed on July 1, 2026 + Clear and comfortable. + Verified Purchase + 12 people found this helpful +
+ + diff --git a/surfsense_backend/tests/unit/platforms/amazon/fixtures/search.html b/surfsense_backend/tests/unit/platforms/amazon/fixtures/search.html new file mode 100644 index 000000000..f565b95c6 --- /dev/null +++ b/surfsense_backend/tests/unit/platforms/amazon/fixtures/search.html @@ -0,0 +1,19 @@ + + + +
+

Example Wireless Headphones

+ $49.99 + 4.6 out of 5 stars + 1,234 + Example Wireless Headphones + Best Seller +
+
+ Sponsored +

Example Earbuds

+ $29.99 + 4.2 out of 5 stars +
+ + diff --git a/surfsense_backend/tests/unit/platforms/amazon/fixtures/seller.html b/surfsense_backend/tests/unit/platforms/amazon/fixtures/seller.html new file mode 100644 index 000000000..c679d1803 --- /dev/null +++ b/surfsense_backend/tests/unit/platforms/amazon/fixtures/seller.html @@ -0,0 +1,11 @@ + + + +

Example Retailer

+
1,875 ratings
+ + + +
4.8 out of 5 stars
1,875 ratings
+ + diff --git a/surfsense_backend/tests/unit/platforms/amazon/test_flows.py b/surfsense_backend/tests/unit/platforms/amazon/test_flows.py new file mode 100644 index 000000000..7aa7433ec --- /dev/null +++ b/surfsense_backend/tests/unit/platforms/amazon/test_flows.py @@ -0,0 +1,220 @@ +from __future__ import annotations + +from pathlib import Path + +from app.proprietary.platforms.amazon import AmazonScrapeInput, scrape_products, scraper +from app.proprietary.platforms.amazon.fetch import FetchResult +from app.proprietary.platforms.amazon.url_resolver import resolve_url + +_FIXTURES = Path(__file__).parent / "fixtures" + + +def _fixture(name: str) -> str: + return (_FIXTURES / name).read_text(encoding="utf-8") + + +def _response(url: str, html: str, status: int = 200) -> FetchResult: + return FetchResult(status=status, html=html, url=url, cookies={}) + + +async def test_product_flow_emits_parsed_item(monkeypatch): + async def fetch_page(url: str, **_kwargs): + return _response(url, _fixture("product.html")) + + monkeypatch.setattr(scraper, "fetch_page", fetch_page) + items = await scrape_products( + AmazonScrapeInput( + categoryOrProductUrls=[{"url": "https://www.amazon.com/dp/B09V3KXJPB"}] + ) + ) + + assert len(items) == 1 + assert items[0]["asin"] == "B09V3KXJPB" + assert items[0]["title"] == "Example Wireless Headphones" + + +async def test_product_flow_maps_not_found_to_error_item(monkeypatch): + async def fetch_page(url: str, **_kwargs): + return _response(url, "", status=404) + + monkeypatch.setattr(scraper, "fetch_page", fetch_page) + items = await scrape_products( + AmazonScrapeInput( + categoryOrProductUrls=[{"url": "https://www.amazon.com/dp/B09V3KXJPB"}] + ) + ) + + assert items[0]["error"] == "product_not_found" + + +async def test_search_flow_honors_cap_and_stops_on_empty_page(monkeypatch): + calls: list[str] = [] + + async def fetch_page(url: str, **_kwargs): + calls.append(url) + html = _fixture("search.html") if "page=1" in url else "" + return _response(url, html) + + monkeypatch.setattr(scraper, "fetch_page", fetch_page) + items = await scrape_products( + AmazonScrapeInput( + categoryOrProductUrls=[{"url": "https://www.amazon.com/s?k=headphones"}], + maxItemsPerStartUrl=1, + maxSearchPagesPerStartUrl=5, + scrapeProductDetails=False, + ) + ) + + assert len(items) == 1 + assert items[0]["categoryPageData"]["position"] == 1 + assert len(calls) == 1 + + +async def test_search_flow_threads_marketplace_locale_to_fetch(monkeypatch): + calls: list[dict[str, object]] = [] + + async def fetch_page(url: str, **kwargs): + calls.append({"url": url, **kwargs}) + return _response(url, _fixture("search.html")) + + monkeypatch.setattr(scraper, "fetch_page", fetch_page) + items = await scrape_products( + AmazonScrapeInput( + categoryOrProductUrls=[{"url": "https://www.amazon.co.uk/s?k=headphones"}], + maxItemsPerStartUrl=1, + scrapeProductDetails=False, + ) + ) + + assert len(items) == 1 + assert calls[0]["country"] == "gb" + assert calls[0]["accept_language"] == "en-GB" + + +async def test_search_flow_returns_no_results_error(monkeypatch): + async def fetch_page(url: str, **_kwargs): + return _response(url, "") + + monkeypatch.setattr(scraper, "fetch_page", fetch_page) + items = await scrape_products( + AmazonScrapeInput( + categoryOrProductUrls=[{"url": "https://www.amazon.com/s?k=missing"}], + scrapeProductDetails=False, + ) + ) + + assert items[0]["error"] == "no_results_found" + + +async def test_product_flow_enriches_offers_and_seller(monkeypatch): + async def fetch_page(url: str, **_kwargs): + return _response(url, _fixture("product.html")) + + async def fetch_aod_html(*_args, **_kwargs): + return _fixture("offers.html") + + async def fetch_seller_html(*_args, **_kwargs): + return _fixture("seller.html") + + monkeypatch.setattr(scraper, "fetch_page", fetch_page) + monkeypatch.setattr(scraper, "fetch_aod_html", fetch_aod_html) + monkeypatch.setattr(scraper, "fetch_seller_html", fetch_seller_html) + items = await scrape_products( + AmazonScrapeInput( + categoryOrProductUrls=[{"url": "https://www.amazon.com/dp/B09V3KXJPB"}], + maxOffers=1, + scrapeSellers=True, + ) + ) + + assert len(items[0]["offers"]) == 1 + assert items[0]["offers"][0]["seller"]["averageRating"] == 4.8 + assert items[0]["seller"]["reviewsCount"] == 1875 + + +async def test_bestsellers_card_only_flow(monkeypatch): + async def fetch_page(url: str, **_kwargs): + return _response(url, _fixture("bestsellers.html")) + + monkeypatch.setattr(scraper, "fetch_page", fetch_page) + items = await scrape_products( + AmazonScrapeInput( + categoryOrProductUrls=[ + { + "url": ( + "https://www.amazon.com/Best-Sellers-Electronics/" + "zgbs/electronics" + ) + } + ], + maxItemsPerStartUrl=2, + scrapeProductDetails=False, + ) + ) + + assert [item["bestsellerPageData"]["rank"] for item in items] == [1, 2] + + +async def test_shortlink_resolves_and_dispatches(monkeypatch): + async def resolve(_url: str): + return "https://www.amazon.com/dp/B09V3KXJPB" + + async def fetch_page(url: str, **_kwargs): + return _response(url, _fixture("product.html")) + + monkeypatch.setattr(scraper, "resolve_shortlink", resolve) + monkeypatch.setattr(scraper, "fetch_page", fetch_page) + items = await scrape_products( + AmazonScrapeInput(categoryOrProductUrls=[{"url": "https://a.co/d/example"}]) + ) + + assert items[0]["asin"] == "B09V3KXJPB" + assert items[0]["input"] == "https://a.co/d/example" + + +async def test_product_flow_expands_variants_and_attaches_prices(monkeypatch): + async def fetch_page(url: str, **_kwargs): + return _response(url, _fixture("product.html")) + + monkeypatch.setattr(scraper, "fetch_page", fetch_page) + items = await scrape_products( + AmazonScrapeInput( + categoryOrProductUrls=[{"url": "https://www.amazon.com/dp/B09V3KXJPB"}], + maxProductVariantsAsSeparateResults=1, + scrapeProductVariantPrices=True, + ) + ) + + assert [item["asin"] for item in items] == ["B09V3KXJPB", "B09V3KXJP1"] + assert items[1]["originalAsin"] == "B09V3KXJPB" + variant = next( + detail + for detail in items[0]["variantDetails"] + if detail["asin"] == "B09V3KXJP1" + ) + assert variant["price"]["value"] == 49.99 + + +async def test_route_outside_location_allowlist_skips_session_mint(monkeypatch): + calls = 0 + + async def get_location_session(*_args, **_kwargs): + nonlocal calls + calls += 1 + return None + + monkeypatch.setattr(scraper, "get_location_session", get_location_session) + resolved = resolve_url("https://www.amazon.com/dp/B09V3KXJPB") + assert resolved is not None + context = await scraper._location_context( + resolved, + AmazonScrapeInput( + categoryOrProductUrls=[{"url": resolved.url}], + zipCode="10001", + locationDeliverableRoutes=["OFFERS"], + ), + "PRODUCT", + ) + + assert context == (None, None, None, None) + assert calls == 0 diff --git a/surfsense_backend/tests/unit/platforms/amazon/test_locale.py b/surfsense_backend/tests/unit/platforms/amazon/test_locale.py new file mode 100644 index 000000000..a0c99d3df --- /dev/null +++ b/surfsense_backend/tests/unit/platforms/amazon/test_locale.py @@ -0,0 +1,33 @@ +from __future__ import annotations + +from app.proprietary.platforms.amazon.locale import ( + accept_language_for, + proxy_country_for, +) + + +def test_proxy_country_for_confirmed_marketplaces(): + assert proxy_country_for("com") == "us" + assert proxy_country_for("co.uk") == "gb" + assert proxy_country_for("de") == "de" + assert proxy_country_for("fr") == "fr" + assert proxy_country_for("it") == "it" + assert proxy_country_for("es") == "es" + + +def test_proxy_country_for_unknown_marketplace_falls_back(): + assert proxy_country_for("co.jp") is None + assert proxy_country_for(None) is None + + +def test_accept_language_for_marketplaces(): + assert accept_language_for("co.uk") == "en-GB" + assert accept_language_for("de") == "de-DE" + assert accept_language_for("fr") == "fr-FR" + assert accept_language_for("it") == "it-IT" + assert accept_language_for("es") == "es-ES" + + +def test_accept_language_for_unknown_marketplace_defaults_to_us_english(): + assert accept_language_for("co.jp") == "en-US" + assert accept_language_for(None) == "en-US" diff --git a/surfsense_backend/tests/unit/platforms/amazon/test_parsers.py b/surfsense_backend/tests/unit/platforms/amazon/test_parsers.py new file mode 100644 index 000000000..e09bbaf54 --- /dev/null +++ b/surfsense_backend/tests/unit/platforms/amazon/test_parsers.py @@ -0,0 +1,176 @@ +from __future__ import annotations + +from pathlib import Path + +from app.proprietary.platforms.amazon import fetch +from app.proprietary.platforms.amazon.fetch import ( + FetchResult, + get_location_session, + is_blocked, + should_localize, +) +from app.proprietary.platforms.amazon.parsers import ( + _float, + parse_aod_offers, + parse_bestsellers_page, + parse_product, + parse_search_page, + parse_seller, +) + +_FIXTURES = Path(__file__).parent / "fixtures" + + +def _fixture(name: str) -> str: + return (_FIXTURES / name).read_text(encoding="utf-8") + + +def test_product_parser_extracts_core_public_fields(): + item = parse_product( + _fixture("product.html"), + asin="B09V3KXJPB", + url="https://www.amazon.com/dp/B09V3KXJPB", + ) + + assert item["title"] == "Example Wireless Headphones" + assert item["price"] == {"value": 49.99, "currency": "USD"} + assert item["listPrice"] == {"value": 79.99, "currency": "USD"} + assert item["stars"] == 4.6 + assert item["reviewsCount"] == 1234 + assert item["inStock"] is True + assert item["features"] == ["Long battery life", "Noise cancelling"] + assert item["starsBreakdown"]["5star"] == 0.8 + assert "B09V3KXJP1" in item["variantAsins"] + assert item["variantAttributes"][0]["name"] == "color_name" + assert item["productPageReviews"][0]["id"] == "R1" + + +def test_block_detection_handles_status_and_markup(): + blocked = _fixture("blocked.html") + assert is_blocked(blocked, 200) + assert is_blocked("", 503) + assert not is_blocked(_fixture("product.html"), 200) + + +def test_block_detection_handles_waf_header_and_body_markers(): + assert is_blocked( + "ordinary status", + 200, + {"X-Amzn-Waf-Action": "challenge"}, + ) + assert is_blocked( + "", 200 + ) + assert is_blocked( + '', 200 + ) + + +def test_float_handles_us_and_eu_price_formats(): + assert _float("1.234,56 €") == 1234.56 + assert _float("12,99 €") == 12.99 + assert _float("$1,234.56") == 1234.56 + assert _float("1234") == 1234.0 + + +def test_block_retry_proxy_uses_fresh_country_session(monkeypatch): + geo_calls: list[str | None] = [] + sticky_calls: list[tuple[str, str | None]] = [] + + def get_geo_proxy_url(country: str | None = None): + geo_calls.append(country) + return f"http://geo-{country}" + + def get_sticky_proxy_url(session_id: str, country: str | None = None): + sticky_calls.append((session_id, country)) + return f"http://sticky-{country}-{session_id}" + + monkeypatch.setattr(fetch, "get_geo_proxy_url", get_geo_proxy_url) + monkeypatch.setattr(fetch, "get_sticky_proxy_url", get_sticky_proxy_url) + + first = fetch._selected_proxy(None, "fr", 1, "https://www.amazon.fr/s?k=x") + second = fetch._selected_proxy(None, "fr", 2, "https://www.amazon.fr/s?k=x") + + assert first == "http://geo-fr" + assert second.startswith("http://sticky-fr-amazon-fr-2-") + assert geo_calls == ["fr"] + assert sticky_calls[0][1] == "fr" + + +def test_search_parser_extracts_cards_and_provenance(): + cards = parse_search_page(_fixture("search.html"), page=2) + + assert [card["asin"] for card in cards] == ["B09V3KXJPB", "B09V3KXJP1"] + assert cards[0]["categoryPageData"] == { + "position": 1, + "page": 2, + "isSponsored": False, + "isBestSeller": True, + } + assert cards[1]["categoryPageData"]["isSponsored"] is True + + +def test_offer_and_seller_parsers(): + offers = parse_aod_offers(_fixture("offers.html")) + seller = parse_seller(_fixture("seller.html"), seller_id="SELLER123") + + assert len(offers) == 2 + assert offers[0]["price"]["value"] == 47.5 + assert offers[0]["seller"]["id"] == "SELLER123" + assert offers[0]["isPinnedOffer"] is True + assert seller["name"] == "Example Retailer" + assert seller["averageRating"] == 4.8 + assert seller["reviewsCount"] == 1875 + + +def test_bestsellers_parser_extracts_ranked_products(): + items = parse_bestsellers_page(_fixture("bestsellers.html")) + + assert [item["asin"] for item in items] == ["B09V3KXJPB", "B09V3KXJP1"] + assert [item["bestsellerPageData"]["rank"] for item in items] == [1, 2] + + +def test_location_route_filter_is_explicit(): + routes = ["PRODUCT", "OFFERS"] + assert should_localize("product", routes) + assert not should_localize("search", routes) + + +async def test_location_session_mints_once_and_reuses_cache(monkeypatch): + fetch._location_sessions.clear() + calls: list[str] = [] + + async def sticky_proxy(_session_id: str, _country: str | None = None): + return "http://sticky-proxy" + + async def fetch_page(url: str, **_kwargs): + calls.append(url) + if len(calls) == 1: + return FetchResult( + status=200, + html='', + url=url, + cookies={"session-id": "session", "ubid-main": "ubid"}, + ) + return FetchResult( + status=200, + html='{"isValidAddress":1}', + url=url, + cookies={"session-id": "session"}, + ) + + monkeypatch.setattr(fetch, "_sticky_proxy", sticky_proxy) + monkeypatch.setattr(fetch, "fetch_page", fetch_page) + + first = await get_location_session( + "www.amazon.com", zip_code="10001", country_code="US" + ) + second = await get_location_session( + "www.amazon.com", zip_code="10001", country_code="US" + ) + + assert first is second + assert first is not None + assert first.proxy == "http://sticky-proxy" + assert first.country_code == "US" + assert len(calls) == 2 diff --git a/surfsense_backend/tests/unit/platforms/amazon/test_proxy.py b/surfsense_backend/tests/unit/platforms/amazon/test_proxy.py new file mode 100644 index 000000000..8302fd1cb --- /dev/null +++ b/surfsense_backend/tests/unit/platforms/amazon/test_proxy.py @@ -0,0 +1,59 @@ +from __future__ import annotations + +import pytest + +from app.config import Config +from app.utils.proxy.providers.dataimpulse import DataImpulseProvider + + +def test_dataimpulse_sticky_url_is_deterministic(monkeypatch): + monkeypatch.setattr( + Config, + "PROXY_URL", + "http://token__cr.us:secret@gw.dataimpulse.com:823", + ) + provider = DataImpulseProvider() + + first = provider.get_sticky_proxy_url("location-123") + second = provider.get_sticky_proxy_url("location-123") + + assert first == second + assert "token__cr.us__sid.location-123" in first + assert first.endswith("@gw.dataimpulse.com:823") + + +def test_dataimpulse_sticky_url_replaces_existing_session(monkeypatch): + monkeypatch.setattr( + Config, + "PROXY_URL", + "http://token__cr.us__sid.old:secret@gw.dataimpulse.com:823", + ) + + result = DataImpulseProvider().get_sticky_proxy_url("new") + + assert "__sid.old" not in result + assert result.count("__sid.new") == 1 + + +def test_dataimpulse_sticky_url_rewrites_country(monkeypatch): + monkeypatch.setattr( + Config, + "PROXY_URL", + "http://token__cr.us:secret@gw.dataimpulse.com:823", + ) + + result = DataImpulseProvider().get_sticky_proxy_url("new", "gb") + + assert "token__cr.gb__sid.new" in result + assert "__cr.us" not in result + + +def test_dataimpulse_sticky_url_rejects_empty_session(monkeypatch): + monkeypatch.setattr( + Config, + "PROXY_URL", + "http://token:secret@gw.dataimpulse.com:823", + ) + + with pytest.raises(ValueError): + DataImpulseProvider().get_sticky_proxy_url("...") diff --git a/surfsense_backend/tests/unit/platforms/amazon/test_skeleton.py b/surfsense_backend/tests/unit/platforms/amazon/test_skeleton.py new file mode 100644 index 000000000..b99644144 --- /dev/null +++ b/surfsense_backend/tests/unit/platforms/amazon/test_skeleton.py @@ -0,0 +1,177 @@ +"""Offline contract tests for the Amazon Product Scraper. + +Deterministic (no network, no live Amazon HTML): asserts the input defaults, the +full output-item serialization contract, the error-item shape, and URL +classification. The live fetch / parse flows are exercised by the e2e script in +later milestones, not here. +""" + +from __future__ import annotations + +from app.proprietary.platforms.amazon import ( + AmazonScrapeInput, + ErrorItem, + ProductItem, + scrape_products, +) +from app.proprietary.platforms.amazon.url_resolver import extract_asin, resolve_url + +# A complete input payload should validate without removing optional fields. +_SPEC_INPUT = { + "categoryOrProductUrls": [{"url": "https://www.amazon.com/s?k=keyboard"}], + "maxItemsPerStartUrl": 100, + "language": "en", + "proxyCountry": "AUTO_SELECT_PROXY_COUNTRY", + "maxSearchPagesPerStartUrl": 9999, + "maxOffers": 0, + "scrapeSellers": False, + "useCaptchaSolver": False, + "scrapeProductVariantPrices": False, + "scrapeProductDetails": True, + "countryCode": "US", + "zipCode": "10001", + "locationDeliverableRoutes": ["PRODUCT", "SEARCH", "OFFERS"], +} + + +def test_input_has_no_auth_fields(): + # Public, anonymous only: no auth-shaped field may exist on the input surface. + forbidden = {"username", "password", "token", "login", "auth", "credentials"} + assert forbidden.isdisjoint(AmazonScrapeInput.model_fields) + + +def test_scrape_input_defaults_match_contract(): + inp = AmazonScrapeInput( + categoryOrProductUrls=[{"url": "https://www.amazon.com/dp/B0"}] + ) + assert inp.maxItemsPerStartUrl is None + assert inp.maxSearchPagesPerStartUrl == 9999 + assert inp.maxProductVariantsAsSeparateResults == 0 + assert inp.maxOffers == 0 + assert inp.proxyCountry == "AUTO_SELECT_PROXY_COUNTRY" + assert inp.language is None + assert inp.countryCode is None + assert inp.zipCode is None + assert inp.locationDeliverableRoutes == ["PRODUCT", "SEARCH", "OFFERS"] + assert inp.scrapeSellers is False + assert inp.useCaptchaSolver is False + assert inp.scrapeProductVariantPrices is False + assert inp.scrapeProductDetails is True + + +def test_complete_payload_validates(): + inp = AmazonScrapeInput(**_SPEC_INPUT) + assert inp.maxItemsPerStartUrl == 100 + assert inp.countryCode == "US" + + +def test_input_allows_extra_inert_fields(): + # extra="allow": unknown add-ons (e.g. upstream proxy config) are accepted. + inp = AmazonScrapeInput( + categoryOrProductUrls=[{"url": "https://www.amazon.com/dp/B0"}], + proxyConfiguration={"useResidentialProxy": True}, + someFutureAddon=123, + ) + assert inp.model_dump().get("someFutureAddon") == 123 + + +def test_output_item_serializes_full_shape(): + item = ProductItem(asin="B08EXAMPLE01").to_output() + assert item["asin"] == "B08EXAMPLE01" + # Unsourced scalars are still present as None (consumers never KeyError). + assert item["title"] is None + assert item["monthlyPurchaseVolume"] is None + # List fields default to []. + assert item["features"] == [] + assert item["offers"] == [] + assert item["productPageReviews"] == [] + # Typed-but-unset nested objects are None. + assert item["price"] is None + assert item["seller"] is None + + +def test_stars_breakdown_round_trips_digit_keys(): + item = ProductItem( + stars=4.8, starsBreakdown={"5star": 0.86, "1star": 0.01} + ).to_output() + # by_alias serialization restores Amazon's digit-prefixed keys. + assert item["starsBreakdown"]["5star"] == 0.86 + assert item["starsBreakdown"]["1star"] == 0.01 + assert item["starsBreakdown"]["4star"] is None + + +def test_error_item_shape(): + err = ErrorItem( + error="product_not_found", + errorDescription="Loaded a 404 page.", + input="https://www.amazon.com/dp/B0XXXXXXXX", + url="https://www.amazon.com/dp/B0XXXXXXXX", + ).model_dump() + assert set(err) >= {"error", "errorDescription", "input", "url"} + assert err["error"] == "product_not_found" + + +def test_extract_asin(): + assert extract_asin("https://www.amazon.com/dp/B09X7MPX8L") == "B09X7MPX8L" + assert extract_asin("https://www.amazon.com/gp/product/B09X7MPX8L/ref=x") == ( + "B09X7MPX8L" + ) + assert extract_asin("https://www.amazon.com/s?k=keyboard") is None + + +def test_resolve_url_classifies_all_kinds(): + product = resolve_url("https://www.amazon.com/dp/B0EXAMPLE1") + assert product is not None + assert product.kind == "product" + assert product.asin == "B0EXAMPLE1" + assert product.marketplace == "com" + + search = resolve_url("https://www.amazon.com/s?k=keyboard") + assert search is not None and search.kind == "search" + + # A category URL with no /s path but a bbn/rh query still classifies as search. + category = resolve_url( + "https://www.amazon.de/s?i=specialty-aps&bbn=16225007011&rh=n%3A16225007011" + ) + assert category is not None + assert category.kind == "search" + assert category.marketplace == "de" + + bestsellers = resolve_url( + "https://www.amazon.com/Best-Sellers-Electronics/zgbs/electronics" + ) + assert bestsellers is not None and bestsellers.kind == "bestsellers" + + shortened = resolve_url("https://a.co/d/abcd123") + assert shortened is not None and shortened.kind == "shortened" + + +def test_resolve_url_rejects_non_amazon_and_unrecognized(): + assert resolve_url("https://example.com/dp/B08EXAMPLE1") is None + # An Amazon host but an unrecognized path (e.g. the homepage) is unrecognized. + assert resolve_url("https://www.amazon.com/") is None + + +async def test_iter_products_unrecognized_yields_error(): + # A junk / non-Amazon URL yields exactly one invalid_url error item. + items = await scrape_products( + AmazonScrapeInput(categoryOrProductUrls=[{"url": "https://example.com/foo"}]) + ) + assert len(items) == 1 + assert items[0]["error"] == "invalid_url" + assert items[0]["input"] == "https://example.com/foo" + + +def test_valid_product_url_is_ready_for_dispatch(): + resolved = resolve_url("https://www.amazon.com/dp/B0EXAMPLE1") + assert resolved is not None + assert resolved.kind == "product" + + +async def test_iter_products_missing_url_key_yields_error(): + # An entry without a "url" key is treated as an invalid start URL, not a crash. + items = await scrape_products( + AmazonScrapeInput(categoryOrProductUrls=[{"noturl": "x"}]) + ) + assert len(items) == 1 + assert items[0]["error"] == "invalid_url" diff --git a/surfsense_backend/tests/unit/tasks/celery_tasks/test_schedule_checker_task.py b/surfsense_backend/tests/unit/tasks/celery_tasks/test_schedule_checker_task.py index 4bca74135..5193d07a0 100644 --- a/surfsense_backend/tests/unit/tasks/celery_tasks/test_schedule_checker_task.py +++ b/surfsense_backend/tests/unit/tasks/celery_tasks/test_schedule_checker_task.py @@ -59,7 +59,7 @@ def _due_connector(connector_type: SearchSourceConnectorType) -> SimpleNamespace return SimpleNamespace( id=42, connector_type=connector_type, - search_space_id=7, + workspace_id=7, user_id="00000000-0000-0000-0000-000000000001", config={}, periodic_indexing_enabled=True, @@ -86,13 +86,8 @@ async def _run_checker(monkeypatch: pytest.MonkeyPatch, connector: SimpleNamespa @pytest.mark.asyncio -async def test_due_bookstack_connector_dispatches_indexing_task(monkeypatch): - """A due BookStack connector must dispatch index_bookstack_pages_task. - - Regression test for the connector type missing from the scheduler's - task_map, which made periodic BookStack syncs silently no-op with only a - "No task found" warning. - """ +async def test_due_bookstack_connector_disables_deprecated_indexing(monkeypatch): + """A due BookStack connector is retired from periodic indexing, not dispatched.""" from app.tasks.celery_tasks import connector_tasks task_mock = MagicMock() @@ -101,16 +96,9 @@ async def test_due_bookstack_connector_dispatches_indexing_task(monkeypatch): connector = _due_connector(SearchSourceConnectorType.BOOKSTACK_CONNECTOR) session = await _run_checker(monkeypatch, connector) - task_mock.delay.assert_called_once_with( - connector.id, - connector.search_space_id, - str(connector.user_id), - None, - None, - ) - # The next run must be rescheduled, otherwise the connector stays "due" - # and is re-examined every minute. - assert connector.next_scheduled_at > datetime.now(UTC) + task_mock.delay.assert_not_called() + assert connector.periodic_indexing_enabled is False + assert connector.next_scheduled_at is None assert session.commits == 1 diff --git a/surfsense_backend/tests/unit/utils/proxy/test_dataimpulse_provider.py b/surfsense_backend/tests/unit/utils/proxy/test_dataimpulse_provider.py index af1288046..a9dd30318 100644 --- a/surfsense_backend/tests/unit/utils/proxy/test_dataimpulse_provider.py +++ b/surfsense_backend/tests/unit/utils/proxy/test_dataimpulse_provider.py @@ -44,6 +44,51 @@ def test_location_stops_at_next_param(monkeypatch: pytest.MonkeyPatch) -> None: assert DataImpulseProvider().get_location() == "de" +def test_geo_proxy_url_rewrites_country_suffix(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(Config, "PROXY_URL", _URL) + + result = DataImpulseProvider().get_geo_proxy_url("gb") + + assert result == "http://tok123__cr.gb:secret@gw.dataimpulse.com:823" + + +def test_geo_proxy_url_replaces_existing_country_once( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr( + Config, + "PROXY_URL", + "http://tok123__cr.us__sid.old:secret@gw.dataimpulse.com:823", + ) + + result = DataImpulseProvider().get_geo_proxy_url("de") + + assert "__cr.us" not in result + assert result.count("__cr.de") == 1 + assert "__sid.old" not in result + + +def test_geo_proxy_url_without_country_keeps_base_url( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(Config, "PROXY_URL", _URL) + + assert DataImpulseProvider().get_geo_proxy_url(None) == _URL + assert DataImpulseProvider().get_geo_proxy_url("") == _URL + + +def test_sticky_proxy_url_composes_country_and_session( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(Config, "PROXY_URL", _URL) + + result = DataImpulseProvider().get_sticky_proxy_url("location-123", "gb") + + assert result == ( + "http://tok123__cr.gb__sid.location-123:secret@gw.dataimpulse.com:823" + ) + + def test_no_country_suffix_yields_empty_location( monkeypatch: pytest.MonkeyPatch, ) -> None: diff --git a/surfsense_mcp/mcp_server/features/scrapers/__init__.py b/surfsense_mcp/mcp_server/features/scrapers/__init__.py index 9aabbe0e5..e6f1055cd 100644 --- a/surfsense_mcp/mcp_server/features/scrapers/__init__.py +++ b/surfsense_mcp/mcp_server/features/scrapers/__init__.py @@ -14,6 +14,7 @@ from ...core.client import SurfSenseClient from ...core.workspace_context import WorkspaceContext from . import run_history from .platforms import ( + amazon, google_maps, google_search, instagram, @@ -31,6 +32,7 @@ _REGISTRARS = ( instagram, tiktok, google_maps, + amazon, run_history, ) diff --git a/surfsense_mcp/mcp_server/features/scrapers/platforms/amazon.py b/surfsense_mcp/mcp_server/features/scrapers/platforms/amazon.py new file mode 100644 index 000000000..09b703157 --- /dev/null +++ b/surfsense_mcp/mcp_server/features/scrapers/platforms/amazon.py @@ -0,0 +1,145 @@ +"""Amazon product scraper tool.""" + +from __future__ import annotations + +from typing import Annotated + +from mcp.server.fastmcp import FastMCP +from pydantic import Field + +from ....core.client import SurfSenseClient +from ....core.rendering import ResponseFormatParam +from ....core.workspace_context import WorkspaceContext, WorkspaceParam +from ..annotations import SCRAPE +from ..capability import run_scraper + + +def register(mcp: FastMCP, client: SurfSenseClient, context: WorkspaceContext) -> None: + """Register the Amazon product tool.""" + + @mcp.tool( + name="surfsense_amazon_scrape", + title="Scrape Amazon products", + annotations=SCRAPE, + structured_output=False, + ) + async def amazon_scrape( + urls: Annotated[ + list[str] | None, + Field( + description="Amazon product, search, category, best-seller, or short " + "(a.co / amzn.to) URLs. Provide urls OR search_terms." + ), + ] = None, + search_terms: Annotated[ + list[str] | None, + Field( + description="Search phrases run on the Amazon domain, e.g. " + "['wireless earbuds']. Provide search_terms OR urls." + ), + ] = None, + max_items: Annotated[ + int, + Field( + ge=1, + le=100, + description="Max products per search term or category/best-seller URL.", + ), + ] = 10, + domain: Annotated[ + str, + Field( + description="Amazon marketplace domain, e.g. 'www.amazon.com', " + "'www.amazon.co.uk'." + ), + ] = "www.amazon.com", + include_details: Annotated[ + bool, + Field( + description="Fetch full product detail pages. False returns faster " + "card-only results from search/category listings." + ), + ] = True, + max_offers: Annotated[ + int, + Field( + ge=0, + le=100, + description="Extra marketplace offers to fetch per product. " + "0 returns the featured offer only.", + ), + ] = 0, + include_sellers: Annotated[ + bool, + Field( + description="Enrich the featured product and each offer with the " + "seller's public profile summary." + ), + ] = False, + max_variants: Annotated[ + int, + Field( + ge=0, + le=100, + description="Product variants (e.g. colors, sizes) to return as " + "separate results. 0 disables variant expansion.", + ), + ] = 0, + include_variant_prices: Annotated[ + bool, + Field( + description="Attach per-variant prices (one extra request per variant)." + ), + ] = False, + country_code: Annotated[ + str | None, + Field(description="Two-letter delivery country for localized pricing, " + "e.g. 'us'."), + ] = None, + zip_code: Annotated[ + str | None, + Field( + description="Delivery ZIP/postal code for localized availability and " + "pricing, e.g. '10001'." + ), + ] = None, + language: Annotated[ + str | None, + Field(description="Content language for the domain, e.g. 'en'."), + ] = None, + workspace: WorkspaceParam = None, + response_format: ResponseFormatParam = "markdown", + ) -> str: + """Scrape public Amazon product data by URL or search term. + + Use this for product research: title, price, list price, rating and + review breakdown, availability, images, features, best-seller ranks, + marketplace offers, sellers, and on-page reviews. Only public, + anonymous data — no login. Provide search_terms to discover products + or urls to target specific products, searches, categories, or + best-seller pages. + Example: search_terms=['mechanical keyboard'], domain='www.amazon.com', + max_items=5, max_offers=3. + """ + return await run_scraper( + client, + context, + platform="amazon", + verb="scrape", + payload={ + "urls": urls, + "search_terms": search_terms, + "max_items": max_items, + "domain": domain, + "include_details": include_details, + "max_offers": max_offers, + "include_sellers": include_sellers, + "max_variants": max_variants, + "include_variant_prices": include_variant_prices, + "country_code": country_code, + "zip_code": zip_code, + "language": language, + }, + workspace=workspace, + response_format=response_format, + ) diff --git a/surfsense_mcp/mcp_server/selfcheck.py b/surfsense_mcp/mcp_server/selfcheck.py index bb61151bc..ffce29058 100644 --- a/surfsense_mcp/mcp_server/selfcheck.py +++ b/surfsense_mcp/mcp_server/selfcheck.py @@ -29,6 +29,7 @@ EXPECTED_TOOLS = { "surfsense_tiktok_trending", "surfsense_google_maps_scrape", "surfsense_google_maps_reviews", + "surfsense_amazon_scrape", "surfsense_instagram_scrape", "surfsense_instagram_details", "surfsense_list_scraper_runs", diff --git a/surfsense_web/app/dashboard/[workspace_id]/playground/components/playground-runner.tsx b/surfsense_web/app/dashboard/[workspace_id]/playground/components/playground-runner.tsx index 6b426b196..32624ca55 100644 --- a/surfsense_web/app/dashboard/[workspace_id]/playground/components/playground-runner.tsx +++ b/surfsense_web/app/dashboard/[workspace_id]/playground/components/playground-runner.tsx @@ -14,6 +14,11 @@ import { AppError } from "@/lib/error"; import { findVerb } from "@/lib/playground/catalog"; import { formatCost, formatDuration, formatPricing } from "@/lib/playground/format"; import { buildPayload, initialFormValues, parseSchemaFields } from "@/lib/playground/json-schema"; +import { + AmazonMarketplaceHint, + getAmazonFieldOptions, + hasAmazonFranceValue, +} from "@/lib/playground/platform-overrides/amazon"; import { ApiReference } from "./api-reference"; import { OutputViewer } from "./output-viewer"; import { RunProgressPanel } from "./run-progress-panel"; @@ -171,6 +176,8 @@ export function PlaygroundRunner({ workspaceId, platform, verb }: PlaygroundRunn [run.detail] ); const endpoint = `POST /workspaces/${workspaceId}/scrapers/${platform}/${verb}`; + const isAmazonScrape = platform === "amazon" && verb === "scrape"; + const hasAmazonFranceUrl = useMemo(() => hasAmazonFranceValue(values), [values]); useEffect(() => { const previousStatus = previousStatusRef.current; @@ -260,7 +267,9 @@ export function PlaygroundRunner({ workspaceId, platform, verb }: PlaygroundRunn values={values} onChange={handleChange} disabled={isRunning} + getFieldOptions={isAmazonScrape ? getAmazonFieldOptions : undefined} /> + {isAmazonScrape ? : null}