"""
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#{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#{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_substitute_notification(
self, substitute: "User", requester: "User", absence, db: AsyncSession
) -> None:
"""Informiert die eingetragene Vertretung über eine genehmigte Abwesenheit."""
cfg = await self._load_smtp(substitute.company_id, db)
start = absence.start_date.strftime("%d.%m.%Y")
end = absence.end_date.strftime("%d.%m.%Y")
zeitraum = start if start == end else f"{start} – {end}"
body = f"""
Du wurdest als Vertretung eingetragen
Hallo {substitute.first_name},
{requester.full_name} ist im Zeitraum {zeitraum}
abwesend und hat dich als Vertretung benannt.
Abwesenheiten ansehen
"""
await self._send(
substitute.email,
f"Vertretung für {requester.full_name} ({zeitraum})",
_html_wrapper("Vertretung", body),
cfg,
)
async def send_pending_approvals_digest(
self, approver: "User", items: list[dict], db: AsyncSession
) -> None:
"""Tägliche Zusammenfassung offener Anträge an eine genehmigende Person."""
cfg = await self._load_smtp(approver.company_id, db)
rows = "".join(
f"{i['user_name']} – {i['type_name']} "
f"({i['start']}{' – ' + i['end'] if i['end'] != i['start'] else ''}, "
f"{i['working_days']} Tag(e))"
for i in items
)
body = f"""
Offene Abwesenheitsanträge
Hallo {approver.first_name}, es warten {len(items)} Anträge auf deine Genehmigung:
Anträge prüfen
"""
await self._send(
approver.email, f"{len(items)} offene Abwesenheitsanträge",
_html_wrapper("Offene Anträge", body), cfg,
)
async def send_carryover_expiry_reminder(
self, user: "User", remaining: int, expires_at, db: AsyncSession
) -> None:
cfg = await self._load_smtp(user.company_id, db)
datum = expires_at.strftime("%d.%m.%Y")
body = f"""
Resturlaub verfällt bald
Hallo {user.first_name}, du hast noch {remaining} Urlaubstage,
die am {datum} verfallen.
Plane deinen Urlaub rechtzeitig ein, damit nichts verloren geht.
Urlaub planen
"""
await self._send(
user.email, f"Dein Resturlaub verfällt am {datum}",
_html_wrapper("Resturlaub", body), cfg,
)
async def send_certificate_overdue_digest(
self, hr_user: "User", items: list[dict], db: AsyncSession
) -> None:
cfg = await self._load_smtp(hr_user.company_id, db)
rows = "".join(
f"{i['user_name']} – krank seit {i['start']}, AU fällig seit {i['due']}"
for i in items
)
body = f"""
Fehlende AU-Bescheinigungen
Hallo {hr_user.first_name}, für {len(items)} Krankmeldung(en)
fehlt die Arbeitsunfähigkeitsbescheinigung:
Abwesenheiten ansehen
"""
await self._send(
hr_user.email, f"{len(items)} fehlende AU-Bescheinigung(en)",
_html_wrapper("AU fehlt", 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()