added alerts collection dsql suggested

This commit is contained in:
TPV committed 2026-07-07 11:03:21 -07:00
1 parent 7e44a5cdb2
commit 6ad9cf5d6d
1 file changed
+12 -10
+12 -10
View File
@@ -78,11 +78,14 @@ async def add_products_to_database(
Rewrites current fields on each upsert. Prices are recorded in a Rewrites current fields on each upsert. Prices are recorded in a
pastPrices array of price/date entries; a price already present in the pastPrices array of price/date entries; a price already present in the
array is never recorded again, so each distinct price keeps only its array is never recorded again, so each distinct price keeps only its
first-seen timestamp. New products and price changes upsert an alert first-seen timestamp. New products and never-before-recorded prices
referencing the product, deduplicated so a product holds at most one upsert an alert referencing the product; a price returning to a value
unclaimed alert regardless of concurrent category or page tasks already in pastPrices emits nothing, so per-source price oscillation
detecting it; alerts write only after the product write succeeds so a (member vs non-member quotes for the same SKU) alerts once per distinct
failed upsert cannot produce an orphaned alert. price instead of every run. Alerts are deduplicated so a product holds
at most one unclaimed alert regardless of concurrent category or page
tasks detecting it, and write only after the product write succeeds so
a failed upsert cannot produce an orphaned alert.
Args: Args:
products: Products to upsert. products: Products to upsert.
@@ -110,11 +113,11 @@ async def add_products_to_database(
document = convert_product_to_mongo(product) document = convert_product_to_mongo(product)
document["_id"] = doc_id document["_id"] = doc_id
existing_doc = existing.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") new_price = document.get("currentPrice")
price_changed = new_price is not None and new_price != old_price recorded_prices = extract_recorded_prices(existing_doc)
never_seen_price = new_price is not None and new_price not in recorded_prices
set_on_insert = {"discoveredDate": now} set_on_insert = {"discoveredDate": now}
if existing_doc is None or price_changed: if existing_doc is None or never_seen_price:
alert = build_alert(monitor_name, document, now) alert = build_alert(monitor_name, document, now)
alert_ops.append( alert_ops.append(
UpdateOne( UpdateOne(
@@ -125,8 +128,7 @@ async def add_products_to_database(
) )
update = {"$set": document, "$setOnInsert": set_on_insert} update = {"$set": document, "$setOnInsert": set_on_insert}
recorded_prices = extract_recorded_prices(existing_doc) if never_seen_price:
if new_price is not None and new_price not in recorded_prices:
update["$push"] = {"pastPrices": {"price": new_price, "date": now}} update["$push"] = {"pastPrices": {"price": new_price, "date": now}}
ops.append(UpdateOne({"_id": doc_id}, update, upsert=True)) ops.append(UpdateOne({"_id": doc_id}, update, upsert=True))