feat(amazon): add scrape capability schema, executor, definition

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

surfsense_backend/app/capabilities/amazon/scrape/__init__.py

@@ -0,0 +1,3 @@
+"""Amazon product scraping capability."""
+
+from __future__ import annotations

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)

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

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"))
+        )