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