refactor: ID-based ad detection with fire-and-forget notifications
- Remove initial_loaded/baseline logic from scheduler - Remove initial_loaded from keywords SELECT query - Replace synchronous notification loops with asyncio.create_task (fire-and-forget) for both new ads and price drops - Add safe_notify_new_ad/safe_notify_price_drop wrapper functions - Scraper already simplified (single page, 30 ads, no cursor)
This commit is contained in:
hermes
+30
-34
@@ -19,11 +19,33 @@ logger = logging.getLogger(__name__)
|
||||
load_dotenv()
|
||||
|
||||
|
||||
async def safe_notify_new_ad(bot, pool, tg_id, notify_fields, ad_uuid):
|
||||
try:
|
||||
msg_id_val = await notify_new_ad(bot, tg_id, notify_fields)
|
||||
if msg_id_val:
|
||||
user_row = await pool.fetchrow("SELECT id FROM users WHERE telegram_id = $1", tg_id)
|
||||
if user_row:
|
||||
await log_notification(pool, str(user_row["id"]), ad_uuid, msg_id_val)
|
||||
except Exception:
|
||||
logger.exception("Failed to send new ad notification for %s", ad_uuid)
|
||||
|
||||
|
||||
async def safe_notify_price_drop(bot, pool, tg_id, notify_fields, ad_uuid):
|
||||
try:
|
||||
msg_id_val = await notify_price_drop(bot, tg_id, notify_fields)
|
||||
if msg_id_val:
|
||||
user_row = await pool.fetchrow("SELECT id FROM users WHERE telegram_id = $1", tg_id)
|
||||
if user_row:
|
||||
await log_notification(pool, str(user_row["id"]), ad_uuid, msg_id_val)
|
||||
except Exception:
|
||||
logger.exception("Failed to send price drop notification for %s", ad_uuid)
|
||||
|
||||
|
||||
async def scheduler_task(pool: object, bot: ExtBot) -> None:
|
||||
while True:
|
||||
try:
|
||||
rows = await pool.fetch(
|
||||
"SELECT id, keyword, interval_minutes, initial_loaded FROM keywords "
|
||||
"SELECT id, keyword, interval_minutes FROM keywords "
|
||||
"WHERE is_active = true "
|
||||
"AND (last_scraped_at IS NULL OR last_scraped_at < now() - (interval_minutes || ' minutes')::interval)"
|
||||
)
|
||||
@@ -31,7 +53,6 @@ async def scheduler_task(pool: object, bot: ExtBot) -> None:
|
||||
for row in rows:
|
||||
kw_id = str(row["id"])
|
||||
keyword = row["keyword"]
|
||||
initial_loaded = row["initial_loaded"]
|
||||
|
||||
subs = await pool.fetch(
|
||||
"SELECT telegram_id FROM users u JOIN keyword_subscriptions ks ON u.id = ks.user_id "
|
||||
@@ -50,15 +71,9 @@ async def scheduler_task(pool: object, bot: ExtBot) -> None:
|
||||
ads_raw, total_hits = await fetch_ads(keyword)
|
||||
new_count = 0
|
||||
|
||||
if not initial_loaded and len(ads_raw) > 0:
|
||||
logger.info("Initial baseline load for '%s' — indexing %d ads, no notifications", keyword, len(ads_raw))
|
||||
|
||||
for ad_data in ads_raw:
|
||||
fields = extract_ad_fields(ad_data)
|
||||
wh_ad_id = fields["wh_ad_id"]
|
||||
is_price_drop = False
|
||||
old_price = None
|
||||
new_price = None
|
||||
|
||||
existing = await pool.fetchrow(
|
||||
"SELECT id, price FROM ads WHERE wh_ad_id = $1",
|
||||
@@ -66,6 +81,7 @@ async def scheduler_task(pool: object, bot: ExtBot) -> None:
|
||||
)
|
||||
|
||||
if not existing:
|
||||
# NEW AD - insert and fire-and-forget notify
|
||||
ad_row = await pool.fetchrow(
|
||||
"INSERT INTO ads (wh_ad_id, raw_json, title, price, location, url, published_at, main_image_url, postcode, modified_at) "
|
||||
"VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) RETURNING id",
|
||||
@@ -75,20 +91,12 @@ async def scheduler_task(pool: object, bot: ExtBot) -> None:
|
||||
)
|
||||
ad_uuid = str(ad_row["id"])
|
||||
|
||||
# Only notify for genuinely new ads after baseline load is done
|
||||
if initial_loaded:
|
||||
notify_fields = {**fields, "keyword": keyword}
|
||||
for tg_id in telegram_ids:
|
||||
msg_id_val = await notify_new_ad(bot, tg_id, notify_fields)
|
||||
if msg_id_val:
|
||||
user_row = await pool.fetchrow("SELECT id FROM users WHERE telegram_id = $1", tg_id)
|
||||
if user_row:
|
||||
try:
|
||||
await log_notification(pool, str(user_row["id"]), ad_uuid, msg_id_val)
|
||||
except Exception:
|
||||
logger.exception("Failed to log new ad notification")
|
||||
asyncio.create_task(safe_notify_new_ad(bot, pool, tg_id, notify_fields, ad_uuid))
|
||||
new_count += 1
|
||||
else:
|
||||
# EXISTING AD - check price drop
|
||||
ad_uuid = str(existing["id"])
|
||||
old_price = existing["price"]
|
||||
new_price = fields["price"]
|
||||
@@ -102,29 +110,17 @@ async def scheduler_task(pool: object, bot: ExtBot) -> None:
|
||||
"INSERT INTO price_history (ad_id, old_price, new_price) VALUES ($1, $2, $3) ON CONFLICT DO NOTHING",
|
||||
ad_uuid, old_price, new_price,
|
||||
)
|
||||
is_price_drop = True
|
||||
notify_fields = {**fields, "keyword": keyword}
|
||||
for tg_id in telegram_ids:
|
||||
asyncio.create_task(safe_notify_price_drop(bot, pool, tg_id, notify_fields, ad_uuid))
|
||||
else:
|
||||
# Update metadata if missing
|
||||
if fields.get("main_image_url") or fields.get("postcode"):
|
||||
await pool.execute(
|
||||
"UPDATE ads SET main_image_url = COALESCE($1, main_image_url), postcode = COALESCE($2, postcode) WHERE id = $3 AND (main_image_url IS NULL OR postcode IS NULL)",
|
||||
fields.get("main_image_url"), fields.get("postcode"), ad_uuid,
|
||||
)
|
||||
|
||||
if is_price_drop:
|
||||
notify_fields = {**fields, "keyword": keyword}
|
||||
for tg_id in telegram_ids:
|
||||
msg_id_val = await notify_price_drop(bot, tg_id, notify_fields)
|
||||
if msg_id_val:
|
||||
user_row = await pool.fetchrow("SELECT id FROM users WHERE telegram_id = $1", tg_id)
|
||||
if user_row:
|
||||
try:
|
||||
await log_notification(pool, str(user_row["id"]), ad_uuid, msg_id_val)
|
||||
except Exception:
|
||||
logger.exception("Failed to log price drop notification")
|
||||
|
||||
if not initial_loaded:
|
||||
await pool.execute("UPDATE keywords SET initial_loaded = true WHERE id = $1", kw_id)
|
||||
|
||||
await pool.execute("UPDATE keywords SET last_scraped_at = now() WHERE id = $1", kw_id)
|
||||
|
||||
await pool.execute(
|
||||
|
||||
Reference in New Issue
Block a user