137 lines
4.3 KiB
Python
137 lines
4.3 KiB
Python
"""
|
|
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: <strong>{cat_name}</strong> — '
|
|
f'{symbol}{spent:,.2f} spent of {symbol}{limit:,.2f} limit ({pct:.0f}%)')
|
|
flash_fn(msg, 'danger')
|
|
else:
|
|
msg = (f'Budget warning: <strong>{cat_name}</strong> — '
|
|
f'{symbol}{spent:,.2f} of {symbol}{limit:,.2f} ({pct:.0f}% used)')
|
|
flash_fn(msg, 'warning')
|
|
|
|
email_lines.append(
|
|
f'<li><b>{cat_name}</b>: {symbol}{spent:,.2f} / {symbol}{limit:,.2f} '
|
|
f'({pct:.0f}%) — <b>{"EXCEEDED" if a["level"]=="over" else "Warning"}</b></li>'
|
|
)
|
|
|
|
if _smtp_configured() and email_lines:
|
|
today = date.today().strftime('%B %Y')
|
|
body = (
|
|
f'<h3>PFM Budget Alert — {today}</h3>'
|
|
f'<p>The following budget categories need your attention:</p>'
|
|
f'<ul>{"".join(email_lines)}</ul>'
|
|
f'<p><a href="{current_app.config.get("APP_URL","")}/budgets">View Budgets →</a></p>'
|
|
)
|
|
_send_email(f'PFM Budget Alert — {today}', body)
|