"""
E-Mail-Versand via SMTP (smtplib + asyncio.to_thread).
Konfiguration pro Firma in smtp_configs. Kein externer Mail-Dienst nötig.
"""
import asyncio
import smtplib
import ssl
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from typing import TYPE_CHECKING
from uuid import UUID
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.config import settings
from app.models.smtp_config import SmtpConfig
if TYPE_CHECKING:
from app.models.user import User
def _html_wrapper(title: str, body: str) -> str:
return f"""
{title}
⏱ {settings.app_name}
{body}
"""
def _decrypt_password(encrypted: str) -> str:
"""Fernet-Entschlüsselung (gleiche Implementierung wie ldap_service)."""
import base64
import hashlib
from cryptography.fernet import Fernet
key = hashlib.sha256(settings.secret_key.encode()).digest()
f = Fernet(base64.urlsafe_b64encode(key))
return f.decrypt(encrypted.encode()).decode()
def _smtp_send_sync(cfg: SmtpConfig, to: str, subject: str, html: str) -> None:
"""Synchroner SMTP-Versand – wird via asyncio.to_thread() aufgerufen."""
password = _decrypt_password(cfg.password_encrypted) if cfg.password_encrypted else None
msg = MIMEMultipart("alternative")
msg["Subject"] = subject
msg["From"] = f"{cfg.from_name} <{cfg.from_email}>"
msg["To"] = to
msg.attach(MIMEText(html, "html", "utf-8"))
if cfg.use_tls:
context = ssl.create_default_context()
with smtplib.SMTP_SSL(cfg.host, cfg.port, context=context) as smtp:
if cfg.username and password:
smtp.login(cfg.username, password)
smtp.send_message(msg)
else:
with smtplib.SMTP(cfg.host, cfg.port) as smtp:
if cfg.use_starttls:
smtp.starttls(context=ssl.create_default_context())
if cfg.username and password:
smtp.login(cfg.username, password)
smtp.send_message(msg)
class EmailService:
async def _load_smtp(self, company_id: UUID, db: AsyncSession) -> SmtpConfig | None:
return await db.scalar(
select(SmtpConfig).where(
SmtpConfig.company_id == company_id,
SmtpConfig.is_enabled == True,
)
)
async def _send(self, to: str, subject: str, html: str, cfg: SmtpConfig | None) -> None:
if not cfg:
print(f"\n{'='*60}")
print(f"EMAIL (kein SMTP konfiguriert – nicht versendet)")
print(f" To: {to}")
print(f" Subject: {subject}")
print(f"{'='*60}\n")
return
try:
await asyncio.to_thread(_smtp_send_sync, cfg, to, subject, html)
except Exception as exc:
print(f"SMTP Fehler: {exc}")
async def send_welcome(self, user: "User", db: AsyncSession) -> None:
cfg = await self._load_smtp(user.company_id, db)
body = f"""
Willkommen bei {settings.app_name}, {user.first_name}!
Dein Firmen-Account wurde erfolgreich erstellt. Du kannst dich ab sofort anmelden.
Zum Login
"""
await self._send(user.email, f"Willkommen bei {settings.app_name}", _html_wrapper("Willkommen", body), cfg)
async def send_invite(self, user: "User", invited_by: "User", raw_token: str, db: AsyncSession) -> None:
cfg = await self._load_smtp(user.company_id, db)
invite_url = f"{settings.frontend_url}/invite/accept?token={raw_token}"
body = f"""
Du wurdest eingeladen!
{invited_by.full_name} hat dich zu {settings.app_name} eingeladen.
Klicke auf den Button, um dein Konto zu aktivieren und ein Passwort festzulegen.
Der Link ist 7 Tage gültig.
Einladung annehmen
Oder kopiere diesen Link: {invite_url}
"""
await self._send(
user.email,
f"{invited_by.full_name} hat dich zu {settings.app_name} eingeladen",
_html_wrapper("Einladung", body),
cfg,
)
async def send_password_reset(self, user: "User", raw_token: str, db: AsyncSession) -> None:
cfg = await self._load_smtp(user.company_id, db)
reset_url = f"{settings.frontend_url}/auth/reset-password?token={raw_token}"
body = f"""
Passwort zurücksetzen
Hallo {user.first_name},
du hast eine Anfrage zum Zurücksetzen deines Passworts gestellt.
Klicke auf den Button – der Link ist 1 Stunde gültig.
Passwort zurücksetzen
Falls du diese Anfrage nicht gestellt hast, kannst du diese E-Mail ignorieren.
"""
await self._send(user.email, "Passwort zurücksetzen", _html_wrapper("Passwort zurücksetzen", body), cfg)
async def send_test(self, cfg: SmtpConfig, to: str) -> None:
"""Test-E-Mail direkt mit übergebenem Konfigurationsobjekt."""
body = f"""
SMTP-Test erfolgreich!
Deine SMTP-Konfiguration für {settings.app_name} funktioniert korrekt.
Server: {cfg.host}:{cfg.port}
"""
await self._send(to, f"{settings.app_name} – SMTP-Test", _html_wrapper("SMTP-Test", body), cfg)
email_service = EmailService()