use get_response_json in get_category_products

Adds the body guard this site lacked; a non-dict body now logs and
returns before the fan-out instead of raising on searchTotalCount.

Dockerfile

@@ -6,7 +6,7 @@ RUN apt-get update && apt-get install -y --no-install-recommends git \
     && rm -rf /var/lib/apt/lists/*
 WORKDIR /app
 COPY pyproject.toml ./
-RUN uv pip install --system --no-cache -r pyproject.toml
+RUN uv pip install --system --no-cache --no-sources -r pyproject.toml
 COPY samsung_shop/ ./samsung_shop/
 COPY configs/proxies.txt ./proxies.txt
 RUN mkdir -p logs && chmod -R a+rwX /app

pyproject.toml

@@ -5,9 +5,11 @@ requires-python = ">=3.13"
 dependencies = [
     "aiohttp>=3.14.1",
     "aioproxies @ git+https://git.rethinkstudios.io/rethink-public/aioproxies.git",
+    "aioweb-tls[curl] @ git+https://git.rethinkstudios.io/rethink-public/aioweb_tls.git",
     "curl-cffi>=0.15.0",
     "log-setup @ git+https://git.rethinkstudios.io/rethink-public/log_setup.git",
-    "monitor-utils @ git+https://git.profitlounge.us/pl_devs/monitor-shared-utils.git",
+    "monitor-utils @ git+https://git.rethinkstudios.io/pl_devs/monitor-shared-utils.git",
+    "motor>=3.7.1",
     "tenacity>=9.1.4",
 ]
 
@@ -27,3 +29,8 @@ dev = [
     "pytest>=9.1.1",
     "pytest-asyncio>=1.4.0",
 ]
+
+# Local dev only: uv sync/run use the sibling checkout; the Dockerfile
+# passes --no-sources so images resolve the git URL above instead.
+[tool.uv.sources]
+monitor-utils = { path = "../monitor-shared-utils", editable = true }

samsung_shop/samsung_shop_monitor.py

@@ -1,53 +1,64 @@
 import asyncio
 import logging
+from urllib.parse import urlencode
 
-from curl_cffi import AsyncSession
+from aioweb_tls import TLSSession
 from monitor_utils.config import DevelopmentMode
-from monitor_utils.constants import EXPECTED_TASK_ERRORS
+from monitor_utils.constants import CYCLE_DELAY
 from monitor_utils.database import add_products_to_database
 from monitor_utils.models import AvailabilityInfo, PricingInfo, Product, ProductMetadata
 from monitor_utils.runner import (
-    execute_monitor_logic,
     execute_pagination,
+    fan_out_group,
     init_monitor,
+    run_bounded_cycle,
     setup_monitor_logging,
 )
-from monitor_utils.sessions import send_request
+from monitor_utils.sessions import get_response_json, send_request
 from motor.motor_asyncio import AsyncIOMotorCollection
 
-setup_monitor_logging(monitor_name="samsung_shop")
+setup_monitor_logging(monitor="samsung_shop")
 DEVELOPMENT_MODE = DevelopmentMode.PRODUCTION
 LOGGER = logging.getLogger(__name__)
 
 PRODUCTS_PER_PAGE = 16
+CATEGORY_CONCURRENCY = 10
+CATEGORY_SEMAPHORE = asyncio.Semaphore(CATEGORY_CONCURRENCY)
 
 
 async def main():
-    """Run the samsung shop monitor loop over all category ids using a base session."""
+    """Run the samsung shop monitor loop over all category ids on bounded cycles."""
     category_paths, collection, proxies = await init_monitor(
-        monitor_name="samsung_shop", development_mode=DEVELOPMENT_MODE
+        monitor="samsung_shop", development_mode=DEVELOPMENT_MODE
-    )
-    await execute_monitor_logic(
-        proxies=proxies,
-        category_identifiers=category_paths,
-        task_func=get_category_products,
-        collection=collection,
     )
+    while True:
+        await run_bounded_cycle(
+            category_paths,
+            get_category_products,
+            proxies,
+            collection,
+            headers={
+                "origin": "https://www.samsung.com",
+                "referer": "https://www.samsung.com/us/shop/offer-program/education/",
+            },
+        )
+        LOGGER.info("tasks completed sleeping", extra={"delay": CYCLE_DELAY})
+        await asyncio.sleep(CYCLE_DELAY)
 
 
 async def get_category_products(
     category_id: str,
-    session: AsyncSession,
+    session: TLSSession,
     collection: AsyncIOMotorCollection,
     offset=0,
 ):
     LOGGER.debug("getting products", extra={"category": category_id, "offset": offset})
-    try:
+    response = await send_request(
-        response = await send_request(
+        session,
-            session,
+        url="https://sribsrch.ecom.samsung.com/estoresearch-api/v1/scom/us/pf_search",
-            url="https://sribsrch.ecom.samsung.com/estoresearch-api/v1/scom/us/pf_search",
+        method="POST",
-            method="POST",
+        data=urlencode(
-            data={
+            {
                 "clientCode": "b2c",
                 "clientName": "scom_pf",
                 "firstSearchYN": "true",
@@ -57,24 +68,26 @@ async def get_category_products(
                 "requestCount": str(PRODUCTS_PER_PAGE),
                 "category_code": category_id,
                 "sort": "newest",
-            },
+            }
-        )
+        ),
-    except EXPECTED_TASK_ERRORS:
+        headers={"content-type": "application/x-www-form-urlencoded"},
-        LOGGER.warning(
+    )
-            "failed to get products",
-            extra={"category": category_id, "offset": offset},
-        )
-        raise
 
-    if offset != 0:
+    data = get_response_json(
-        await process_page(response, category_id, offset, collection)
+        response, logger=LOGGER, category=category_id, offset=offset
+    )
+    if data is None:
         return
 
-    page_task = asyncio.create_task(
+    if offset != 0:
-        process_page(response, category_id, offset, collection)
+        await process_page(data, category_id, offset, collection)
-    )
+        return
-    try:
+
-        total_products = response.get("searchTotalCount")
+    # offset zero owns the fan-out: the group binds this page's processing to the
+    # pagination it spawns, so a cancelled cycle takes both down together.
+    async with fan_out_group() as group:
+        group.create_task(process_page(data, category_id, offset, collection))
+        total_products = data.get("searchTotalCount")
         if total_products is None:
             LOGGER.warning("missing searchTotalCount", extra={"category": category_id})
             return
@@ -85,11 +98,10 @@ async def get_category_products(
             products_per_page=PRODUCTS_PER_PAGE,
             func=get_category_products,
             session=session,
+            semaphore=CATEGORY_SEMAPHORE,
             category_id=category_id,
             collection=collection,
         )
-    finally:
-        await page_task
 
 
 async def process_page(
@@ -100,34 +112,27 @@ async def process_page(
 ):
     """Extract and persist one page of category search results."""
 
-    try:
+    products = extract_products(response)
-        products = extract_products(response)
+    if products:
-        if products:
+        await add_products_to_database(
-            await add_products_to_database(
+            products,
-                products,
+            collection,
-                collection,
+            category_identifier=category_id,
-                category_identifier=category_id,
+            monitor="samsung_shop",
-                monitor_name="samsung_shop",
+        )
-            )
+        LOGGER.info(
-            LOGGER.info(
+            "added products",
-                "added products",
+            extra={
-                extra={
+                "category": category_id,
-                    "category": category_id,
+                "offset": offset,
-                    "offset": offset,
+                "count": len(products),
-                    "count": len(products),
+            },
-                },
+        )
-            )
+    else:
-        else:
+        LOGGER.debug(
-            LOGGER.debug(
+            "no products found",
-                "no products found",
-                extra={"category": category_id, "offset": offset},
-            )
-    except EXPECTED_TASK_ERRORS:
-        LOGGER.warning(
-            "failed to build products",
             extra={"category": category_id, "offset": offset},
         )
-        raise
 
 
 def extract_products(response: dict) -> list[Product]:
@@ -155,4 +160,5 @@ def extract_products(response: dict) -> list[Product]:
     return products
 
 
-asyncio.run(main())
+if __name__ == "__main__":
+    asyncio.run(main())