feat(settings): add SMTP admin configuration

This commit is contained in:
Schubert Ferenc 2026-07-04 23:54:20 +02:00
parent c58e212e3a
commit 884a20e043
32 changed files with 1200 additions and 42 deletions

View file

@ -169,6 +169,10 @@ def action_title(action: str) -> str:
"repairs.public_link.regenerate": "Reparatur-Statuslink erneut erstellt",
"repairs.status_mail.sent": "Statusmail versendet",
"repairs.status_mail.failed": "Statusmail fehlgeschlagen",
"system_settings.smtp.update": "SMTP-Konfiguration geändert",
"system_settings.smtp.test_sent": "SMTP-Testmail versendet",
"system_settings.smtp.test_failed": "SMTP-Testmail fehlgeschlagen",
"system_settings.public_links.update": "Öffentliche Link-Konfiguration geändert",
}
return labels.get(action, action)

View file

@ -1,13 +1,10 @@
import smtplib
from dataclasses import dataclass
from datetime import UTC, datetime
from email.message import EmailMessage
from html import escape
from sqlalchemy.orm import Session
from starlette.requests import Request
from app.core.config import settings
from app.models.repair import Repair, RepairNotificationEvent
from app.models.user import User
from app.repositories.repair_repository import RepairRepository
@ -15,6 +12,7 @@ from app.schemas.repair import RepairNotificationOverviewResponse, RepairNotific
from app.services.audit_service import write_audit_log
from app.services.repair_public_link_service import RepairPublicLinkService
from app.services.repair_status_labels import MAIL_STATUS_LABELS, STATUS_LABELS
from app.services.system_settings_service import SystemSettingsService
@dataclass(frozen=True)
@ -98,16 +96,6 @@ def _repair_label(repair: Repair) -> str:
return f"{repair.repair_number} · {repair.customer_name}"
def _smtp_configured() -> bool:
return bool(settings.smtp_host and settings.smtp_from_email)
def _from_header() -> str:
if settings.smtp_from_name:
return f"{settings.smtp_from_name} <{settings.smtp_from_email}>"
return str(settings.smtp_from_email)
def _plain_status_mail(repair: Repair, *, status_label: str, public_status_url: str) -> str:
return (
f"Hallo {repair.customer_name},\n\n"
@ -235,8 +223,9 @@ class RepairNotificationService:
text = _plain_status_mail(repair, status_label=status_label, public_status_url=public_status_url)
html = _html_status_mail(repair, status_label=status_label, public_status_url=public_status_url)
smtp_config = SystemSettingsService.get_smtp_runtime_config(db)
if not _smtp_configured():
if not smtp_config.is_configured:
event = RepairRepository.create_notification_event(
db,
repair_id=repair.id,
@ -253,7 +242,7 @@ class RepairNotificationService:
return event
try:
RepairNotificationService._send_email(recipient=recipient, subject=subject, text=text, html=html)
SystemSettingsService.send_email(smtp_config, recipient=recipient, subject=subject, text=text, html=html)
except Exception:
event = RepairRepository.create_notification_event(
db,
@ -294,22 +283,6 @@ class RepairNotificationService:
)
return event
@staticmethod
def _send_email(*, recipient: str, subject: str, text: str, html: str) -> None:
message = EmailMessage()
message["Subject"] = subject
message["From"] = _from_header()
message["To"] = recipient
message.set_content(text)
message.add_alternative(html, subtype="html")
with smtplib.SMTP(str(settings.smtp_host), settings.smtp_port, timeout=15) as smtp:
if settings.smtp_use_tls:
smtp.starttls()
if settings.smtp_username and settings.smtp_password:
smtp.login(settings.smtp_username, settings.smtp_password)
smtp.send_message(message)
@staticmethod
def _audit_failure(db: Session, repair: Repair, *, actor: User | None, request: Request | None, reason: str) -> None:
write_audit_log(

View file

@ -19,6 +19,7 @@ from app.schemas.repair import (
)
from app.services.audit_service import write_audit_log
from app.services.repair_status_labels import STATUS_LABELS
from app.services.system_settings_service import SystemSettingsService
def normalize_token(token: str) -> str:
@ -29,10 +30,10 @@ def repair_label(repair: Repair) -> str:
return f"{repair.repair_number} · {repair.customer_name}"
def public_status_path(token: str) -> str:
def public_status_path(db: Session, token: str) -> str:
normalized_token = normalize_token(token)
path = f"/status/{normalized_token}"
base_url = (settings.public_repair_status_base_url or "").strip().rstrip("/")
base_url = SystemSettingsService.get_public_repair_status_base_url(db)
if not base_url:
return path
return f"{base_url}/{normalized_token}"
@ -109,7 +110,7 @@ class RepairPublicLinkService:
last_used_at=public_link.last_used_at,
created_at=public_link.created_at,
revoked_at=public_link.revoked_at,
public_status_path=public_status_path(token),
public_status_path=public_status_path(db, token),
)
@staticmethod

View file

@ -0,0 +1,316 @@
import smtplib
from dataclasses import dataclass
from email.message import EmailMessage
from pydantic import EmailStr
from sqlalchemy.orm import Session
from starlette.requests import Request
from app.core.config import settings
from app.models.user import User
from app.repositories.system_settings_repository import SystemSettingsRepository
from app.schemas.system_setting import (
PublicLinksSettingsResponse,
PublicLinksSettingsUpdate,
SmtpSettingsResponse,
SmtpSettingsUpdate,
SmtpTestResponse,
SettingsSource,
)
from app.services.audit_service import write_audit_log
SMTP_KEYS = (
"smtp.host",
"smtp.port",
"smtp.username",
"smtp.password",
"smtp.from_email",
"smtp.from_name",
"smtp.use_tls",
"smtp.enabled",
)
SECRET_KEYS = {"smtp.password"}
@dataclass(frozen=True)
class SmtpRuntimeConfig:
host: str
port: int
username: str
password: str
from_email: str
from_name: str
use_tls: bool
enabled: bool
source: SettingsSource
@property
def is_configured(self) -> bool:
return self.enabled and bool(self.host and self.from_email)
@property
def password_is_set(self) -> bool:
return bool(self.password)
@property
def from_header(self) -> str:
if self.from_name:
return f"{self.from_name} <{self.from_email}>"
return self.from_email
def parse_bool(value: str | None, *, default: bool = False) -> bool:
if value is None:
return default
return value.strip().lower() in {"1", "true", "yes", "on"}
def parse_int(value: str | None, *, default: int) -> int:
try:
parsed = int(str(value or "").strip())
except ValueError:
return default
if parsed < 1 or parsed > 65535:
return default
return parsed
class SystemSettingsService:
@staticmethod
def get_smtp_settings(db: Session) -> SmtpSettingsResponse:
return SystemSettingsService.smtp_response(SystemSettingsService.get_smtp_runtime_config(db))
@staticmethod
def get_smtp_runtime_config(db: Session) -> SmtpRuntimeConfig:
values = SystemSettingsService._get_values(db, SMTP_KEYS)
database_enabled = parse_bool(values.get("smtp.enabled"), default=False)
database_host = values.get("smtp.host", "")
database_from_email = values.get("smtp.from_email", "")
if database_enabled and database_host and database_from_email:
return SmtpRuntimeConfig(
host=database_host,
port=parse_int(values.get("smtp.port"), default=587),
username=values.get("smtp.username", ""),
password=values.get("smtp.password", ""),
from_email=database_from_email,
from_name=values.get("smtp.from_name", "Funktechnik Schubert") or "Funktechnik Schubert",
use_tls=parse_bool(values.get("smtp.use_tls"), default=True),
enabled=True,
source="database",
)
if settings.smtp_host and settings.smtp_from_email:
return SmtpRuntimeConfig(
host=settings.smtp_host,
port=settings.smtp_port,
username=settings.smtp_username or "",
password=settings.smtp_password or "",
from_email=settings.smtp_from_email,
from_name=settings.smtp_from_name,
use_tls=settings.smtp_use_tls,
enabled=True,
source="environment",
)
return SmtpRuntimeConfig(
host=database_host or settings.smtp_host or "",
port=parse_int(values.get("smtp.port"), default=settings.smtp_port),
username=values.get("smtp.username", "") or settings.smtp_username or "",
password=values.get("smtp.password", "") or settings.smtp_password or "",
from_email=database_from_email or settings.smtp_from_email or "",
from_name=values.get("smtp.from_name", "") or settings.smtp_from_name,
use_tls=parse_bool(values.get("smtp.use_tls"), default=settings.smtp_use_tls),
enabled=False,
source="missing",
)
@staticmethod
def update_smtp_settings(
db: Session,
payload: SmtpSettingsUpdate,
*,
actor: User,
request: Request,
) -> SmtpSettingsResponse:
current_values = SystemSettingsService._get_values(db, SMTP_KEYS)
password = payload.password if payload.password else current_values.get("smtp.password", "")
updates = {
"smtp.host": payload.host,
"smtp.port": str(payload.port),
"smtp.username": payload.username,
"smtp.password": password,
"smtp.from_email": payload.from_email,
"smtp.from_name": payload.from_name or "Funktechnik Schubert",
"smtp.use_tls": "true" if payload.use_tls else "false",
"smtp.enabled": "true" if payload.enabled else "false",
}
for key, value in updates.items():
SystemSettingsRepository.upsert(db, key=key, value=value, is_secret=key in SECRET_KEYS)
db.commit()
write_audit_log(
db,
action="system_settings.smtp.update",
entity_type="system_settings",
entity_label="SMTP-Konfiguration",
actor=actor,
request=request,
metadata={
"enabled": payload.enabled,
"host": payload.host,
"from_email": payload.from_email,
"password_changed": bool(payload.password),
},
)
return SystemSettingsService.get_smtp_settings(db)
@staticmethod
def get_public_links_settings(db: Session) -> PublicLinksSettingsResponse:
value = SystemSettingsService.get_public_repair_status_base_url(db)
source: SettingsSource = "missing"
if value:
db_value = SystemSettingsRepository.get(db, "public.repair_status_base_url")
source = "database" if db_value and db_value.value.strip() else "environment"
return PublicLinksSettingsResponse(repair_status_base_url=value, source=source)
@staticmethod
def get_public_repair_status_base_url(db: Session) -> str:
setting = SystemSettingsRepository.get(db, "public.repair_status_base_url")
if setting and setting.value.strip():
return setting.value.strip().rstrip("/")
return (settings.public_repair_status_base_url or "").strip().rstrip("/")
@staticmethod
def update_public_links_settings(
db: Session,
payload: PublicLinksSettingsUpdate,
*,
actor: User,
request: Request,
) -> PublicLinksSettingsResponse:
SystemSettingsRepository.upsert(
db,
key="public.repair_status_base_url",
value=payload.repair_status_base_url,
is_secret=False,
)
db.commit()
write_audit_log(
db,
action="system_settings.public_links.update",
entity_type="system_settings",
entity_label="Öffentliche Links",
actor=actor,
request=request,
metadata={"repair_status_base_url": payload.repair_status_base_url},
)
return SystemSettingsService.get_public_links_settings(db)
@staticmethod
def send_test_mail(
db: Session,
recipient: EmailStr,
*,
actor: User,
request: Request,
) -> SmtpTestResponse:
smtp_config = SystemSettingsService.get_smtp_runtime_config(db)
if not smtp_config.is_configured:
write_audit_log(
db,
action="system_settings.smtp.test_failed",
entity_type="system_settings",
entity_label="SMTP-Testmail",
actor=actor,
request=request,
metadata={"reason": "smtp_not_configured", "source": smtp_config.source},
)
return SmtpTestResponse(success=False, message="SMTP ist nicht vollständig konfiguriert.", source=smtp_config.source)
try:
SystemSettingsService.send_email(
smtp_config,
recipient=str(recipient),
subject="Olympus CRM SMTP-Test",
text=(
"Hallo,\n\n"
"dies ist eine Testmail aus Olympus CRM. "
"Die SMTP-Konfiguration ist grundsätzlich erreichbar.\n\n"
"Olympus CRM"
),
html=(
"<p>Hallo,</p>"
"<p>dies ist eine Testmail aus Olympus CRM. "
"Die SMTP-Konfiguration ist grundsätzlich erreichbar.</p>"
"<p>Olympus CRM</p>"
),
)
except Exception:
write_audit_log(
db,
action="system_settings.smtp.test_failed",
entity_type="system_settings",
entity_label="SMTP-Testmail",
actor=actor,
request=request,
metadata={"reason": "smtp_send_failed", "source": smtp_config.source},
)
return SmtpTestResponse(success=False, message="Testmail konnte nicht versendet werden.", source=smtp_config.source)
write_audit_log(
db,
action="system_settings.smtp.test_sent",
entity_type="system_settings",
entity_label="SMTP-Testmail",
actor=actor,
request=request,
metadata={"source": smtp_config.source},
)
return SmtpTestResponse(success=True, message="Testmail wurde versendet.", source=smtp_config.source)
@staticmethod
def send_email(
smtp_config: SmtpRuntimeConfig,
*,
recipient: str,
subject: str,
text: str,
html: str,
) -> None:
message = EmailMessage()
message["Subject"] = subject
message["From"] = smtp_config.from_header
message["To"] = recipient
message.set_content(text)
message.add_alternative(html, subtype="html")
with smtplib.SMTP(smtp_config.host, smtp_config.port, timeout=15) as smtp:
if smtp_config.use_tls:
smtp.starttls()
if smtp_config.username and smtp_config.password:
smtp.login(smtp_config.username, smtp_config.password)
smtp.send_message(message)
@staticmethod
def smtp_response(smtp_config: SmtpRuntimeConfig) -> SmtpSettingsResponse:
return SmtpSettingsResponse(
host=smtp_config.host,
port=smtp_config.port,
username=smtp_config.username,
password_is_set=smtp_config.password_is_set,
from_email=smtp_config.from_email,
from_name=smtp_config.from_name,
use_tls=smtp_config.use_tls,
enabled=smtp_config.enabled,
source=smtp_config.source,
)
@staticmethod
def _get_values(db: Session, keys: tuple[str, ...]) -> dict[str, str]:
settings_by_key = SystemSettingsRepository.get_many(db, keys)
return {key: settings_by_key[key].value.strip() for key in keys if key in settings_by_key}