"""Email: Mailjet (HTTPS) → Resend → Brevo SMTP (port 2525) → SMTP fallback.""" import os import smtplib import base64 from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText def _build_body(username: str, reset_url: str) -> str: return ( f"Hi {username},\n\n" f"We received a request to reset your YapStation password.\n\n" f"Open this link (valid for 1 hour):\n{reset_url}\n\n" f"If you did not request this, ignore this email.\n\n" f"— YapStation" ) def _build_msg(mail_from: str, to_email: str, username: str, reset_url: str) -> MIMEMultipart: msg = MIMEMultipart() msg["Subject"] = "YapStation — reset your password" msg["From"] = mail_from msg["To"] = to_email msg.attach(MIMEText(_build_body(username, reset_url), "plain", "utf-8")) return msg def _send_via_mailjet(to_email: str, username: str, reset_url: str) -> bool: api_key = os.environ.get("MAILJET_API_KEY", "").strip() secret_key = os.environ.get("MAILJET_SECRET_KEY", "").strip() if not api_key or not secret_key: return False try: import httpx mail_from = os.environ.get("MAIL_FROM", "yapstation.official@gmail.com").strip() resp = httpx.post( "https://api.mailjet.com/v3.1/send", auth=(api_key, secret_key), json={ "Messages": [{ "From": {"Email": mail_from, "Name": "YapStation"}, "To": [{"Email": to_email}], "Subject": "YapStation — reset your password", "TextPart": _build_body(username, reset_url), }] }, timeout=15, ) if resp.status_code == 200: return True print(f"[mail] Mailjet error {resp.status_code}: {resp.text}") return False except Exception as e: print(f"[mail] Mailjet failed: {e}") return False def _send_via_resend(to_email: str, username: str, reset_url: str) -> bool: api_key = os.environ.get("RESEND_API_KEY", "").strip() if not api_key: return False try: import httpx mail_from = os.environ.get("MAIL_FROM", "YapStation ").strip() resp = httpx.post( "https://api.resend.com/emails", headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}, json={"from": mail_from, "to": [to_email], "subject": "YapStation — reset your password", "text": _build_body(username, reset_url)}, timeout=15, ) if resp.status_code in (200, 201): return True print(f"[mail] Resend error {resp.status_code}: {resp.text}") return False except Exception as e: print(f"[mail] Resend failed: {e}") return False def _send_via_brevo(to_email: str, username: str, reset_url: str) -> bool: smtp_user = os.environ.get("BREVO_SMTP_USER", "").strip().strip('"').strip("'") smtp_key = os.environ.get("BREVO_API_KEY", "").strip().strip('"').strip("'").replace(" ", "") if not smtp_user or not smtp_key: return False mail_from = os.environ.get("MAIL_FROM", smtp_user).strip() msg = _build_msg(mail_from, to_email, username, reset_url) try: with smtplib.SMTP("smtp-relay.brevo.com", 2525, timeout=30) as smtp: smtp.ehlo(); smtp.starttls(); smtp.ehlo() smtp.login(smtp_user, smtp_key) smtp.sendmail(mail_from, [to_email], msg.as_string()) return True except Exception as e: print(f"[mail] Brevo SMTP failed ({type(e).__name__}): {e}") return False def _send_via_smtp(to_email: str, username: str, reset_url: str) -> bool: host = os.environ.get("SMTP_HOST", "").strip() if not host or not to_email: return False port = int(os.environ.get("SMTP_PORT", "587")) user = os.environ.get("SMTP_USER", "").strip() password = "".join(os.environ.get("SMTP_PASSWORD", "").split()).strip('"').strip("'") mail_from = os.environ.get("MAIL_FROM", user or "noreply@localhost").strip() msg = _build_msg(mail_from, to_email, username, reset_url) try: if port == 465: with smtplib.SMTP_SSL(host, port, timeout=30) as smtp: if user and password: smtp.login(user, password) smtp.sendmail(mail_from, [to_email], msg.as_string()) else: with smtplib.SMTP(host, port, timeout=30) as smtp: smtp.ehlo(); smtp.starttls(); smtp.ehlo() if user and password: smtp.login(user, password) smtp.sendmail(mail_from, [to_email], msg.as_string()) return True except Exception as e: print(f"[mail] SMTP send failed ({type(e).__name__}): {e}") return False def send_password_reset_email(to_email: str, username: str, reset_url: str) -> bool: return _send_via_mailjet(to_email, username, reset_url) or _send_via_resend(to_email, username, reset_url) or _send_via_brevo(to_email, username, reset_url) or _send_via_smtp(to_email, username, reset_url) def smtp_configured() -> bool: return bool( os.environ.get("MAILJET_API_KEY", "").strip() or os.environ.get("RESEND_API_KEY", "").strip() or os.environ.get("BREVO_SMTP_USER", "").strip() or os.environ.get("SMTP_HOST", "").strip() )