bug fix to avoid swallowing exceptions/failed pages inside fetch_all_pages/small refactor

2026-08-12 17:45:59 -07:00
1 parent 321546f9b9
commit b071ac35bf
@@ -6,3 +6,4 @@ __pycache__
 monitor_utils.egg-info
 uv.lock
 .claude
 build/
@@ -1,63 +0,0 @@
 """Config loading and mode helpers shared across monitors and the bot."""
 import logging
 from enum import Enum
 from monitor_utils.constants import LOGGERS
 from monitor_utils.database import get_collection
 CONFIGS_COLLECTION_NAME = "configs"
 class ConfigNotFoundError(Exception):
    """Raised when a monitor has no config document in Mongo."""
 class DevelopmentMode(str, Enum):
    """Runtime mode selecting production or development behavior."""
    PRODUCTION = "PRODUCTION"
    DEVELOPMENT = "DEVELOPMENT"
 def is_production(development_mode: DevelopmentMode) -> bool:
    """Return True when the mode is PRODUCTION."""
    return development_mode == DevelopmentMode.PRODUCTION
 async def get_config(monitor_name: str, use_production: bool = False) -> 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_name: Monitor identifier (the config document _id).
        use_production: Read the production configs collection instead of
            configs_testing.
    Returns:
        The config document, including its categories and channels.
    Raises:
        ConfigNotFoundError: When no document exists for the monitor.
    """
    collection_name = (
        CONFIGS_COLLECTION_NAME
        if use_production
        else f"{CONFIGS_COLLECTION_NAME}_testing"
    )
    config = await get_collection(collection_name).find_one({"_id": monitor_name})
    if config is None:
        raise ConfigNotFoundError(monitor_name)
    return config
 def set_logger_levels(loggers: list[str] = LOGGERS, level: int = logging.WARNING):
    """Set the given loggers to the given level to quiet noisy libraries."""
    for name in loggers:
        logging.getLogger(name).setLevel(level)
@@ -1,70 +0,0 @@
 """Shared constants for monitors and the consumer bot.
 Holds Mongo connection settings, bot tokens, retry and timeout tuning,
 client fingerprint identifiers, webhook routing maps, product status values,
 and the expected error tuple used for known-failure handling.
 """
 import asyncio
 from enum import Enum
 from json import JSONDecodeError
 import aiohttp
 import requests
 from aioweb import AiowebError
 from curl_cffi.requests.errors import RequestsError
 from pymongo.errors import PyMongoError
 MONGO_DATABASE_NAME = "tpv_woot"
 MONGO_URL = (
    "mongodb://tpv:n3%23QI07GnbZCbvGJ@zelda.disqualifier.me:27017/?authSource=tpv_woot"
 )
 LOGGING_LEVEL = "INFO"
 LOGGERS = [
    "discord",
    "pymongo",
    "aiohttp",
 ]
 RETRY_LIMIT = 5
 RETRY_WAIT = 3
 CYCLE_DELAY = 0.30
 MIN_BACKOFF_DELAY = 0.1
 MAX_BACKOFF_DELAY = 2.0
 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 = ["chrome", "chrome_android", "safari", "safari_ios"]
 CLIENT_TIMEOUT = aiohttp.ClientTimeout(
    total=TIMEOUT_SECONDS, sock_connect=30, sock_read=30
 )
 EXPECTED_TASK_ERRORS = (
    RequestsError,
    aiohttp.ClientError,
    asyncio.TimeoutError,
    requests.Timeout,
    requests.HTTPError,
    PyMongoError,
    JSONDecodeError,
    AiowebError,
 )
 class ProductStatus(str, Enum):
    """Lifecycle status for a monitored product document."""
    PENDING = "pending"
    SENT = "sent"
    SKIPPED = "skipped"
    FAILED = "failed"
@@ -1,369 +0,0 @@
 """Database helpers for upserting monitored products and tracking price history.
 Provides the shared Motor client, collection access, and the upsert path that
 rewrites current product fields while preserving prior prices in a pastPrices
 array deduplicated by price, stamping lastUpdated on every write, and emitting
 alert references for new products and price changes.
 """
 import logging
 from dataclasses import asdict
 from datetime import datetime, timezone
 from functools import cache
 from motor.motor_asyncio import AsyncIOMotorClient, AsyncIOMotorCollection
 from pymongo import UpdateOne
 from pymongo.errors import BulkWriteError, PyMongoError
 from monitor_utils.constants import MONGO_DATABASE_NAME, MONGO_URL
 from monitor_utils.models import Product
 LOGGER = logging.getLogger(__name__)
 def get_collection(collection_name: str) -> AsyncIOMotorCollection:
    """Return the named collection from the shared client's database.
    Args:
        collection_name: Name of the collection to return.
    Returns:
        The requested collection.
    """
    return get_mongo_client()[MONGO_DATABASE_NAME][collection_name]
 def resolve_alerts_collection(
    product_collection: AsyncIOMotorCollection,
 ) -> AsyncIOMotorCollection:
    """Return the alerts collection beside a product collection.
    Alerts co-locate with the products they reference: the same database and
    the same development mode. The mode is read from the product collection's
    own name (setup_config appends a _testing suffix outside production), so a
    monitor writing products to <name>_testing also writes alerts to
    alerts_testing and the development bot draining alerts_testing sees them.
    Args:
        product_collection: The mode-resolved product collection the monitor
            already writes to.
    Returns:
        The matching alerts collection in the same database and mode.
    """
    suffix = "_testing" if product_collection.name.endswith("_testing") else ""
    return product_collection.database[f"alerts{suffix}"]
@cache
 def get_mongo_client() -> AsyncIOMotorClient:
    """Return a cached Motor client configured for the shared cluster.
    Returns:
        The pooled, timezone-aware Motor client.
    """
    return AsyncIOMotorClient(
        MONGO_URL,
        minPoolSize=10,
        connectTimeoutMS=20_000,
        socketTimeoutMS=60_000,
        serverSelectionTimeoutMS=30_000,
        retryWrites=True,
        retryReads=True,
        tz_aware=True,
        tzinfo=timezone.utc,
    )
 async def get_product_from_database(
    collection: AsyncIOMotorCollection,
    product_id: str,
 ) -> dict:
    """Fetch a single product document by its _id."""
    return await collection.find_one({"_id": product_id})
 async def add_products_to_database(
    products: list[Product],
    collection: AsyncIOMotorCollection,
    category_identifier: str,
    monitor_name: str,
    use_composite: bool = False,
 ):
    """Upsert products, tracking deduplicated price history and emitting alerts.
    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. 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
    upsert an alert referencing the product; a price returning to a value
    already in pastPrices emits nothing, so per-source price oscillation
    (member vs non-member quotes for the same SKU) alerts once per distinct
    price instead of every run. Alerts are deduplicated so a product holds
    at most one unclaimed alert regardless of concurrent category or page
    tasks detecting it, and write only after the product write succeeds so
    a failed upsert cannot produce an orphaned alert.
    Args:
        products: Products to upsert.
        collection: Target MongoDB collection.
        category_identifier: Category suffix appended to composite ids.
        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
            product_id.
    """
    # Callers are expected to filter empty batches, but guard here too: an empty
    # products list would make collection.bulk_write([]) raise InvalidOperation,
    # which callers would then log as a spurious category failure every cycle.
    if not products:
        return
    doc_ids = [
        build_doc_id(p.product_id, category_identifier, use_composite) for p in products
    ]
    cursor = collection.find(
        {"_id": {"$in": doc_ids}}, {"currentPrice": 1, "pastPrices": 1}
    )
    existing = {str(doc["_id"]): doc async for doc in cursor}
    ops = []
    alert_ops = []
    now = datetime.now(timezone.utc)
    for product, doc_id in zip(products, doc_ids):
        document = convert_product_to_mongo(product)
        document["_id"] = doc_id
        existing_doc = existing.get(doc_id)
        new_price = document.get("currentPrice")
        previous_price = existing_doc.get("currentPrice") if existing_doc else None
        document["priceBeforeDrop"] = resolve_price_before_drop(
            previous_price, new_price
        )
        recorded_prices = extract_recorded_prices(existing_doc)
        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_name, document, now)
            alert_ops.append(
                UpdateOne(
                    {"product_id": doc_id, "claimed": False},
                    {"$setOnInsert": alert},
                    upsert=True,
                )
            )
        update = {"$set": document, "$setOnInsert": set_on_insert}
        if never_seen_price:
            update["$push"] = {"pastPrices": {"price": new_price, "date": now}}
        ops.append(UpdateOne({"_id": doc_id}, update, upsert=True))
    try:
        await collection.bulk_write(ops, ordered=False)
        if alert_ops:
            await write_alert_ops(alert_ops, resolve_alerts_collection(collection))
    except PyMongoError:
        LOGGER.exception(
            "failed to add products to database",
            extra={"category": category_identifier},
        )
        raise
 async def write_alert_ops(
    alert_ops: list[UpdateOne], alerts_collection: AsyncIOMotorCollection
 ):
    """Write alert upserts, swallowing duplicate-key races.
    Concurrent tasks upserting an alert for the same product can both pass
    the upsert's match phase and race the insert; the partial unique index
    on unclaimed product_id makes the loser fail with a duplicate-key
    error, which is the dedupe working as intended and is ignored. Any
    other write error re-raises.
    Args:
        alert_ops: Alert upsert operations to write.
        alerts_collection: The mode-resolved alerts collection to write to,
            supplied by the caller so the writer stays mode-agnostic.
    """
    try:
        await alerts_collection.bulk_write(alert_ops, ordered=False)
    except BulkWriteError as exc:
        real_errors = [
            write_error
            for write_error in exc.details["writeErrors"]
            if write_error.get("code") != 11000
        ]
        if real_errors:
            raise
 def resolve_price_before_drop(
    previous_price: float | None, new_price: float | None
 ) -> float | None:
    """Return the pre-drop price when new_price is a drop from previous_price.
    Owns price-drop detection so monitors do not: a drop is a new price strictly
    below the previously stored one, and its priceBeforeDrop is that prior price.
    Any other case (no prior price, unchanged, or a rise) is not a drop.
    Args:
        previous_price: The product's last stored currentPrice, or None when new.
        new_price: The incoming currentPrice.
    Returns:
        previous_price on a drop, otherwise None.
    """
    if (
        previous_price is not None
        and new_price is not None
        and new_price < previous_price
    ):
        return previous_price
    return None
 def extract_recorded_prices(existing_doc: dict | None) -> set[float]:
    """Return the set of prices already recorded in a document's pastPrices.
    Args:
        existing_doc: Existing product document, or None for new products.
    Returns:
        Prices present in the pastPrices array, empty when none exist.
    """
    if not existing_doc:
        return set()
    return {entry["price"] for entry in existing_doc.get("pastPrices", [])}
 def build_alert(monitor_name: str, document: dict, now: datetime) -> dict:
    """Build an alert referencing a product for consumer delivery.
    Args:
        monitor_name: Monitor that produced the product, used by the
            consumer to resolve the owning collection and thresholds.
        document: Product document supplying the id and pricing for the
            priority calculation.
        now: Detection timestamp stored as createdAt.
    Returns:
        The alert document for upsert via $setOnInsert.
    """
    return {
        "monitor": monitor_name,
        "product_id": document["_id"],
        "priority": calculate_priority(document),
        "claimed": False,
        "attempts": 0,
        "createdAt": now,
    }
 def calculate_priority(document: dict) -> int:
    """Derive an alert's priority bucket from its discount percentage.
    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, and price increases
    resolve to priority 0.
    Args:
        document: Product document supplying the prices.
    Returns:
        Priority from 0 to 10, one bucket per 10 percent of discount.
    """
    current_price = document.get("currentPrice")
    original_price = document.get("originalPrice")
    if current_price is None or not original_price:
        return 0
    percentage = int(((original_price - current_price) / original_price) * 100)
    return min(max(percentage // 10, 0), 10)
 def build_doc_id(
    product_id: str, category_identifier: str, use_composite: bool = False
 ) -> str:
    """Return the composite id when use_composite is set, else the bare product id.
    Args:
        product_id: Retailer-specific product identifier.
        category_identifier: Category suffix for multi-variant monitors.
        use_composite: Append category_identifier to disambiguate variants.
    Returns:
        The composite id when use_composite is set, otherwise product_id.
    """
    return f"{product_id}_{category_identifier}" if use_composite else product_id
 def convert_product_to_mongo(product: Product) -> dict:
    """Flatten a Product into its MongoDB document representation.
    Stamps lastUpdated with the current UTC time. Prices are already floats via
    the Product's pydantic validation (numeric strings coerced, comma strings
    rejected at construction), so no coercion happens here. The _id is assigned
    by the caller via build_doc_id.
    Args:
        product: Product to convert.
    Returns:
        The MongoDB document dict for the product.
    """
    data = asdict(product)
    metadata = data["metadata"]
    timing = data["timing"]
    pricing = data["pricing"]
    availability = data["availability"]
    current_price = pricing["current_price"]
    original_price = pricing["original_price"]
    return {
        "title": data["title"],
        "url": data["url"],
        "currentPrice": current_price,
        "originalPrice": original_price,
        "imageUrl": metadata["image_url"],
        "customizationIndex": metadata["customization_index"],
        "memberNonMemberComparison": metadata["member_non_member_comparison"],
        "overstockText": metadata["overstock_text"],
        "affiliateUrl": metadata["affiliate_url"],
        "promoCode": metadata["promo_code"],
        "promoPercentage": metadata["promo_percentage"],
        "promoDollarOff": metadata["promo_dollar_off"],
        "promoMinSpend": metadata["promo_min_spend"],
        "startDate": timing["start_date"],
        "endDate": timing["end_date"],
        "priceBeforeDrop": pricing["price_before_drop"],
        "eduPrice": pricing["edu_price"],
        "hasMsrp": pricing["has_msrp"],
        "available": availability["available"],
        "availabilityLastChecked": availability["availability_last_checked"],
        "condition": availability["condition"],
        "appExclusive": availability["app_exclusive"],
        "memberPick": availability["member_pick"],
        "memberDeal": availability["member_deal"],
        "lastUpdated": datetime.now(timezone.utc),
    }
@@ -1,98 +0,0 @@
 # Idk who made this i was provided it by louis
 import asyncio
 from typing import Dict
 import httpx
 """
 Minimal usage:
    from dsql_shortlinks import generate_short_link_sync
    result = generate_short_link_sync(
        "https://example.com/product",
        api_key="YOUR_DSQL_KEY",
    )
    print(result["short_url"])
 """
 DEFAULT_BASE_URL = "https://api.profitlounge.us/link/affiliate"
 class DsqlShortlinks:
    def __init__(
        self,
        api_key: str,
        *,
        base_url: str = DEFAULT_BASE_URL,
        timeout: int = 60,
    ) -> None:
        self.api_key = api_key
        self.base_url = base_url
        self.timeout = timeout
    async def create_short_link(self, url: str) -> Dict[str, str]:
        headers = {"x-api-key": self.api_key}
        payload = {"url": url}
        try:
            async with httpx.AsyncClient() as client:
                response = await client.post(
                    self.base_url,
                    headers=headers,
                    json=payload,
                    timeout=self.timeout,
                )
                response.raise_for_status()
                data = response.json()
                if isinstance(data, dict) and data.get("short_url"):
                    return data
        except Exception:
            pass
        return {"short_url": url, "long_url": url}
    def create_short_link_sync(self, url: str) -> Dict[str, str]:
        try:
            running_loop = asyncio.get_running_loop()
        except RuntimeError:
            running_loop = None
        if running_loop and running_loop.is_running():
            raise RuntimeError(
                "create_short_link_sync() cannot run inside an active event loop. "
                "Use await create_short_link(...) instead."
            )
        return asyncio.run(self.create_short_link(url))
 async def generate_short_link(
    url: str, api_key: str, timeout: int = 60
 ) -> Dict[str, str]:
    client = DsqlShortlinks(api_key, timeout=timeout)
    return await client.create_short_link(url)
 def generate_short_link_sync(
    url: str, api_key: str, timeout: int = 60
 ) -> Dict[str, str]:
    client = DsqlShortlinks(api_key, timeout=timeout)
    return client.create_short_link_sync(url)
 if __name__ == "__main__":
    EXAMPLE_URL = "https://woot.com/"
    EXAMPLE_API_KEY = "YOUR_DSQL_KEY"
    print("Example:")
    print("from dsql_shortlinks import generate_short_link_sync")
    print()
    print("result = generate_short_link_sync(")
    print(f'    "{EXAMPLE_URL}",')
    print(f'    api_key="{EXAMPLE_API_KEY}",')
    print(")")
    print()
    print('print(result["short_url"])')
@@ -1,163 +0,0 @@
 """Dataclass models shared across monitors and the consumer bot."""
 from dataclasses import field
 from datetime import datetime
 from pydantic.dataclasses import dataclass
@dataclass
 class CycleResult:
    """Tallied outcome of one monitor cycle.
    Pure counts for the caller to log or act on as it sees fit; no policy is
    baked in.
    Attributes:
        successes: Category tasks that completed without an expected error.
        failures: Category tasks that raised an expected error.
        skipped: True when pre_task_func failed and no category tasks ran.
    """
    successes: int = 0
    failures: int = 0
    skipped: bool = False
@dataclass
 class ConsumerConfig:
    """Configuration for a single retailer's consumer.
    Attributes:
        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.
        ping_threshold: Price floor for the secondary 40% rule (e.g. Woot pings
            on 40%+ discounts if the item is >= this price). None disables
            the secondary rule.
    """
    monitor_name: str
    collection: object
    channel_ids: dict
    min_percentage: int
    ping_threshold: int | None = None
@dataclass
 class PricingInfo:
    """Price fields for a product.
    Attributes:
        original_price: Pre-discount price.
        current_price: Current listed price.
        price_before_drop: Stored price prior to the last detected drop.
        edu_price: Education-store price when offered.
        past_prices: Prior price history as price/date entries, oldest first.
    """
    original_price: float
    current_price: float
    price_before_drop: float | None = None
    edu_price: float | None = None
    past_prices: list[dict] | None = None
    has_msrp: bool = True
@dataclass
 class AvailabilityInfo:
    """Availability and membership flags for a product.
    Attributes:
        available: Whether the product is purchasable.
        condition: Product condition such as new or open box.
        app_exclusive: Only purchasable through the retailer's app.
        member_pick: Flagged as a member pick offer.
        member_deal: Flagged as a member deal offer.
        availability_last_checked: When available was last verified against
            the retailer, letting monitors skip rechecks within an interval.
    """
    available: bool | None = None
    condition: str | None = None
    app_exclusive: bool | None = None
    member_pick: bool | None = None
    member_deal: bool | None = None
    availability_last_checked: datetime | None = None
@dataclass
 class ProductMetadata:
    """Presentation and routing metadata for a product.
    Attributes:
        image_url: Product image URL.
        channel_id: Resolved Discord channel id for the ping.
        customization_index: Config number when priced via customization.
        member_non_member_comparison: Display string comparing member and
            non-member pricing.
        overstock_text: Overstock label text when present.
        percentage: Computed discount percentage.
        affiliate_url: Affiliate link when available.
        promo_code: Event-level checkout code (e.g. "BIRTHDAY26") when the
            product's event advertises one, else None.
        promo_percentage: Advertised percent off from a percentage code
            ("extra 22% off" -> 22); None for dollar codes or no code. This is
            the claimed discount, not the actual price delta.
        promo_dollar_off: Advertised dollar amount off from a spend-threshold
            code ("get an extra $22 off" -> 22); None for percentage codes or
            no code. Pairs with promo_min_spend.
        promo_min_spend: Spend threshold a dollar code requires ("Spend $100"
            -> 100); None unless promo_dollar_off is set.
    """
    image_url: str | None = None
    channel_id: str | None = None
    customization_index: str | None = None
    member_non_member_comparison: str | None = None
    overstock_text: str | None = None
    percentage: int | None = None
    affiliate_url: str | None = None
    promo_code: str | None = None
    promo_percentage: int | None = None
    promo_dollar_off: int | None = None
    promo_min_spend: int | None = None
@dataclass
 class ProductTiming:
    """Lifecycle timestamps for a product.
    Attributes:
        discovered_date: When the product was first stored.
        start_date: Offer start when the retailer provides one.
        end_date: Offer end when the retailer provides one.
    """
    discovered_date: datetime | None = None
    start_date: datetime | None = None
    end_date: datetime | None = None
@dataclass
 class Product:
    """A monitored product assembled from retailer data.
    Attributes:
        product_id: Retailer-specific product identifier.
        title: Product title.
        url: Product page URL.
        pricing: Price fields.
        metadata: Presentation and routing metadata.
        timing: Lifecycle timestamps.
        availability: Availability and membership flags.
    """
    product_id: str
    title: str
    url: str
    pricing: PricingInfo
    metadata: ProductMetadata = field(default_factory=ProductMetadata)
    timing: ProductTiming = field(default_factory=ProductTiming)
    availability: AvailabilityInfo = field(default_factory=AvailabilityInfo)
@@ -1,691 +0,0 @@
 import asyncio
 import logging
 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 log_setup import setup_logging
 from motor.motor_asyncio import AsyncIOMotorCollection
 from monitor_utils.config import (
    DevelopmentMode,
    get_config,
    is_production,
    set_logger_levels,
 )
 from monitor_utils.constants import (
    CYCLE_TIMEOUT_SECONDS,
    EXPECTED_TASK_ERRORS,
    LOGGING_LEVEL,
    MAX_CURL_HANDLES,
    TIMEOUT_SECONDS,
 )
 from monitor_utils.database import get_collection
 from monitor_utils.sessions import get_default_session
 from monitor_utils.models import (
    AvailabilityInfo,
    CycleResult,
    PricingInfo,
    Product,
    ProductMetadata,
    ProductTiming,
 )
 LOGGER = logging.getLogger(__name__)
 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_name.
    Args:
        monitor_name: Monitor name used as the rotated-history base.
        logging_level: Root logging level.
    """
    setup_logging(
        name="logs/latest.log",
        history_name=monitor_name,
        level=logging_level,
        output="json",
    )
    set_logger_levels()
 async def init_monitor(
    monitor_name: str,
    development_mode: DevelopmentMode,
    production_proxy_path: str = "proxies.txt",
    testing_proxy_path: str = "tests/configs/proxies.txt",
 ) -> tuple[list[str] | dict[str, str], AsyncIOMotorCollection, AioProxies]:
    """Load a monitor's config, collection, and proxies for the mode.
    Async because config now lives in Mongo; call it from inside the monitor's
    event loop (in main), not at module import. Logging is configured separately
    by setup_monitor_logging at module scope and is not touched here.
    Args:
        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.
        testing_proxy_path: Proxy file path used in testing mode.
    Returns:
        Tuple of category names, Mongo collection, and loaded proxies.
    """
    use_production = is_production(development_mode)
    category_names, _, collection = await setup_config(
        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)
    return category_names, collection, proxies
 async def setup_config(
    monitor_name: str, development_mode: DevelopmentMode
 ) -> tuple[
    list[str] | dict[str, str], dict[str, str | list[str]], AsyncIOMotorCollection
 ]:
    """Load monitor config and resolve the Mongo collection for the mode.
    Testing mode reads the configs_testing collection and appends a _testing
    suffix to the product collection name so production data is never touched
    during development.
    Args:
        monitor_name: Monitor identifier used for config and collection lookup.
        development_mode: Selects the production or testing config and
            collection.
    Returns:
        Tuple of category names, channel id mapping, and Mongo collection.
    """
    use_production = is_production(development_mode)
    config = await get_config(
        monitor_name=monitor_name,
        use_production=use_production,
    )
    category_names = config["categories"]
    collection = get_collection(
        monitor_name if use_production else f"{monitor_name}_testing"
    )
    return category_names, config["channels"], collection
 def combined_discount(base_percentage: int, promo_percentage: int | None) -> int:
    """Combine an on-sale discount with a percentage code applied at checkout.
    A "extra N% off" code discounts the already-reduced sale price, so the two
    compound rather than add: 20% off then an extra 30% code is 44% off list,
    not 50%. Returns base_percentage unchanged when there is no percentage code.
    Args:
        base_percentage: The on-sale discount off list, as a whole percent.
        promo_percentage: The code's advertised percent off, or None.
    Returns:
        The combined discount off list, rounded to a whole percent.
    """
    if not promo_percentage:
        return base_percentage
    return round(100 - (100 - base_percentage) * (100 - promo_percentage) / 100)
 def build_product_details(
    product: dict,
    min_percentage: int,
    min_price_drop: float | None = None,
 ) -> Product | None:
    """Build a Product from a Mongo document if it passes discount thresholds.
    Skips documents with a missing or zero original price. Passing requires
    meeting min_percentage, or min_price_drop when provided. 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.
    Returns:
        The built Product, or None when skipped or below thresholds.
    """
    product_id = product["_id"]
    current_price = product["currentPrice"]
    original_price = product["originalPrice"]
    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:
        if not original_price:
            LOGGER.warning(
                "missing or zero original price, skipping",
                extra={"product_id": product_id},
            )
            return
        percentage = int(((original_price - current_price) / original_price) * 100)
        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.
        # Dollar codes do not map to a percent and never affect the threshold.
        effective_percentage = combined_discount(
            percentage, product.get("promoPercentage")
        )
        passes_percentage = effective_percentage >= min_percentage
        passes_price_drop = min_price_drop is not None and price_drop >= min_price_drop
        if not passes_percentage and not passes_price_drop:
            return
    built = Product(
        product_id=product_id,
        title=product["title"],
        url=product["url"],
        metadata=ProductMetadata(
            image_url=product["imageUrl"],
            customization_index=product["customizationIndex"],
            member_non_member_comparison=product["memberNonMemberComparison"],
            overstock_text=product["overstockText"],
            percentage=percentage,
            affiliate_url=product.get("affiliateUrl"),
            promo_code=product.get("promoCode"),
            promo_percentage=product.get("promoPercentage"),
            promo_dollar_off=product.get("promoDollarOff"),
            promo_min_spend=product.get("promoMinSpend"),
        ),
        timing=ProductTiming(
            discovered_date=product["discoveredDate"],
            start_date=product["startDate"],
            end_date=product["endDate"],
        ),
        pricing=PricingInfo(
            current_price=current_price,
            original_price=original_price,
            price_before_drop=product["priceBeforeDrop"],
            edu_price=product["eduPrice"],
            past_prices=product.get("pastPrices"),
            has_msrp=has_msrp,
        ),
        availability=AvailabilityInfo(
            available=product["available"],
            condition=product["condition"],
            app_exclusive=product["appExclusive"],
            member_pick=product["memberPick"],
            member_deal=product["memberDeal"],
        ),
    )
    return built
 async def run_monitor_cycle(
    category_identifiers: list[str] | dict[str, str],
    task_func: Callable[..., Awaitable[None]],
    session: aiohttp.ClientSession | AsyncSession,
    collection: AsyncIOMotorCollection,
    *,
    pre_task_func: Callable[..., Awaitable[None]] | None = None,
    concurrent_pre_task_func: Callable[..., Awaitable[None]] | None = None,
    **kwargs,
 ) -> CycleResult:
    """Run one monitor cycle over a caller-supplied session and report the outcome.
    Awaits pre_task_func if given, dispatches task_func per category
    concurrently, logs expected failures, and raises unexpected ones. The
    caller owns the loop, the session lifetime, and the inter-cycle delay: wrap
    this in a `while True` that opens a fresh session per cycle and sleeps
    between cycles — or use run_bounded_cycle, which owns the session and a
    hard cycle budget and leaves only the loop and the sleep to the caller.
    The returned CycleResult is pure counts for the caller to log or act on;
    this function applies no policy to it.
    session and collection are the two per-cycle resources, passed to every
    callback (pre_task_func, concurrent_pre_task_func, and each task_func) just
    like each other. A callback that does not need one still receives it, so
    every callback must accept **kwargs to absorb the ones it ignores.
    Args:
        category_identifiers: List of category ids, or a mapping of category
            id to query id; mapping entries pass query_category_id to
            task_func.
        task_func: Awaitable invoked per category with the identifier, session,
            collection, and **kwargs.
        session: Open HTTP session for this cycle, built and closed by the
            caller. Proxies are the caller's responsibility, bound to the
            session (see get_default_session).
        collection: Mongo collection the tasks persist products to; forwarded to
            every callback so any that writes has it.
        pre_task_func: Optional awaitable run once at the start of the cycle,
            awaited before any category tasks are dispatched and receiving
            session, collection, and **kwargs. Use it for setup the category
            tasks depend on (session headers, cookies, fetched state); if it
            raises an expected task error the cycle is skipped and a skipped
            CycleResult returned.
        concurrent_pre_task_func: Optional awaitable run once per cycle, gathered
            alongside the category tasks and receiving session, collection, and
            **kwargs. Category tasks must not depend on its side effects
            (session headers, cookies, fetched state) since no ordering is
            guaranteed.
        **kwargs: Forwarded verbatim to pre_task_func, concurrent_pre_task_func,
            and every task_func call.
    Returns:
        A CycleResult tallying task successes and expected-error failures, or a
        skipped result when pre_task_func failed.
    Raises:
        Exception: Any unexpected (non-EXPECTED_TASK_ERRORS) error raised by a
            task or pre_task_func propagates, failing loud so the caller's loop
            crashes rather than silently retrying a broken monitor.
    """
    if pre_task_func:
        try:
            await pre_task_func(session=session, collection=collection, **kwargs)
        except EXPECTED_TASK_ERRORS as error:
            LOGGER.warning("pre task failed, skipping cycle", exc_info=error)
            return CycleResult(skipped=True)
    LOGGER.info("starting tasks")
    if isinstance(category_identifiers, dict):
        tasks = [
            task_func(
                category_id, session, collection, query_category_id=query, **kwargs
            )
            for category_id, query in category_identifiers.items()
        ]
        category_keys = list(category_identifiers.keys())
    else:
        tasks = [
            task_func(category_identifier, session, collection, **kwargs)
            for category_identifier in category_identifiers
        ]
        category_keys = list(category_identifiers)
    if concurrent_pre_task_func:
        tasks.append(
            concurrent_pre_task_func(session=session, collection=collection, **kwargs)
        )
        category_keys.append("pre_task")
    results = await asyncio.gather(*tasks, return_exceptions=True)
    successes = 0
    failures = 0
    for category_identifier, result in zip(category_keys, results):
        if isinstance(result, EXPECTED_TASK_ERRORS):
            LOGGER.warning(
                "failed to get products",
                extra={"category": category_identifier},
                exc_info=result,
            )
            failures += 1
        elif isinstance(result, Exception):
            LOGGER.error(
                "unexpected error",
                extra={"category": category_identifier},
                exc_info=result,
            )
            raise result
        else:
            successes += 1
    return CycleResult(successes=successes, failures=failures)
 async def run_bounded_cycle(
    category_identifiers: list[str] | dict[str, str],
    task_func: Callable[..., Awaitable[None]],
    proxies: AioProxies,
    collection: AsyncIOMotorCollection,
    *,
    headers: dict | None = None,
    session_timeout: int = TIMEOUT_SECONDS,
    max_clients: int = MAX_CURL_HANDLES,
    cycle_timeout: int = CYCLE_TIMEOUT_SECONDS,
    **kwargs,
 ) -> CycleResult | None:
    """Run one monitor cycle on a fresh session under a hard time budget.
    The session is opened and closed inside the call, so curl multi-handle
    state cannot outlive one cycle: a wedged transfer dies with its session
    instead of parking the process. The budget turns an engine-level lost
    wakeup into a caught TimeoutError and an abandoned cycle, never a
    permanent wedge; the session's own close is bounded by
    get_default_session, so the exit cannot park past the budget either.
    The caller still owns the loop and the inter-cycle delay:
        while True:
            await run_bounded_cycle(ids, task_func, proxies, collection)
            await asyncio.sleep(CYCLE_DELAY)
    Args:
        category_identifiers: Categories, passed through to run_monitor_cycle.
        task_func: Awaitable invoked per category.
        proxies: Pool bound to each cycle's fresh session.
        collection: Mongo collection forwarded to every callback.
        headers: Monitor-wide static headers merged into every request.
        session_timeout: Per-request timeout on the cycle's session.
        max_clients: Curl handle pool size. Keep at or above the monitor's
            whole fan-out; see get_default_session for why.
        cycle_timeout: Hard budget in seconds for the whole cycle.
        **kwargs: Forwarded to run_monitor_cycle (pre_task_func,
            concurrent_pre_task_func, and per-task kwargs).
    Returns:
        The cycle's CycleResult, or None when the budget expired and the
        cycle was abandoned.
    """
    async with get_default_session(
        proxies, headers, timeout=session_timeout, max_clients=max_clients
    ) as session:
        try:
            async with asyncio.timeout(cycle_timeout):
                return await run_monitor_cycle(
                    category_identifiers,
                    task_func,
                    session=session,
                    collection=collection,
                    **kwargs,
                )
        except TimeoutError:
            LOGGER.error(
                "cycle exceeded budget, abandoning cycle",
                extra={"timeout": cycle_timeout},
            )
            return None
 async def execute_pagination(
    category_identifier: str,
    total_products: int,
    products_per_page: int,
    func: Callable[..., Awaitable[None]],
    session: aiohttp.ClientSession | AsyncSession,
    *,
    semaphore: asyncio.Semaphore | None = None,
    start_offset: int = 0,
    pagination_mode: str = "offset",
    **kwargs,
 ):
    """Paginate a category by dispatching concurrent page or offset tasks to func.
    Runs pages 2..N under a semaphore, retries failed pages with exponential
    backoff, and logs any pages that still fail after retries. Page 1 is
    assumed to have been fetched by the caller.
    Args:
        category_identifier: Label used for logging only; not passed to func.
        total_products: Total product count used to derive page count.
        products_per_page: Page size used to derive page count and offsets.
        func: Awaitable task invoked per page; receives session plus either
            offset or page_number depending on pagination_mode.
        session: HTTP session forwarded to each func call.
        semaphore: Optional bound on concurrent page fetches. Omit it to let a
            monitor with few pages fan out unbounded; pass one to cap monitors
            with many pages. Pass a single shared semaphore across all categories
            so the limit caps total fan-out; a per-call semaphore would instead
            bound each category in isolation, making the effective limit this
            value times the category count.
        start_offset: Additive base applied to every offset, for APIs whose
            offset space does not begin at zero. Page N requests offset
            start_offset + (N - 1) * products_per_page.
        pagination_mode: Either offset or page, selecting the kwarg sent to func.
        **kwargs: Forwarded verbatim to every func call. Any identifier func
            needs (category_id, category_name, category_path, or bestbuy's
            query_category_id) must be supplied here, since naming varies
            per monitor.
    """
    total_pages = (int(total_products) + products_per_page - 1) // products_per_page
    LOGGER.info(
        "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(
                    session=session,
                    offset=current_offset,
                    **kwargs,
                )
            else:
                await func(
                    session=session,
                    page_number=page_index,
                    **kwargs,
                )
    failed_pages = await fetch_all_pages(
        page_range, fetch_page, category_identifier, pagination_label
    )
    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 after retries",
            extra={
                "category": category_identifier,
                "label": pagination_label,
                "failed": failed_pages,
            },
        )
    if stop_event.is_set():
        LOGGER.info("pagination stopped early", extra={"category": category_identifier})
 async def fetch_all_pages(
    page_range: Iterable[int],
    fetch_func: Callable[[int], Awaitable[None]],
    category_identifier: str,
    pagination_label: str,
 ) -> list[int]:
    """Dispatch all page fetches concurrently and return failed page numbers.
    Args:
        page_range: Page numbers to fetch.
        fetch_func: Awaitable invoked per page number.
        category_identifier: Label used for logging.
        pagination_label: Either offset or page, used for logging.
    Returns:
        Page numbers whose fetch raised an expected error.
    """
    tasks = [fetch_func(page) for page in page_range]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    return handle_page_results(
        list(page_range), results, category_identifier, pagination_label
    )
 def handle_page_results(
    page_numbers: list[int],
    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 for
    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.
    """
    failed_pages = []
    for page_number, result in zip(page_numbers, results):
        if isinstance(result, EXPECTED_TASK_ERRORS):
            LOGGER.warning(
                "failed",
                extra={
                    "category": category_identifier,
                    "label": pagination_label,
                    "page": page_number,
                    "retry": retry_attempt,
                },
                exc_info=result,
            )
            failed_pages.append(page_number)
        elif isinstance(result, Exception):
            LOGGER.error(
                "unexpected error",
                extra={
                    "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
 def first_error(group_exc: BaseExceptionGroup) -> BaseException:
    """Return a leaf exception of a possibly nested group, preferring an
    unexpected one so a real bug is never masked by a routine network error."""
    _, unexpected = group_exc.split(EXPECTED_TASK_ERRORS)
    if unexpected is not None:
        group_exc = unexpected
    error = group_exc.exceptions[0]
    while isinstance(error, BaseExceptionGroup):
        error = error.exceptions[0]
    return error
@asynccontextmanager
 async def fan_out_group():
    """A TaskGroup for page-one fan-out whose failure surfaces unwrapped.
    Use it where one page's processing must be bound to the pagination that
    page spawns: create the processing task on the yielded group, then await
    execute_pagination in the body. The group cancels whichever sibling is
    still running when the other fails, so a dead category stops doing work
    nobody will use, and a cancelled cycle takes both down together.
    The unwrapping is the reason this is shared rather than inlined. A bare
    TaskGroup always raises an ExceptionGroup, and run_monitor_cycle classifies
    task results with isinstance against EXPECTED_TASK_ERRORS -- an
    ExceptionGroup matches neither branch as intended, so a routine page
    failure would be reclassified as an unexpected one and kill the monitor.
    first_error unwraps to a leaf, preferring an unexpected exception so a real
    bug is never masked by a network error.
    Note the caller's own body is inside the group, so a plain exception raised
    there (not from a child) propagates unchanged rather than being wrapped,
    and an early `return` still awaits the child tasks.
    Example:
        async with fan_out_group() as group:
            group.create_task(process_page(data, ...))
            await execute_pagination(...)
    """
    try:
        async with asyncio.TaskGroup() as group:
            yield group
    except BaseExceptionGroup as group_exc:
        raise first_error(group_exc) from None
@@ -1,173 +0,0 @@
 """Shared HTTP session helpers for all monitors.
 A single aioweb_tls TLSSession (curl_cffi backend, random client fingerprint per
 session) replaces the old aiohttp/curl_cffi split. send_request wraps aioweb's
 request_with_retries, adding per-request proxy rotation and old-style JSON/text
 return values so monitor call sites change as little as possible.
 """
 import asyncio
 import logging
 import random
 from aioproxies import AioProxies
 from aioweb import Response
 from aioweb_tls import CurlCffi, TLSSession
 from monitor_utils.constants import (
    CLIENT_IDENTIFIERS,
    MAX_CURL_HANDLES,
    RETRY_LIMIT,
    SESSION_CLOSE_TIMEOUT_SECONDS,
    TIMEOUT_SECONDS,
 )
 LOGGER = logging.getLogger(__name__)
 class BoundedCloseSession(TLSSession):
    """A TLSSession whose close cannot outlive SESSION_CLOSE_TIMEOUT_SECONDS.
    curl_cffi's close awaits the multi handle with no timeout of its own, so a
    wedged transfer parks whatever is closing the session. A caller-side budget
    cannot cover this reliably: asyncio.timeout cancels its task once, so a
    cycle budget that has already fired leaves the close running as a fresh,
    uncancelled await, and the process parks in the `async with` exit that was
    meant to reclaim it.
    Bounding it here keeps that guarantee attached to the session itself, so
    every `async with get_default_session(...)` gets it without the call site
    having to arrange anything. Abandoning a session leaks its handle until the
    process restarts, which is strictly better than a wedged monitor.
    Known inconsistency, deliberate: an abandoned session reports itself closed
    while its curl handle is still open. TLSSession.close sets _closed before
    awaiting the backend, so _is_closed() returns True on this path, and since
    aioweb's __del__ only warns when _is_closed() is False, its leak warning is
    suppressed for exactly the sessions that leaked. The "session close exceeded
    budget" error logged below is therefore the only signal that it happened —
    alert on it rather than on the library's warning.
    Correcting the flag would mean overriding _is_closed too, and aioweb_tls is
    a third-party library: this class deliberately overrides one documented
    lifecycle method and nothing private, so an upstream release has one narrow
    way to break it. A stale flag on a session discarded the same instant is the
    cheaper side of that trade.
    """
    async def close(self) -> None:
        try:
            async with asyncio.timeout(SESSION_CLOSE_TIMEOUT_SECONDS):
                await super().close()
        except TimeoutError:
            LOGGER.error(
                "session close exceeded budget, abandoning session",
                extra={"timeout": SESSION_CLOSE_TIMEOUT_SECONDS},
            )
 def get_default_session(
    proxies: AioProxies,
    headers: dict | None = None,
    timeout: int = TIMEOUT_SECONDS,
    max_clients: int = MAX_CURL_HANDLES,
 ) -> TLSSession:
    """Return a TLSSession impersonating a random client fingerprint.
    Fingerprint rotation lives here so every monitor rotates the same way; the
    caller owns the session lifetime via ``async with``.
    Cookies are not set here: aioweb has no constructor cookie path, and
    monitors usually obtain them at runtime, so call session.set_cookie(name,
    value, domain=...) where they are fetched (e.g. a pre_task_func) instead.
    Args:
        proxies: Pool attached to the session as proxy_pool, which send_request
            rotates through per request. Required so a session can never send
            un-proxied.
        headers: Default headers merged into every request by aioweb. Use for
            monitor-wide static headers; per-request headers still go through
            send_request. The curl_cffi impersonation already supplies a
            matching User-Agent, so one is not defaulted here.
        timeout: Total request timeout in seconds, defaulting to TIMEOUT_SECONDS.
            aioweb's own default is only 10s, so this restores the wider budget
            the old sessions used and slow retailers depend on.
        max_clients: Size of the curl handle pool, defaulting to
            MAX_CURL_HANDLES. This is a hard concurrency ceiling: a request
            finding the pool empty waits in pop_curl's untimed pool.get(), and
            `timeout` does not cover that wait, only the transfer after
            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.
    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 = random.choice(CLIENT_IDENTIFIERS)
    session = BoundedCloseSession(
        backend=CurlCffi(impersonate=impersonate),
        headers=headers,
        timeout=timeout,
        max_clients=max_clients,
    )
    session.proxy_pool = proxies
    return session
 async def send_request(
    session: TLSSession,
    url: str,
    method: str,
    *,
    params: dict | None = None,
    json_data: dict | None = None,
    data: dict | None = None,
    headers: dict | None = None,
 ) -> Response:
    """Send a request through a TLSSession with retries and proxy rotation.
    Wraps aioweb's request_with_retries, which retries transport errors,
    timeouts, and retryable statuses (429/5xx) up to RETRY_LIMIT attempts with
    exponential backoff. The next proxy is drawn from the pool per call and
    applied to both schemes. A non-2xx response or a fully failed request raises
    AiowebError, an expected task error, so a failed fetch surfaces the same way
    the old HTTPError path did. On success the aioweb Response is returned for
    the caller to read via response.json(), response.text(), etc.
    Args:
        session: The TLSSession to send through.
        url: Request URL.
        method: HTTP method.
        params: Query parameters.
        json_data: JSON body. A dict here is sent as JSON with the
            application/json content-type set by aioweb.
        data: Raw request body, passed through unchanged (e.g. a url-encoded
            form string). The caller owns the matching content-type via headers.
            A dict is JSON-encoded by aioweb, so use json_data for JSON bodies.
        headers: Per-request headers merged over the session defaults.
    Returns:
        The aioweb Response, guaranteed 2xx (non-2xx raises). Read the body with
        response.json() (dict/list, or None if not valid JSON) or
        response.text().
    Raises:
        AiowebError: On a non-2xx response or after all retries fail.
    """
    proxy_url = session.proxy_pool.next().url()
    proxy = {"http": proxy_url, "https": proxy_url}
    body = json_data if json_data is not None else data
    response = await session.request_with_retries(
        method.upper(),
        url,
        params=params,
        data=body,
        headers=headers,
        proxies=proxy,
        attempts=RETRY_LIMIT,
    )
    response.raise_for_status()
    return response
@@ -5,7 +5,7 @@ build-backend = "setuptools.build_meta"
 [project]
 name = "monitor-utils"
 version = "0.1.0"
-requires-python = ">=3.10"
+requires-python = ">=3.11"  # asyncio.timeout/TaskGroup, BaseExceptionGroup
 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 = "py310"
+target-version = "py311"
 # dsql_shortlinks.py is vendored third-party code; leave it untouched.
 extend-exclude = ["src/monitor_utils/dsql_shortlinks.py"]
@@ -24,6 +24,9 @@ LOGGER = logging.getLogger(__name__)
 # 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:
@@ -104,7 +107,9 @@ 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. Prices are recorded in a pastPrices array
+    the incoming product is ignored. newListing is likewise stamped here, not
    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
@@ -148,6 +153,10 @@ 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(
@@ -260,11 +269,16 @@ def extract_recorded_prices(existing_doc: dict | None) -> set[float]:
 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: the
    product doc is rewritten every cycle, so its flag flips False one cycle
    after discovery, while a discovery alert may sit in a backed-up queue
    longer than that. The alert's copy is written once and cannot rot.
    Args:
        monitor_name: Monitor that produced the product, used by the
            consumer to resolve the owning collection and thresholds.
-        document: Product document supplying the id and pricing for the
+        document: Product document supplying the id, pricing for the
-            priority calculation.
+            priority calculation, and the stamped newListing flag.
        now: Detection timestamp stored as createdAt.
    Returns:
@@ -278,6 +292,7 @@ def build_alert(monitor_name: str, document: dict, now: datetime) -> dict:
        "claimed": False,
        "attempts": 0,
        "createdAt": now,
        "newListing": document.get("newListing", False),
    }
@@ -289,7 +304,9 @@ def calculate_priority(document: dict) -> int:
    percentage. Missing prices, zero original prices, and price increases
    resolve to priority 0. 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.
+    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.
@@ -307,7 +324,11 @@ def calculate_priority(document: dict) -> int:
        return NO_MSRP_PRIORITY
    percentage = int(((original_price - current_price) / original_price) * 100)
-    return min(max(percentage // 10, 0), 10)
+    priority = min(max(percentage // 10, 0), 10)
    if document.get("newListing"):
        return max(priority, NEW_LISTING_PRIORITY)
    return priority
 def build_doc_id(
@@ -123,6 +123,7 @@ class ProductMetadata:
    promo_percentage: int | None = None
    promo_dollar_off: int | None = None
    promo_min_spend: int | None = None
    new_listing: bool | None = None
@dataclass
@@ -148,17 +148,24 @@ 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. Passing requires
-    meeting min_percentage, or min_price_drop when provided. Channel routing
+    meeting min_percentage, or min_price_drop when provided. The caller may
-    is owned by the consumer bot; no channel id is attached here.
+    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-price skip still applies.
    Returns:
        The built Product, or None when skipped or below thresholds.
@@ -167,6 +174,14 @@ def build_product_details(
    product_id = product["_id"]
    current_price = product["currentPrice"]
    original_price = product["originalPrice"]
    new_listing = product.get("newListing")
    if not original_price:
        LOGGER.warning(
            "missing or zero original price, skipping",
            extra={"product_id": product_id},
        )
        return
    percentage = 0
    # Defaulted, not indexed: documents written before the field existed have no
@@ -175,13 +190,6 @@ def build_product_details(
    # pre-existing threshold path rather than silently routing them as MSRP-less.
    has_msrp = product.get("hasMsrp", True)
    if has_msrp:
        if not original_price:
            LOGGER.warning(
                "missing or zero original price, skipping",
                extra={"product_id": product_id},
            )
            return
        percentage = int(((original_price - current_price) / original_price) * 100)
        price_drop = original_price - current_price
        # A percentage code compounds onto the sale price, so an item can clear the
@@ -193,7 +201,7 @@ 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 passes_percentage and not passes_price_drop:
+        if not bypass_thresholds and not passes_percentage and not passes_price_drop:
            return
    built = Product(
@@ -211,6 +219,7 @@ 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,
        ),
        timing=ProductTiming(
            discovered_date=product["discoveredDate"],
@@ -513,11 +522,11 @@ async def fetch_all_pages(
        Page numbers whose fetch raised an expected error.
    """
-    tasks = [fetch_func(page) for page in page_range]
+    # list once: a one-shot iterable would be exhausted before the result pairing
    pages = list(page_range)
    tasks = [fetch_func(page) for page in pages]
    results = await asyncio.gather(*tasks, return_exceptions=True)
-    return handle_page_results(
+    return handle_page_results(pages, results, category_identifier, pagination_label)
        list(page_range), results, category_identifier, pagination_label
    )
 def handle_page_results(
@@ -8,7 +8,11 @@ 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 NO_MSRP_PRIORITY, calculate_priority
+from monitor_utils.database import (
    NEW_LISTING_PRIORITY,
    NO_MSRP_PRIORITY,
    calculate_priority,
 )
 def document(current, original, **overrides) -> dict:
@@ -44,3 +48,21 @@ def test_broken_prices_still_outrank_nothing():
    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_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
@@ -122,3 +122,73 @@ 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
@@ -0,0 +1,43 @@
 """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]
@@ -0,0 +1,119 @@
 """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_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_flag_to_false():
    """Direct callers passing an unstamped document must not KeyError."""
    alert = build_alert("bestbuy", {"_id": "p1"}, NOW)
    assert alert["newListing"] is False