From dee090edaf4973c0684b327e1f0c18e9a24ac3a3 Mon Sep 17 00:00:00 2001 From: TPV Date: Tue, 4 Aug 2026 13:53:32 -0700 Subject: [PATCH] added no msrp prirority so they don't starve in queue --- build/lib/monitor_utils/__init__.py | 0 build/lib/monitor_utils/config.py | 63 ++ build/lib/monitor_utils/constants.py | 70 +++ build/lib/monitor_utils/database.py | 369 +++++++++++ build/lib/monitor_utils/dsql_shortlinks.py | 98 +++ build/lib/monitor_utils/models.py | 163 +++++ build/lib/monitor_utils/runner.py | 691 +++++++++++++++++++++ build/lib/monitor_utils/sessions.py | 173 ++++++ src/monitor_utils/database.py | 12 +- tests/test_alert_priority.py | 46 ++ 10 files changed, 1684 insertions(+), 1 deletion(-) create mode 100644 build/lib/monitor_utils/__init__.py create mode 100644 build/lib/monitor_utils/config.py create mode 100644 build/lib/monitor_utils/constants.py create mode 100644 build/lib/monitor_utils/database.py create mode 100644 build/lib/monitor_utils/dsql_shortlinks.py create mode 100644 build/lib/monitor_utils/models.py create mode 100644 build/lib/monitor_utils/runner.py create mode 100644 build/lib/monitor_utils/sessions.py create mode 100644 tests/test_alert_priority.py diff --git a/build/lib/monitor_utils/__init__.py b/build/lib/monitor_utils/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/build/lib/monitor_utils/config.py b/build/lib/monitor_utils/config.py new file mode 100644 index 0000000..d189712 --- /dev/null +++ b/build/lib/monitor_utils/config.py @@ -0,0 +1,63 @@ +"""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) diff --git a/build/lib/monitor_utils/constants.py b/build/lib/monitor_utils/constants.py new file mode 100644 index 0000000..c48f0b9 --- /dev/null +++ b/build/lib/monitor_utils/constants.py @@ -0,0 +1,70 @@ +"""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" diff --git a/build/lib/monitor_utils/database.py b/build/lib/monitor_utils/database.py new file mode 100644 index 0000000..f1a4f29 --- /dev/null +++ b/build/lib/monitor_utils/database.py @@ -0,0 +1,369 @@ +"""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 _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), + } diff --git a/build/lib/monitor_utils/dsql_shortlinks.py b/build/lib/monitor_utils/dsql_shortlinks.py new file mode 100644 index 0000000..763d98b --- /dev/null +++ b/build/lib/monitor_utils/dsql_shortlinks.py @@ -0,0 +1,98 @@ +# 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"])') diff --git a/build/lib/monitor_utils/models.py b/build/lib/monitor_utils/models.py new file mode 100644 index 0000000..56049f3 --- /dev/null +++ b/build/lib/monitor_utils/models.py @@ -0,0 +1,163 @@ +"""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) diff --git a/build/lib/monitor_utils/runner.py b/build/lib/monitor_utils/runner.py new file mode 100644 index 0000000..bc689d4 --- /dev/null +++ b/build/lib/monitor_utils/runner.py @@ -0,0 +1,691 @@ +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 diff --git a/build/lib/monitor_utils/sessions.py b/build/lib/monitor_utils/sessions.py new file mode 100644 index 0000000..c909372 --- /dev/null +++ b/build/lib/monitor_utils/sessions.py @@ -0,0 +1,173 @@ +"""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 diff --git a/src/monitor_utils/database.py b/src/monitor_utils/database.py index f1a4f29..4b2daa3 100644 --- a/src/monitor_utils/database.py +++ b/src/monitor_utils/database.py @@ -20,6 +20,11 @@ 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 + def get_collection(collection_name: str) -> AsyncIOMotorCollection: """Return the named collection from the shared client's database. @@ -282,7 +287,9 @@ 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, and price increases - resolve to priority 0. + 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. Args: document: Product document supplying the prices. @@ -296,6 +303,9 @@ def calculate_priority(document: dict) -> int: if current_price is None or not original_price: return 0 + if not document.get("hasMsrp", True): + return NO_MSRP_PRIORITY + percentage = int(((original_price - current_price) / original_price) * 100) return min(max(percentage // 10, 0), 10) diff --git a/tests/test_alert_priority.py b/tests/test_alert_priority.py new file mode 100644 index 0000000..750a227 --- /dev/null +++ b/tests/test_alert_priority.py @@ -0,0 +1,46 @@ +"""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 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