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(
         session,
         url="https://sribsrch.ecom.samsung.com/estoresearch-api/v1/scom/us/pf_search",
         method="POST",
-            data={
+        data=urlencode(
+            {
                 "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",
-            },
+            }
+        ),
+        headers={"content-type": "application/x-www-form-urlencoded"},
     )
-    except EXPECTED_TASK_ERRORS:
-        LOGGER.warning(
-            "failed to get products",
-            extra={"category": category_id, "offset": offset},
-        )
-        raise
 
-    if offset != 0:
-        await process_page(response, category_id, offset, collection)
+    data = get_response_json(
+        response, logger=LOGGER, category=category_id, offset=offset
+    )
+    if data is None:
         return
 
-    page_task = asyncio.create_task(
-        process_page(response, category_id, offset, collection)
-    )
-    try:
-        total_products = response.get("searchTotalCount")
+    if offset != 0:
+        await process_page(data, category_id, offset, collection)
+        return
+
+    # 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,14 +112,13 @@ async def process_page(
 ):
     """Extract and persist one page of category search results."""
 
-    try:
     products = extract_products(response)
     if products:
         await add_products_to_database(
             products,
             collection,
             category_identifier=category_id,
-                monitor_name="samsung_shop",
+            monitor="samsung_shop",
         )
         LOGGER.info(
             "added products",
@@ -122,17 +133,12 @@ async def process_page(
             "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]:
     products = []
     for current_product in response["searchResults"]:
+        current_price = current_product["sale_price"]
         product = Product(
             product_id=current_product["modelCode"],
             title=current_product["name"],
@@ -142,7 +148,8 @@ def extract_products(response: dict) -> list[Product]:
             ),
             pricing=PricingInfo(
                 original_price=current_product["msrp_price"],
-                current_price=current_product["sale_price"],
+                current_price=current_price,
+                edu_price=current_price,
             ),
             availability=AvailabilityInfo(
                 available=(current_product["stockFlag"] == "Y")
@@ -153,4 +160,5 @@ def extract_products(response: dict) -> list[Product]:
     return products
 
 
+if __name__ == "__main__":
     asyncio.run(main())