From 136a67c16f57b98fc960c6a1b6697272780e39e0 Mon Sep 17 00:00:00 2001 From: tpv Date: Mon, 6 Jul 2026 09:46:07 -0700 Subject: [PATCH] testing price tracking/composite key changes --- src/monitor_utils/config.py | 4 ++ src/monitor_utils/database.py | 91 ++++++++++++++++++++++++++++++----- src/monitor_utils/sessions.py | 4 ++ 3 files changed, 88 insertions(+), 11 deletions(-) diff --git a/src/monitor_utils/config.py b/src/monitor_utils/config.py index 1161774..454e0c0 100644 --- a/src/monitor_utils/config.py +++ b/src/monitor_utils/config.py @@ -17,11 +17,13 @@ class DevelopmentMode(str, Enum): def is_production(development_mode: DevelopmentMode) -> bool: """Return True when the mode is PRODUCTION.""" + return development_mode == DevelopmentMode.PRODUCTION def get_bot_token(is_production: bool) -> str: """Return the production or development bot token for the mode.""" + if is_production: return PRODUCTION_BOT_TOKEN @@ -45,6 +47,7 @@ def get_config( Returns: Tuple of the parsed config dict and the resolved config path. """ + monitor_name = monitor_name.replace(".json", "") base = config_dir if use_production_path else test_config_dir config_path = pathlib.Path(base) / f"{monitor_name}.json" @@ -54,5 +57,6 @@ def get_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/src/monitor_utils/database.py b/src/monitor_utils/database.py index dd5a43c..82a6525 100644 --- a/src/monitor_utils/database.py +++ b/src/monitor_utils/database.py @@ -14,11 +14,26 @@ 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] @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, maxPoolSize=200, @@ -34,6 +49,8 @@ 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}) @@ -41,18 +58,42 @@ async def add_products_to_database( products: list[Product], collection: AsyncIOMotorCollection, category_identifier: str, + use_composite: bool = False, ): + """Upsert products, tracking price history and last-updated time. + + Rewrites current fields on each upsert while preserving prior prices in a + pastPrices map keyed by epoch-millisecond timestamps, appending only when + the price differs from the stored value. Flags dropped prices as pending + and records the drop's prior price. + + Args: + products: Products to upsert. + collection: Target MongoDB collection. + category_identifier: Category suffix appended to composite ids. + use_composite: Build composite ids so multi-variant monitors (Woot) + keep one doc per variant; other monitors collapse to one doc per + product_id. + """ if not products: return - composite_ids = [f"{p.product_id}_{category_identifier}" for p in products] - cursor = collection.find({"_id": {"$in": composite_ids}}, {"currentPrice": 1}) + doc_ids = { + id(p): build_doc_id(p.product_id, category_identifier, use_composite) + for p in products + } + cursor = collection.find( + {"_id": {"$in": list(doc_ids.values())}}, {"currentPrice": 1} + ) existing_prices = {str(doc["_id"]): doc.get("currentPrice") async for doc in cursor} ops = [] + now = datetime.now(timezone.utc) + price_key = str(int(now.timestamp() * 1000)) for product in products: + doc_id = doc_ids[id(product)] document = convert_product_to_mongo(product, category_identifier) - doc_id = document["_id"] + document["_id"] = doc_id old_price = existing_prices.get(doc_id) new_price = document.get("currentPrice") @@ -60,20 +101,18 @@ async def add_products_to_database( new_price is not None and new_price < old_price ) - set_on_insert = {"discoveredDate": datetime.now(timezone.utc)} + set_on_insert = {"discoveredDate": now} if price_dropped: document["status"] = STATUS_PENDING document["priceBeforeDrop"] = old_price else: set_on_insert["status"] = STATUS_PENDING - ops.append( - UpdateOne( - {"_id": doc_id}, - {"$set": document, "$setOnInsert": set_on_insert}, - upsert=True, - ) - ) + update = {"$set": document, "$setOnInsert": set_on_insert} + if new_price is not None and new_price != old_price: + update["$set"][f"pastPrices.{price_key}"] = new_price + + ops.append(UpdateOne({"_id": doc_id}, update, upsert=True)) try: await collection.bulk_write(ops, ordered=False) @@ -85,7 +124,36 @@ async def add_products_to_database( raise +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, category_identifier: str) -> dict: + """Flatten a Product into its MongoDB document representation. + + Coerces string prices to floats and stamps lastUpdated with the current + UTC time. + + Args: + product: Product to convert. + category_identifier: Category suffix appended to the composite _id. + + Returns: + The MongoDB document dict for the product. + """ + data = asdict(product) metadata = data["metadata"] timing = data["timing"] @@ -117,4 +185,5 @@ def convert_product_to_mongo(product: Product, category_identifier: str) -> dict "appExclusive": availability["app_exclusive"], "memberPick": availability["member_pick"], "memberDeal": availability["member_deal"], + "lastUpdated": datetime.now(timezone.utc), } diff --git a/src/monitor_utils/sessions.py b/src/monitor_utils/sessions.py index dd96fad..43a0689 100644 --- a/src/monitor_utils/sessions.py +++ b/src/monitor_utils/sessions.py @@ -63,6 +63,7 @@ async def get_base_session( Yields: The configured aiohttp session. """ + headers = headers if headers else {} headers.setdefault("User-Agent", UA.random) session = aiohttp.ClientSession( @@ -95,6 +96,7 @@ async def get_tls_session( Yields: The configured curl_cffi session. """ + headers = headers if headers else {} headers.setdefault("User-Agent", UA.chrome) async with AsyncSession( @@ -115,6 +117,7 @@ def build_session_details( session: Session whose headers are updated in place. headers: Headers to merge; User-Agent is defaulted if absent. """ + headers = headers or {} headers.setdefault("User-Agent", UA.random) session.headers.update(headers) @@ -174,6 +177,7 @@ async def send_request( json.JSONDecodeError: When parsing fails and reraise_json_error is True. """ + verify = True if development_proxy: proxy = development_proxy