From b071ac35bf4e04e630768aaf0d80d161a6243b4f Mon Sep 17 00:00:00 2001 From: TPV Date: Wed, 12 Aug 2026 17:45:59 -0700 Subject: [PATCH] bug fix to avoid swallowing exceptions/failed pages inside fetch_all_pages/small refactor --- .gitignore | 3 +- 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 ------ pyproject.toml | 4 +- src/monitor_utils/database.py | 31 +- src/monitor_utils/models.py | 1 + src/monitor_utils/runner.py | 37 +- tests/test_alert_priority.py | 24 +- tests/test_build_product_details.py | 70 +++ tests/test_fetch_all_pages.py | 43 ++ tests/test_new_listing_stamp.py | 119 ++++ 17 files changed, 309 insertions(+), 1650 deletions(-) delete mode 100644 build/lib/monitor_utils/__init__.py delete mode 100644 build/lib/monitor_utils/config.py delete mode 100644 build/lib/monitor_utils/constants.py delete mode 100644 build/lib/monitor_utils/database.py delete mode 100644 build/lib/monitor_utils/dsql_shortlinks.py delete mode 100644 build/lib/monitor_utils/models.py delete mode 100644 build/lib/monitor_utils/runner.py delete mode 100644 build/lib/monitor_utils/sessions.py create mode 100644 tests/test_fetch_all_pages.py create mode 100644 tests/test_new_listing_stamp.py diff --git a/.gitignore b/.gitignore index 88940e7..a4fbbb8 100644 --- a/.gitignore +++ b/.gitignore @@ -5,4 +5,5 @@ __pycache__ .idea monitor_utils.egg-info uv.lock -.claude \ No newline at end of file +.claude +build/ \ No newline at end of file diff --git a/build/lib/monitor_utils/__init__.py b/build/lib/monitor_utils/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/build/lib/monitor_utils/config.py b/build/lib/monitor_utils/config.py deleted file mode 100644 index d189712..0000000 --- a/build/lib/monitor_utils/config.py +++ /dev/null @@ -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) diff --git a/build/lib/monitor_utils/constants.py b/build/lib/monitor_utils/constants.py deleted file mode 100644 index c48f0b9..0000000 --- a/build/lib/monitor_utils/constants.py +++ /dev/null @@ -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" diff --git a/build/lib/monitor_utils/database.py b/build/lib/monitor_utils/database.py deleted file mode 100644 index f1a4f29..0000000 --- a/build/lib/monitor_utils/database.py +++ /dev/null @@ -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 _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 deleted file mode 100644 index 763d98b..0000000 --- a/build/lib/monitor_utils/dsql_shortlinks.py +++ /dev/null @@ -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"])') diff --git a/build/lib/monitor_utils/models.py b/build/lib/monitor_utils/models.py deleted file mode 100644 index 56049f3..0000000 --- a/build/lib/monitor_utils/models.py +++ /dev/null @@ -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) diff --git a/build/lib/monitor_utils/runner.py b/build/lib/monitor_utils/runner.py deleted file mode 100644 index bc689d4..0000000 --- a/build/lib/monitor_utils/runner.py +++ /dev/null @@ -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 diff --git a/build/lib/monitor_utils/sessions.py b/build/lib/monitor_utils/sessions.py deleted file mode 100644 index c909372..0000000 --- a/build/lib/monitor_utils/sessions.py +++ /dev/null @@ -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 diff --git a/pyproject.toml b/pyproject.toml index ab305f9..fbfccbf 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -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"] diff --git a/src/monitor_utils/database.py b/src/monitor_utils/database.py index 4b2daa3..c582f7e 100644 --- a/src/monitor_utils/database.py +++ b/src/monitor_utils/database.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 - priority calculation. + document: Product document supplying the id, pricing for the + 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( diff --git a/src/monitor_utils/models.py b/src/monitor_utils/models.py index 56049f3..9c67563 100644 --- a/src/monitor_utils/models.py +++ b/src/monitor_utils/models.py @@ -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 diff --git a/src/monitor_utils/runner.py b/src/monitor_utils/runner.py index 73dcab4..0f4f5d4 100644 --- a/src/monitor_utils/runner.py +++ b/src/monitor_utils/runner.py @@ -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 - is owned by the consumer bot; no channel id is attached here. + meeting min_percentage, or min_price_drop when provided. The caller may + bypass both thresholds (new listings send at any discount for monitors + that opt in); which alerts qualify is the caller's decision, not read + from the document, so per-monitor rules stay out of this function. + Channel routing is owned by the consumer bot; no channel id is attached + here. Args: product: Mongo document for the product. min_percentage: Minimum discount percentage to pass. min_price_drop: Optional minimum absolute price drop to pass. + bypass_thresholds: Build the product even when it fails both + thresholds. The missing-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( - list(page_range), results, category_identifier, pagination_label - ) + return handle_page_results(pages, results, category_identifier, pagination_label) def handle_page_results( diff --git a/tests/test_alert_priority.py b/tests/test_alert_priority.py index 750a227..19cd01e 100644 --- a/tests/test_alert_priority.py +++ b/tests/test_alert_priority.py @@ -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 diff --git a/tests/test_build_product_details.py b/tests/test_build_product_details.py index aac1220..1ef0c25 100644 --- a/tests/test_build_product_details.py +++ b/tests/test_build_product_details.py @@ -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 diff --git a/tests/test_fetch_all_pages.py b/tests/test_fetch_all_pages.py new file mode 100644 index 0000000..4fb2992 --- /dev/null +++ b/tests/test_fetch_all_pages.py @@ -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] diff --git a/tests/test_new_listing_stamp.py b/tests/test_new_listing_stamp.py new file mode 100644 index 0000000..838bb3f --- /dev/null +++ b/tests/test_new_listing_stamp.py @@ -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 -- 2.54.0