feat: Phase 2 — user experience & advanced filtering

- Add price_min/price_max filters per keyword (stored in cents) - Add allowed_postcodes filter per keyword (text[] array) - Add mute hours per user with cross-midnight support - Add digest mode with configurable interval and buffered summaries - 10 new bot commands: /price_min, /price_max, /clear_price, /postcode, /clear_postcode, /mute_hours, /mute_off, /digest_on, /digest_off, /status - Updated keyword card display to show price range and postcodes - New migrations: 05-keyword-filters.sql, 06-user-settings.sql
2026-07-10 21:19:34 +02:00
parent 3e7e5d0b32
commit 5d792e8ae7
5 changed files with 574 additions and 4 deletions
@@ -189,10 +189,23 @@ def _format_kw_card(kw: dict) -> str:
    status_icon = "🟢 Active" if kw["is_active"] else "🔴 Stopped"
    subs_line = f"\nSubscribers: <code>{kw['subs']}</code>" 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: <code>{'–'.join(parts)}</code>"
+    
+    postcode_line = ""
+    if kw.get("allowed_postcodes"):
+        postcode_line = f"\nPostcodes: <code>{', '.join(kw['allowed_postcodes'])}</code>"
+
    return (
        f"<b>🔍 {kw['keyword']}</b>\n"
        f"{status_icon} | Interval: <code>{kw['interval_minutes']} min</code>\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 <kw> <€>"),
+        ("price_max", "Set max price: /price_max <kw> <€>"),
+        ("clear_price", "Remove price filter: /clear_price <kw>"),
+        ("postcode", "Set postcodes: /postcode <kw> p1,p2"),
+        ("clear_postcode", "Remove postcode filter: /clear_postcode <kw>"),
+        ("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 <keyword> <amount in €>")  # 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"✅ <b>{kw['keyword']}</b>: 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 <keyword> <amount in €>")  # 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"✅ <b>{kw['keyword']}</b>: 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 <keyword>")  # 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"✅ <b>{kw['keyword']}</b>: 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 <keyword> 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"✅ <b>{kw['keyword']}</b>: 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 <keyword>")  # 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"✅ <b>{kw['keyword']}</b>: 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 = ["📦 <b>Pending Digest Summary</b>"]
+        for keyword, items in by_keyword.items():
+            lines.append(f"\n<b>🔍 {keyword}</b> ({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 = ["<b>⚙️ Your Settings</b>"]
+    
+    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:
@@ -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 = ["📦 <b>Digest Summary</b>"]
+        
+        for keyword, items in by_keyword.items():
+            lines.append(f"\n<b>🔍 {keyword}</b> ({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)
@@ -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.';
@@ -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';
@@ -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"):