"""Email service — SMTP delivery with HTML templates and delivery logging.""" import smtplib import logging import time import uuid from datetime import datetime, timezone from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart from pathlib import Path from typing import Optional from app.config import settings logger = logging.getLogger(__name__) TEMPLATE_DIR = Path(__file__).parent.parent / "templates" def _render_template(template_name: str, context: dict) -> str: """Render a Jinja2 HTML email template. Returns empty string on failure.""" try: from jinja2 import Environment, FileSystemLoader env = Environment(loader=FileSystemLoader(str(TEMPLATE_DIR)), autoescape=True) tmpl = env.get_template(template_name) return tmpl.render(**context) except Exception as e: logger.warning("Template render failed (%s): %s", template_name, e) return "" def _log_email( to: str, subject: str, email_type: str, status: str, school_id: Optional[str] = None, error_message: Optional[str] = None, ) -> None: """Write an email delivery record to email_logs. Best-effort — never raises.""" try: import os from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from app.models.email_log import EmailLog, EmailStatus, EmailType db_url = os.getenv("DATABASE_URL", "postgresql://postgres:postgres@db:5432/taptrack_hub") engine = create_engine( db_url.replace("postgresql+asyncpg://", "postgresql://"), pool_pre_ping=True, pool_size=1, max_overflow=0, ) Session = sessionmaker(bind=engine) db = Session() try: # Map type string to enum, default to 'other' try: etype = EmailType(email_type) except ValueError: etype = EmailType.other log = EmailLog( id=str(uuid.uuid4()), school_id=school_id, to_email=to, subject=subject, email_type=etype, status=EmailStatus(status), error_message=error_message, sent_at=datetime.now(timezone.utc) if status == "sent" else None, created_at=datetime.now(timezone.utc), ) db.add(log) db.commit() finally: db.close() engine.dispose() except Exception as e: logger.debug("Email log write failed: %s", e) def send_email( to: str, subject: str, body: str, html: Optional[str] = None, template_name: Optional[str] = None, context: Optional[dict] = None, email_type: str = "other", school_id: Optional[str] = None, max_retries: int = 3, ) -> bool: """ Send an email via configured SMTP. Priority order for HTML content: 1. `html` parameter (raw HTML string) 2. `template_name` + `context` (rendered Jinja2 template) 3. `body` only (plain text) Logs every attempt to email_logs table. Retries up to `max_retries` times on transient SMTP failure. Returns True on success. """ if not settings.SMTP_HOST: logger.warning("SMTP not configured — would send to %s: %s", to, subject) _log_email(to, subject, email_type, "failed", school_id, "SMTP not configured") return False # Resolve HTML content html_content = html if not html_content and template_name: ctx = context or {} ctx.setdefault("subject", subject) ctx.setdefault("portal_url", settings.HUB_BASE_URL) html_content = _render_template(template_name, ctx) last_error: Optional[str] = None for attempt in range(1, max_retries + 1): try: msg = MIMEMultipart("alternative") msg["Subject"] = subject msg["From"] = settings.SMTP_FROM msg["To"] = to msg.attach(MIMEText(body, "plain")) if html_content: msg.attach(MIMEText(html_content, "html")) with smtplib.SMTP(settings.SMTP_HOST, settings.SMTP_PORT) as smtp: smtp.ehlo() smtp.starttls() smtp.ehlo() if settings.SMTP_USER: smtp.login(settings.SMTP_USER, settings.SMTP_PASSWORD) smtp.sendmail(settings.SMTP_FROM, [to], msg.as_string()) logger.info("Email sent to %s: %s", to, subject) _log_email(to, subject, email_type, "sent", school_id) return True except smtplib.SMTPException as e: last_error = str(e) logger.warning("SMTP attempt %d/%d failed for %s: %s", attempt, max_retries, to, e) if attempt < max_retries: time.sleep(2 ** attempt) # exponential backoff: 2s, 4s except Exception as e: last_error = str(e) logger.error("Email send failed for %s: %s", to, e) break # Non-SMTP errors don't retry _log_email(to, subject, email_type, "failed", school_id, last_error) return False