Merge pull request 'added has_msrp for no msrp based products(new lists)' (#7) from development into main

Reviewed-on: #7

.gitignore

@@ -5,6 +5,4 @@ __pycache__
 .idea
 monitor_utils.egg-info
 uv.lock
-.claude/
+.claude
-build/
-logs/

pyproject.toml

@@ -5,7 +5,7 @@ build-backend = "setuptools.build_meta"
 [project]
 name = "monitor-utils"
 version = "0.1.0"
-requires-python = ">=3.11"  # asyncio.timeout/TaskGroup, BaseExceptionGroup
+requires-python = ">=3.10"
 dependencies = [
     "aiohttp>=3.14.1",
     "aioproxies @ git+https://git.rethinkstudios.io/rethink-public/aioproxies.git",
@@ -36,7 +36,7 @@ testpaths = ["tests"]
 
 [tool.ruff]
 line-length = 88
-target-version = "py311"
+target-version = "py310"
 # dsql_shortlinks.py is vendored third-party code; leave it untouched.
 extend-exclude = ["src/monitor_utils/dsql_shortlinks.py"]
 

src/monitor_utils/config.py

@@ -3,8 +3,6 @@
 import logging
 from enum import Enum
 
-from motor.motor_asyncio import AsyncIOMotorClient
-
 from monitor_utils.constants import LOGGERS
 from monitor_utils.database import get_collection
 
@@ -28,22 +26,16 @@ def is_production(development_mode: DevelopmentMode) -> bool:
     return development_mode == DevelopmentMode.PRODUCTION
 
 
-async def get_config(
+async def get_config(monitor_name: str, use_production: bool = False) -> dict:
-    monitor: str,
-    use_production: bool = False,
-    mongo_client: AsyncIOMotorClient = None,
-) -> dict:
     """Load a monitor's config document from Mongo.
 
     Reads the configs collection in production and configs_testing in
     development, so a dev run never reads production routing.
 
     Args:
-        monitor: Monitor identifier (the config document _id).
+        monitor_name: Monitor identifier (the config document _id).
         use_production: Read the production configs collection instead of
             configs_testing.
-        mongo_client: Client to read through; the shared cluster client when
-            None.
 
     Returns:
         The config document, including its categories and channels.
@@ -57,11 +49,9 @@ async def get_config(
         if use_production
         else f"{CONFIGS_COLLECTION_NAME}_testing"
     )
-    config = await get_collection(collection_name, mongo_client=mongo_client).find_one(
+    config = await get_collection(collection_name).find_one({"_id": monitor_name})
-        {"_id": monitor}
-    )
     if config is None:
-        raise ConfigNotFoundError(monitor)
+        raise ConfigNotFoundError(monitor_name)
 
     return config
 

src/monitor_utils/constants.py

@@ -15,8 +15,10 @@ from aioweb import AiowebError
 from curl_cffi.requests.errors import RequestsError
 from pymongo.errors import PyMongoError
 
-MONGO_DATABASE_NAME = "tpv"
+MONGO_DATABASE_NAME = "tpv_woot"
-MONGO_URL = "mongodb://tpv:mvVuBhhUiCabeNikmUt71L8VqJQHh7IF@ridley.data.dev.ca.rethinkstudios.io:27017/tpv?authSource=admin"
+MONGO_URL = (
+    "mongodb://tpv:n3%23QI07GnbZCbvGJ@zelda.disqualifier.me:27017/?authSource=tpv_woot"
+)
 LOGGING_LEVEL = "INFO"
 LOGGERS = [
     "discord",
@@ -33,17 +35,17 @@ FAILURE_DELAY = 3
 
 TIMEOUT_SECONDS = 30
 CYCLE_TIMEOUT_SECONDS = 600
+# curl_cffi's AsyncSession defaults to 10 handles and blocks on an untimed
+# pool.get() past that, so a monitor fanning out wider silently serialises with
+# no timeout covering the wait. Handles are allocated lazily (the pool holds
+# None placeholders until popped), so a generous ceiling costs a low-concurrency
+# monitor nothing. Monitors fanning out past this pass max_clients explicitly.
 MAX_CURL_HANDLES = 64
+# Budget for closing a session. curl_cffi's close awaits the multi handle with
+# no timeout of its own, so a wedged transfer would otherwise park whatever is
+# closing the session, including an `async with` exit.
 SESSION_CLOSE_TIMEOUT_SECONDS = 10
-CLIENT_IDENTIFIERS = [
+CLIENT_IDENTIFIERS = ["chrome", "chrome_android", "safari", "safari_ios"]
-    "chrome119",
-    "chrome120",
-    "chrome124",
-    "chrome131",
-    "chrome131_android",
-    "safari",
-    "safari_ios",
-]
 CLIENT_TIMEOUT = aiohttp.ClientTimeout(
     total=TIMEOUT_SECONDS, sock_connect=30, sock_read=30
 )

src/monitor_utils/database.py

@@ -7,7 +7,6 @@ alert references for new products and price changes.
 """
 
 import logging
-import math
 from dataclasses import asdict
 from datetime import datetime, timezone
 from functools import cache
@@ -21,36 +20,18 @@ from monitor_utils.models import Product
 
 LOGGER = logging.getLogger(__name__)
 
-# Drain rank for products with no MSRP. Their percentage is 0 by construction, so
-# without this they sort dead last and starve behind any backlog. 6 puts them
-# directly under the 70%+ band, tying with 60-69% and broken by createdAt.
-NO_MSRP_PRIORITY = 6
-# Floor rank for new listings, which send regardless of discount and would
-# otherwise sort near 0 and starve the same way MSRP-less products did.
-NEW_LISTING_PRIORITY = 6
 
-
+def get_collection(collection_name: str) -> AsyncIOMotorCollection:
-def get_collection(
-    collection_name: str,
-    mongo_client: AsyncIOMotorClient = None,
-    database_name=MONGO_DATABASE_NAME,
-) -> AsyncIOMotorCollection:
-    # rename mongo_client to client and add optional database name arg
     """Return the named collection from the shared client's database.
 
     Args:
         collection_name: Name of the collection to return.
-        mongo_client: optional client for projects with already configured client
-        database_name: optional name to use besides the default mongo database name
 
     Returns:
         The requested collection.
     """
 
-    if mongo_client is None:
+    return get_mongo_client()[MONGO_DATABASE_NAME][collection_name]
-        mongo_client = get_mongo_client()
-
-    return mongo_client[database_name][collection_name]
 
 
 def resolve_alerts_collection(
@@ -77,18 +58,15 @@ def resolve_alerts_collection(
 
 
 @cache
-def get_mongo_client(mongo_url: str = MONGO_URL) -> AsyncIOMotorClient:
+def get_mongo_client() -> AsyncIOMotorClient:
     """Return a cached Motor client configured for the shared cluster.
 
-    Args:
-        mongo_url: The mongo url to use for connections.
-
     Returns:
         The pooled, timezone-aware Motor client.
     """
 
     return AsyncIOMotorClient(
-        mongo_url,
+        MONGO_URL,
         minPoolSize=10,
         connectTimeoutMS=20_000,
         socketTimeoutMS=60_000,
@@ -113,7 +91,7 @@ async def add_products_to_database(
     products: list[Product],
     collection: AsyncIOMotorCollection,
     category_identifier: str,
-    monitor: str,
+    monitor_name: str,
     use_composite: bool = False,
 ):
     """Upsert products, tracking deduplicated price history and emitting alerts.
@@ -121,9 +99,7 @@ async def add_products_to_database(
     Rewrites current fields on each upsert. priceBeforeDrop is derived here,
     not by monitors: it is set to the previous stored currentPrice when the new
     price is lower (a drop) and cleared otherwise, so any price_before_drop on
-    the incoming product is ignored. newListing is likewise stamped here, not
+    the incoming product is ignored. Prices are recorded in a pastPrices array
-    by monitors: True only on the cycle whose write inserts the document, and
-    copied onto the discovery alert, whose copy never flips. Prices are recorded in a pastPrices array
     of price/date entries; a price already present in the array is never
     recorded again, so each distinct price keeps only its first-seen timestamp.
     New products and never-before-recorded prices
@@ -139,7 +115,7 @@ async def add_products_to_database(
         products: Products to upsert.
         collection: Target MongoDB collection.
         category_identifier: Category suffix appended to composite ids.
-        monitor: Monitor name stamped onto emitted alerts for consumer
+        monitor_name: Monitor name stamped onto emitted alerts for consumer
             routing.
         use_composite: Build composite ids so multi-variant monitors (Woot)
             keep one doc per variant; other monitors collapse to one doc per
@@ -167,10 +143,6 @@ async def add_products_to_database(
         document = convert_product_to_mongo(product)
         document["_id"] = doc_id
         existing_doc = existing.get(doc_id)
-        # Stamped here, not by monitors: only the database knows whether the
-        # doc already exists. Must precede build_alert so the alert's copy of
-        # the flag is taken from the stamped value.
-        document["newListing"] = existing_doc is None
         new_price = document.get("currentPrice")
         previous_price = existing_doc.get("currentPrice") if existing_doc else None
         document["priceBeforeDrop"] = resolve_price_before_drop(
@@ -181,7 +153,7 @@ async def add_products_to_database(
         never_seen_price = new_price is not None and new_price not in recorded_prices
         set_on_insert = {"discoveredDate": now}
         if existing_doc is None or never_seen_price:
-            alert = build_alert(monitor, document, now)
+            alert = build_alert(monitor_name, document, now)
             alert_ops.append(
                 UpdateOne(
                     {"product_id": doc_id, "claimed": False},
@@ -280,22 +252,14 @@ def extract_recorded_prices(existing_doc: dict | None) -> set[float]:
     return {entry["price"] for entry in existing_doc.get("pastPrices", [])}
 
 
-def build_alert(monitor: str, document: dict, now: datetime) -> dict:
+def build_alert(monitor_name: str, document: dict, now: datetime) -> dict:
     """Build an alert referencing a product for consumer delivery.
 
-    The alert carries its own copy of the document's newListing flag and
-    priceBeforeDrop: the product doc is rewritten every cycle, so its flag
-    flips False one cycle after discovery and its priceBeforeDrop is
-    cleared by the next cycle that sees no drop, while an alert may sit in
-    a backed-up queue longer than that. The alert's copies are written once
-    and cannot rot.
-
     Args:
-        monitor: Monitor that produced the product, used by the
+        monitor_name: Monitor that produced the product, used by the
             consumer to resolve the owning collection and thresholds.
-        document: Product document supplying the id, pricing for the
+        document: Product document supplying the id and pricing for the
-            priority calculation, and the stamped newListing flag and
+            priority calculation.
-            priceBeforeDrop.
         now: Detection timestamp stored as createdAt.
 
     Returns:
@@ -303,14 +267,12 @@ def build_alert(monitor: str, document: dict, now: datetime) -> dict:
     """
 
     return {
-        "monitor": monitor,
+        "monitor": monitor_name,
         "product_id": document["_id"],
         "priority": calculate_priority(document),
         "claimed": False,
         "attempts": 0,
         "createdAt": now,
-        "newListing": document.get("newListing", False),
-        "priceBeforeDrop": document.get("priceBeforeDrop"),
     }
 
 
@@ -319,15 +281,8 @@ def calculate_priority(document: dict) -> int:
 
     Always recalculated from current and original price so the priority
     reflects the prices at alert time rather than any stale stored
-    percentage. Missing prices, zero original prices, non-finite prices, a
+    percentage. Missing prices, zero original prices, and price increases
-    discount too large to represent, and price increases resolve to priority
+    resolve to priority 0.
-    0, so one product's bad pricing can never raise out of the whole cycle (a
-    retailer field like "Infinity" survives float() and used to crash the
-    monitor with an OverflowError). A product with no MSRP has no percentage
-    to rank on and takes NO_MSRP_PRIORITY instead, since a real 0 would leave
-    it permanently behind every other alert in the claim query's sort. A new
-    listing is floored at NEW_LISTING_PRIORITY for the same reason: it may
-    carry any discount at all, and its percentage rank still wins when higher.
 
     Args:
         document: Product document supplying the prices.
@@ -341,22 +296,8 @@ def calculate_priority(document: dict) -> int:
     if current_price is None or not original_price:
         return 0
 
-    if not (math.isfinite(current_price) and math.isfinite(original_price)):
+    percentage = int(((original_price - current_price) / original_price) * 100)
-        return 0
+    return min(max(percentage // 10, 0), 10)
-
-    if not document.get("hasMsrp", True):
-        return NO_MSRP_PRIORITY
-
-    discount = ((original_price - current_price) / original_price) * 100
-    if not math.isfinite(discount):
-        return 0
-
-    percentage = int(discount)
-    priority = min(max(percentage // 10, 0), 10)
-    if document.get("newListing"):
-        return max(priority, NEW_LISTING_PRIORITY)
-
-    return priority
 
 
 def build_doc_id(
@@ -413,7 +354,6 @@ def convert_product_to_mongo(product: Product) -> dict:
         "promoPercentage": metadata["promo_percentage"],
         "promoDollarOff": metadata["promo_dollar_off"],
         "promoMinSpend": metadata["promo_min_spend"],
-        "asin": metadata["asin"],
         "startDate": timing["start_date"],
         "endDate": timing["end_date"],
         "priceBeforeDrop": pricing["price_before_drop"],

src/monitor_utils/models.py

@@ -29,7 +29,7 @@ class ConsumerConfig:
     """Configuration for a single retailer's consumer.
 
     Attributes:
-        monitor: Retailer name.
+        monitor_name: Retailer name (matches setup_config key).
         collection: MongoDB collection for this retailer's products.
         channel_ids: Discord webhook/channel routing config.
         min_percentage: Minimum discount % to trigger a ping.
@@ -38,7 +38,7 @@ class ConsumerConfig:
             the secondary rule.
     """
 
-    monitor: str
+    monitor_name: str
     collection: object
     channel_ids: dict
     min_percentage: int
@@ -123,8 +123,6 @@ class ProductMetadata:
     promo_percentage: int | None = None
     promo_dollar_off: int | None = None
     promo_min_spend: int | None = None
-    new_listing: bool | None = None
-    asin: str | None = None
 
 
 @dataclass

src/monitor_utils/runner.py

@@ -1,16 +1,13 @@
 import asyncio
 import logging
-import math
 from collections.abc import Awaitable, Callable, Iterable
 from contextlib import asynccontextmanager, nullcontext
 
 import aiohttp
 from aioproxies import AioProxies
 from curl_cffi.requests import AsyncSession
-from curl_cffi.requests.impersonate import BrowserTypeLiteral
 from log_setup import setup_logging
 from motor.motor_asyncio import AsyncIOMotorCollection
-from motor.motor_asyncio import AsyncIOMotorClient
 
 from monitor_utils.config import (
     DevelopmentMode,
@@ -39,22 +36,22 @@ from monitor_utils.models import (
 LOGGER = logging.getLogger(__name__)
 
 
-def setup_monitor_logging(monitor: str, logging_level=LOGGING_LEVEL):
+def setup_monitor_logging(monitor_name: str, logging_level=LOGGING_LEVEL):
     """Configure JSON logging for a monitor and quiet noisy libraries.
 
     Call once at module import, at global scope, before creating module loggers,
     so logging is configured synchronously at startup rather than inside the
     async init_monitor. Logs go to logs/latest.log with rotated history keyed by
-    monitor.
+    monitor_name.
 
     Args:
-        monitor: Monitor name used as the rotated-history base.
+        monitor_name: Monitor name used as the rotated-history base.
         logging_level: Root logging level.
     """
 
     setup_logging(
         name="logs/latest.log",
-        history_name=monitor,
+        history_name=monitor_name,
         level=logging_level,
         output="json",
     )
@@ -62,7 +59,7 @@ def setup_monitor_logging(monitor: str, logging_level=LOGGING_LEVEL):
 
 
 async def init_monitor(
-    monitor: str,
+    monitor_name: str,
     development_mode: DevelopmentMode,
     production_proxy_path: str = "proxies.txt",
     testing_proxy_path: str = "tests/configs/proxies.txt",
@@ -74,7 +71,7 @@ async def init_monitor(
     by setup_monitor_logging at module scope and is not touched here.
 
     Args:
-        monitor: Monitor identifier used for config and collection lookup.
+        monitor_name: Monitor identifier used for config and collection lookup.
         development_mode: Selects production or testing config, collection,
             and proxy path.
         production_proxy_path: Proxy file path used in production mode.
@@ -86,7 +83,7 @@ async def init_monitor(
 
     use_production = is_production(development_mode)
     category_names, _, collection = await setup_config(
-        monitor=monitor, development_mode=development_mode
+        monitor_name=monitor_name, development_mode=development_mode
     )
     proxy_path = production_proxy_path if use_production else testing_proxy_path
     proxies = AioProxies.from_file(path=proxy_path)
@@ -94,9 +91,7 @@ async def init_monitor(
 
 
 async def setup_config(
-    monitor: str,
+    monitor_name: str, development_mode: DevelopmentMode
-    development_mode: DevelopmentMode,
-    mongo_client: AsyncIOMotorClient = None,
 ) -> tuple[
     list[str] | dict[str, str], dict[str, str | list[str]], AsyncIOMotorCollection
 ]:
@@ -107,11 +102,9 @@ async def setup_config(
     during development.
 
     Args:
-        monitor: Monitor identifier used for config and collection lookup.
+        monitor_name: Monitor identifier used for config and collection lookup.
         development_mode: Selects the production or testing config and
             collection.
-        mongo_client: Client the config read and the returned collection use;
-            the shared cluster client when None.
 
     Returns:
         Tuple of category names, channel id mapping, and Mongo collection.
@@ -119,15 +112,13 @@ async def setup_config(
 
     use_production = is_production(development_mode)
     config = await get_config(
-        monitor=monitor,
+        monitor_name=monitor_name,
         use_production=use_production,
-        mongo_client=mongo_client,
     )
 
     category_names = config["categories"]
     collection = get_collection(
-        monitor if use_production else f"{monitor}_testing",
+        monitor_name if use_production else f"{monitor_name}_testing"
-        mongo_client=mongo_client,
     )
     return category_names, config["channels"], collection
 
@@ -157,27 +148,17 @@ def build_product_details(
     product: dict,
     min_percentage: int,
     min_price_drop: float | None = None,
-    bypass_thresholds: bool = False,
 ) -> Product | None:
     """Build a Product from a Mongo document if it passes discount thresholds.
 
-    Skips documents with a missing or zero original price, a non-finite
+    Skips documents with a missing or zero original price. Passing requires
-    price, or a discount too large to represent: a retailer field like
+    meeting min_percentage, or min_price_drop when provided. Channel routing
-    "Infinity" survives float(), and the percentage math used to raise
+    is owned by the consumer bot; no channel id is attached here.
-    OverflowError out of every caller. Passing requires
-    meeting min_percentage, or min_price_drop when provided. The caller may
-    bypass both thresholds (new listings send at any discount for monitors
-    that opt in); which alerts qualify is the caller's decision, not read
-    from the document, so per-monitor rules stay out of this function.
-    Channel routing is owned by the consumer bot; no channel id is attached
-    here.
 
     Args:
         product: Mongo document for the product.
         min_percentage: Minimum discount percentage to pass.
         min_price_drop: Optional minimum absolute price drop to pass.
-        bypass_thresholds: Build the product even when it fails both
-            thresholds. The missing- and non-finite-price skips still apply.
 
     Returns:
         The built Product, or None when skipped or below thresholds.
@@ -186,8 +167,10 @@ def build_product_details(
     product_id = product["_id"]
     current_price = product["currentPrice"]
     original_price = product["originalPrice"]
-    new_listing = product.get("newListing")
 
+    percentage = 0
+    has_msrp = product["has_msrp"]
+    if has_msrp:
         if not original_price:
             LOGGER.warning(
                 "missing or zero original price, skipping",
@@ -195,35 +178,7 @@ def build_product_details(
             )
             return
 
-    # Skipped on BOTH paths, not just the percentage one: an MSRP-less product
+        percentage = int(((original_price - current_price) / original_price) * 100)
-    # would otherwise carry the infinite price into its payload. Only floats can
-    # be non-finite, so ints and other types keep their existing behaviour.
-    if any(
-        isinstance(price, float) and not math.isfinite(price)
-        for price in (current_price, original_price)
-    ):
-        LOGGER.warning(
-            "non-finite price, skipping",
-            extra={"product_id": product_id},
-        )
-        return
-
-    percentage = 0
-    # Defaulted, not indexed: documents written before the field existed have no
-    # hasMsrp at all, and stale ones that drop out of a monitor's feed are never
-    # re-upserted, so they would KeyError here forever. True keeps them on the
-    # pre-existing threshold path rather than silently routing them as MSRP-less.
-    has_msrp = product.get("hasMsrp", True)
-    if has_msrp:
-        discount = ((original_price - current_price) / original_price) * 100
-        if not math.isfinite(discount):
-            LOGGER.warning(
-                "discount not representable, skipping",
-                extra={"product_id": product_id},
-            )
-            return
-
-        percentage = int(discount)
         price_drop = original_price - current_price
         # A percentage code compounds onto the sale price, so an item can clear the
         # threshold on the combined discount even when the base discount would not.
@@ -234,10 +189,10 @@ def build_product_details(
         )
         passes_percentage = effective_percentage >= min_percentage
         passes_price_drop = min_price_drop is not None and price_drop >= min_price_drop
-        if not bypass_thresholds and not passes_percentage and not passes_price_drop:
+        if not passes_percentage and not passes_price_drop:
             return
 
-    built_product = Product(
+    built = Product(
         product_id=product_id,
         title=product["title"],
         url=product["url"],
@@ -252,8 +207,6 @@ def build_product_details(
             promo_percentage=product.get("promoPercentage"),
             promo_dollar_off=product.get("promoDollarOff"),
             promo_min_spend=product.get("promoMinSpend"),
-            new_listing=new_listing,
-            asin=product.get("asin"),
         ),
         timing=ProductTiming(
             discovered_date=product["discoveredDate"],
@@ -265,7 +218,7 @@ def build_product_details(
             original_price=original_price,
             price_before_drop=product["priceBeforeDrop"],
             edu_price=product["eduPrice"],
-            past_prices=product.get("pastPrices"),
+            past_prices=product["pastPrices"],
             has_msrp=has_msrp,
         ),
         availability=AvailabilityInfo(
@@ -276,7 +229,7 @@ def build_product_details(
             member_deal=product["memberDeal"],
         ),
     )
-    return built_product
+    return built
 
 
 async def run_monitor_cycle(
@@ -403,7 +356,6 @@ async def run_bounded_cycle(
     session_timeout: int = TIMEOUT_SECONDS,
     max_clients: int = MAX_CURL_HANDLES,
     cycle_timeout: int = CYCLE_TIMEOUT_SECONDS,
-    impersonate: BrowserTypeLiteral | None = None,
     **kwargs,
 ) -> CycleResult | None:
     """Run one monitor cycle on a fresh session under a hard time budget.
@@ -505,12 +457,16 @@ async def execute_pagination(
         "executing pagination",
         extra={"category": category_identifier, "total": total_pages},
     )
+    stop_event = asyncio.Event()
     pagination_label = "offset" if pagination_mode == "offset" else "page"
     page_range = range(2, total_pages + 1)
 
     async def fetch_page(page_index: int):
         """Fetch a single page or offset, optionally under the shared semaphore."""
         async with semaphore or nullcontext():
+            if stop_event.is_set():
+                return
+
             if pagination_mode == "offset":
                 current_offset = start_offset + (page_index - 1) * products_per_page
                 await func(
@@ -528,9 +484,14 @@ async def execute_pagination(
     failed_pages = await fetch_all_pages(
         page_range, fetch_page, category_identifier, pagination_label
     )
-    if failed_pages:
+    if failed_pages and not stop_event.is_set():
+        failed_pages = await retry_failed_pages(
+            failed_pages, fetch_page, category_identifier, pagination_label
+        )
+
+    if failed_pages and not stop_event.is_set():
         LOGGER.error(
-            "failed to get products from failed pages",
+            "failed to get products after retries",
             extra={
                 "category": category_identifier,
                 "label": pagination_label,
@@ -538,6 +499,9 @@ async def execute_pagination(
             },
         )
 
+    if stop_event.is_set():
+        LOGGER.info("pagination stopped early", extra={"category": category_identifier})
+
 
 async def fetch_all_pages(
     page_range: Iterable[int],
@@ -557,11 +521,11 @@ async def fetch_all_pages(
         Page numbers whose fetch raised an expected error.
     """
 
-    # list once: a one-shot iterable would be exhausted before the result pairing
+    tasks = [fetch_func(page) for page in page_range]
-    pages = list(page_range)
-    tasks = [fetch_func(page) for page in pages]
     results = await asyncio.gather(*tasks, return_exceptions=True)
-    return handle_page_results(pages, results, category_identifier, pagination_label)
+    return handle_page_results(
+        list(page_range), results, category_identifier, pagination_label
+    )
 
 
 def handle_page_results(
@@ -569,17 +533,20 @@ def handle_page_results(
     results: list,
     category_identifier: str,
     pagination_label: str,
+    retry_attempt: int | None = None,
 ) -> list[int]:
     """Log page results and collect failures, raising on unexpected errors.
 
-    Expected errors are logged as warnings and their pages collected and
+    Expected errors are logged as warnings and their pages collected for
-    returned. Unexpected exceptions are logged and re-raised.
+    retry. Unexpected exceptions are logged and re-raised. Successful pages
+    are logged only when part of a retry attempt.
 
     Args:
         page_numbers: Page numbers matching results by position.
         results: Gathered results or exceptions per page.
         category_identifier: Label used for logging.
         pagination_label: Either offset or page, used for logging.
+        retry_attempt: Retry attempt number when handling retry results.
 
     Returns:
         Page numbers whose fetch raised an expected error.
@@ -594,6 +561,7 @@ def handle_page_results(
                     "category": category_identifier,
                     "label": pagination_label,
                     "page": page_number,
+                    "retry": retry_attempt,
                 },
                 exc_info=result,
             )
@@ -605,10 +573,67 @@ def handle_page_results(
                     "category": category_identifier,
                     "label": pagination_label,
                     "page": page_number,
+                    "retry": retry_attempt,
                 },
                 exc_info=result,
             )
             raise result
+        elif retry_attempt:
+            LOGGER.info(
+                "retry succeeded",
+                extra={
+                    "category": category_identifier,
+                    "label": pagination_label,
+                    "page": page_number,
+                    "retry": retry_attempt,
+                },
+            )
+
+    return failed_pages
+
+
+async def retry_failed_pages(
+    failed_pages: list[int],
+    fetch_func: Callable[[int], Awaitable[None]],
+    category_identifier: str,
+    pagination_label: str,
+    max_retries: int = 3,
+) -> list[int]:
+    """Retry failed pages with exponential backoff until exhausted or clear.
+
+    Sleeps 2**attempt seconds before each round and re-dispatches only the
+    pages that failed the previous round.
+
+    Args:
+        failed_pages: Page numbers to retry.
+        fetch_func: Awaitable invoked per page number.
+        category_identifier: Label used for logging.
+        pagination_label: Either offset or page, used for logging.
+        max_retries: Maximum retry rounds.
+
+    Returns:
+        Page numbers still failing after all retries.
+    """
+
+    retry_attempt = 1
+    while failed_pages and retry_attempt <= max_retries:
+        await asyncio.sleep(2**retry_attempt)
+        LOGGER.info(
+            "retrying failed pages",
+            extra={
+                "category": category_identifier,
+                "label": pagination_label,
+                "count": len(failed_pages),
+                "attempt": retry_attempt,
+                "max_retries": max_retries,
+            },
+        )
+        tasks = [fetch_func(page_number) for page_number in failed_pages]
+        results = await asyncio.gather(*tasks, return_exceptions=True)
+        failed_pages = handle_page_results(
+            failed_pages, results, category_identifier, pagination_label, retry_attempt
+        )
+        retry_attempt += 1
 
     return failed_pages
 

src/monitor_utils/sessions.py

@@ -13,7 +13,6 @@ import random
 from aioproxies import AioProxies
 from aioweb import Response
 from aioweb_tls import CurlCffi, TLSSession
-from curl_cffi.requests.impersonate import BrowserTypeLiteral
 from monitor_utils.constants import (
     CLIENT_IDENTIFIERS,
     MAX_CURL_HANDLES,
@@ -71,7 +70,6 @@ def get_default_session(
     headers: dict | None = None,
     timeout: int = TIMEOUT_SECONDS,
     max_clients: int = MAX_CURL_HANDLES,
-    impersonate: BrowserTypeLiteral | None = None,
 ) -> TLSSession:
     """Return a TLSSession impersonating a random client fingerprint.
 
@@ -100,17 +98,13 @@ def get_default_session(
             checkout. Keep it at or above the monitor's real fan-out (the sum
             of its pagination semaphores plus any unsemaphored per-cycle
             requests) so the wait is never reachable.
-        impersonate: A curl_cffi profile to pin instead of drawing one from
-            CLIENT_IDENTIFIERS, for a retailer whose bot protection only
-            tolerates some profiles from a given IP (B&H's Cloudflare refuses
-            every chrome profile on a flagged IP but still accepts safari).
 
     Returns:
         A BoundedCloseSession (a TLSSession whose close is bounded) over a
         curl_cffi backend with a random impersonation, with the proxy pool
         attached as proxy_pool.
     """
-    impersonate = impersonate or random.choice(CLIENT_IDENTIFIERS)
+    impersonate = random.choice(CLIENT_IDENTIFIERS)
     session = BoundedCloseSession(
         backend=CurlCffi(impersonate=impersonate),
         headers=headers,
@@ -177,43 +171,3 @@ async def send_request(
     )
     response.raise_for_status()
     return response
-
-
-def get_response_json(
-    response: Response,
-    *,
-    logger: logging.Logger = LOGGER,
-    expected: type | tuple[type, ...] = dict,
-    **context,
-) -> dict | list | None:
-    """Parse a response body as JSON, or log it and return None.
-
-    aioweb's response.json() yields None for a non-JSON body, so a single
-    isinstance check covers both the empty/invalid case and a body of the
-    wrong shape. Callers stop on None the same way they did with the inline
-    check this replaces.
-
-    Args:
-        response: A 2xx aioweb Response from send_request.
-        logger: Logger the warning is emitted on; pass the monitor's own so
-            the record stays attributed to that monitor.
-        expected: Type(s) the parsed body must be an instance of.
-        **context: Extra fields for the warning record (category, page, ...).
-
-    Returns:
-        The parsed body when it matches `expected`, else None.
-    """
-
-    data = response.json()
-    if isinstance(data, expected):
-        return data
-
-    logger.warning(
-        "unexpected response",
-        extra={
-            **context,
-            "body_type": type(data).__name__,
-            "body": str(data)[:200],
-        },
-    )
-    return None

tests/test_alert_priority.py

@@ -1,85 +0,0 @@
-"""unit test: calculate_priority ranks MSRP-less products instead of burying them
-
-run from the repo root:
-    uv run --with pytest pytest tests/test_alert_priority.py
-
-pure and offline — passes plain dicts shaped like product documents; touches no
-Mongo. The claim query sorts priority descending, so a priority of 0 means an
-alert is only drained once everything else is.
-"""
-
-from monitor_utils.database import (
-    NEW_LISTING_PRIORITY,
-    NO_MSRP_PRIORITY,
-    calculate_priority,
-)
-
-
-def document(current, original, **overrides) -> dict:
-    doc = {"_id": "p1", "currentPrice": current, "originalPrice": original}
-    doc.update(overrides)
-    return doc
-
-
-def test_no_msrp_takes_its_own_rank():
-    assert calculate_priority(document(50.0, 50.0, hasMsrp=False)) == NO_MSRP_PRIORITY
-
-
-def test_no_msrp_rank_sits_directly_under_the_seventy_band():
-    """It must outrank every sub-70% product and lose to every 70%+ one."""
-    assert calculate_priority(document(25.0, 100.0)) == 7  # 75% off
-    assert NO_MSRP_PRIORITY < 7
-    assert NO_MSRP_PRIORITY > calculate_priority(document(45.0, 100.0))  # 55% off
-
-
-def test_msrp_products_are_unaffected():
-    assert calculate_priority(document(25.0, 100.0, hasMsrp=True)) == 7
-    assert calculate_priority(document(50.0, 100.0, hasMsrp=True)) == 5
-
-
-def test_documents_without_the_field_keep_the_percentage_path():
-    """Documents written before hasMsrp existed must not all jump the queue."""
-    assert calculate_priority(document(50.0, 100.0)) == 5
-    assert calculate_priority(document(100.0, 100.0)) == 0
-
-
-def test_broken_prices_still_outrank_nothing():
-    """A missing or zero original price is a data defect, not a missing MSRP."""
-    assert calculate_priority(document(50.0, None, hasMsrp=False)) == 0
-    assert calculate_priority(document(50.0, 0, hasMsrp=False)) == 0
-    assert calculate_priority(document(None, 100.0, hasMsrp=False)) == 0
-
-
-def test_non_finite_prices_resolve_to_zero_instead_of_raising():
-    """float("Infinity") from a retailer field once crashed the whole monitor."""
-    inf, nan = float("inf"), float("nan")
-    assert calculate_priority(document(50.0, inf)) == 0
-    assert calculate_priority(document(inf, 100.0)) == 0
-    assert calculate_priority(document(-inf, 100.0)) == 0
-    assert calculate_priority(document(nan, 100.0)) == 0
-    assert calculate_priority(document(50.0, nan)) == 0
-    assert calculate_priority(document(50.0, inf, newListing=True)) == 0
-
-
-def test_a_discount_too_large_to_represent_resolves_to_zero():
-    """Finite prices can still overflow the ratio (a subnormal original)."""
-    assert calculate_priority(document(-1e308, 5e-324)) == 0
-    assert calculate_priority(document(1e308, 5e-324)) == 0
-
-
-def test_new_listing_is_floored_not_buried():
-    """A low-discount new listing sends regardless, so it must not sort last."""
-    assert (
-        calculate_priority(document(95.0, 100.0, newListing=True))
-        == NEW_LISTING_PRIORITY
-    )
-
-
-def test_new_listing_keeps_a_higher_percentage_rank():
-    """The floor is a floor: a 75%-off new listing still ranks as 75% off."""
-    assert calculate_priority(document(25.0, 100.0, newListing=True)) == 7
-
-
-def test_repeat_alerts_keep_the_percentage_path():
-    assert calculate_priority(document(95.0, 100.0, newListing=False)) == 0
-    assert calculate_priority(document(95.0, 100.0)) == 0

tests/test_build_product_details.py

@@ -78,17 +78,6 @@ def test_promo_fields_read_through():
     assert product.metadata.percentage == 20
 
 
-def test_asin_reads_through():
-    """asin round-trips: convert_product_to_mongo writes it, so the consumer
-    must read it back for the SellerAmp lookup"""
-    product = build_product_details(_doc(asin="B0CHX3QBCH"), min_percentage=0)
-    assert product.metadata.asin == "B0CHX3QBCH"
-
-
-def test_missing_asin_defaults_to_none():
-    assert build_product_details(_doc(), min_percentage=0).metadata.asin is None
-
-
 def test_missing_promo_keys_default_to_none():
     """a doc from a lagging monitor lacks promo keys -> None, not KeyError"""
     product = build_product_details(_doc(), min_percentage=0)
@@ -133,111 +122,3 @@ def test_past_prices_read_through():
     product = build_product_details(_doc(pastPrices=history), min_percentage=0)
 
     assert product.pricing.past_prices == history
-
-
-# ----------------------------------------------------------- bypass_thresholds
-
-
-def test_bypass_builds_a_below_threshold_product():
-    """A new listing sends at any discount when the caller says so."""
-
-    doc = _doc(currentPrice=95.0, originalPrice=100.0)  # 5%, way under 40
-    product = build_product_details(doc, min_percentage=40, bypass_thresholds=True)
-    assert product is not None
-    assert product.metadata.percentage == 5
-
-
-def test_bypass_defaults_off():
-    doc = _doc(currentPrice=95.0, originalPrice=100.0)
-    assert build_product_details(doc, min_percentage=40) is None
-
-
-def test_bypass_does_not_rescue_broken_prices():
-    """The missing-price skip is a data guard, not a threshold."""
-
-    doc = _doc(currentPrice=95.0, originalPrice=None)
-    assert build_product_details(doc, min_percentage=40, bypass_thresholds=True) is None
-
-
-# --------------------------------------------------------------- newListing
-
-
-def test_new_listing_flag_reads_through():
-    product = build_product_details(_doc(newListing=True), min_percentage=0)
-    assert product.metadata.new_listing is True
-
-
-def test_legacy_docs_without_the_flag_build_as_none():
-    """Documents written before newListing existed must not KeyError."""
-
-    product = build_product_details(_doc(), min_percentage=0)
-    assert product.metadata.new_listing is None
-
-
-# ------------------------------------------------------------ msrp-less guard
-
-
-def test_msrpless_doc_with_missing_original_price_skips_not_crashes():
-    """Regression: the guard lived inside the has_msrp branch, so this doc
-    shape raised ValidationError (not an expected error) and killed the
-    consumer."""
-
-    doc = _doc(hasMsrp=False, originalPrice=None)
-    assert build_product_details(doc, min_percentage=40) is None
-
-
-def test_msrpless_doc_with_zero_original_price_skips():
-    doc = _doc(hasMsrp=False, originalPrice=0)
-    assert build_product_details(doc, min_percentage=40) is None
-
-
-def test_bypass_does_not_rescue_msrpless_broken_prices():
-    doc = _doc(hasMsrp=False, originalPrice=None)
-    assert build_product_details(doc, min_percentage=40, bypass_thresholds=True) is None
-
-
-def test_msrpless_doc_with_prices_builds_unthresholded():
-    """A priced MSRP-less doc still skips the threshold gate entirely."""
-
-    doc = _doc(hasMsrp=False, currentPrice=95.0, originalPrice=100.0)  # 5% < 40
-    product = build_product_details(doc, min_percentage=40)
-    assert product is not None
-    assert product.pricing.has_msrp is False
-
-
-# ------------------------------------------------------------ non-finite prices
-
-
-def test_infinite_price_skips_instead_of_raising():
-    """Regression (2026-10-09 06:25Z): float("Infinity") from a retailer field
-    raised OverflowError here and stopped the consumer's API deal loop."""
-
-    inf = float("inf")
-    assert build_product_details(_doc(originalPrice=inf), min_percentage=40) is None
-    assert build_product_details(_doc(currentPrice=inf), min_percentage=40) is None
-    assert build_product_details(_doc(currentPrice=-inf), min_percentage=40) is None
-
-
-def test_nan_price_skips_instead_of_raising():
-    nan = float("nan")
-    assert build_product_details(_doc(currentPrice=nan), min_percentage=40) is None
-    assert build_product_details(_doc(originalPrice=nan), min_percentage=40) is None
-
-
-def test_msrpless_infinite_price_skips_rather_than_reaching_the_payload():
-    """Without the skip the MSRP-less path would carry inf into the Product."""
-
-    doc = _doc(hasMsrp=False, currentPrice=float("inf"))
-    assert build_product_details(doc, min_percentage=40) is None
-
-
-def test_bypass_does_not_rescue_a_non_finite_price():
-    doc = _doc(originalPrice=float("inf"))
-    assert build_product_details(doc, min_percentage=40, bypass_thresholds=True) is None
-
-
-def test_unrepresentable_discount_skips_instead_of_raising():
-    """Finite prices can still overflow the ratio (a subnormal original)."""
-
-    doc = _doc(currentPrice=-1e308, originalPrice=5e-324)
-    assert build_product_details(doc, min_percentage=40) is None

tests/test_fetch_all_pages.py

@@ -1,43 +0,0 @@
-"""unit test: fetch_all_pages result accounting for any iterable input
-
-run from the repo root:
-    uv run pytest tests/test_fetch_all_pages.py
-
-pure and offline — no HTTP. Regression: a generator page_range was exhausted
-building the tasks, so the result pairing dropped every failure and swallowed
-unexpected exceptions.
-"""
-
-import asyncio
-
-import pytest
-from aioweb import AiowebError
-
-from monitor_utils.runner import fetch_all_pages
-
-
-async def _failing_fetch(page: int):
-    raise AiowebError(f"page {page} failed")
-
-
-async def _crashing_fetch(page: int):
-    raise ValueError(f"page {page} crashed")
-
-
-def test_generator_input_still_reports_failed_pages():
-    failed = asyncio.run(
-        fetch_all_pages((p for p in [2, 3]), _failing_fetch, "cat", "page")
-    )
-    assert failed == [2, 3]
-
-
-def test_generator_input_still_raises_unexpected_errors():
-    with pytest.raises(ValueError):
-        asyncio.run(fetch_all_pages((p for p in [2]), _crashing_fetch, "cat", "page"))
-
-
-def test_range_input_reports_failed_pages():
-    """The re-iterable path the in-repo caller uses keeps working."""
-
-    failed = asyncio.run(fetch_all_pages(range(2, 4), _failing_fetch, "cat", "page"))
-    assert failed == [2, 3]

tests/test_get_response_json.py

@@ -1,80 +0,0 @@
-"""unit tests: sessions.get_response_json parses a body or logs and returns None
-
-run from the repo root:
-    uv run pytest tests/test_response_json.py
-
-offline — the aioweb Response is faked at the leaf; nothing is sent.
-"""
-
-import logging
-
-import pytest
-
-from monitor_utils.sessions import get_response_json
-
-
-class FakeResponse:
-    def __init__(self, body):
-        self.body = body
-
-    def json(self):
-        return self.body
-
-
-def warnings(caplog):
-    return [r for r in caplog.records if r.levelno == logging.WARNING]
-
-
-def test_dict_body_is_returned_without_logging(caplog):
-    with caplog.at_level(logging.WARNING, logger="monitor_utils.sessions"):
-        assert get_response_json(FakeResponse({"a": 1})) == {"a": 1}
-    assert warnings(caplog) == []
-
-
-def test_none_body_logs_and_returns_none(caplog):
-    with caplog.at_level(logging.WARNING, logger="monitor_utils.sessions"):
-        assert get_response_json(FakeResponse(None), category="c1", page=2) is None
-    (record,) = warnings(caplog)
-    assert record.message == "unexpected response"
-    assert record.body_type == "NoneType"
-    assert record.body == "None"
-    assert (record.category, record.page) == ("c1", 2)
-
-
-@pytest.mark.parametrize("body", [[1, 2], "text", 42, b"bytes"])
-def test_wrong_shape_logs_type_and_truncated_body(body, caplog):
-    with caplog.at_level(logging.WARNING, logger="monitor_utils.sessions"):
-        assert get_response_json(FakeResponse(body)) is None
-    (record,) = warnings(caplog)
-    assert record.body_type == type(body).__name__
-    assert record.body == str(body)[:200]
-
-
-def test_body_preview_is_capped_at_200_chars(caplog):
-    with caplog.at_level(logging.WARNING, logger="monitor_utils.sessions"):
-        get_response_json(FakeResponse("x" * 1000))
-    (record,) = warnings(caplog)
-    assert len(record.body) == 200
-
-
-def test_expected_accepts_other_types(caplog):
-    with caplog.at_level(logging.WARNING, logger="monitor_utils.sessions"):
-        assert get_response_json(FakeResponse([1]), expected=list) == [1]
-        assert get_response_json(FakeResponse([1]), expected=(dict, list)) == [1]
-        assert get_response_json(FakeResponse({}), expected=list) is None
-    assert len(warnings(caplog)) == 1
-
-
-def test_warning_goes_to_the_passed_logger(caplog):
-    with caplog.at_level(logging.WARNING):
-        get_response_json(
-            FakeResponse(None), logger=logging.getLogger("woot.woot_monitor")
-        )
-    (record,) = warnings(caplog)
-    assert record.name == "woot.woot_monitor"
-
-
-def test_empty_dict_counts_as_a_dict(caplog):
-    with caplog.at_level(logging.WARNING, logger="monitor_utils.sessions"):
-        assert get_response_json(FakeResponse({})) == {}
-    assert warnings(caplog) == []

tests/test_new_listing_stamp.py

@@ -1,146 +0,0 @@
-"""unit test: add_products_to_database stamps newListing and copies it onto alerts
-
-run from the repo root:
-    uv run pytest tests/test_new_listing_stamp.py
-
-offline — drives add_products_to_database against fake collections that record
-the UpdateOne ops instead of touching Mongo. The recorded ops are inspected via
-UpdateOne's private _doc/_filter, the same fields its own __eq__ compares.
-
-The stamp is the database layer's job: only it knows whether the doc already
-exists, so monitors never set the flag. The alert carries its own copy because
-the product doc is rewritten every cycle (its flag flips False one cycle after
-discovery) while a discovery alert may outlive that in a backed-up queue.
-"""
-
-import asyncio
-
-from monitor_utils.database import add_products_to_database, build_alert
-from monitor_utils.models import PricingInfo, Product
-
-from datetime import datetime, timezone
-
-NOW = datetime(2026, 8, 10, tzinfo=timezone.utc)
-
-
-class FakeCursor:
-    def __init__(self, docs):
-        self._docs = list(docs)
-
-    def __aiter__(self):
-        return self
-
-    async def __anext__(self):
-        if not self._docs:
-            raise StopAsyncIteration
-        return self._docs.pop(0)
-
-
-class FakeCollection:
-    """Records bulk_write ops; doubles as its own .database for alerts lookup."""
-
-    def __init__(self, existing_docs=(), name="products"):
-        self.name = name
-        self._existing = list(existing_docs)
-        self.bulk_writes = []
-        self.database = self
-        self.alerts = None
-
-    def __getitem__(self, name):
-        if self.alerts is None:
-            self.alerts = FakeCollection(name=name)
-        return self.alerts
-
-    def find(self, _filter, _projection):
-        return FakeCursor(self._existing)
-
-    async def bulk_write(self, ops, ordered=True):
-        self.bulk_writes.append(list(ops))
-
-
-def _product(product_id="p1", current=95.0, original=100.0) -> Product:
-    return Product(
-        product_id=product_id,
-        title="Widget",
-        url="http://x/p1",
-        pricing=PricingInfo(current_price=current, original_price=original),
-    )
-
-
-def run(collection, products):
-    asyncio.run(
-        add_products_to_database(products, collection, "tools", "bestbuy")
-    )
-
-
-def product_update(collection):
-    return collection.bulk_writes[0][0]._doc
-
-
-def alert_doc(collection):
-    return collection.alerts.bulk_writes[0][0]._doc["$setOnInsert"]
-
-
-def test_insert_stamps_new_listing_true_on_doc_and_alert():
-    collection = FakeCollection()
-    run(collection, [_product()])
-
-    assert product_update(collection)["$set"]["newListing"] is True
-    assert alert_doc(collection)["newListing"] is True
-
-
-def test_existing_doc_with_new_price_stamps_false():
-    existing = {"_id": "p1", "currentPrice": 100.0, "pastPrices": []}
-    collection = FakeCollection([existing])
-    run(collection, [_product(current=95.0)])
-
-    assert product_update(collection)["$set"]["newListing"] is False
-    # a never-seen price still alerts, but not as a new listing
-    assert alert_doc(collection)["newListing"] is False
-
-
-def test_drop_alert_carries_price_before_drop():
-    """the doc's priceBeforeDrop is cleared by the next no-drop cycle, so the
-    alert keeps its own copy from emit time — a drop claimed late still knows
-    what it dropped from"""
-
-    existing = {"_id": "p1", "currentPrice": 100.0, "pastPrices": []}
-    collection = FakeCollection([existing])
-    run(collection, [_product(current=95.0)])
-
-    assert alert_doc(collection)["priceBeforeDrop"] == 100.0
-
-
-def test_non_drop_alert_carries_none():
-    """a never-seen higher price alerts without a drop; discoveries have no
-    prior price at all"""
-
-    existing = {"_id": "p1", "currentPrice": 100.0, "pastPrices": []}
-    collection = FakeCollection([existing])
-    run(collection, [_product(current=120.0)])
-    assert alert_doc(collection)["priceBeforeDrop"] is None
-
-    discovery = FakeCollection()
-    run(discovery, [_product()])
-    assert alert_doc(discovery)["priceBeforeDrop"] is None
-
-
-def test_existing_doc_with_known_price_emits_no_alert():
-    existing = {
-        "_id": "p1",
-        "currentPrice": 95.0,
-        "pastPrices": [{"price": 95.0, "date": NOW}],
-    }
-    collection = FakeCollection([existing])
-    run(collection, [_product(current=95.0)])
-
-    assert product_update(collection)["$set"]["newListing"] is False
-    assert collection.alerts is None
-
-
-def test_build_alert_defaults_missing_fields():
-    """Direct callers passing an unstamped document must not KeyError."""
-
-    alert = build_alert("bestbuy", {"_id": "p1"}, NOW)
-    assert alert["newListing"] is False
-    assert alert["priceBeforeDrop"] is None