diff --git a/src/monitor_utils/runner.py b/src/monitor_utils/runner.py index 7819bc5..80cb0fd 100644 --- a/src/monitor_utils/runner.py +++ b/src/monitor_utils/runner.py @@ -41,6 +41,23 @@ def init_monitor( testing_proxy_path: str = "configs/proxies.txt", logging_level=LOGGING_LEVEL, ) -> tuple[list[str], AsyncIOMotorCollection, AioProxies]: + """Run shared monitor startup and return categories, collection, and proxies. + + Configures JSON logging under the monitor's name, applies logger levels, + loads the monitor config and Mongo collection, and loads proxies from the + production or testing path based on development_mode. + + Args: + monitor_name: Monitor identifier used for logging and config 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. + logging_level: Level passed to setup_logging. + + Returns: + Tuple of category names, Mongo collection, and loaded proxies. + """ use_production = is_production(development_mode) setup_logging( name="logs/latest.log", @@ -60,6 +77,19 @@ def init_monitor( def setup_config( monitor_name: str, development_mode: DevelopmentMode ) -> tuple[list[str], dict[int], AsyncIOMotorCollection]: + """Load monitor config and resolve the Mongo collection for the mode. + + Testing mode appends a _testing suffix to the 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 path and + collection. + + Returns: + Tuple of category names, channel id mapping, and Mongo collection. + """ use_production = is_production(development_mode) config, _ = get_config( monitor_name=monitor_name, @@ -76,7 +106,22 @@ def setup_config( def get_channel_id( product: Product, channel_ids: dict, use_price_drop_bucket: bool = False ) -> str: - # KeyError is not a issue here bc the percentage filtering per monitor handles it + """Resolve the Discord channel id for a product based on its attributes. + + Checks flags in priority order: app exclusive, sold out, event ending + within a day, availability flag mappings, then the percentage or price + drop bucket. A KeyError on bucket lookup is acceptable because each + monitor's percentage filtering guarantees the bucket exists. + + Args: + product: Built product whose attributes select the channel. + channel_ids: Mapping of channel keys to a channel id or list of ids. + use_price_drop_bucket: Route to the price drop bucket instead of the + percentage bucket. + + Returns: + The resolved channel id. + """ if product.availability.app_exclusive: return get_single_channel_id(channel_ids["appExclusive"]) @@ -101,6 +146,7 @@ def get_channel_id( def get_single_channel_id(value: str | list[str]) -> str: + """Return the first channel id when given a list, else the value itself.""" if isinstance(value, list): return value[0] @@ -113,6 +159,21 @@ def build_product_details( 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. The resolved + channel id is attached to the product's metadata before returning. + + Args: + product: Mongo document for the product. + channel_ids: Mapping of channel keys passed to get_channel_id. + 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"] @@ -178,6 +239,26 @@ async def execute_monitor_logic( use_tls: bool = False, **kwargs, ): + """Run the monitor's category tasks in a loop with a fresh session per cycle. + + Each cycle opens a TLS or base session, optionally runs pre_task_func, + dispatches task_func per category concurrently, logs expected failures, + raises unexpected ones, then sleeps briefly before the next cycle. + + 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, and **kwargs. + proxies: Proxies applied to the session. + headers: Headers applied to the session. + cookies: Cookies applied to the base session; ignored for TLS. + pre_task_func: Optional awaitable run once per cycle before category + tasks, receiving session and **kwargs. + use_tls: Use the TLS fingerprint session instead of the base session. + **kwargs: Forwarded verbatim to pre_task_func and every task_func call. + """ while True: session_ctx = ( get_tls_session( @@ -263,7 +344,6 @@ async def execute_pagination( query_category_id) must be supplied here, since naming varies per monitor. """ - LOGGER.info("executing pagination", extra={"category": category_identifier}) total_pages = (int(total_products) + products_per_page - 1) // products_per_page semaphore = asyncio.Semaphore(max_concurrent_tasks) @@ -272,6 +352,7 @@ async def execute_pagination( page_range = range(2, total_pages + 1) async def fetch_page(page_index: int): + """Fetch a single page or offset under the shared semaphore.""" async with semaphore: if stop_event.is_set(): return @@ -318,6 +399,17 @@ async def fetch_all_pages( 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( @@ -332,6 +424,22 @@ def handle_page_results( 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): @@ -380,6 +488,21 @@ async def retry_failed_pages( 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)