import asyncio import json import logging import os import signal import sys from contextlib import suppress from dotenv import load_dotenv from telegram import Update from telegram.ext import Application, ExtBot from db import close_pool, get_pool from scraper import extract_ad_fields, fetch_ads from notifier import log_notification, notify_new_ad, notify_price_drop 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 FROM keywords " "WHERE is_active = true " "AND (last_scraped_at IS NULL OR last_scraped_at < now() - (interval_minutes || ' minutes')::interval)" ) for row in rows: kw_id = str(row["id"]) keyword = row["keyword"] subs = await pool.fetch( "SELECT telegram_id FROM users u JOIN keyword_subscriptions ks ON u.id = ks.user_id " "WHERE ks.keyword_id = $1 AND u.is_active = true", kw_id, ) if not subs: await pool.execute("UPDATE keywords SET is_active = false WHERE id = $1", kw_id) continue telegram_ids = [sub["telegram_id"] for sub in subs] logger.info("Scraping keyword '%s' (%d subscriber(s))", keyword, len(telegram_ids)) try: ads_raw, total_hits = await fetch_ads(keyword) new_count = 0 for ad_data in ads_raw: fields = extract_ad_fields(ad_data) wh_ad_id = fields["wh_ad_id"] existing = await pool.fetchrow( "SELECT id, price FROM ads WHERE wh_ad_id = $1", wh_ad_id, ) 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", wh_ad_id, json.dumps(ad_data), fields["title"], fields["price"], fields["location"], fields["url"], fields.get("published_at"), fields.get("main_image_url"), fields.get("postcode"), fields.get("modified_at"), ) ad_uuid = str(ad_row["id"]) notify_fields = {**fields, "keyword": keyword} for tg_id in telegram_ids: 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"] if old_price is not None and new_price is not None and new_price < old_price: await pool.execute( "UPDATE ads SET price = $1, main_image_url = $2, postcode = $3, modified_at = $4 WHERE id = $5", new_price, fields.get("main_image_url"), fields.get("postcode"), fields.get("modified_at"), ad_uuid, ) await pool.execute( "INSERT INTO price_history (ad_id, old_price, new_price) VALUES ($1, $2, $3) ON CONFLICT DO NOTHING", ad_uuid, old_price, new_price, ) 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, ) await pool.execute("UPDATE keywords SET last_scraped_at = now() WHERE id = $1", kw_id) await pool.execute( "INSERT INTO scrape_logs (keyword_id, status, ads_found, new_ads) VALUES ($1, 'success', $2, $3)", kw_id, len(ads_raw), new_count, ) except Exception: logger.exception("Error scraping keyword '%s' (%s)", keyword, kw_id) await pool.execute( "INSERT INTO scrape_logs (keyword_id, status, error_message) VALUES ($1, 'error', $2)", kw_id, str(sys.exc_info()[1]), ) await asyncio.sleep(5) except Exception: logger.exception("Scheduler iteration error") await asyncio.sleep(30) async def main() -> None: logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", ) if not os.getenv("TELEGRAM_BOT_TOKEN"): logger.error("TELEGRAM_BOT_TOKEN is required") sys.exit(1) pool = await get_pool() app = Application.builder().token(os.getenv("TELEGRAM_BOT_TOKEN")).build() from bot import register_handlers, setup_global_commands # noqa: E402 await setup_global_commands(app) register_handlers(app) scheduler = asyncio.ensure_future(scheduler_task(pool, app.bot)) loop = asyncio.get_running_loop() stop = loop.create_future() def _signal_handler() -> None: if not stop.done(): stop.set_result(True) for sig in (signal.SIGINT, signal.SIGTERM): loop.add_signal_handler(sig, _signal_handler) try: await app.initialize() await app.start() logger.info("Bot started with long polling") poll_task = asyncio.ensure_future(app.updater.start_polling()) # type: ignore[attr-defined] await stop logger.info("Shutting down...") finally: scheduler.cancel() with suppress(asyncio.CancelledError): await scheduler poll_task.cancel() with suppress(asyncio.CancelledError): await poll_task await app.shutdown() await close_pool() logger.info("Shutdown complete") if __name__ == "__main__": asyncio.run(main())