added aioweb/refactor

samsung_shop/samsung_shop_monitor.py

@@ -1,18 +1,19 @@
 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, EXPECTED_TASK_ERRORS
 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,
+    run_monitor_cycle,
     execute_pagination,
     init_monitor,
     setup_monitor_logging,
 )
-from monitor_utils.sessions import send_request
+from monitor_utils.sessions import get_default_session, send_request
 from motor.motor_asyncio import AsyncIOMotorCollection
 
 setup_monitor_logging(monitor_name="samsung_shop")
@@ -20,6 +21,8 @@ DEVELOPMENT_MODE = DevelopmentMode.PRODUCTION
 LOGGER = logging.getLogger(__name__)
 
 PRODUCTS_PER_PAGE = 16
+CATEGORY_CONCURRENCY = 5
+CATEGORY_SEMAPHORE = asyncio.Semaphore(CATEGORY_CONCURRENCY)
 
 
 async def main():
@@ -27,17 +30,21 @@ async def main():
     category_paths, collection, proxies = await init_monitor(
         monitor_name="samsung_shop", development_mode=DEVELOPMENT_MODE
     )
-    await execute_monitor_logic(
+    async with get_default_session(proxies) as session:
-        proxies=proxies,
+        while True:
-        category_identifiers=category_paths,
+            await run_monitor_cycle(
+                category_paths,
                 task_func=get_category_products,
+                session=session,
                 collection=collection,
             )
+            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,
 ):
@@ -47,7 +54,8 @@ async def get_category_products(
             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,7 +65,9 @@ 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(
@@ -66,15 +76,14 @@ async def get_category_products(
         )
         raise
 
+    data = response.json()
     if offset != 0:
-        await process_page(response, category_id, offset, collection)
+        await process_page(data, category_id, offset, collection)
         return
 
-    page_task = asyncio.create_task(
+    page_task = asyncio.create_task(process_page(data, category_id, offset, collection))
-        process_page(response, category_id, offset, collection)
-    )
     try:
-        total_products = response.get("searchTotalCount")
+        total_products = data.get("searchTotalCount")
         if total_products is None:
             LOGGER.warning("missing searchTotalCount", extra={"category": category_id})
             return
@@ -85,6 +94,7 @@ 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,
         )