""" Notification dispatcher — sends pending notification events via ntfy. RSS is pull-based so no dispatch is needed for it; events are simply marked dispatched once ntfy is sent (or immediately if the user has no ntfy configured but has an RSS token, so the feed can clean up old items). Runs every 5 minutes on Celery Beat. """ import base64 import logging from collections import defaultdict from datetime import datetime, timedelta, timezone import requests from app.database import get_sync_db from app.models.follow import Follow from app.models.notification import NotificationEvent from app.models.user import User from app.workers.celery_app import celery_app logger = logging.getLogger(__name__) NTFY_TIMEOUT = 10 _EVENT_TITLES = { "new_document": "New Bill Text", "new_amendment": "Amendment Filed", "bill_updated": "Bill Updated", "weekly_digest": "Weekly Digest", } _EVENT_TAGS = { "new_document": "page_facing_up", "new_amendment": "memo", "bill_updated": "rotating_light", } # Milestone events are more urgent than LLM brief events _EVENT_PRIORITY = { "bill_updated": "high", "new_document": "default", "new_amendment": "default", } def _in_quiet_hours(prefs: dict, now: datetime) -> bool: """Return True if the current local time falls within the user's quiet window. Quiet hours are stored as local-time hour integers. If the user has a stored IANA timezone name we convert ``now`` (UTC) to that zone before comparing. Falls back to UTC if the timezone is absent or unrecognised. """ start = prefs.get("quiet_hours_start") end = prefs.get("quiet_hours_end") if start is None or end is None: return False tz_name = prefs.get("timezone") if tz_name: try: from zoneinfo import ZoneInfo h = now.astimezone(ZoneInfo(tz_name)).hour except Exception: h = now.hour # unrecognised timezone — degrade gracefully to UTC else: h = now.hour if start <= end: return start <= h < end # Wraps midnight (e.g. 22 → 8) return h >= start or h < end @celery_app.task(bind=True, name="app.workers.notification_dispatcher.dispatch_notifications") def dispatch_notifications(self): """Fan out pending notification events to ntfy and mark dispatched.""" db = get_sync_db() try: pending = ( db.query(NotificationEvent) .filter(NotificationEvent.dispatched_at.is_(None)) .order_by(NotificationEvent.created_at) .limit(200) .all() ) sent = 0 failed = 0 held = 0 now = datetime.now(timezone.utc) for event in pending: user = db.get(User, event.user_id) if not user: event.dispatched_at = now db.commit() continue # Look up follow mode for this (user, bill) pair follow = db.query(Follow).filter_by( user_id=event.user_id, follow_type="bill", follow_value=event.bill_id ).first() follow_mode = follow.follow_mode if follow else "neutral" # Pocket Veto: only milestone (bill_updated) events; skip LLM brief events if follow_mode == "pocket_veto" and event.event_type in ("new_document", "new_amendment"): event.dispatched_at = now db.commit() continue # Referral-tier events (committee referrals) are noisy for loose topic follows; # suppress them only for topic-follow events so direct bill/member followers # still get notified when their bill is referred to committee. payload = event.payload or {} is_topic_follow = payload.get("source") == "topic_follow" if follow_mode == "neutral" and payload.get("milestone_tier") == "referral" and is_topic_follow: event.dispatched_at = now db.commit() continue prefs = user.notification_prefs or {} ntfy_url = prefs.get("ntfy_topic_url", "").strip() ntfy_auth_method = prefs.get("ntfy_auth_method", "none") ntfy_token = prefs.get("ntfy_token", "").strip() ntfy_username = prefs.get("ntfy_username", "").strip() ntfy_password = prefs.get("ntfy_password", "").strip() ntfy_enabled = prefs.get("ntfy_enabled", False) rss_enabled = prefs.get("rss_enabled", False) digest_enabled = prefs.get("digest_enabled", False) ntfy_configured = ntfy_enabled and bool(ntfy_url) # Hold events when ntfy is configured but delivery should be deferred in_quiet = _in_quiet_hours(prefs, now) if ntfy_configured else False hold = ntfy_configured and (in_quiet or digest_enabled) if hold: held += 1 continue # Leave undispatched — digest task or next run after quiet hours if ntfy_configured: try: _send_ntfy( event, ntfy_url, ntfy_auth_method, ntfy_token, ntfy_username, ntfy_password, follow_mode=follow_mode, ) sent += 1 except Exception as e: logger.warning(f"ntfy dispatch failed for event {event.id}: {e}") failed += 1 # Mark dispatched: ntfy was attempted, or user has no ntfy (RSS-only or neither) event.dispatched_at = now db.commit() logger.info( f"dispatch_notifications: {sent} sent, {failed} failed, " f"{held} held (quiet hours/digest), {len(pending)} total pending" ) return {"sent": sent, "failed": failed, "held": held, "total": len(pending)} finally: db.close() @celery_app.task(bind=True, name="app.workers.notification_dispatcher.send_notification_digest") def send_notification_digest(self): """ Send a bundled ntfy digest for users with digest mode enabled. Runs daily; weekly-frequency users only receive on Mondays. """ db = get_sync_db() try: now = datetime.now(timezone.utc) users = db.query(User).all() digest_users = [ u for u in users if (u.notification_prefs or {}).get("digest_enabled", False) and (u.notification_prefs or {}).get("ntfy_enabled", False) and (u.notification_prefs or {}).get("ntfy_topic_url", "").strip() ] sent = 0 for user in digest_users: prefs = user.notification_prefs or {} frequency = prefs.get("digest_frequency", "daily") # Weekly digests only fire on Mondays (weekday 0) if frequency == "weekly" and now.weekday() != 0: continue lookback_hours = 168 if frequency == "weekly" else 24 cutoff = now - timedelta(hours=lookback_hours) events = ( db.query(NotificationEvent) .filter_by(user_id=user.id) .filter( NotificationEvent.dispatched_at.is_(None), NotificationEvent.created_at > cutoff, ) .order_by(NotificationEvent.created_at.desc()) .all() ) if not events: continue try: ntfy_url = prefs.get("ntfy_topic_url", "").strip() _send_digest_ntfy(events, ntfy_url, prefs) for event in events: event.dispatched_at = now db.commit() sent += 1 except Exception as e: logger.warning(f"Digest send failed for user {user.id}: {e}") logger.info(f"send_notification_digest: digests sent to {sent} users") return {"sent": sent} finally: db.close() def _send_ntfy( event: NotificationEvent, topic_url: str, auth_method: str = "none", token: str = "", username: str = "", password: str = "", follow_mode: str = "neutral", ) -> None: payload = event.payload or {} bill_label = payload.get("bill_label", event.bill_id.upper()) bill_title = payload.get("bill_title", "") event_label = _EVENT_TITLES.get(event.event_type, "Bill Update") title = f"{event_label}: {bill_label}" lines = [bill_title] if bill_title else [] if payload.get("brief_summary"): lines.append("") lines.append(payload["brief_summary"][:300]) message = "\n".join(lines) or bill_label headers = { "Title": title, "Priority": _EVENT_PRIORITY.get(event.event_type, "default"), "Tags": _EVENT_TAGS.get(event.event_type, "bell"), } if payload.get("bill_url"): headers["Click"] = payload["bill_url"] if follow_mode == "pocket_boost": headers["Actions"] = ( f"view, View Bill, {payload.get('bill_url', '')}; " "view, Find Your Rep, https://www.house.gov/representatives/find-your-representative" ) if auth_method == "token" and token: headers["Authorization"] = f"Bearer {token}" elif auth_method == "basic" and username: creds = base64.b64encode(f"{username}:{password}".encode()).decode() headers["Authorization"] = f"Basic {creds}" resp = requests.post(topic_url, data=message.encode("utf-8"), headers=headers, timeout=NTFY_TIMEOUT) resp.raise_for_status() @celery_app.task(bind=True, name="app.workers.notification_dispatcher.send_weekly_digest") def send_weekly_digest(self): """ Proactive week-in-review summary for followed bills. Runs every Monday at 8:30 AM UTC. Queries bills followed by each user for any activity in the past 7 days and dispatches a low-noise summary via ntfy and/or creates a NotificationEvent for the RSS feed. Unlike send_notification_digest (which bundles queued events), this task generates a fresh summary independent of the notification event queue. """ from app.config import settings as app_settings from app.models.bill import Bill db = get_sync_db() try: now = datetime.now(timezone.utc) cutoff = now - timedelta(days=7) base_url = (app_settings.PUBLIC_URL or app_settings.LOCAL_URL).rstrip("/") users = db.query(User).all() ntfy_sent = 0 rss_created = 0 for user in users: prefs = user.notification_prefs or {} ntfy_enabled = prefs.get("ntfy_enabled", False) ntfy_url = prefs.get("ntfy_topic_url", "").strip() rss_enabled = prefs.get("rss_enabled", False) ntfy_configured = ntfy_enabled and bool(ntfy_url) if not ntfy_configured and not rss_enabled: continue bill_follows = db.query(Follow).filter_by( user_id=user.id, follow_type="bill" ).all() if not bill_follows: continue bill_ids = [f.follow_value for f in bill_follows] active_bills = ( db.query(Bill) .filter( Bill.bill_id.in_(bill_ids), Bill.updated_at >= cutoff, ) .order_by(Bill.updated_at.desc()) .limit(20) .all() ) if not active_bills: continue count = len(active_bills) anchor = active_bills[0] summary_lines = [] for bill in active_bills[:10]: lbl = _format_bill_label(bill) action = (bill.latest_action_text or "")[:80] summary_lines.append(f"• {lbl}: {action}" if action else f"• {lbl}") if count > 10: summary_lines.append(f" …and {count - 10} more") summary = "\n".join(summary_lines) # Mark dispatched_at immediately so dispatch_notifications skips this event; # it still appears in the RSS feed since that endpoint reads all events. event = NotificationEvent( user_id=user.id, bill_id=anchor.bill_id, event_type="weekly_digest", dispatched_at=now, payload={ "bill_label": "Weekly Digest", "bill_title": f"{count} followed bill{'s' if count != 1 else ''} had activity this week", "brief_summary": summary, "bill_count": count, "bill_url": f"{base_url}/bills/{anchor.bill_id}", }, ) db.add(event) rss_created += 1 if ntfy_configured: try: _send_weekly_digest_ntfy(count, summary, ntfy_url, prefs) ntfy_sent += 1 except Exception as e: logger.warning(f"Weekly digest ntfy failed for user {user.id}: {e}") db.commit() logger.info(f"send_weekly_digest: {ntfy_sent} ntfy sent, {rss_created} events created") return {"ntfy_sent": ntfy_sent, "rss_created": rss_created} finally: db.close() def _format_bill_label(bill) -> str: _TYPE_MAP = { "hr": "H.R.", "s": "S.", "hjres": "H.J.Res.", "sjres": "S.J.Res.", "hconres": "H.Con.Res.", "sconres": "S.Con.Res.", "hres": "H.Res.", "sres": "S.Res.", } prefix = _TYPE_MAP.get(bill.bill_type.lower(), bill.bill_type.upper()) return f"{prefix} {bill.bill_number}" def _send_weekly_digest_ntfy(count: int, summary: str, ntfy_url: str, prefs: dict) -> None: auth_method = prefs.get("ntfy_auth_method", "none") ntfy_token = prefs.get("ntfy_token", "").strip() ntfy_username = prefs.get("ntfy_username", "").strip() ntfy_password = prefs.get("ntfy_password", "").strip() headers = { "Title": f"PocketVeto Weekly — {count} bill{'s' if count != 1 else ''} updated", "Priority": "low", "Tags": "newspaper,calendar", } if auth_method == "token" and ntfy_token: headers["Authorization"] = f"Bearer {ntfy_token}" elif auth_method == "basic" and ntfy_username: creds = base64.b64encode(f"{ntfy_username}:{ntfy_password}".encode()).decode() headers["Authorization"] = f"Basic {creds}" resp = requests.post(ntfy_url, data=summary.encode("utf-8"), headers=headers, timeout=NTFY_TIMEOUT) resp.raise_for_status() def _send_digest_ntfy(events: list, ntfy_url: str, prefs: dict) -> None: auth_method = prefs.get("ntfy_auth_method", "none") ntfy_token = prefs.get("ntfy_token", "").strip() ntfy_username = prefs.get("ntfy_username", "").strip() ntfy_password = prefs.get("ntfy_password", "").strip() headers = { "Title": f"PocketVeto Digest — {len(events)} update{'s' if len(events) != 1 else ''}", "Priority": "default", "Tags": "newspaper", } if auth_method == "token" and ntfy_token: headers["Authorization"] = f"Bearer {ntfy_token}" elif auth_method == "basic" and ntfy_username: creds = base64.b64encode(f"{ntfy_username}:{ntfy_password}".encode()).decode() headers["Authorization"] = f"Basic {creds}" # Group by bill, show up to 10 by_bill: dict = defaultdict(list) for event in events: by_bill[event.bill_id].append(event) lines = [] for bill_id, bill_events in list(by_bill.items())[:10]: payload = bill_events[0].payload or {} bill_label = payload.get("bill_label", bill_id.upper()) event_labels = list({_EVENT_TITLES.get(e.event_type, "Update") for e in bill_events}) lines.append(f"• {bill_label}: {', '.join(event_labels)}") if len(by_bill) > 10: lines.append(f" …and {len(by_bill) - 10} more bills") message = "\n".join(lines) resp = requests.post(ntfy_url, data=message.encode("utf-8"), headers=headers, timeout=NTFY_TIMEOUT) resp.raise_for_status()