added doc strings

This commit is contained in:
tpv committed 2026-07-03 19:34:16 -07:00
1 parent 99d660ee52
commit 201721ef3b
1 file changed
+125 -2
+125 -2
View File
@@ -41,6 +41,23 @@ def init_monitor(
testing_proxy_path: str = "configs/proxies.txt", testing_proxy_path: str = "configs/proxies.txt",
logging_level=LOGGING_LEVEL, logging_level=LOGGING_LEVEL,
) -> tuple[list[str], AsyncIOMotorCollection, AioProxies]: ) -> 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) use_production = is_production(development_mode)
setup_logging( setup_logging(
name="logs/latest.log", name="logs/latest.log",
@@ -60,6 +77,19 @@ def init_monitor(
def setup_config( def setup_config(
monitor_name: str, development_mode: DevelopmentMode monitor_name: str, development_mode: DevelopmentMode
) -> tuple[list[str], dict[int], AsyncIOMotorCollection]: ) -> 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) use_production = is_production(development_mode)
config, _ = get_config( config, _ = get_config(
monitor_name=monitor_name, monitor_name=monitor_name,
@@ -76,7 +106,22 @@ def setup_config(
def get_channel_id( def get_channel_id(
product: Product, channel_ids: dict, use_price_drop_bucket: bool = False product: Product, channel_ids: dict, use_price_drop_bucket: bool = False
) -> str: ) -> 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: if product.availability.app_exclusive:
return get_single_channel_id(channel_ids["appExclusive"]) 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: 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): if isinstance(value, list):
return value[0] return value[0]
@@ -113,6 +159,21 @@ def build_product_details(
min_percentage: int, min_percentage: int,
min_price_drop: float | None = None, min_price_drop: float | None = None,
) -> Product | 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"] product_id = product["_id"]
current_price = product["currentPrice"] current_price = product["currentPrice"]
original_price = product["originalPrice"] original_price = product["originalPrice"]
@@ -178,6 +239,26 @@ async def execute_monitor_logic(
use_tls: bool = False, use_tls: bool = False,
**kwargs, **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: while True:
session_ctx = ( session_ctx = (
get_tls_session( get_tls_session(
@@ -263,7 +344,6 @@ async def execute_pagination(
query_category_id) must be supplied here, since naming varies query_category_id) must be supplied here, since naming varies
per monitor. per monitor.
""" """
LOGGER.info("executing pagination", extra={"category": category_identifier}) LOGGER.info("executing pagination", extra={"category": category_identifier})
total_pages = (int(total_products) + products_per_page - 1) // products_per_page total_pages = (int(total_products) + products_per_page - 1) // products_per_page
semaphore = asyncio.Semaphore(max_concurrent_tasks) semaphore = asyncio.Semaphore(max_concurrent_tasks)
@@ -272,6 +352,7 @@ async def execute_pagination(
page_range = range(2, total_pages + 1) page_range = range(2, total_pages + 1)
async def fetch_page(page_index: int): async def fetch_page(page_index: int):
"""Fetch a single page or offset under the shared semaphore."""
async with semaphore: async with semaphore:
if stop_event.is_set(): if stop_event.is_set():
return return
@@ -318,6 +399,17 @@ async def fetch_all_pages(
category_identifier: str, category_identifier: str,
pagination_label: str, pagination_label: str,
) -> list[int]: ) -> 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] tasks = [fetch_func(page) for page in page_range]
results = await asyncio.gather(*tasks, return_exceptions=True) results = await asyncio.gather(*tasks, return_exceptions=True)
return handle_page_results( return handle_page_results(
@@ -332,6 +424,22 @@ def handle_page_results(
pagination_label: str, pagination_label: str,
retry_attempt: int | None = None, retry_attempt: int | None = None,
) -> list[int]: ) -> 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 = [] failed_pages = []
for page_number, result in zip(page_numbers, results): for page_number, result in zip(page_numbers, results):
@@ -380,6 +488,21 @@ async def retry_failed_pages(
pagination_label: str, pagination_label: str,
max_retries: int = 3, max_retries: int = 3,
) -> list[int]: ) -> 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 retry_attempt = 1
while failed_pages and retry_attempt <= max_retries: while failed_pages and retry_attempt <= max_retries:
await asyncio.sleep(2**retry_attempt) await asyncio.sleep(2**retry_attempt)