""" BAN YARO — E-Mail-Versand Unterstützt zwei Backends (wird automatisch gewählt): 1. Brevo REST-API — wenn BREVO_API_KEY gesetzt (bevorzugt) 2. SMTP — wenn SMTP_HOST gesetzt (Fallback) """ import imaplib import os import base64 import smtplib import asyncio import logging import ssl from datetime import datetime from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart from email.mime.application import MIMEApplication from email.utils import formatdate import httpx logger = logging.getLogger(__name__) # Brevo REST-API BREVO_API_KEY = os.getenv("BREVO_API_KEY", "") BREVO_API_URL = "https://api.brevo.com/v3/smtp/email" # SMTP Fallback SMTP_HOST = os.getenv("SMTP_HOST", "") SMTP_PORT = int(os.getenv("SMTP_PORT", "587")) SMTP_USER = os.getenv("SMTP_USER", "") SMTP_PASS = os.getenv("SMTP_PASS", "") or os.getenv("SMTP_SUPPORT_PASS", "") # IMAP für Gesendet-Ordner IMAP_HOST = os.getenv("IMAP_HOST", SMTP_HOST) IMAP_PORT = int(os.getenv("IMAP_PORT", "993")) _SENT_CANDIDATES = ["Sent", "Sent Messages", "Sent Items", "INBOX.Sent", "Gesendete Objekte"] SMTP_FROM = os.getenv("SMTP_FROM", "Ban Yaro ") APP_URL = os.getenv("APP_URL", "https://banyaro.app") # ------------------------------------------------------------------ # IMAP: Mail in Gesendet-Ordner speichern # ------------------------------------------------------------------ def _imap_save_sent(msg_bytes: bytes): if not IMAP_HOST or not SMTP_USER or not SMTP_PASS: return try: ctx = ssl.create_default_context() with imaplib.IMAP4_SSL(IMAP_HOST, IMAP_PORT, ssl_context=ctx) as imap: imap.login(SMTP_USER, SMTP_PASS) _, raw_folders = imap.list() available = [f.decode(errors="replace") for f in (raw_folders or [])] folder = None for line in available: name = line.rsplit('"." ', 1)[-1].strip().strip('"') for candidate in _SENT_CANDIDATES: if candidate.lower() in name.lower(): folder = name break if folder: break if not folder: folder = "INBOX.Sent" imap.append( folder, r"\Seen", imaplib.Time2Internaldate(datetime.now().timestamp()), msg_bytes, ) except Exception as e: logger.warning(f"IMAP Gesendet-Speicherung fehlgeschlagen: {e}") def _build_mime_copy(to: str, subject: str, html: str, plain: str, attachments: list | None) -> MIMEMultipart: """Baut eine MIME-Nachricht für die Gesendet-Ablage (Brevo-Pfad).""" if attachments: msg = MIMEMultipart("mixed") alt = MIMEMultipart("alternative") alt.attach(MIMEText(plain, "plain", "utf-8")) alt.attach(MIMEText(html, "html", "utf-8")) msg.attach(alt) for a in attachments: part = MIMEApplication(a["content"], Name=a["filename"]) part["Content-Disposition"] = f'attachment; filename="{a["filename"]}"' msg.attach(part) else: msg = MIMEMultipart("alternative") msg.attach(MIMEText(plain, "plain", "utf-8")) msg.attach(MIMEText(html, "html", "utf-8")) msg["Subject"] = subject msg["From"] = SMTP_FROM msg["To"] = to msg["Date"] = formatdate(localtime=False) return msg # ------------------------------------------------------------------ # Brevo REST-API # ------------------------------------------------------------------ async def _send_brevo(to: str, subject: str, html: str, plain: str, attachments: list | None = None): from_raw = SMTP_FROM if "<" in from_raw: from_name = from_raw[:from_raw.index("<")].strip() from_email = from_raw[from_raw.index("<")+1:from_raw.index(">")].strip() else: from_name = "Ban Yaro" from_email = from_raw.strip() payload = { "sender": {"name": from_name, "email": from_email}, "to": [{"email": to}], "subject": subject, "htmlContent": html, "textContent": plain, "headers": {"X-Mailin-Track-Click": "0", "X-Mailin-Track-Opens": "0"}, } if attachments: payload["attachment"] = [ { "name": a["filename"], "content": base64.b64encode(a["content"]).decode("ascii"), } for a in attachments ] async with httpx.AsyncClient(timeout=15) as client: resp = await client.post( BREVO_API_URL, json=payload, headers={"api-key": BREVO_API_KEY, "Content-Type": "application/json"}, ) resp.raise_for_status() # ------------------------------------------------------------------ # SMTP Fallback # ------------------------------------------------------------------ def _send_smtp_sync(to: str, subject: str, html: str, plain: str, attachments: list | None = None): if attachments: msg = MIMEMultipart("mixed") alt = MIMEMultipart("alternative") alt.attach(MIMEText(plain, "plain", "utf-8")) alt.attach(MIMEText(html, "html", "utf-8")) msg.attach(alt) for a in attachments: part = MIMEApplication(a["content"], Name=a["filename"]) part["Content-Disposition"] = f'attachment; filename="{a["filename"]}"' msg.attach(part) else: msg = MIMEMultipart("alternative") msg.attach(MIMEText(plain, "plain", "utf-8")) msg.attach(MIMEText(html, "html", "utf-8")) msg["Subject"] = subject msg["From"] = SMTP_FROM msg["To"] = to msg["Date"] = formatdate(localtime=False) msg_bytes = msg.as_bytes() with smtplib.SMTP(SMTP_HOST, SMTP_PORT, timeout=10) as s: s.ehlo() s.starttls() if SMTP_USER: s.login(SMTP_USER, SMTP_PASS) s.sendmail(SMTP_FROM, [to], msg_bytes) _imap_save_sent(msg_bytes) # ------------------------------------------------------------------ # Öffentliche Funktion # ------------------------------------------------------------------ async def send_email(to: str, subject: str, html: str, plain: str = "", attachments: list | None = None): if BREVO_API_KEY: try: await _send_brevo(to, subject, html, plain, attachments) logger.info(f"Mail via Brevo gesendet: «{subject}» → {to}") # MIME-Kopie für Gesendet-Ordner konstruieren loop = asyncio.get_event_loop() await loop.run_in_executor(None, lambda: _imap_save_sent( _build_mime_copy(to, subject, html, plain, attachments).as_bytes() )) return except Exception as e: logger.error(f"Brevo-Fehler: {e}") raise if SMTP_HOST: loop = asyncio.get_event_loop() try: await loop.run_in_executor( None, _send_smtp_sync, to, subject, html, plain, attachments ) logger.info(f"Mail via SMTP gesendet: «{subject}» → {to}") return except Exception as e: logger.error(f"SMTP-Fehler: {e}") raise logger.warning(f"Kein Mail-Backend konfiguriert — Mail NICHT gesendet: «{subject}» → {to}") def email_html( body_html: str, cta_url: str = None, cta_label: str = None, footer_text: str = None, ) -> str: """Shared branded HTML email template (matches Status-Report design).""" cta_block = "" if cta_url and cta_label: cta_block = f"""

{cta_label}

""" footer = footer_text or "Ban Yaro · banyaro.app" return f"""\
🐾 Ban Yaro
{body_html}{cta_block}
{footer}
""" async def send_verify_email(to: str, name: str, token: str): url = f"{APP_URL}/api/auth/verify/{token}" subject = "Ban Yaro — E-Mail-Adresse bestätigen" body = f"""

Hallo {name},

bitte bestätige deine E-Mail-Adresse, damit dein Konto vollständig eingerichtet ist.

Der Link ist 48 Stunden gültig.

Falls du dich nicht bei Ban Yaro registriert hast, ignoriere diese E-Mail einfach.

""" html = email_html(body, cta_url=url, cta_label="E-Mail bestätigen") plain = f"Ban Yaro — E-Mail bestätigen\n\nHallo {name},\n\nbitte bestätige:\n{url}\n\nLink ist 48h gültig.\n" await send_email(to, subject, html, plain)