""" Alert Service — budget threshold alerts triggered after expense transactions. Checks budgeted categories and flashes/emails when spending crosses 80% or 100%. Dedup: alert_sent_80 / alert_sent_100 flags on each Budget row prevent repeat emails within the same month. Flags default to False on new Budget rows. 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 from app.extensions import db log = logging.getLogger('app.alert_service') 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'] pw = 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: ctx = ssl.create_default_context() with smtplib.SMTP(host, port) as srv: srv.ehlo() srv.starttls(context=ctx) srv.login(user, pw) srv.sendmail(user, to_addr, msg.as_string()) log.info('[alert] email sent: %s → %s', subject, to_addr) return True except Exception as exc: log.error('[alert] email failed: %s', exc, exc_info=True) return False def send_test_email(): """Send a test email to verify SMTP config. Returns (ok: bool, message: str).""" if not _smtp_configured(): return False, ('SMTP not configured. ' 'Set SMTP_HOST, SMTP_USER, SMTP_PASSWORD, and ALERT_EMAIL in .env.') to_addr = current_app.config['ALERT_EMAIL'] body = ( '
This is a test email from your Personal Finance Management app.
' 'Budget alerts are configured correctly.
' ) ok = _send_email('PFM — Test Email', body) if ok: return True, f'Test email sent to {to_addr}.' return False, 'SMTP error — check server logs for details.' def get_budget_alerts(month_str=None): """ Return list of alert dicts for budget categories at or over threshold. Does NOT check dedup flags — use for display purposes only. Each dict: {category, spent, limit, pct, level} level: 'over' (≥100%) | 'warning' (≥80%) """ from app.services.budget_service import get_budget_summary if month_str is None: month_str = date.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. For each category that newly crossed 80% or 100%, flash a message and (if SMTP configured and user has alerts enabled) send one email. Dedup flags on the Budget row prevent repeat sends. Call this after committing an expense transaction. flash_fn: Flask's flash() (passed in to avoid circular imports at module level) """ from app.models.budget import Budget from app.models.user import User from app.services.budget_service import get_month_spending user = User.query.first() alerts_enabled = user.budget_alerts_enabled if user else False symbol = (user.currency_symbol or '$') if user else '$' month_str = date.today().strftime('%Y-%m') try: budgets = Budget.query.filter_by(month=month_str).all() except Exception as exc: log.error('[alert] failed to query budgets: %s', exc) return dirty_budgets = [] for b in budgets: try: limit = float(b.limit_amount) + float(b.rollover_amount or 0) if limit <= 0: continue spent = get_month_spending(b.category_id, month_str) pct = spent / limit * 100 cat = b.category cat_name = cat.name if cat else 'Unknown' # 100% threshold — only fire if not already sent this month if pct >= 100 and not b.alert_sent_100: msg = (f'Budget exceeded: {cat_name} — ' f'{symbol}{spent:,.2f} of {symbol}{limit:,.2f} ({pct:.0f}%)') flash_fn(msg, 'danger') if alerts_enabled and _smtp_configured(): _send_single_alert_email(cat_name, spent, limit, pct, symbol, threshold=100) b.alert_sent_100 = True dirty_budgets.append(b) # 80% threshold — only fire if not already sent (and 100% not yet hit) elif pct >= 80 and not b.alert_sent_80: msg = (f'Budget warning: {cat_name} — ' f'{symbol}{spent:,.2f} of {symbol}{limit:,.2f} ({pct:.0f}% used)') flash_fn(msg, 'warning') if alerts_enabled and _smtp_configured(): _send_single_alert_email(cat_name, spent, limit, pct, symbol, threshold=80) b.alert_sent_80 = True dirty_budgets.append(b) except Exception as exc: log.error('[alert] error checking budget %s: %s', b.id, exc) if dirty_budgets: try: db.session.commit() except Exception as exc: log.error('[alert] failed to save alert flags: %s', exc) db.session.rollback() def _send_single_alert_email(cat_name, spent, limit, pct, symbol, threshold): label = '100%+' if threshold >= 100 else '80%' remaining = max(limit - spent, 0) color = '#ef4444' if threshold >= 100 else '#f59e0b' emoji = '🚨' if threshold >= 100 else '⚠️' app_url = current_app.config.get('APP_URL', '') body = f"""Your {cat_name} budget has reached {pct:.0f}%.
| Spent | {symbol}{spent:,.2f} |
| Budget limit | {symbol}{limit:,.2f} |
| Remaining | {symbol}{remaining:,.2f} |
Disable alerts in Settings → Profile.