diff --git a/worker/src/bot.py b/worker/src/bot.py
index fa94116..1a7d536 100644
--- a/worker/src/bot.py
+++ b/worker/src/bot.py
@@ -188,11 +188,23 @@ def _admin_menu_keyboard() -> InlineKeyboardMarkup:
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 ""
-
+
+ filters = ""
+ price_min = kw.get("price_min")
+ price_max = kw.get("price_max")
+ if price_min is not None or price_max is not None:
+ min_str = f"{(price_min or 0) / 100:.0f}" if price_min is not None else "0"
+ max_str = f"{price_max / 100:.0f}" if price_max is not None else "ā"
+ filters = f"\nš° Price: ā¬{min_str}āā¬{max_str}"
+
+ postcodes = kw.get("allowed_postcodes")
+ if postcodes:
+ filters += f"\nš Postcodes: {','.join(str(p) for p in 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'))}{filters}{subs_line}"
)
@@ -201,7 +213,7 @@ async def _refresh_kw_card(bot, chat_id: int, message_id: int, kw_id_full: str):
pool = await get_pool()
# Accept either full UUID or truncated prefix ā both work as LIKE match
kw = await pool.fetchrow(
- "SELECT kw.id, kw.keyword, kw.is_active, kw.interval_minutes, kw.last_scraped_at, COUNT(ks.user_id) AS subs "
+ "SELECT kw.id, kw.keyword, kw.is_active, kw.interval_minutes, kw.last_scraped_at, kw.price_min, kw.price_max, kw.allowed_postcodes, COUNT(ks.user_id) AS subs "
"FROM keywords kw LEFT JOIN keyword_subscriptions ks ON ks.keyword_id = kw.id "
"WHERE kw.id::text LIKE $1 || '%' GROUP BY kw.id",
kw_id_full + "%",
@@ -226,12 +238,399 @@ 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 for keyword"),
+ ("price_max", "Set max price for keyword"),
+ ("clear_price", "Clear price filters"),
+ ("postcode", "Set postcodes for keyword"),
+ ("clear_postcode", "Clear postcode filter"),
+ ("mute_hours", "Set mute hours"),
+ ("mute_off", "Disable mute hours"),
+ ("digest_on", "Enable digest mode"),
+ ("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))
+ 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))
+ app.add_handler(CommandHandler("postcode", postcode_handler))
+ app.add_handler(CommandHandler("clear_postcode", clear_postcode_handler))
+ app.add_handler(CommandHandler("mute_hours", mute_hours_handler))
+ app.add_handler(CommandHandler("mute_off", mute_off_handler))
+ app.add_handler(CommandHandler("digest_on", digest_on_handler))
+ app.add_handler(CommandHandler("digest_off", digest_off_handler))
+ 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))
+
+
+# āā /start āāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāā
+
+async def start_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+ row = await _auto_register(update)
+ if not row:
+ return
+
+ name = update.effective_user.first_name or "there"
+ chat_id = update.effective_chat.id # type: ignore[union-attr]
+ await context.bot.send_message(
+ chat_id=chat_id,
+ text=f"Hello {name}! I'll notify you about new willhaben listings.",
+ parse_mode="HTML",
+ reply_markup=_main_menu_keyboard(),
+ )
+
+
+# āā /admin āāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāā
+
+async def admin_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+ row = await _require_admin(update)
+ if not row:
+ return
+
+ chat_id = update.effective_chat.id # type: ignore[union-attr]
+ await context.bot.send_message(
+ chat_id=chat_id, text="āļø Admin Panel",
+ parse_mode="HTML", reply_markup=_admin_menu_keyboard(),
+ )
+
+
+# āā price filter handlers āāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāā
+
+async def price_min_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+ """Set minimum price for a keyword. Usage: /price_min keyword 50"""
+ await _set_price_filter(update, context, "price_min")
+
+
+async def price_max_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+ """Set maximum price for a keyword. Usage: /price_max keyword 1000"""
+ await _set_price_filter(update, context, "price_max")
+
+
+async def _set_price_filter(update: Update, context: ContextTypes.DEFAULT_TYPE, column: str) -> None:
+ user = await _require_user(update)
+ if not user or not update.message:
+ return
+
+ text = (update.message.text or "").strip() # type: ignore[union-attr]
+ parts = text.split()
+ if len(parts) < 3:
+ await update.message.reply_text( # type: ignore[union-attr]
+ f"Usage: /{update.message.text.split()[0][1:]} keyword 50 (price in ā¬)",
+ parse_mode="HTML")
+ return
+
+ keyword = " ".join(parts[1:-1])
+ try:
+ price_euros = float(parts[-1])
+ except ValueError:
+ await update.message.reply_text("Invalid price. Enter a number in ā¬.") # type: ignore[union-attr]
+ return
+
+ price_cents = int(round(price_euros * 100))
+ pool = await get_pool()
+
+ # Find keyword ā non-admins can only modify their own keywords
+ if user.get("is_admin"):
+ kw = await pool.fetchrow("SELECT id FROM keywords WHERE LOWER(keyword) = LOWER($1)", keyword)
+ else:
+ kw = await pool.fetchrow(
+ "SELECT kw.id 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, user["id"])
+
+ if not kw:
+ await update.message.reply_text(f"Keyword {keyword} not found.") # type: ignore[union-attr]
+ return
+
+ await pool.execute(f"UPDATE keywords SET {column} = $1 WHERE id = $2", price_cents, kw["id"])
+ label = "Min" if column == "price_min" else "Max"
+ await update.message.reply_text( # type: ignore[union-attr]
+ f"ā
{label} price for {keyword} set to ā¬{price_euros:.0f}.",
+ parse_mode="HTML")
+
+
+async def clear_price_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+ """Clear price filters for a keyword. Usage: /clear_price keyword"""
+ user = await _require_user(update)
+ if not user or not update.message:
+ return
+
+ text = (update.message.text or "").strip() # type: ignore[union-attr]
+ parts = text.split()
+ if len(parts) < 2:
+ await update.message.reply_text("Usage: /clear_price keyword", parse_mode="HTML")
+ return
+
+ keyword = " ".join(parts[1:])
+ pool = await get_pool()
+
+ if user.get("is_admin"):
+ kw = await pool.fetchrow("SELECT id FROM keywords WHERE LOWER(keyword) = LOWER($1)", keyword)
+ else:
+ kw = await pool.fetchrow(
+ "SELECT kw.id 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, user["id"])
+
+ if not kw:
+ await update.message.reply_text(f"Keyword {keyword} 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"ā
Price filters cleared for {keyword}.", parse_mode="HTML")
+
+
+# āā postcode filter handlers āāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāā
+
+async def postcode_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+ """Set allowed postcodes for a keyword. Usage: /postcode keyword 1010,1020,1030"""
+ user = await _require_user(update)
+ if not user or not update.message:
+ return
+
+ text = (update.message.text or "").strip() # type: ignore[union-attr]
+ parts = text.split()
+ if len(parts) < 3:
+ await update.message.reply_text("Usage: /postcode keyword 1010,1020", parse_mode="HTML")
+ return
+
+ keyword = " ".join(parts[1:-1])
+ pc_str = parts[-1]
+ postcodes = [p.strip() for p in pc_str.split(",")]
+
+ # Validate postcodes
+ for p in postcodes:
+ if not p.isdigit() or len(p) < 3:
+ await update.message.reply_text(f"Invalid postcode: {p}. Use numeric postcodes.") # type: ignore[union-attr]
+ return
+
+ pool = await get_pool()
+
+ if user.get("is_admin"):
+ kw = await pool.fetchrow("SELECT id FROM keywords WHERE LOWER(keyword) = LOWER($1)", keyword)
+ else:
+ kw = await pool.fetchrow(
+ "SELECT kw.id 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, user["id"])
+
+ if not kw:
+ await update.message.reply_text(f"Keyword {keyword} 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"ā
Postcodes for {keyword} set to: {','.join(postcodes)}.",
+ parse_mode="HTML")
+
+
+async def clear_postcode_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+ """Clear postcode filter for a keyword. Usage: /clear_postcode keyword"""
+ user = await _require_user(update)
+ if not user or not update.message:
+ return
+
+ text = (update.message.text or "").strip() # type: ignore[union-attr]
+ parts = text.split()
+ if len(parts) < 2:
+ await update.message.reply_text("Usage: /clear_postcode keyword", parse_mode="HTML")
+ return
+
+ keyword = " ".join(parts[1:])
+ pool = await get_pool()
+
+ if user.get("is_admin"):
+ kw = await pool.fetchrow("SELECT id FROM keywords WHERE LOWER(keyword) = LOWER($1)", keyword)
+ else:
+ kw = await pool.fetchrow(
+ "SELECT kw.id 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, user["id"])
+
+ if not kw:
+ await update.message.reply_text(f"Keyword {keyword} 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"ā
Postcode filter cleared for {keyword}.", parse_mode="HTML")
+
+
+# āā mute handlers āāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāā
+
+async def mute_hours_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+ """Set mute hours. Usage: /mute_hours 22:00-07:00"""
+ user = await _require_user(update)
+ if not user or not update.message:
+ return
+
+ text = (update.message.text or "").strip() # type: ignore[union-attr]
+ parts = text.split()
+ if len(parts) < 2:
+ await update.message.reply_text("Usage: /mute_hours 22:00-07:00 (times in UTC)", parse_mode="HTML")
+ return
+
+ time_range = parts[1]
+ try:
+ start_str, end_str = time_range.split("-")
+ # Validate time format
+ _ = datetime.strptime(start_str.strip(), "%H:%M")
+ _ = datetime.strptime(end_str.strip(), "%H:%M")
+ except (ValueError, AttributeError):
+ await update.message.reply_text("Invalid format. Use HH:MM-HH:MM (e.g., 22:00-07:00).") # type: ignore[union-attr]
+ 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.strip(), end_str.strip())
+
+ await update.message.reply_text( # type: ignore[union-attr]
+ f"ā
Mute hours set to {start_str.strip()}ā{end_str.strip()} UTC.\nNotifications during this window will be suppressed.",
+ parse_mode="HTML")
+
+
+async def mute_off_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+ """Disable mute hours."""
+ user = await _require_user(update)
+ if not user:
+ return
+
+ pool = await get_pool()
+ await pool.execute(
+ """INSERT INTO user_settings (telegram_id) VALUES ($1)
+ 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")
+
+
+# āā digest handlers āāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāā
+
+async def digest_on_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+ """Enable digest mode. Usage: /digest_on [minutes]"""
+ user = await _require_user(update)
+ if not user:
+ return
+
+ interval = 60 # default
+ text = (update.message.text or "").strip() # type: ignore[union-attr]
+ parts = text.split()
+ if len(parts) > 1:
+ try:
+ interval = int(parts[1])
+ if interval < 5 or interval > 1440:
+ raise ValueError
+ except ValueError:
+ msg = update.message or update.callback_query # type: ignore[union-attr]
+ await msg.reply_text("Interval must be between 5 and 1440 minutes.") # type: ignore[union-attr]
+ 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)
+
+ msg = update.message or update.callback_query # type: ignore[union-attr]
+ await msg.reply_text( # type: ignore[union-attr]
+ f"ā
Digest mode enabled. Notifications will be bundled every {interval} min.",
+ parse_mode="HTML")
+
+
+async def digest_off_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+ """Disable digest mode and flush remaining buffer."""
+ user = await _require_user(update)
+ if not user:
+ return
+
+ pool = await get_pool()
+ 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"]))
+
+ # Flush remaining digest buffer ā send remaining items as individual notifications
+ # This is handled by the scheduler's flush_digests function, but we also clear the buffer here
+ await pool.execute("DELETE FROM digest_buffer WHERE telegram_id = $1", str(user["telegram_id"]))
+
+ msg = update.message or update.callback_query # type: ignore[union-attr]
+ await msg.reply_text("ā
Digest mode disabled. Notifications will be sent immediately again.") # type: ignore[union-attr]
+
+
+# āā status handler āāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāā
+
+async def status_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+ """Show current user settings (mute + digest)."""
+ 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"] is not None):
+ 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 ({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") # type: ignore[union-attr]
+
+
+# āā handler registration + global commands āāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāā
+
+async def setup_global_commands(app: Application) -> None:
+ """Call this from main() after building the app to register bot commands."""
+ await app.bot.set_my_commands([
+ ("start", "Open main menu"),
+ ("admin", "Admin panel (admins only)"),
+ ("price_min", "Set min price for keyword"),
+ ("price_max", "Set max price for keyword"),
+ ("clear_price", "Clear price filters"),
+ ("postcode", "Set postcodes for keyword"),
+ ("clear_postcode", "Clear postcode filter"),
+ ("mute_hours", "Set mute hours"),
+ ("mute_off", "Disable mute hours"),
+ ("digest_on", "Enable digest mode"),
+ ("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))
+ 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))
+ app.add_handler(CommandHandler("postcode", postcode_handler))
+ app.add_handler(CommandHandler("clear_postcode", clear_postcode_handler))
+ app.add_handler(CommandHandler("mute_hours", mute_hours_handler))
+ app.add_handler(CommandHandler("mute_off", mute_off_handler))
+ app.add_handler(CommandHandler("digest_on", digest_on_handler))
+ app.add_handler(CommandHandler("digest_off", digest_off_handler))
+ 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))
@@ -485,7 +884,8 @@ async def _handle_callback(update: Update, context: ContextTypes.DEFAULT_TYPE) -
elif sub == "list":
subs = await pool.fetch(
"SELECT kw.id, kw.keyword, kw.is_active, kw.interval_minutes, "
- "kw.last_scraped_at, COUNT(ks2.user_id) AS subs "
+ "kw.last_scraped_at, kw.price_min, kw.price_max, kw.allowed_postcodes, "
+ "COUNT(ks2.user_id) AS subs "
"FROM keyword_subscriptions ks "
"JOIN keywords kw ON kw.id = ks.keyword_id "
"LEFT JOIN keyword_subscriptions ks2 ON ks2.keyword_id = ks.keyword_id "
diff --git a/worker/src/main.py b/worker/src/main.py
index f7789e2..cacafc3 100644
--- a/worker/src/main.py
+++ b/worker/src/main.py
@@ -4,6 +4,7 @@ import logging
import os
import signal
import sys
+from collections import defaultdict
from contextlib import suppress
from datetime import datetime, timedelta, timezone
from typing import Any
@@ -24,6 +25,30 @@ 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 +114,83 @@ 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 u
+ digested = await flush_digests(pool, bot)
+ if digested:
+ logger.info("Flushed %d digest summaries", digested)
+ except Exception:
+ logger.exception("Error flushing digests")
+
+ try:sers:
+ 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
@@ -101,7 +203,7 @@ async def scheduler_task(pool: object, bot: ExtBot) -> None:
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,9 @@ 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 match keyword filters
+ if not _ad_passes_filters(fields, row):
+ continue
wh_ad_id = fields["wh_ad_id"]
is_price_drop = False
old_price = None
@@ -163,7 +268,7 @@ async def scheduler_task(pool: object, bot: ExtBot) -> None:
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, ad_uuid=ad_uuid)
+ msg_id_val = await notify_new_ad(bot, tg_id, notify_fields, ad_uuid=ad_uuid, pool=pool)
if msg_id_val:
user_row = await pool.fetchrow("SELECT id FROM users WHERE telegram_id = $1", tg_id)
if user_row:
@@ -197,7 +302,7 @@ async def scheduler_task(pool: object, bot: ExtBot) -> None:
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, ad_uuid=ad_uuid)
+ msg_id_val = await notify_price_drop(bot, tg_id, notify_fields, ad_uuid=ad_uuid, pool=pool)
if msg_id_val:
user_row = await pool.fetchrow("SELECT id FROM users WHERE telegram_id = $1", tg_id)
if user_row:
diff --git a/worker/src/migrations/05-keyword-filters.sql b/worker/src/migrations/05-keyword-filters.sql
new file mode 100644
index 0000000..e22a3d6
--- /dev/null
+++ b/worker/src/migrations/05-keyword-filters.sql
@@ -0,0 +1,9 @@
+-- Price and postcode filters for keywords
+
+ALTER TABLE keywords ADD COLUMN IF NOT EXISTS price_min int;
+ALTER TABLE keywords ADD COLUMN IF NOT EXISTS price_max int;
+ALTER TABLE keywords ADD COLUMN IF NOT EXISTS allowed_postcodes text[];
+
+COMMENT ON COLUMN keywords.price_min IS 'Minimum price filter in cents (e.g., 5000 = ā¬50.00)';
+COMMENT ON COLUMN keywords.price_max IS 'Maximum price filter in cents (e.g., 5000 = ā¬50.00)';
+COMMENT ON COLUMN keywords.allowed_postcodes IS 'Array of allowed postcode strings (e.g., ARRAY[''1010'', ''1020''])';
\ 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..8c6c381
--- /dev/null
+++ b/worker/src/migrations/06-user-settings.sql
@@ -0,0 +1,49 @@
+-- User settings and digest buffer tables
+
+CREATE TABLE IF NOT EXISTS user_settings (
+ telegram_id text PRIMARY KEY,
+ mute_start time,
+ mute_end time,
+ digest_mode boolean 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
+ 'Per-user preferences for mute hours and digest notifications';
+
+COMMENT ON COLUMN user_settings.mute_start IS
+ 'UTC time at which notifications are silenced each day';
+
+COMMENT ON COLUMN user_settings.mute_end IS
+ 'UTC time at which notifications are re-enabled each day';
+
+COMMENT ON COLUMN user_settings.digest_mode IS
+ 'When true, notifications are batched into periodic digest messages';
+
+COMMENT ON COLUMN user_settings.digest_interval IS
+ 'Interval in minutes between digest flushes (default 60)';
+
+COMMENT ON COLUMN user_settings.last_digest_flush IS
+ 'Timestamp of the last digest sent to this user';
+
+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,
+ 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
+ 'Temporary holding area for notifications pending digest flush';
\ No newline at end of file
diff --git a/worker/src/notifier.py b/worker/src/notifier.py
index 2658c22..074c373 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, time, timezone
from typing import Any
import asyncpg
@@ -10,6 +10,46 @@ from telegram.ext import ExtBot
logger = logging.getLogger(__name__)
+async def is_user_muted(pool: asyncpg.Pool, telegram_id: str) -> bool:
+ """Check if a user is currently in their mute window (UTC)."""
+ settings = await pool.fetchrow(
+ "SELECT mute_start, mute_end FROM user_settings WHERE telegram_id = $1",
+ telegram_id)
+
+ if not settings or not settings["mute_start"]:
+ return False
+
+ now_utc = datetime.now(tz=timezone.utc).time()
+ start_t = settings["mute_start"]
+ end_t = settings["mute_end"]
+
+ # Handle cross-midnight windows (e.g., 22:00-07:00)
+ if start_t > end_t:
+ return now_utc >= start_t or now_utc < end_t
+ else:
+ return start_t <= now_utc < end_t
+
+
+async def get_user_settings(pool: asyncpg.Pool, telegram_id: str) -> dict | None:
+ """Get user settings row, or None if not set."""
+ return await pool.fetchrow(
+ "SELECT mute_start, mute_end, digest_mode, digest_interval FROM user_settings WHERE telegram_id = $1",
+ telegram_id)
+
+
+async def buffer_for_digest(pool: asyncpg.Pool, telegram_id: str, ad: dict, ad_uuid: str) -> None:
+ """Buffer a notification for digest mode instead of sending immediately."""
+ await pool.execute(
+ """INSERT INTO digest_buffer (telegram_id, ad_id, keyword, title, price, url)
+ VALUES ($1, $2, $3, $4, $5, $6)""",
+ telegram_id, ad_uuid,
+ ad.get("keyword", ""),
+ ad.get("title", "Untitled"),
+ int(round(ad["price"] * 100)) if ad.get("price") else None,
+ ad.get("url"))
+ logger.info("Buffered notification for digest: user=%s ad=%s", telegram_id, ad.get("wh_ad_id", ""))
+
+
def _build_keyboard(ad: dict[str, Any]) -> InlineKeyboardMarkup | None:
keyboard: list[list[InlineKeyboardButton]] = []
if ad.get("url"):
@@ -76,7 +116,21 @@ async def notify_new_ad(
telegram_id: int,
ad: dict[str, Any],
ad_uuid: str | None = None,
+ pool: asyncpg.Pool | None = None,
) -> int | None:
+ # Check mute hours and digest mode
+ if pool:
+ tg_id_str = str(telegram_id)
+ muted = await is_user_muted(pool, tg_id_str)
+ if muted:
+ return None # Muted ā discard (mute takes precedence over digest)
+
+ # Not muted, check digest mode
+ settings = await get_user_settings(pool, tg_id_str)
+ if settings and settings["digest_mode"] and ad_uuid:
+ await buffer_for_digest(pool, tg_id_str, ad, ad_uuid)
+ return None # Buffered ā will be sent in next digest flush
+
text = _format_text("š New listing found!", ad)
reply_markup = _build_keyboard(ad)
@@ -127,7 +181,21 @@ async def notify_price_drop(
telegram_id: int,
ad: dict[str, Any],
ad_uuid: str | None = None,
+ pool: asyncpg.Pool | None = None,
) -> int | None:
+ # Check mute hours and digest mode
+ if pool:
+ tg_id_str = str(telegram_id)
+ muted = await is_user_muted(pool, tg_id_str)
+ if muted:
+ return None # Muted ā discard (mute takes precedence over digest)
+
+ # Not muted, check digest mode
+ settings = await get_user_settings(pool, tg_id_str)
+ if settings and settings["digest_mode"] and ad_uuid:
+ await buffer_for_digest(pool, tg_id_str, ad, ad_uuid)
+ return None # Buffered ā will be sent in next digest flush
+
text = _format_text("ā ļø Price drop!", ad)
reply_markup = _build_keyboard(ad)