feat(walmart): add scrape and reviews capabilities

surfsense_backend/app/capabilities/walmart/__init__.py

@@ -0,0 +1,6 @@
+"""``walmart.*`` namespace: platform-native Walmart data verbs."""
+
+from __future__ import annotations
+
+from app.capabilities.walmart.reviews import definition as _reviews  # noqa: F401
+from app.capabilities.walmart.scrape import definition as _scrape  # noqa: F401

surfsense_backend/app/capabilities/walmart/reviews/__init__.py

@@ -0,0 +1,3 @@
+"""Walmart deep review scraping capability."""
+
+from __future__ import annotations

surfsense_backend/app/capabilities/walmart/reviews/definition.py

@@ -0,0 +1,24 @@
+"""``walmart.reviews`` capability registration (billed per review; see config
+``WALMART_MICROS_PER_REVIEW``)."""
+
+from __future__ import annotations
+
+from app.capabilities.core import BillingUnit, Capability, register_capability
+from app.capabilities.walmart.reviews.executor import build_reviews_executor
+from app.capabilities.walmart.reviews.schemas import ReviewsInput, ReviewsOutput
+
+WALMART_REVIEWS = Capability(
+    name="walmart.reviews",
+    description=(
+        "Fetch deep paginated public Walmart product reviews with ratings, text, "
+        "authors, verified-purchase flags, images, and seller responses. Use "
+        "product urls or item ids."
+    ),
+    input_schema=ReviewsInput,
+    output_schema=ReviewsOutput,
+    executor=build_reviews_executor(),
+    billing_unit=BillingUnit.WALMART_REVIEW,
+    docs_url="/docs/connectors/native/walmart",
+)
+
+register_capability(WALMART_REVIEWS)

surfsense_backend/app/capabilities/walmart/reviews/executor.py

@@ -0,0 +1,40 @@
+"""``walmart.reviews`` executor: verb input → scraper → review items."""
+
+from __future__ import annotations
+
+from collections.abc import Awaitable, Callable
+
+from app.capabilities.core import Executor
+from app.capabilities.core.progress import emit_progress
+from app.capabilities.walmart.reviews.schemas import ReviewsInput, ReviewsOutput
+from app.proprietary.platforms.walmart import WalmartReviewsInput, scrape_reviews
+
+ReviewsFn = Callable[..., Awaitable[list[dict]]]
+
+
+def build_reviews_executor(scrape_fn: ReviewsFn | None = None) -> Executor:
+    """Bind the executor to a reviews scraper fn (defaults to the proprietary actor)."""
+    scrape_fn = scrape_fn or scrape_reviews
+
+    async def execute(payload: ReviewsInput) -> ReviewsOutput:
+        input_model = WalmartReviewsInput(
+            itemIds=payload.sources(),
+            maxReviews=payload.max_reviews,
+            sort=payload.sort_by,
+        )
+        emit_progress(
+            "starting",
+            "Fetching Walmart reviews",
+            total=payload.estimated_units,
+            unit="review",
+        )
+        items = await scrape_fn(input_model, limit=payload.estimated_units)
+        emit_progress(
+            "done",
+            f"Scraped {sum('error' not in item for item in items)} review(s)",
+            current=len(items),
+            unit="review",
+        )
+        return ReviewsOutput(items=items)
+
+    return execute

surfsense_backend/app/capabilities/walmart/reviews/schemas.py

@@ -0,0 +1,72 @@
+"""``walmart.reviews`` I/O contracts.
+
+A lean surface over ``WalmartReviewsInput``; the scraper's ``ReviewItem`` is
+reused verbatim as the output element. Accepts product URLs or bare item ids —
+both resolve to a ``usItemId`` the reviews page is keyed on.
+"""
+
+from __future__ import annotations
+
+from typing import Literal
+
+from pydantic import BaseModel, Field, model_validator
+
+from app.proprietary.platforms.walmart import ReviewItem
+
+MAX_WALMART_REVIEW_SOURCES = 20
+
+
+class ReviewsInput(BaseModel):
+    urls: list[str] = Field(
+        default_factory=list,
+        max_length=MAX_WALMART_REVIEW_SOURCES,
+        description=(
+            "Walmart product URLs (/ip/...) or reviews URLs to fetch reviews for. "
+            "Provide these OR item_ids (at least one is required)."
+        ),
+    )
+    item_ids: list[str] = Field(
+        default_factory=list,
+        max_length=MAX_WALMART_REVIEW_SOURCES,
+        description="Walmart numeric item ids (usItemId) to fetch reviews for.",
+    )
+    max_reviews: int = Field(
+        default=200,
+        ge=1,
+        le=5000,
+        description="Max reviews to return per product (10 per page).",
+    )
+    sort_by: Literal["most-recent", "most-helpful", "rating-high", "rating-low"] = (
+        Field(default="most-recent", description="Review ordering.")
+    )
+
+    @model_validator(mode="after")
+    def _require_a_source(self) -> ReviewsInput:
+        if not (self.urls or self.item_ids):
+            raise ValueError("Provide at least one of 'urls' or 'item_ids'.")
+        return self
+
+    def sources(self) -> list[str]:
+        """URLs and item ids merged; the scraper resolves each to a usItemId."""
+        return [*self.urls, *self.item_ids]
+
+    @property
+    def estimated_units(self) -> int:
+        """Worst-case billable reviews: up to ``max_reviews`` per source."""
+        return (len(self.urls) + len(self.item_ids)) * self.max_reviews
+
+
+class ReviewsOutput(BaseModel):
+    items: list[ReviewItem] = Field(
+        default_factory=list,
+        description="One item per review, in the scraper's emission order.",
+    )
+
+    @property
+    def billable_units(self) -> int:
+        """One returned review = one billable unit; error items are not billed."""
+        return sum(
+            1
+            for item in self.items
+            if not (item.model_extra and item.model_extra.get("error"))
+        )

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

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

surfsense_backend/app/capabilities/walmart/scrape/definition.py

@@ -0,0 +1,22 @@
+"""Registration for the ``walmart.scrape`` capability."""
+
+from __future__ import annotations
+
+from app.capabilities.core import BillingUnit, Capability, register_capability
+from app.capabilities.walmart.scrape.executor import build_scrape_executor
+from app.capabilities.walmart.scrape.schemas import ScrapeInput, ScrapeOutput
+
+WALMART_SCRAPE = Capability(
+    name="walmart.scrape",
+    description=(
+        "Scrape public Walmart product details, search/category listings, "
+        "prices, sellers, variants, availability, and a sample of on-page reviews."
+    ),
+    input_schema=ScrapeInput,
+    output_schema=ScrapeOutput,
+    executor=build_scrape_executor(),
+    billing_unit=BillingUnit.WALMART_PRODUCT,
+    docs_url="/docs/connectors/native/walmart",
+)
+
+register_capability(WALMART_SCRAPE)

surfsense_backend/app/capabilities/walmart/scrape/executor.py

@@ -0,0 +1,46 @@
+"""Executor for the ``walmart.scrape`` capability."""
+
+from __future__ import annotations
+
+from collections.abc import Awaitable, Callable
+
+from app.capabilities.core import Executor
+from app.capabilities.core.progress import emit_progress
+from app.capabilities.walmart.scrape.schemas import (
+    MAX_WALMART_RESULTS,
+    ScrapeInput,
+    ScrapeOutput,
+)
+from app.proprietary.platforms.walmart import WalmartScrapeInput, 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:
+        input_model = WalmartScrapeInput(
+            startUrls=payload.start_urls(),
+            maxItemsPerStartUrl=payload.max_items,
+            includeDetails=payload.include_details,
+            includeReviewsSample=payload.include_reviews_sample,
+        )
+        emit_progress(
+            "starting",
+            "Scraping Walmart products",
+            total=payload.estimated_units,
+            unit="product",
+        )
+        items = await scrape_fn(input_model, limit=MAX_WALMART_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/walmart/scrape/schemas.py

@@ -0,0 +1,64 @@
+"""Input and output contracts for ``walmart.scrape``."""
+
+from __future__ import annotations
+
+from urllib.parse import quote_plus
+
+from pydantic import BaseModel, Field, model_validator
+
+from app.proprietary.platforms.walmart import ProductItem
+
+MAX_WALMART_SOURCES = 20
+MAX_WALMART_RESULTS = 1000
+
+
+class ScrapeInput(BaseModel):
+    """Agent-facing controls for public Walmart product discovery and enrichment."""
+
+    urls: list[str] = Field(default_factory=list, max_length=MAX_WALMART_SOURCES)
+    search_terms: list[str] = Field(
+        default_factory=list, max_length=MAX_WALMART_SOURCES
+    )
+    max_items: int = Field(default=10, ge=1, le=100)
+    include_details: bool = True
+    include_reviews_sample: bool = True
+
+    @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_WALMART_SOURCES:
+            raise ValueError(
+                f"Provide no more than {MAX_WALMART_SOURCES} combined sources."
+            )
+        return self
+
+    def start_urls(self) -> list[str]:
+        """Direct URLs plus a search URL synthesized per search term."""
+        searches = [
+            f"https://www.walmart.com/search?q={quote_plus(term)}"
+            for term in self.search_terms
+        ]
+        return [*self.urls, *searches]
+
+    @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) * self.max_items
+        return min(search_products + direct_products, MAX_WALMART_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"))
+        )