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..5100aaf01
--- /dev/null
+++ b/surfsense_backend/app/proprietary/platforms/amazon/scraper.py
@@ -0,0 +1,485 @@
+"""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 .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 _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,
+    )
+    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"
+    )
+    response = await fetch_page(resolved.url, cookies=cookies, proxy=proxy)
+    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,
+        )
+        if offers_html:
+            fields["offers"] = parse_aod_offers(offers_html, domain=resolved.domain)[
+                : 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
+            )
+            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")
+    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
+        )
+        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
+    )
+    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))
+        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