diff --git a/worker/src/bot.py b/worker/src/bot.py index fa94116..5c3acaa 100644 --- a/worker/src/bot.py +++ b/worker/src/bot.py @@ -189,10 +189,23 @@ def _format_kw_card(kw: dict) -> str: status_icon = "🟢 Active" if kw["is_active"] else "šŸ”“ Stopped" subs_line = f"\nSubscribers: {kw['subs']}" if kw.get("subs", 1) > 1 else "" + price_line = "" + if kw.get("price_min") is not None or kw.get("price_max") is not None: + parts = [] + if kw.get("price_min") is not None: + parts.append(f"€{kw['price_min'] / 100:.0f}") + if kw.get("price_max") is not None: + parts.append(f"€{kw['price_max'] / 100:.0f}") + price_line = f"\nPrice: {'–'.join(parts)}" + + postcode_line = "" + if kw.get("allowed_postcodes"): + postcode_line = f"\nPostcodes: {', '.join(kw['allowed_postcodes'])}" + return ( f"šŸ” {kw['keyword']}\n" f"{status_icon} | Interval: {kw['interval_minutes']} min\n" - f"Last scrape: {_vienna_time(kw.get('last_scraped_at'))}{subs_line}" + f"Last scrape: {_vienna_time(kw.get('last_scraped_at'))}{price_line}{postcode_line}{subs_line}" ) @@ -226,12 +239,37 @@ async def setup_global_commands(app: Application) -> None: await app.bot.set_my_commands([ ("start", "Open main menu"), ("admin", "Admin panel (admins only)"), + ("price_min", "Set min price: /price_min <€>"), + ("price_max", "Set max price: /price_max <€>"), + ("clear_price", "Remove price filter: /clear_price "), + ("postcode", "Set postcodes: /postcode p1,p2"), + ("clear_postcode", "Remove postcode filter: /clear_postcode "), + ("mute_hours", "Set mute window: /mute_hours HH:MM-HH:MM"), + ("mute_off", "Disable mute hours"), + ("digest_on", "Enable digest: /digest_on [minutes]"), + ("digest_off", "Disable digest mode"), + ("status", "Show your settings"), ]) def register_handlers(app: Application) -> None: app.add_handler(CommandHandler("start", start_handler)) app.add_handler(CommandHandler("admin", admin_handler)) + # Phase 2: Price filters + app.add_handler(CommandHandler("price_min", price_min_handler)) + app.add_handler(CommandHandler("price_max", price_max_handler)) + app.add_handler(CommandHandler("clear_price", clear_price_handler)) + # Phase 2: Postcode filters + app.add_handler(CommandHandler("postcode", postcode_handler)) + app.add_handler(CommandHandler("clear_postcode", clear_postcode_handler)) + # Phase 2: Mute hours + app.add_handler(CommandHandler("mute_hours", mute_hours_handler)) + app.add_handler(CommandHandler("mute_off", mute_off_handler)) + # Phase 2: Digest mode + app.add_handler(CommandHandler("digest_on", digest_on_handler)) + app.add_handler(CommandHandler("digest_off", digest_off_handler)) + # Phase 2: Status + app.add_handler(CommandHandler("status", status_handler)) app.add_handler(CallbackQueryHandler(callback_router)) # Catch all non-command text messages for conversation flows app.add_handler(MessageHandler(TEXT_FILTER, text_input_handler)) @@ -268,6 +306,326 @@ async def admin_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> N ) +# ── Phase 2: Price filter commands ──────────────────────────────────────── + +async def _find_keyword_for_user(pool, user_id: str, keyword_text: str) -> dict | None: + """Find a keyword matching the text that the user subscribes to (or any if admin).""" + # First try exact match for the user's keywords + row = await pool.fetchrow( + """SELECT kw.* FROM keywords kw + JOIN keyword_subscriptions ks ON ks.keyword_id = kw.id + WHERE LOWER(kw.keyword) = LOWER($1) AND ks.user_id = $2""", + keyword_text.lower(), user_id, + ) + if row: + return dict(row) + + # Admin can access any keyword + row = await pool.fetchrow( + "SELECT * FROM keywords WHERE LOWER(keyword) = LOWER($1)", + keyword_text.lower(), + ) + return dict(row) if row else None + + +async def price_min_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: + if not update.message or not update.message.text: # type: ignore[union-attr] + return + + parts = update.message.text.split(maxsplit=2) # type: ignore[union-attr] + if len(parts) < 3: + await update.message.reply_text("Usage: /price_min ") # type: ignore[union-attr] + return + + keyword_text = parts[1] + try: + amount = float(parts[2]) + if amount < 0: + raise ValueError + except ValueError: + await update.message.reply_text("Enter a valid positive amount.") # type: ignore[union-attr] + return + + user = await _require_user(update) + if not user: + return + + pool = await get_pool() + kw = await _find_keyword_for_user(pool, user["id"], keyword_text) + if not kw: + await update.message.reply_text(f"Keyword '{keyword_text}' not found.") # type: ignore[union-attr] + return + + price_cents = int(round(amount * 100)) + await pool.execute("UPDATE keywords SET price_min = $1 WHERE id = $2", price_cents, kw["id"]) + await update.message.reply_text( # type: ignore[union-attr] + f"āœ… {kw['keyword']}: min price set to €{amount:.2f}", parse_mode="HTML") + + +async def price_max_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: + if not update.message or not update.message.text: # type: ignore[union-attr] + return + + parts = update.message.text.split(maxsplit=2) # type: ignore[union-attr] + if len(parts) < 3: + await update.message.reply_text("Usage: /price_max ") # type: ignore[union-attr] + return + + keyword_text = parts[1] + try: + amount = float(parts[2]) + if amount < 0: + raise ValueError + except ValueError: + await update.message.reply_text("Enter a valid positive amount.") # type: ignore[union-attr] + return + + user = await _require_user(update) + if not user: + return + + pool = await get_pool() + kw = await _find_keyword_for_user(pool, user["id"], keyword_text) + if not kw: + await update.message.reply_text(f"Keyword '{keyword_text}' not found.") # type: ignore[union-attr] + return + + price_cents = int(round(amount * 100)) + await pool.execute("UPDATE keywords SET price_max = $1 WHERE id = $2", price_cents, kw["id"]) + await update.message.reply_text( # type: ignore[union-attr] + f"āœ… {kw['keyword']}: max price set to €{amount:.2f}", parse_mode="HTML") + + +async def clear_price_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: + if not update.message or not update.message.text: # type: ignore[union-attr] + return + + parts = update.message.text.split(maxsplit=1) # type: ignore[union-attr] + if len(parts) < 2: + await update.message.reply_text("Usage: /clear_price ") # type: ignore[union-attr] + return + + user = await _require_user(update) + if not user: + return + + pool = await get_pool() + kw = await _find_keyword_for_user(pool, user["id"], parts[1]) + if not kw: + await update.message.reply_text(f"Keyword '{parts[1]}' not found.") # type: ignore[union-attr] + return + + await pool.execute("UPDATE keywords SET price_min = NULL, price_max = NULL WHERE id = $1", kw["id"]) + await update.message.reply_text( # type: ignore[union-attr] + f"āœ… {kw['keyword']}: price filters cleared", parse_mode="HTML") + + +# ── Phase 2: Postcode filter commands ───────────────────────────────────── + +async def postcode_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: + if not update.message or not update.message.text: # type: ignore[union-attr] + return + + parts = update.message.text.split(maxsplit=1) # type: ignore[union-attr] + if len(parts) < 2: + await update.message.reply_text("Usage: /postcode p1,p2,p3") # type: ignore[union-attr] + return + + user = await _require_user(update) + if not user: + return + + postcodes = [p.strip() for p in parts[1].split(",")] + for p in postcodes: + if not p.isdigit() or len(p) != 4: + await update.message.reply_text(f"Invalid postcode '{p}'. Use 4-digit codes (e.g., 1010,1020).") # type: ignore[union-attr] + return + + pool = await get_pool() + kw = await _find_keyword_for_user(pool, user["id"], parts[0]) + if not kw: + await update.message.reply_text(f"Keyword '{parts[0]}' not found.") # type: ignore[union-attr] + return + + await pool.execute("UPDATE keywords SET allowed_postcodes = $1 WHERE id = $2", postcodes, kw["id"]) + await update.message.reply_text( # type: ignore[union-attr] + f"āœ… {kw['keyword']}: postcodes set to {', '.join(postcodes)}", parse_mode="HTML") + + +async def clear_postcode_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: + if not update.message or not update.message.text: # type: ignore[union-attr] + return + + parts = update.message.text.split(maxsplit=1) # type: ignore[union-attr] + if len(parts) < 2: + await update.message.reply_text("Usage: /clear_postcode ") # type: ignore[union-attr] + return + + user = await _require_user(update) + if not user: + return + + pool = await get_pool() + kw = await _find_keyword_for_user(pool, user["id"], parts[1]) + if not kw: + await update.message.reply_text(f"Keyword '{parts[1]}' not found.") # type: ignore[union-attr] + return + + await pool.execute("UPDATE keywords SET allowed_postcodes = NULL WHERE id = $1", kw["id"]) + await update.message.reply_text( # type: ignore[union-attr] + f"āœ… {kw['keyword']}: postcode filter cleared", parse_mode="HTML") + + +# ── Phase 2: Mute hours commands ────────────────────────────────────────── + +async def mute_hours_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: + if not update.message or not update.message.text: # type: ignore[union-attr] + return + + parts = update.message.text.split(maxsplit=1) # type: ignore[union-attr] + if len(parts) < 2: + await update.message.reply_text("Usage: /mute_hours HH:MM-HH:MM (UTC)") # type: ignore[union-attr] + return + + try: + start_str, end_str = parts[1].split("-", 1) + # Validate time format + datetime.strptime(start_str, "%H:%M") + datetime.strptime(end_str, "%H:%M") + except (ValueError, AttributeError): + await update.message.reply_text("Usage: /mute_hours HH:MM-HH:MM (UTC), e.g. /mute_hours 22:00-07:00") # type: ignore[union-attr] + return + + user = await _require_user(update) + if not user: + return + + pool = await get_pool() + await pool.execute( + """INSERT INTO user_settings (telegram_id, mute_start, mute_end) VALUES ($1, $2, $3) + ON CONFLICT (telegram_id) DO UPDATE SET mute_start = $2, mute_end = $3""", + str(user["telegram_id"]), start_str, end_str, + ) + await update.message.reply_text( # type: ignore[union-attr] + f"āœ… Mute hours set: {start_str}–{end_str} UTC", parse_mode="HTML") + + +async def mute_off_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: + user = await _require_user(update) + if not user: + return + + pool = await get_pool() + await pool.execute( + """INSERT INTO user_settings (telegram_id, mute_start, mute_end) VALUES ($1, NULL, NULL) + ON CONFLICT (telegram_id) DO UPDATE SET mute_start = NULL, mute_end = NULL""", + str(user["telegram_id"]), + ) + msg = update.message or update.callback_query # type: ignore[union-attr] + await msg.reply_text("āœ… Mute hours disabled.", parse_mode="HTML") + + +# ── Phase 2: Digest mode commands ───────────────────────────────────────── + +async def digest_on_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: + if not update.message or not update.message.text: # type: ignore[union-attr] + return + + parts = update.message.text.split(maxsplit=1) # type: ignore[union-attr] + interval = 60 # default + + if len(parts) > 1: + try: + interval = int(parts[1]) + if interval < 5 or interval > 1440: + raise ValueError + except ValueError: + await update.message.reply_text("Enter a number between 5 and 1440 minutes.") # type: ignore[union-attr] + return + + user = await _require_user(update) + if not user: + return + + pool = await get_pool() + await pool.execute( + """INSERT INTO user_settings (telegram_id, digest_mode, digest_interval) VALUES ($1, true, $2) + ON CONFLICT (telegram_id) DO UPDATE SET digest_mode = true, digest_interval = $2""", + str(user["telegram_id"]), interval, + ) + await update.message.reply_text( # type: ignore[union-attr] + f"āœ… Digest mode enabled (every {interval} min)", parse_mode="HTML") + + +async def digest_off_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: + user = await _require_user(update) + if not user: + return + + pool = await get_pool() + # Flush any pending digest items before disabling + buffered = await pool.fetch( + "SELECT db.id, db.keyword, db.title, db.price, db.url FROM digest_buffer db WHERE db.telegram_id = $1", + str(user["telegram_id"]), + ) + + if buffered: + # Send immediate summary of pending items + by_keyword = {} + for item in buffered: + by_keyword.setdefault(item["keyword"], []).append(item) + + lines = ["šŸ“¦ Pending Digest Summary"] + for keyword, items in by_keyword.items(): + lines.append(f"\nšŸ” {keyword} ({len(items)} ads)") + for item in items: + price_str = f"€{item['price'] / 100:.0f}" if item["price"] else "N/A" + lines.append(f" • {item['title']} — {price_str}") + + try: + await update.message.reply_text("\n".join(lines), parse_mode="HTML") # type: ignore[union-attr] + except Exception: + pass + await pool.execute("DELETE FROM digest_buffer WHERE telegram_id = $1", str(user["telegram_id"])) + + await pool.execute( + """INSERT INTO user_settings (telegram_id, digest_mode) VALUES ($1, false) + ON CONFLICT (telegram_id) DO UPDATE SET digest_mode = false""", + str(user["telegram_id"]), + ) + msg = update.message or update.callback_query # type: ignore[union-attr] + await msg.reply_text("āœ… Digest mode disabled.", parse_mode="HTML") + + +# ── Phase 2: Status command ─────────────────────────────────────────────── + +async def status_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: + user = await _require_user(update) + if not user: + return + + pool = await get_pool() + settings = await pool.fetchrow( + "SELECT mute_start, mute_end, digest_mode, digest_interval FROM user_settings WHERE telegram_id = $1", + str(user["telegram_id"]), + ) + + lines = ["āš™ļø Your Settings"] + + if settings and settings["mute_start"]: + lines.append(f"\nšŸ”‡ Mute hours: {settings['mute_start']}–{settings['mute_end']} UTC") + else: + lines.append("\nšŸ”‡ Mute hours: off") + + if settings and settings["digest_mode"]: + lines.append(f"šŸ“¦ Digest: on (every {settings['digest_interval']} min)") + else: + lines.append("šŸ“¦ Digest: off") + + msg = update.message or update.callback_query # type: ignore[union-attr] + await msg.reply_text("\n".join(lines), parse_mode="HTML") + + # ── text input handler (keyword name, custom interval, admin flows) ─────── async def text_input_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: diff --git a/worker/src/main.py b/worker/src/main.py index f7789e2..00a2214 100644 --- a/worker/src/main.py +++ b/worker/src/main.py @@ -17,13 +17,37 @@ from telegram.ext import Application, ExtBot from db import close_pool, get_pool from health import create_health_app, record_scheduler_run, set_start_time, set_telegram_polling from scraper import extract_ad_fields, fetch_ads -from notifier import log_notification, notify_new_ad, notify_price_drop +from notifier import log_notification, notify_new_ad, notify_price_drop, is_user_muted, buffer_for_digest logger = logging.getLogger(__name__) load_dotenv() +def _ad_passes_filters(fields: dict, kw_row: dict) -> bool: + """Check if an ad passes the keyword's price and postcode filters.""" + price = fields.get("price") + + # Price filter (stored in cents, ad price is in euros as float) + if price is not None: + price_cents = int(round(price * 100)) + price_min = kw_row.get("price_min") + price_max = kw_row.get("price_max") + if price_min is not None and price_cents < price_min: + return False + if price_max is not None and price_cents > price_max: + return False + + # Postcode filter + allowed_postcodes = kw_row.get("allowed_postcodes") + if allowed_postcodes is not None: + ad_postcode = fields.get("postcode") + if ad_postcode is None or ad_postcode not in allowed_postcodes: + return False + + return True + + async def process_notification_queue(pool: asyncpg.Pool, bot: ExtBot) -> int: """Process pending notifications from the retry queue.""" rows = await pool.fetch(""" @@ -89,6 +113,76 @@ async def process_notification_queue(pool: asyncpg.Pool, bot: ExtBot) -> int: return processed +async def flush_digests(pool: asyncpg.Pool, bot: ExtBot) -> int: + """Flush digest buffers for users whose interval has elapsed.""" + users = await pool.fetch(""" + SELECT us.telegram_id, us.digest_interval, us.last_digest_flush + FROM user_settings us + WHERE us.digest_mode = true + AND (us.last_digest_flush IS NULL + OR us.last_digest_flush < now() - (us.digest_interval || ' minutes')::interval) + """) + + if not users: + return 0 + + flushed = 0 + + for user_row in users: + tg_id = user_row["telegram_id"] + + # Get all buffered notifications for this user + buffered = await pool.fetch(""" + SELECT db.id, db.ad_id, db.keyword, db.title, db.price, db.url + FROM digest_buffer db + WHERE db.telegram_id = $1 + ORDER BY db.created_at DESC + """, tg_id) + + if not buffered: + # Update flush time even if no items (to keep tracking) + await pool.execute( + """INSERT INTO user_settings (telegram_id, last_digest_flush) VALUES ($1, now()) + ON CONFLICT (telegram_id) DO UPDATE SET last_digest_flush = now()""", + tg_id) + continue + + # Build digest message grouped by keyword + by_keyword = defaultdict(list) + for item in buffered: + by_keyword[item["keyword"]].append(item) + + lines = ["šŸ“¦ Digest Summary"] + + for keyword, items in by_keyword.items(): + lines.append(f"\nšŸ” {keyword} ({len(items)} ad{'s' if len(items) != 1 else ''})") + for item in items[:20]: # Limit to 20 ads per keyword to avoid message too long + price_str = f"€{item['price'] / 100:.0f}" if item["price"] else "N/A" + lines.append(f" • {item['title']} — {price_str}") + + text = "\n".join(lines) + + try: + await bot.send_message( + chat_id=int(tg_id), + text=text, + parse_mode="HTML", + ) + flushed += 1 + logger.info("Sent digest to %s (%d items)", tg_id, len(buffered)) + except Exception as e: + logger.error("Failed to send digest to %s: %s", tg_id, e) + + # Clear buffered items and update flush time + await pool.execute("DELETE FROM digest_buffer WHERE telegram_id = $1", tg_id) + await pool.execute( + """INSERT INTO user_settings (telegram_id, last_digest_flush) VALUES ($1, now()) + ON CONFLICT (telegram_id) DO UPDATE SET last_digest_flush = now()""", + tg_id) + + return flushed + + async def scheduler_task(pool: object, bot: ExtBot) -> None: while True: record_scheduler_run() # mark this cycle as started @@ -99,9 +193,17 @@ async def scheduler_task(pool: object, bot: ExtBot) -> None: except Exception: logger.exception("Error processing notification queue") + try: + digested = await flush_digests(pool, bot) + if digested: + logger.info("Flushed %d digest summaries", digested) + except Exception: + logger.exception("Error flushing digests") + try: rows = await pool.fetch( - "SELECT id, keyword, interval_minutes, initial_loaded, ads_cursor FROM keywords " + "SELECT id, keyword, interval_minutes, initial_loaded, ads_cursor, " + "price_min, price_max, allowed_postcodes FROM keywords " "WHERE is_active = true " "AND (last_scraped_at IS NULL OR last_scraped_at < now() - (interval_minutes || ' minutes')::interval)" ) @@ -135,6 +237,11 @@ async def scheduler_task(pool: object, bot: ExtBot) -> None: for ad_data in ads_raw: fields = extract_ad_fields(ad_data) + + # Skip ads that don't pass price/postcode filters + if not _ad_passes_filters(fields, dict(row)): + continue + wh_ad_id = fields["wh_ad_id"] is_price_drop = False old_price = None @@ -163,6 +270,13 @@ async def scheduler_task(pool: object, bot: ExtBot) -> None: if initial_loaded: notify_fields = {**fields, "keyword": keyword} for tg_id in telegram_ids: + # Check mute hours first + muted = await is_user_muted(pool, tg_id) + if muted: + # Check digest mode — buffer instead of discard + await buffer_for_digest(pool, tg_id, notify_fields, ad_uuid) + continue + msg_id_val = await notify_new_ad(bot, tg_id, notify_fields, ad_uuid=ad_uuid) if msg_id_val: user_row = await pool.fetchrow("SELECT id FROM users WHERE telegram_id = $1", tg_id) @@ -197,6 +311,12 @@ async def scheduler_task(pool: object, bot: ExtBot) -> None: if is_price_drop: notify_fields = {**fields, "keyword": keyword} for tg_id in telegram_ids: + # Check mute hours first + muted = await is_user_muted(pool, tg_id) + if muted: + await buffer_for_digest(pool, tg_id, notify_fields, ad_uuid) + continue + msg_id_val = await notify_price_drop(bot, tg_id, notify_fields, ad_uuid=ad_uuid) if msg_id_val: user_row = await pool.fetchrow("SELECT id FROM users WHERE telegram_id = $1", tg_id) diff --git a/worker/src/migrations/05-keyword-filters.sql b/worker/src/migrations/05-keyword-filters.sql new file mode 100644 index 0000000..20f5f25 --- /dev/null +++ b/worker/src/migrations/05-keyword-filters.sql @@ -0,0 +1,9 @@ +-- Phase 2: Price and postcode filters per keyword +ALTER TABLE keywords + ADD COLUMN IF NOT EXISTS price_min int, + ADD COLUMN IF NOT EXISTS price_max int, + ADD COLUMN IF NOT EXISTS allowed_postcodes text[]; + +COMMENT ON COLUMN keywords.price_min IS 'Minimum price in cents (e.g. 5000 = €50). NULL = no limit.'; +COMMENT ON COLUMN keywords.price_max IS 'Maximum price in cents (e.g. 500000 = €5000). NULL = no limit.'; +COMMENT ON COLUMN keywords.allowed_postcodes IS 'Austrian postcodes (4-digit strings). Only ads matching these are notified.'; \ No newline at end of file diff --git a/worker/src/migrations/06-user-settings.sql b/worker/src/migrations/06-user-settings.sql new file mode 100644 index 0000000..cdea942 --- /dev/null +++ b/worker/src/migrations/06-user-settings.sql @@ -0,0 +1,31 @@ +-- Phase 2: User settings (mute hours + digest mode) and digest buffer +CREATE TABLE IF NOT EXISTS user_settings ( + telegram_id text PRIMARY KEY, + mute_start time, + mute_end time, + digest_mode bool NOT NULL DEFAULT false, + digest_interval int NOT NULL DEFAULT 60, + last_digest_flush timestamptz, + CONSTRAINT chk_mute_hours CHECK ( + mute_start IS NULL AND mute_end IS NULL + OR mute_start IS NOT NULL AND mute_end IS NOT NULL + ) +); + +COMMENT ON TABLE user_settings IS 'User-specific settings for notification behavior'; + +CREATE TABLE IF NOT EXISTS digest_buffer ( + id uuid PRIMARY KEY DEFAULT gen_random_uuid(), + telegram_id text NOT NULL, + ad_id uuid REFERENCES ads(id) ON DELETE CASCADE, + keyword text NOT NULL, + title text NOT NULL, + price int, -- in cents + url text, + created_at timestamptz NOT NULL DEFAULT now() +); + +CREATE INDEX IF NOT EXISTS idx_digest_buffer + ON digest_buffer(telegram_id, created_at DESC); + +COMMENT ON TABLE digest_buffer IS 'Buffer for digest-mode notifications, flushed at intervals'; \ No newline at end of file diff --git a/worker/src/notifier.py b/worker/src/notifier.py index 2658c22..b3e50ab 100644 --- a/worker/src/notifier.py +++ b/worker/src/notifier.py @@ -1,5 +1,5 @@ import logging -from datetime import datetime +from datetime import datetime, timezone from typing import Any import asyncpg @@ -10,6 +10,58 @@ from telegram.ext import ExtBot logger = logging.getLogger(__name__) +async def is_user_muted(pool: asyncpg.Pool, telegram_id: int) -> bool: + """Check if the current time is within the user's mute window.""" + settings = await pool.fetchrow( + "SELECT mute_start, mute_end FROM user_settings WHERE telegram_id = $1", + str(telegram_id), + ) + + if not settings or not settings["mute_start"] or not settings["mute_end"]: + return False + + now_utc = datetime.now(tz=timezone.utc).time() + start = settings["mute_start"] + end = settings["mute_end"] + + if start < end: + # Normal window (e.g., 08:00–12:00) + return start <= now_utc <= end + else: + # Window crosses midnight (e.g., 22:00 to 07:00 next day) + return now_utc >= start or now_utc <= end + + +async def buffer_for_digest( + pool: asyncpg.Pool, + telegram_id: int, + ad: dict[str, Any], + ad_uuid: str | None = None, +) -> None: + """Buffer a notification for digest-mode users, or discard if not in digest mode.""" + settings = await pool.fetchrow( + "SELECT digest_mode FROM user_settings WHERE telegram_id = $1", + str(telegram_id), + ) + + if not settings or not settings["digest_mode"]: + # Not in digest mode — during mute hours, just discard + return + + # In digest mode — buffer the notification + price_cents = int(round(ad["price"] * 100)) if ad.get("price") else None + try: + await pool.execute( + """INSERT INTO digest_buffer (telegram_id, ad_id, keyword, title, price, url) + VALUES ($1, $2, $3, $4, $5, $6)""", + str(telegram_id), ad_uuid, ad.get("keyword", ""), ad.get("title", ""), + price_cents, ad.get("url"), + ) + logger.info("Buffered digest for %s: %s", telegram_id, ad.get("title", "")) + except Exception: + logger.exception("Failed to buffer digest for %s", telegram_id) + + def _build_keyboard(ad: dict[str, Any]) -> InlineKeyboardMarkup | None: keyboard: list[list[InlineKeyboardButton]] = [] if ad.get("url"):