113 lines
3.7 KiB
Python
113 lines
3.7 KiB
Python
"""Business logic for contact submissions."""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import smtplib
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
from email.message import EmailMessage
|
|
from typing import Any, Dict, Tuple
|
|
|
|
from .. import settings
|
|
from ..database import save_contact
|
|
from ..metrics import record_submission
|
|
from ..utils import is_valid_email
|
|
|
|
|
|
@dataclass
|
|
class ContactSubmission:
|
|
name: str
|
|
email: str
|
|
company: str | None
|
|
message: str
|
|
timeline: str | None
|
|
created_at: str = datetime.now(timezone.utc).isoformat()
|
|
|
|
|
|
def validate_submission(raw: Dict[str, Any]) -> Tuple[ContactSubmission | None, Dict[str, str]]:
|
|
"""Validate the incoming payload and return a submission object."""
|
|
name = (raw.get("name") or "").strip()
|
|
email = (raw.get("email") or "").strip()
|
|
message = (raw.get("message") or "").strip()
|
|
consent = raw.get("consent")
|
|
company = (raw.get("company") or "").strip()
|
|
|
|
errors: Dict[str, str] = {}
|
|
if not name:
|
|
errors["name"] = "Name is required."
|
|
elif len(name) > 200:
|
|
errors["name"] = "Name is too long (max 200 chars)."
|
|
if not is_valid_email(email):
|
|
errors["email"] = "Valid email is required."
|
|
if not message:
|
|
errors["message"] = "Message is required."
|
|
elif len(message) > 5000:
|
|
errors["message"] = "Message is too long (max 5000 chars)."
|
|
if not consent:
|
|
errors["consent"] = "Consent is required."
|
|
if company and len(company) > 200:
|
|
errors["company"] = "Organisation name is too long (max 200 chars)."
|
|
|
|
if errors:
|
|
return None, errors
|
|
|
|
submission = ContactSubmission(
|
|
name=name,
|
|
email=email,
|
|
company=company or None,
|
|
message=message,
|
|
timeline=(raw.get("timeline") or "").strip() or None,
|
|
)
|
|
return submission, {}
|
|
|
|
|
|
def persist_submission(submission: ContactSubmission) -> int:
|
|
"""Persist the submission and update metrics."""
|
|
record_id = save_contact(submission)
|
|
record_submission()
|
|
return record_id
|
|
|
|
|
|
def send_notification(submission: ContactSubmission) -> bool:
|
|
"""Send an email notification for the submission if SMTP is configured."""
|
|
if not settings.SMTP_SETTINGS["host"] or not settings.SMTP_SETTINGS["recipients"]:
|
|
logging.info("SMTP not configured; skipping email notification")
|
|
return False
|
|
|
|
sender = settings.SMTP_SETTINGS["sender"] or "no-reply@example.com"
|
|
recipients = settings.SMTP_SETTINGS["recipients"]
|
|
|
|
msg = EmailMessage()
|
|
msg["Subject"] = f"Neue Kontaktanfrage von {submission.name}"
|
|
msg["From"] = sender
|
|
msg["To"] = ", ".join(recipients)
|
|
msg.set_content(
|
|
"\n".join(
|
|
[
|
|
f"Name: {submission.name}",
|
|
f"E-Mail: {submission.email}",
|
|
f"Organisation: {submission.company or '—'}",
|
|
f"Zeithorizont: {submission.timeline or '—'}",
|
|
"",
|
|
"Nachricht:",
|
|
submission.message,
|
|
"",
|
|
f"Eingang: {submission.created_at}",
|
|
]
|
|
)
|
|
)
|
|
|
|
try:
|
|
with smtplib.SMTP(settings.SMTP_SETTINGS["host"], settings.SMTP_SETTINGS["port"], timeout=15) as server:
|
|
if settings.SMTP_SETTINGS["use_tls"]:
|
|
server.starttls()
|
|
if settings.SMTP_SETTINGS["username"]:
|
|
server.login(
|
|
settings.SMTP_SETTINGS["username"], settings.SMTP_SETTINGS["password"] or "")
|
|
server.send_message(msg)
|
|
logging.info("Notification email dispatched to %s", recipients)
|
|
return True
|
|
except Exception as exc: # pragma: no cover - SMTP failures are logged only
|
|
logging.error("Failed to send notification email: %s", exc)
|
|
return False
|