init commit

pyproject.toml

@@ -0,0 +1,29 @@
+[project]
+name = "samsung-shop-monitor"
+version = "0.1.0"
+requires-python = ">=3.13"
+dependencies = [
+    "aiohttp>=3.14.1",
+    "aioproxies @ git+https://git.rethinkstudios.io/rethink-public/aioproxies.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",
+    "tenacity>=9.1.4",
+]
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[tool.hatch.metadata]
+allow-direct-references = true
+
+[tool.hatch.build.targets.wheel]
+packages = ["samsung_shop"]
+
+[dependency-groups]
+dev = [
+    "flake8>=7.3.0",
+    "pytest>=9.1.1",
+    "pytest-asyncio>=1.4.0",
+]

samsung_shop/samsung_shop_monitor.py

@@ -0,0 +1,109 @@
+import asyncio
+import logging
+import re
+from urllib.parse import quote
+
+from curl_cffi import AsyncSession
+from monitor_utils.config import DevelopmentMode
+from monitor_utils.constants import 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, execute_pagination, init_monitor
+from monitor_utils.sessions import send_request
+from motor.motor_asyncio import AsyncIOMotorCollection
+
+DEVELOPMENT_MODE = DevelopmentMode.DEVELOPMENT
+LOGGER = logging.getLogger(__name__)
+
+
+async def main():
+    """Run the samsung shop monitor loop over all category ids using a base session."""
+    category_paths, collection, proxies = await init_monitor(
+        monitor_name="hp", development_mode=DEVELOPMENT_MODE
+    )
+    await execute_monitor_logic(
+        proxies=proxies,
+        category_identifiers=category_paths,
+        task_func=get_category_products,
+        collection=collection,
+    )
+
+
+async def get_category_products(
+    category_id: str,
+    session: AsyncSession,
+    collection: AsyncIOMotorCollection,
+    offset=30,
+):
+    # update offset to actual offset
+    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={
+                "clientCode": "b2c",
+                "clientName": "scom_pf",
+                "firstSearchYN": "true",
+                "countryCode": "us",
+                "storeID": "us_education",
+                "startIndex": "0",
+                "requestCount": "16",
+                "category_code": category_id,
+                "sort": "newest",
+            },
+        )
+        products = extract_products(response)
+        if products:
+            await add_products_to_database(
+                products,
+                collection,
+                category_identifier=category_id,
+                monitor_name="samsung_shop",
+            )
+            LOGGER.info(
+                "added products",
+                extra={
+                    "category": category_id,
+                    "offset": offset,
+                    "count": len(products),
+                },
+            )
+        else:
+            LOGGER.debug(
+                "no products found",
+                extra={"category": category_id, "offset": offset},
+            )
+    except EXPECTED_TASK_ERRORS:
+        LOGGER.warning(
+            "failed to get products",
+            extra={"category": category_id, "offset": offset},
+        )
+        raise
+
+
+def extract_products(response: dict) -> list[Product]:
+    products = []
+    for current_product in response["searchResults"]:
+        product = Product(
+            product_id=current_product["modelCode"],
+            title=current_product["name"],
+            url=f"https://www.samsung.com{current_product['mlpUrl']}",
+            metadata=ProductMetadata(
+                image_url=current_product["images"]["largeImage"]["url"]
+            ),
+            pricing=PricingInfo(
+                original_price=current_product["msrp_price"],
+                current_price=current_product["sale_price"],
+            ),
+            availability=AvailabilityInfo(
+                available=(current_product["stockFlag"] == "Y")
+            ),
+        )
+        products.append(product)
+
+    return products
+
+
+asyncio.run(main())

test.py

@@ -0,0 +1,42 @@
+import requests
+
+headers = {
+    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:152.0) Gecko/20100101 Firefox/152.0",
+    "Accept": "*/*",
+    "Accept-Language": "en-US,en;q=0.9",
+    # 'Accept-Encoding': 'gzip, deflate, br, zstd',
+    "Content-Type": "application/x-www-form-urlencoded",
+    "Origin": "https://www.samsung.com",
+    "Connection": "keep-alive",
+    "Referer": "https://www.samsung.com/",
+    "Sec-Fetch-Dest": "empty",
+    "Sec-Fetch-Mode": "cors",
+    "Sec-Fetch-Site": "same-site",
+    "Priority": "u=0",
+    # Requests doesn't support trailers
+    # 'TE': 'trailers',
+}
+
+data = {
+    "clientCode": "b2c",
+    "clientName": "scom_pf",
+    "firstSearchYN": "true",
+    "countryCode": "us",
+    "storeID": "us_education",
+    "startIndex": "0",
+    "requestCount": "16",
+    "category_code": "08080000",
+    # "filters": '[{"name":"Type","values":["Wall Ovens"],"value_norm":["Wall Ovens"]}]',
+    "sort": "newest",
+}
+
+response = requests.post(
+    "https://sribsrch.ecom.samsung.com/estoresearch-api/v1/scom/us/pf_search",
+    headers=headers,
+    data=data,
+)
+print(response.json())
+with open("resp.json", "w") as write_resp:
+    import json
+
+    json.dump(response.json(), write_resp, indent=4)