""" Alert Service — budget threshold alerts triggered after expense transactions. Checks all budgeted categories for the current month and returns alert dicts for any that just crossed 80% or 100%. Also sends an email if SMTP is configured. Usage (in routes after db.session.commit()): from app.services.alert_service import check_and_flash_budget_alerts check_and_flash_budget_alerts(flash_fn) """ import logging import smtplib import ssl from datetime import date from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart from flask import current_app log = logging.getLogger(__name__) def _smtp_configured(): cfg = current_app.config return all([ cfg.get('SMTP_HOST'), cfg.get('SMTP_USER'), cfg.get('SMTP_PASSWORD'), cfg.get('ALERT_EMAIL'), ]) def _send_email(subject, body_html): cfg = current_app.config host = cfg['SMTP_HOST'] port = int(cfg.get('SMTP_PORT', 587)) user = cfg['SMTP_USER'] password = cfg['SMTP_PASSWORD'] to_addr = cfg['ALERT_EMAIL'] msg = MIMEMultipart('alternative') msg['Subject'] = subject msg['From'] = user msg['To'] = to_addr msg.attach(MIMEText(body_html, 'html')) try: context = ssl.create_default_context() with smtplib.SMTP(host, port) as srv: srv.ehlo() srv.starttls(context=context) srv.login(user, password) srv.sendmail(user, to_addr, msg.as_string()) log.info('[alert] email sent: %s → %s', subject, to_addr) except Exception as exc: log.error('[alert] email failed: %s', exc, exc_info=True) def get_budget_alerts(month_str=None): """ Return list of alert dicts for budget categories at or over threshold. Each dict: {category, spent, limit, pct, level} level: 'over' (≥100%) | 'warning' (≥80%) Only returns categories that have a budget set. """ from app.services.budget_service import get_budget_summary if month_str is None: today = date.today() month_str = today.strftime('%Y-%m') summary = get_budget_summary(month_str) alerts = [] for item in summary: if not item['has_budget'] or item['pct'] is None: continue pct = item['pct'] if pct >= 100: alerts.append({**item, 'level': 'over'}) elif pct >= 80: alerts.append({**item, 'level': 'warning'}) return alerts def check_and_flash_budget_alerts(flash_fn): """ Check current month's budgets and flash any threshold alerts. Optionally sends an email summary if SMTP is configured. Call this after committing an expense transaction. flash_fn: Flask's flash() function (passed in to avoid circular imports) """ try: alerts = get_budget_alerts() except Exception as exc: log.error('[alert] check failed: %s', exc, exc_info=True) return if not alerts: return email_lines = [] for a in alerts: cat_name = a['category'].name if a['category'] else 'Unknown' pct = a['pct'] spent = a['spent'] limit = a['limit'] symbol = current_app.config.get('APP_CURRENCY_SYMBOL', '$') if a['level'] == 'over': msg = (f'Budget exceeded: {cat_name} — ' f'{symbol}{spent:,.2f} spent of {symbol}{limit:,.2f} limit ({pct:.0f}%)') flash_fn(msg, 'danger') else: msg = (f'Budget warning: {cat_name} — ' f'{symbol}{spent:,.2f} of {symbol}{limit:,.2f} ({pct:.0f}% used)') flash_fn(msg, 'warning') email_lines.append( f'
  • {cat_name}: {symbol}{spent:,.2f} / {symbol}{limit:,.2f} ' f'({pct:.0f}%) — {"EXCEEDED" if a["level"]=="over" else "Warning"}
  • ' ) if _smtp_configured() and email_lines: today = date.today().strftime('%B %Y') body = ( f'

    PFM Budget Alert — {today}

    ' f'

    The following budget categories need your attention:

    ' f'' f'

    View Budgets →

    ' ) _send_email(f'PFM Budget Alert — {today}', body)