From 1e53904cc69388930f0da8d43f265385295ad95b Mon Sep 17 00:00:00 2001 From: TPV Date: Mon, 27 Jul 2026 12:02:41 -0700 Subject: [PATCH] added max_clients arg to get_default_session/cycle timeout seconds --- build/lib/monitor_utils/__init__.py | 0 build/lib/monitor_utils/config.py | 63 +++ build/lib/monitor_utils/constants.py | 66 +++ build/lib/monitor_utils/database.py | 368 +++++++++++++ build/lib/monitor_utils/dsql_shortlinks.py | 98 ++++ build/lib/monitor_utils/models.py | 160 ++++++ build/lib/monitor_utils/runner.py | 569 +++++++++++++++++++++ build/lib/monitor_utils/sessions.py | 126 +++++ src/monitor_utils/constants.py | 6 + src/monitor_utils/sessions.py | 20 +- 10 files changed, 1474 insertions(+), 2 deletions(-) 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 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..f1f801f --- /dev/null +++ b/build/lib/monitor_utils/constants.py @@ -0,0 +1,66 @@ +"""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 +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..bf85560 --- /dev/null +++ b/build/lib/monitor_utils/database.py @@ -0,0 +1,368 @@ +"""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"], + "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..ebd5c13 --- /dev/null +++ b/build/lib/monitor_utils/models.py @@ -0,0 +1,160 @@ +"""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. + """ + + original_price: float + current_price: float + price_before_drop: float | None = None + edu_price: float | None = None + + +@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..bbc34d9 --- /dev/null +++ b/build/lib/monitor_utils/runner.py @@ -0,0 +1,569 @@ +import asyncio +import logging +from collections.abc import Awaitable, Callable, Iterable +from contextlib import 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 EXPECTED_TASK_ERRORS, LOGGING_LEVEL +from monitor_utils.database import get_collection +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"] + 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"], + ), + 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. 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 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 diff --git a/build/lib/monitor_utils/sessions.py b/build/lib/monitor_utils/sessions.py new file mode 100644 index 0000000..377eeee --- /dev/null +++ b/build/lib/monitor_utils/sessions.py @@ -0,0 +1,126 @@ +"""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 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, + 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 TLSSession over a curl_cffi backend with a random impersonation, with + the proxy pool attached as proxy_pool. + """ + impersonate = random.choice(CLIENT_IDENTIFIERS) + session = TLSSession( + 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/constants.py b/src/monitor_utils/constants.py index abe46f5..f1f801f 100644 --- a/src/monitor_utils/constants.py +++ b/src/monitor_utils/constants.py @@ -35,6 +35,12 @@ 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 CLIENT_IDENTIFIERS = ["chrome", "chrome_android", "safari", "safari_ios"] CLIENT_TIMEOUT = aiohttp.ClientTimeout( total=TIMEOUT_SECONDS, sock_connect=30, sock_read=30 diff --git a/src/monitor_utils/sessions.py b/src/monitor_utils/sessions.py index 0bc78b9..377eeee 100644 --- a/src/monitor_utils/sessions.py +++ b/src/monitor_utils/sessions.py @@ -11,13 +11,19 @@ import random from aioproxies import AioProxies from aioweb import Response from aioweb_tls import CurlCffi, TLSSession -from monitor_utils.constants import CLIENT_IDENTIFIERS, RETRY_LIMIT, TIMEOUT_SECONDS +from monitor_utils.constants import ( + CLIENT_IDENTIFIERS, + MAX_CURL_HANDLES, + RETRY_LIMIT, + 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. @@ -39,6 +45,13 @@ def get_default_session( 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 TLSSession over a curl_cffi backend with a random impersonation, with @@ -46,7 +59,10 @@ def get_default_session( """ impersonate = random.choice(CLIENT_IDENTIFIERS) session = TLSSession( - backend=CurlCffi(impersonate=impersonate), headers=headers, timeout=timeout + backend=CurlCffi(impersonate=impersonate), + headers=headers, + timeout=timeout, + max_clients=max_clients, ) session.proxy_pool = proxies return session