""" Email utilities — send verification codes via Resend HTTP API (primary) with SMTP fallback. """ import os import random import string import socket import smtplib import logging import threading import requests as http_requests from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart # ── Force IPv4 strictly to avoid Docker 'Errno 101 Network is unreachable' on IPv6 loops _orig_getaddrinfo = socket.getaddrinfo def _ipv4_only_getaddrinfo(*args, **kwargs): responses = _orig_getaddrinfo(*args, **kwargs) return [r for r in responses if r[0] == socket.AF_INET] socket.getaddrinfo = _ipv4_only_getaddrinfo logger = logging.getLogger(__name__) # ── Brevo Configuration (preferred — bypasses HF SMTP firewall) ────── BREVO_API_KEY = os.getenv("BREVO_API_KEY", "") BREVO_FROM_EMAIL = os.getenv("BREVO_FROM_EMAIL", "bantaypahayag@gmail.com") BREVO_FROM_NAME = os.getenv("BREVO_FROM_NAME", "BantayPahayag Verification") # ── SMTP Configuration (fallback) ──────────────────────────────────── SMTP_HOST = os.getenv("SMTP_HOST", "smtp.gmail.com") SMTP_PORT = int(os.getenv("SMTP_PORT", "587")) SMTP_USER = os.getenv("SMTP_USER", "") SMTP_PASSWORD = os.getenv("SMTP_PASSWORD", "") SMTP_FROM_NAME = os.getenv("SMTP_FROM_NAME", "BantayPahayag") SMTP_TIMEOUT = int(os.getenv("SMTP_TIMEOUT", "10")) # seconds def generate_verification_code(length: int = 6) -> str: """Generate a random numeric verification code.""" return "".join(random.choices(string.digits, k=length)) # ═══════════════════════════════════════════════════════════════════════ # BREVO HTTP API (Primary — works on Hugging Face Free Tier) # ═══════════════════════════════════════════════════════════════════════ def _send_via_brevo(to_email: str, subject: str, html_body: str) -> bool: """Send email via Brevo HTTP API (port 443 — not blocked by HF).""" if not BREVO_API_KEY: logger.warning("BREVO_API_KEY not configured. Cannot send via Brevo.") return False try: resp = http_requests.post( "https://api.brevo.com/v3/smtp/email", headers={ "api-key": BREVO_API_KEY, "Content-Type": "application/json", "Accept": "application/json" }, json={ "sender": {"name": BREVO_FROM_NAME, "email": BREVO_FROM_EMAIL}, "to": [{"email": to_email}], "subject": subject, "htmlContent": html_body, }, timeout=15, ) if resp.status_code in (200, 201): logger.info("✅ Email sent via Brevo to %s (id: %s)", to_email, resp.json().get("messageId")) return True else: logger.error("Brevo API error %s: %s", resp.status_code, resp.text) return False except Exception as exc: logger.error("Brevo request failed: %s", exc) return False try: resp = http_requests.post( "https://api.resend.com/emails", headers={ "Authorization": f"Bearer {RESEND_API_KEY}", "Content-Type": "application/json", }, json={ "from": RESEND_FROM, "to": [to_email], "subject": subject, "html": html_body, }, timeout=15, ) if resp.status_code in (200, 201): logger.info("✅ Email sent via Resend to %s (id: %s)", to_email, resp.json().get("id")) return True else: logger.error("Resend API error %s: %s", resp.status_code, resp.text) return False except Exception as exc: logger.error("Resend request failed: %s", exc) return False # ═══════════════════════════════════════════════════════════════════════ # SMTP FALLBACK (works on local dev, blocked on HF Free Tier) # ═══════════════════════════════════════════════════════════════════════ def _send_via_smtp(to_email: str, subject: str, html_body: str, text_body: str) -> bool: """Send email via traditional SMTP. Works locally, blocked on HF.""" if not SMTP_USER or not SMTP_PASSWORD: return False try: msg = MIMEMultipart("alternative") msg["Subject"] = subject msg["From"] = f"{SMTP_FROM_NAME} <{SMTP_USER}>" msg["To"] = to_email msg.attach(MIMEText(text_body, "plain")) msg.attach(MIMEText(html_body, "html")) with smtplib.SMTP(SMTP_HOST, SMTP_PORT, timeout=SMTP_TIMEOUT) as server: server.starttls() server.login(SMTP_USER, SMTP_PASSWORD) server.sendmail(SMTP_USER, to_email, msg.as_string()) logger.info("✅ Email sent via SMTP to %s", to_email) return True except Exception as exc: logger.warning("SMTP failed for %s: %s", to_email, exc) return False # ═══════════════════════════════════════════════════════════════════════ # VERIFICATION EMAIL # ═══════════════════════════════════════════════════════════════════════ def _build_verification_html(code: str, username: str = "") -> str: return f"""
BantayPahayag

Verify Your Email

Hi{' ' + username if username else ''}! Use the code below to verify your email address and activate your BantayPahayag account.

{code}

This code expires in 10 minutes.

If you didn't create a BantayPahayag account, you can safely ignore this email.

BantayPahayag \u2014 News Verification System
Cavite State University Thesis Project

""" def _send_verification_email_worker(to_email: str, code: str, username: str = "") -> None: """Background worker: tries Resend first, then SMTP fallback.""" # Always log the code for demo/debugging purposes logger.warning("==================================================================") logger.warning(" [VERIFICATION CODE]") logger.warning(" USER: %s", to_email) logger.warning(" CODE: %s", code) logger.warning("==================================================================") subject = f"BantayPahayag \u2014 Your Verification Code: {code}" html = _build_verification_html(code, username) # Try Brevo first (works on HF) if _send_via_brevo(to_email, subject, html): return # Fallback to SMTP (works locally) text_body = ( f"BantayPahayag \u2014 Email Verification\n\n" f"Hi{' ' + username if username else ''}!\n\n" f"Your verification code is: {code}\n\n" f"This code expires in 10 minutes.\n\n" f"If you didn't create a BantayPahayag account, ignore this email." ) if _send_via_smtp(to_email, subject, html, text_body): return logger.warning("All email channels failed for %s. Code is in logs above.", to_email) def send_verification_email(to_email: str, code: str, username: str = "") -> bool: """Send verification email in a background thread (fire-and-forget).""" thread = threading.Thread( target=_send_verification_email_worker, args=(to_email, code, username), daemon=True, ) thread.start() logger.info("Verification email queued for %s", to_email) return True # ═══════════════════════════════════════════════════════════════════════ # PASSWORD RESET EMAIL # ═══════════════════════════════════════════════════════════════════════ def _build_reset_html(code: str, username: str = "") -> str: return f"""
BantayPahayag

Reset Your Password

Hi{' ' + username if username else ''}! You recently requested to reset your password. Use the code below to securely change it.

{code}

This code expires in 10 minutes.

If you did not request a password reset, you can safely ignore this email.

BantayPahayag \u2014 News Verification System
Cavite State University Thesis Project

""" def _send_reset_email_worker(to_email: str, code: str, username: str = "") -> None: """Background worker: tries Resend first, then SMTP fallback.""" # Always log the code for demo/debugging purposes logger.warning("==================================================================") logger.warning(" [PASSWORD RESET CODE]") logger.warning(" USER: %s", to_email) logger.warning(" CODE: %s", code) logger.warning("==================================================================") subject = f"BantayPahayag \u2014 Password Reset Code: {code}" html = _build_reset_html(code, username) # Try Brevo first (works on HF) if _send_via_brevo(to_email, subject, html): return # Fallback to SMTP (works locally) text_body = ( f"BantayPahayag \u2014 Password Reset\n\n" f"Hi{' ' + username if username else ''}!\n\n" f"Your password reset code is: {code}\n\n" f"This code expires in 10 minutes.\n\n" f"If you did not request this, ignore this email." ) if _send_via_smtp(to_email, subject, html, text_body): return logger.warning("All email channels failed for %s. Code is in logs above.", to_email) def send_password_reset_email(to_email: str, code: str, username: str = "") -> bool: """Send password reset email in a background thread (fire-and-forget).""" thread = threading.Thread( target=_send_reset_email_worker, args=(to_email, code, username), daemon=True, ) thread.start() logger.info("Password reset email queued for %s", to_email) return True