From 9e2ede16b1300a42dd4aa174f42bffd4f0966bec Mon Sep 17 00:00:00 2001 From: TPV Date: Mon, 6 Jul 2026 19:55:43 -0700 Subject: [PATCH] added alerts collection dsql suggested --- src/monitor_utils/constants.py | 2 +- src/monitor_utils/database.py | 102 +++++++++++++++++++++++++++------ 2 files changed, 85 insertions(+), 19 deletions(-) diff --git a/src/monitor_utils/constants.py b/src/monitor_utils/constants.py index e44e325..881a9a2 100644 --- a/src/monitor_utils/constants.py +++ b/src/monitor_utils/constants.py @@ -18,6 +18,7 @@ MONGO_DATABASE_NAME = "tpv_woot" MONGO_URL = ( "mongodb://tpv:n3%23QI07GnbZCbvGJ@zelda.disqualifier.me:27017/?authSource=tpv_woot" ) +ALERTS_COLLECTION_NAME = "alerts" PRODUCTION_BOT_TOKEN = ( "MTQwMDE4NDk2MzY4NDYzMDUyOQ.GOrqOb.6frT3upUQ5B9XubVOIT9pdzVcYVXSrnEMK7RlA" ) @@ -36,7 +37,6 @@ RETRY_LIMIT = 5 RETRY_WAIT = 3 MIN_BACKOFF_DELAY = 0.1 MAX_BACKOFF_DELAY = 2.0 -STOP_POLL_DELAY = 5 FAILURE_DELAY = 3 TIMEOUT_SECONDS = 30 diff --git a/src/monitor_utils/database.py b/src/monitor_utils/database.py index 5199667..d5c6b06 100644 --- a/src/monitor_utils/database.py +++ b/src/monitor_utils/database.py @@ -2,7 +2,8 @@ Provides the shared Motor client, collection access, and the upsert path that rewrites current product fields while preserving prior prices in a pastPrices -map and stamping lastUpdated on every write. +array deduplicated by price, stamping lastUpdated on every write, and emitting +alert references for new products and price changes. """ import logging @@ -14,7 +15,11 @@ from motor.motor_asyncio import AsyncIOMotorClient, AsyncIOMotorCollection from pymongo import UpdateOne from pymongo.errors import PyMongoError -from monitor_utils.constants import ProductStatus, MONGO_DATABASE_NAME, MONGO_URL +from monitor_utils.constants import ( + ALERTS_COLLECTION_NAME, + MONGO_DATABASE_NAME, + MONGO_URL, +) from monitor_utils.models import Product LOGGER = logging.getLogger(__name__) @@ -65,53 +70,65 @@ async def add_products_to_database( products: list[Product], collection: AsyncIOMotorCollection, category_identifier: str, + monitor_name: str, use_composite: bool = False, ): - """Upsert products, tracking price history and last-updated time. + """Upsert products, tracking deduplicated price history and emitting alerts. - 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. + Rewrites current fields on each upsert. 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 price changes + insert an alert referencing the product; alerts insert 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. """ + 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}) - existing_prices = {str(doc["_id"]): doc.get("currentPrice") async for doc in cursor} + cursor = collection.find( + {"_id": {"$in": doc_ids}}, {"currentPrice": 1, "pastPrices": 1} + ) + existing = {str(doc["_id"]): doc async for doc in cursor} ops = [] + alerts = [] now = datetime.now(timezone.utc) - price_key = str(int(now.timestamp() * 1000)) for product, doc_id in zip(products, doc_ids): document = convert_product_to_mongo(product) document["_id"] = doc_id - old_price = existing_prices.get(doc_id) + existing_doc = existing.get(doc_id) + old_price = existing_doc.get("currentPrice") if existing_doc else None new_price = document.get("currentPrice") price_changed = new_price is not None and new_price != old_price set_on_insert = {"discoveredDate": now} - if old_price is None: - set_on_insert["status"] = ProductStatus.PENDING - elif price_changed: - document["status"] = ProductStatus.PENDING - document["priceBeforeDrop"] = old_price + if existing_doc is None or price_changed: + alerts.append(build_alert(monitor_name, document, now)) update = {"$set": document, "$setOnInsert": set_on_insert} - if price_changed: - update["$set"][f"pastPrices.{price_key}"] = new_price + recorded_prices = extract_recorded_prices(existing_doc) + if new_price is not None and new_price not in recorded_prices: + 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 alerts: + await get_collection(ALERTS_COLLECTION_NAME).insert_many( + alerts, ordered=False + ) except PyMongoError: LOGGER.exception( "failed to add products to database", @@ -120,6 +137,54 @@ async def add_products_to_database( raise +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 priority percentage. + now: Detection timestamp stored as createdAt. + + Returns: + The alert document for insertion. + """ + + percentage = document.get("percentage") or 0 + data = { + "monitor": monitor_name, + "product_id": document["_id"], + "priority": min(percentage // 10, 10), + "claimed": False, + "attempts": 0, + "createdAt": now, + } + return { + "monitor": monitor_name, + "product_id": document["_id"], + "priority": min(percentage // 10, 10), + "claimed": False, + "attempts": 0, + "createdAt": now, + } + + def build_doc_id( product_id: str, category_identifier: str, use_composite: bool = False ) -> str: @@ -133,6 +198,7 @@ def build_doc_id( Returns: The composite id when use_composite is set, otherwise product_id. """ + return f"{product_id}_{category_identifier}" if use_composite else product_id