"""Outbound email for onboarding (access requests + approval notices). Gmail SMTP + App Password by default. Deliberately fail-soft: if SMTP isn't configured or a send errors, we log and return False so the caller can fall back to in-app notifications — email is never load-bearing for the app's core flow. """ import logging import smtplib import ssl from email.message import EmailMessage from email.utils import formatdate from app import sysconfig from app.config import settings from app.db import SessionLocal log = logging.getLogger("subfeed.email") def _smtp() -> dict: """Effective SMTP config (admin DB override over env defaults). Opens its own short session because email is often sent from a BackgroundTask, outside a request's db.""" with SessionLocal() as db: return { "host": sysconfig.get_str(db, "smtp_host"), "port": sysconfig.get_int(db, "smtp_port"), "user": sysconfig.get_str(db, "smtp_user"), "password": sysconfig.get_str(db, "smtp_password"), "from": sysconfig.get_str(db, "smtp_from"), } def email_enabled() -> bool: c = _smtp() return bool(c["host"] and c["user"] and c["password"]) def _admin_contact() -> str | None: return next(iter(sorted(settings.admin_email_set)), None) def _send(to: list[str], subject: str, body: str, reply_to: str | None = None) -> bool: recipients = [e for e in to if e] if not recipients: return False c = _smtp() if not (c["host"] and c["user"] and c["password"]): log.info("SMTP not configured; skipping email %r to %s", subject, recipients) return False msg = EmailMessage() msg["Subject"] = subject msg["From"] = c["from"] or c["user"] msg["To"] = ", ".join(recipients) # A real Date and a Reply-To (so it's a conversation, not a no-reply blast) both # nudge spam filters the right way; reputation still does most of the work. msg["Date"] = formatdate(localtime=True) if reply_to: msg["Reply-To"] = reply_to msg.set_content(body) try: with smtplib.SMTP(c["host"], c["port"], timeout=20) as s: s.starttls(context=ssl.create_default_context()) s.login(c["user"], c["password"]) s.send_message(msg) log.info("Sent email %r to %s", subject, recipients) return True except Exception: log.exception("SMTP send failed (%r to %s)", subject, recipients) return False def send_access_approved(email: str) -> bool: admin = _admin_contact() body = ( "Hi,\n\n" "Your request to use Siftlode has been approved — welcome aboard.\n\n" "Just open Siftlode and sign in with this Google account, and your YouTube\n" "subscriptions feed will be ready.\n\n" "If you weren't expecting this, you can ignore the message.\n\n" "— Siftlode" + (f"\n\nQuestions? Just reply to this email ({admin})." if admin else "") ) return _send([email], "You're in — your Siftlode access is approved", body, reply_to=admin) def send_admin_new_request(admins: list[str], requester: str) -> bool: body = ( f"{requester} just requested access to Siftlode.\n\n" "Open Siftlode → Settings → Account → Access requests to approve or deny it.\n\n" "(Reply to this email to reach the requester directly.)\n" ) return _send(admins, f"Siftlode access request from {requester}", body, reply_to=requester) def send_test(to: str) -> bool: body = ( "This is a test email from Siftlode.\n\n" "If you're reading this, your SMTP settings work.\n\n" "— Siftlode" ) return _send([to], "Siftlode SMTP test", body)