from __future__ import annotations import base64 from datetime import datetime from email.message import EmailMessage from html import escape import json import smtplib from urllib.error import HTTPError, URLError from urllib.request import Request, urlopen from app.config import settings from app.schemas.booking import BookingRequestCreate class MailConfigurationError(RuntimeError): pass class MailDeliveryError(RuntimeError): pass def _build_booking_subject(payload: BookingRequestCreate) -> str: return f"{settings.booking_email_subject_prefix} Nuova richiesta visita - {payload.name}" def _build_booking_html(payload: BookingRequestCreate) -> str: submitted_at = datetime.now().strftime("%d/%m/%Y %H:%M") name = escape(payload.name) email = escape(payload.email) phone = escape(payload.phone) pet_name = escape(payload.pet_name or "-") pet_type = escape(payload.pet_type) doctor = escape(payload.doctor) service = escape(payload.service) date = escape(payload.date) time = escape(payload.time or "Qualsiasi orario") notes = escape(payload.notes or "-") return f"""
È stata inviata una nuova richiesta di prenotazione non vincolante dal sito web.
| Nome e cognome | {name} |
| {email} | |
| Telefono | {phone} |
| Nome animale | {pet_name} |
| Tipo animale | {pet_type} |
| Medico richiesto | {doctor} |
| Tipo di visita | {service} |
| Periodo preferito | {date} |
| Fascia oraria preferita | {time} |
| Note | {notes} |
| Inviata il | {submitted_at} |
Questa richiesta non costituisce conferma automatica dell'appuntamento e richiede presa in carico da parte dello staff.
""" def _build_booking_text(payload: BookingRequestCreate) -> str: submitted_at = datetime.now().strftime("%d/%m/%Y %H:%M") return ( "Nuova richiesta di prenotazione visita\n\n" f"Nome e cognome: {payload.name}\n" f"Email: {payload.email}\n" f"Telefono: {payload.phone}\n" f"Nome animale: {payload.pet_name or '-'}\n" f"Tipo animale: {payload.pet_type}\n" f"Medico richiesto: {payload.doctor}\n" f"Tipo di visita: {payload.service}\n" f"Periodo preferito: {payload.date}\n" f"Fascia oraria preferita: {payload.time or 'Qualsiasi orario'}\n" f"Note: {payload.notes or '-'}\n" f"Inviata il: {submitted_at}\n\n" "La richiesta non costituisce conferma automatica dell'appuntamento." ) def send_booking_request_email(payload: BookingRequestCreate) -> None: if not settings.booking_mail_enabled: raise MailConfigurationError(f"Configurazione email incompleta per provider {settings.mail_provider}.") if settings.mail_provider == "mailjet": _send_booking_request_email_mailjet(payload) return if settings.mail_provider == "smtp": _send_booking_request_email_smtp(payload) return raise MailConfigurationError(f"Provider email non supportato: {settings.mail_provider}") def _send_booking_request_email_smtp(payload: BookingRequestCreate) -> None: if not settings.booking_mail_enabled: raise MailConfigurationError("Configurazione SMTP incompleta.") message = EmailMessage() message["Subject"] = _build_booking_subject(payload) message["From"] = settings.booking_email_from message["To"] = settings.booking_email_to message["Reply-To"] = payload.email message.set_content(_build_booking_text(payload)) message.add_alternative(_build_booking_html(payload), subtype="html") with smtplib.SMTP(settings.smtp_host, settings.smtp_port, timeout=30) as server: if settings.smtp_use_tls: server.starttls() server.login(settings.smtp_username, settings.smtp_password) server.send_message(message) def _send_booking_request_email_mailjet(payload: BookingRequestCreate) -> None: mailjet_payload = { "Messages": [ { "From": { "Email": settings.booking_email_from, "Name": settings.booking_email_from_name, }, "To": [ { "Email": settings.booking_email_to, "Name": "Clinica Veterinaria Formiginese", } ], "ReplyTo": { "Email": payload.email, "Name": payload.name, }, "Subject": _build_booking_subject(payload), "TextPart": _build_booking_text(payload), "HTMLPart": _build_booking_html(payload), } ] } if settings.mailjet_sandbox_mode: mailjet_payload["SandboxMode"] = True auth_token = base64.b64encode( f"{settings.mailjet_api_key}:{settings.mailjet_api_secret}".encode("utf-8") ).decode("ascii") request = Request( settings.mailjet_api_url, data=json.dumps(mailjet_payload).encode("utf-8"), headers={ "Content-Type": "application/json", "Accept": "application/json", "Authorization": f"Basic {auth_token}", }, method="POST", ) try: with urlopen(request, timeout=30) as response: response_body = response.read().decode("utf-8") response_status = response.status except HTTPError as exc: error_body = exc.read().decode("utf-8", errors="replace") raise MailDeliveryError(f"Mailjet API HTTP {exc.code}: {error_body[:500]}") from exc except URLError as exc: raise MailDeliveryError(f"Mailjet API connection error: {exc.reason}") from exc if response_status < 200 or response_status >= 300: raise MailDeliveryError(f"Mailjet API HTTP {response_status}: {response_body[:500]}") try: response_payload = json.loads(response_body) except json.JSONDecodeError as exc: raise MailDeliveryError(f"Mailjet API response non JSON: {response_body[:500]}") from exc messages = response_payload.get("Messages", []) message_status = messages[0].get("Status") if messages else None if message_status != "success": raise MailDeliveryError(f"Mailjet API delivery status {message_status}: {response_body[:500]}")