Add Vorlagen editor, upload portal, onboarding, and participant import command
Some checks failed
CI/CD Pipeline / test (push) Has been cancelled
CI/CD Pipeline / deploy (push) Has been cancelled
Code Quality / quality (push) Has been cancelled

- Dokument-Vorlagen-Editor: create/edit/reset document templates (admin)
- Upload-Portal: public portal for Nachweis uploads via token
- Onboarding: invite Destinatäre via email with multi-step wizard
- Bestätigungsschreiben: preview and send confirmation letters
- Email settings: SMTP configuration UI
- Management command: import_veranstaltung_teilnehmer for bulk participant import

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
SysAdmin Agent
2026-03-21 09:25:18 +00:00
parent fdf078fa10
commit aed540fe4b
51 changed files with 5335 additions and 33 deletions

View File

@@ -418,3 +418,427 @@ def poll_emails(self, search_all_recent_days=0):
# Backward-compatible alias for existing Celery Beat schedules
poll_destinataer_emails = poll_emails
# =============================================================================
# SMTP-Ausgangs-Tasks: Nachweis-Aufforderungen und Token-Erinnerungen
# =============================================================================
import secrets # noqa: E402 (wird hier benötigt)
from datetime import timedelta # noqa: E402
def _get_smtp_connection():
"""
Erstellt eine Django-E-Mail-Verbindung mit SMTP-Einstellungen aus der DB.
"""
from django.core.mail import get_connection
from stiftung.utils.config import get_config
return get_connection(
backend="django.core.mail.backends.smtp.EmailBackend",
host=get_config("smtp_host", "smtp.ionos.de"),
port=int(get_config("smtp_port", 465)),
username=get_config("smtp_user", ""),
password=get_config("smtp_password", ""),
use_ssl=bool(get_config("smtp_use_ssl", True)),
use_tls=False,
fail_silently=False,
)
def _get_smtp_from_email():
"""Gibt die konfigurierte Absenderadresse zurück."""
from stiftung.utils.config import get_config
return get_config("smtp_from_email", "buero@vhtv-stiftung.de")
@shared_task(bind=True, max_retries=3, default_retry_delay=300)
def send_nachweis_aufforderung(self, destinataer_id, nachweis_id, base_url=None):
"""
Erstellt einen UploadToken und sendet eine Nachweis-Aufforderungs-E-Mail
mit Einmallink und QR-Code an den Destinatär.
Args:
destinataer_id: UUID des Destinatärs
nachweis_id: UUID des VierteljahresNachweises
base_url: Basis-URL der Anwendung (z.B. 'https://vhtv-stiftung.de')
"""
from django.conf import settings
from django.core.mail import EmailMultiAlternatives
from django.template.loader import render_to_string
from django.utils import timezone
import io
try:
import qrcode
from PIL import Image
import base64
qr_available = True
except ImportError:
qr_available = False
from stiftung.models import Destinataer, VierteljahresNachweis, UploadToken
try:
destinataer = Destinataer.objects.get(id=destinataer_id)
nachweis = VierteljahresNachweis.objects.get(id=nachweis_id)
except (Destinataer.DoesNotExist, VierteljahresNachweis.DoesNotExist) as exc:
logger.error("send_nachweis_aufforderung: Objekt nicht gefunden: %s", exc)
return {"status": "error", "message": str(exc)}
if not destinataer.email:
logger.warning(
"send_nachweis_aufforderung: Destinatär %s hat keine E-Mail-Adresse",
destinataer_id,
)
return {"status": "skipped", "reason": "no_email"}
# Bestehende aktive Tokens für diesen Nachweis deaktivieren
UploadToken.objects.filter(
destinataer=destinataer,
nachweis=nachweis,
ist_aktiv=True,
).update(ist_aktiv=False)
# Neuen Token erstellen
token_str = secrets.token_urlsafe(48)
gueltig_bis = timezone.now() + timedelta(days=30)
upload_token = UploadToken.objects.create(
token=token_str,
destinataer=destinataer,
nachweis=nachweis,
gueltig_bis=gueltig_bis,
)
if base_url is None:
base_url = getattr(settings, "SITE_URL", "https://vhtv-stiftung.de")
upload_url = f"{base_url}/portal/upload/{token_str}/"
# QR-Code generieren
qr_code_base64 = None
if qr_available:
try:
qr = qrcode.QRCode(
version=1,
error_correction=qrcode.constants.ERROR_CORRECT_M,
box_size=6,
border=4,
)
qr.add_data(upload_url)
qr.make(fit=True)
img = qr.make_image(fill_color="black", back_color="white")
buffer = io.BytesIO()
img.save(buffer, format="PNG")
qr_code_base64 = base64.b64encode(buffer.getvalue()).decode()
except Exception as qr_exc:
logger.warning("QR-Code-Generierung fehlgeschlagen: %s", qr_exc)
# Halbjahr bestimmen (Q1+Q2 = 1. Halbjahr, Q3+Q4 = 2. Halbjahr)
halbjahr = 1 if nachweis.quartal in [1, 2] else 2
halbjahr_label = f"{halbjahr}. Halbjahr {nachweis.jahr}"
quartal_label = f"Q{nachweis.quartal} {nachweis.jahr}"
context = {
"destinataer": destinataer,
"nachweis": nachweis,
"upload_url": upload_url,
"qr_code_base64": qr_code_base64,
"gueltig_bis": gueltig_bis,
"halbjahr_label": halbjahr_label,
"quartal_label": quartal_label,
}
subject = f"Nachweis-Aufforderung: {quartal_label} ({halbjahr_label}) vHTV-Stiftung"
from_email = _get_smtp_from_email()
to_email = destinataer.email
from stiftung.utils.vorlagen import render_vorlage
text_body = render_vorlage("email/nachweis_aufforderung.txt", context)
html_body = render_vorlage("email/nachweis_aufforderung.html", context)
try:
connection = _get_smtp_connection()
msg = EmailMultiAlternatives(subject, text_body, from_email, [to_email], connection=connection)
msg.attach_alternative(html_body, "text/html")
msg.send()
logger.info(
"Nachweis-Aufforderung gesendet an %s (Token %s)",
to_email,
upload_token.id,
)
return {
"status": "sent",
"destinataer_id": str(destinataer_id),
"nachweis_id": str(nachweis_id),
"token_id": str(upload_token.id),
}
except Exception as exc:
logger.exception("E-Mail-Versand fehlgeschlagen für %s: %s", to_email, exc)
raise self.retry(exc=exc)
@shared_task(bind=True, max_retries=3, default_retry_delay=300)
def send_nachweis_erinnerung(self, token_id, base_url=None):
"""
Sendet eine Erinnerungs-E-Mail für einen bald ablaufenden Upload-Token.
Wird durch Celery Beat ausgelöst (7 Tage vor Ablauf).
"""
from django.conf import settings
from django.core.mail import EmailMultiAlternatives
from django.template.loader import render_to_string
from stiftung.models import UploadToken
try:
upload_token = UploadToken.objects.select_related(
"destinataer", "nachweis"
).get(id=token_id, ist_aktiv=True)
except UploadToken.DoesNotExist:
return {"status": "skipped", "reason": "token_not_found_or_inactive"}
if not upload_token.ist_gueltig():
return {"status": "skipped", "reason": "token_invalid"}
if not upload_token.destinataer.email:
return {"status": "skipped", "reason": "no_email"}
if base_url is None:
base_url = getattr(settings, "SITE_URL", "https://vhtv-stiftung.de")
upload_url = f"{base_url}/portal/upload/{upload_token.token}/"
nachweis = upload_token.nachweis
halbjahr = 1 if nachweis.quartal in [1, 2] else 2
halbjahr_label = f"{halbjahr}. Halbjahr {nachweis.jahr}"
context = {
"destinataer": upload_token.destinataer,
"nachweis": nachweis,
"upload_url": upload_url,
"gueltig_bis": upload_token.gueltig_bis,
"halbjahr_label": halbjahr_label,
"ist_erinnerung": True,
}
subject = f"Erinnerung: Nachweis-Upload noch ausstehend {halbjahr_label}"
from_email = _get_smtp_from_email()
to_email = upload_token.destinataer.email
from stiftung.utils.vorlagen import render_vorlage
text_body = render_vorlage("email/nachweis_aufforderung.txt", context)
html_body = render_vorlage("email/nachweis_aufforderung.html", context)
try:
connection = _get_smtp_connection()
msg = EmailMultiAlternatives(subject, text_body, from_email, [to_email], connection=connection)
msg.attach_alternative(html_body, "text/html")
msg.send()
upload_token.erinnerung_gesendet = True
upload_token.save(update_fields=["erinnerung_gesendet"])
logger.info("Erinnerung gesendet an %s (Token %s)", to_email, token_id)
return {"status": "sent", "token_id": str(token_id)}
except Exception as exc:
logger.exception("Erinnerungs-E-Mail fehlgeschlagen für %s: %s", to_email, exc)
raise self.retry(exc=exc)
@shared_task(bind=True, max_retries=3, default_retry_delay=300)
def send_onboarding_einladung(self, einladung_id, base_url=None):
"""
Sendet eine Onboarding-Einladungs-E-Mail an eine neue potenzielle Destinatärin/
einen neuen potenziellen Destinatär.
Args:
einladung_id: UUID der OnboardingEinladung
base_url: Basis-URL der Anwendung
"""
from django.conf import settings
from django.core.mail import EmailMultiAlternatives
from django.template.loader import render_to_string
from stiftung.models import OnboardingEinladung
try:
einladung = OnboardingEinladung.objects.get(id=einladung_id)
except OnboardingEinladung.DoesNotExist as exc:
logger.error("send_onboarding_einladung: Einladung %s nicht gefunden", einladung_id)
return {"status": "error", "message": str(exc)}
if not einladung.ist_gueltig():
return {"status": "skipped", "reason": "einladung_ungueltig"}
if base_url is None:
base_url = getattr(settings, "SITE_URL", "https://vhtv-stiftung.de")
onboarding_url = f"{base_url}/portal/onboarding/{einladung.token}/"
context = {
"einladung": einladung,
"onboarding_url": onboarding_url,
"gueltig_bis": einladung.gueltig_bis,
}
subject = "Einladung zum Onboarding van Hees-Theyssen-Vogel'sche Stiftung"
from_email = _get_smtp_from_email()
to_email = einladung.email
from stiftung.utils.vorlagen import render_vorlage
text_body = render_vorlage("email/onboarding_einladung.txt", context)
html_body = render_vorlage("email/onboarding_einladung.html", context)
try:
connection = _get_smtp_connection()
msg = EmailMultiAlternatives(subject, text_body, from_email, [to_email], connection=connection)
msg.attach_alternative(html_body, "text/html")
msg.send()
logger.info(
"Onboarding-Einladung gesendet an %s (Einladung %s)",
to_email,
einladung_id,
)
return {"status": "sent", "einladung_id": str(einladung_id), "email": to_email}
except Exception as exc:
logger.exception("Onboarding-E-Mail-Versand fehlgeschlagen für %s: %s", to_email, exc)
raise self.retry(exc=exc)
@shared_task(bind=True, max_retries=3, default_retry_delay=300)
def send_bestaetigung(self, destinataer_id, base_url=None):
"""
Generiert ein Bestätigungsschreiben (PDF) für einen Destinatär und sendet es
per E-Mail. Das PDF wird zusätzlich im DMS unter Kontext "korrespondenz" abgelegt.
Args:
destinataer_id: UUID des Destinatärs
base_url: Basis-URL der Anwendung (für Konsistenz mit anderen Tasks)
"""
from decimal import Decimal
from django.core.files.base import ContentFile
from django.template.loader import render_to_string
from django.core.mail import EmailMultiAlternatives
from django.utils import timezone
from stiftung.models import Destinataer, DestinataerUnterstuetzung, DokumentDatei
try:
destinataer = Destinataer.objects.get(id=destinataer_id)
except Destinataer.DoesNotExist as exc:
logger.error("send_bestaetigung: Destinatär %s nicht gefunden", destinataer_id)
return {"status": "error", "message": str(exc)}
if not destinataer.email:
logger.warning("send_bestaetigung: Destinatär %s hat keine E-Mail-Adresse", destinataer_id)
return {"status": "skipped", "reason": "no_email"}
# Alle abgeschlossenen Unterstützungen laden
unterstuetzungen = list(DestinataerUnterstuetzung.objects.filter(
destinataer=destinataer,
status__in=["ausgezahlt", "abgeschlossen"],
).order_by("faellig_am"))
gesamtbetrag = sum(u.betrag for u in unterstuetzungen) if unterstuetzungen else Decimal("0")
zeitraum = None
if unterstuetzungen:
erste = unterstuetzungen[0].faellig_am
letzte = unterstuetzungen[-1].faellig_am
if erste == letzte:
zeitraum = erste.strftime("%d.%m.%Y")
else:
zeitraum = f"{erste.strftime('%d.%m.%Y')} {letzte.strftime('%d.%m.%Y')}"
betrag_quartal = destinataer.vierteljaehrlicher_betrag
betrag_jaehrlich = (betrag_quartal * 4) if betrag_quartal else None
zweck = destinataer.ausbildungsstand or (destinataer.get_berufsgruppe_display() if destinataer.berufsgruppe else None)
datum = timezone.now().date()
context = {
"destinataer": destinataer,
"unterstuetzungen": unterstuetzungen,
"gesamtbetrag": gesamtbetrag,
"datum": datum,
"zeitraum": zeitraum,
"betrag_quartal": betrag_quartal,
"betrag_jaehrlich": betrag_jaehrlich,
"zweck": zweck,
}
# PDF generieren via WeasyPrint
pdf_bytes = None
try:
from weasyprint import HTML
from stiftung.utils.vorlagen import render_vorlage
html_content = render_vorlage("pdf/bestaetigung.html", context)
pdf_bytes = HTML(string=html_content).write_pdf()
except Exception as exc:
logger.error("send_bestaetigung: PDF-Generierung fehlgeschlagen: %s", exc)
raise self.retry(exc=exc)
# PDF im DMS ablegen
filename = (
f"bestaetigung_{destinataer.nachname}_{destinataer.vorname}"
f"_{datum.strftime('%Y%m%d')}.pdf"
)
try:
doc = DokumentDatei(
titel=f"Bestätigungsschreiben {datum.strftime('%d.%m.%Y')} {destinataer.get_full_name()}",
beschreibung="Automatisch generiertes Bestätigungsschreiben über Förderleistungen.",
kontext="korrespondenz",
dateiname_original=filename,
dateityp="application/pdf",
dateigroesse=len(pdf_bytes),
destinataer=destinataer,
)
doc.datei.save(filename, ContentFile(pdf_bytes), save=False)
doc.save()
logger.info("Bestätigung im DMS gespeichert (ID: %s).", doc.pk)
except Exception as exc:
logger.error("send_bestaetigung: DMS-Speicherung fehlgeschlagen: %s", exc)
# Weiter mit E-Mail-Versand auch wenn DMS-Speicherung schlägt fehl
# E-Mail senden
from stiftung.utils.vorlagen import render_vorlage
html_body = render_vorlage("email/bestaetigung.html", context)
subject = "Bestätigung Ihrer Stiftungsförderung van Hees-Theyssen-Vogel'sche Stiftung"
from_email = _get_smtp_from_email()
to_email = destinataer.email
try:
connection = _get_smtp_connection()
msg = EmailMultiAlternatives(subject, "", from_email, [to_email], connection=connection)
msg.attach_alternative(html_body, "text/html")
if pdf_bytes:
msg.attach(filename, pdf_bytes, "application/pdf")
msg.send()
logger.info("Bestätigung gesendet an %s (Destinatär %s)", to_email, destinataer_id)
return {"status": "sent", "destinataer_id": str(destinataer_id), "email": to_email}
except Exception as exc:
logger.exception("send_bestaetigung: E-Mail-Versand fehlgeschlagen für %s: %s", to_email, exc)
raise self.retry(exc=exc)
@shared_task
def check_ablaufende_tokens():
"""
Prüft täglich Upload-Tokens, die in 7 Tagen ablaufen,
und sendet Erinnerungs-E-Mails (falls noch nicht gesendet).
Wird durch Celery Beat aufgerufen.
"""
from django.utils import timezone
from stiftung.models import UploadToken
grenze = timezone.now() + timedelta(days=7)
tokens = UploadToken.objects.filter(
ist_aktiv=True,
eingeloest_am__isnull=True,
erinnerung_gesendet=False,
gueltig_bis__lte=grenze,
gueltig_bis__gt=timezone.now(),
)
count = 0
for token in tokens:
send_nachweis_erinnerung.delay(str(token.id))
count += 1
logger.info("check_ablaufende_tokens: %d Erinnerungen angestoßen", count)
return {"triggered": count}