Files
hms-licht-ton/backend/app/services/email_service.py
T

111 lines
3.8 KiB
Python
Raw Normal View History

"""Email service using aiosmtplib for contact and rental confirmation emails."""
import logging
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from typing import Any
import aiosmtplib
from app.config import get_settings
logger = logging.getLogger(__name__)
settings = get_settings()
class EmailService:
"""Send transactional emails via SMTP."""
async def send_email(
self,
to_addr: str,
subject: str,
html_body: str,
text_body: str = "",
reply_to: str | None = None,
) -> bool:
"""Send an email via aiosmtplib. Returns True on success."""
msg = MIMEMultipart("alternative")
msg["From"] = settings.smtp_from
msg["To"] = to_addr
msg["Subject"] = subject
if reply_to:
msg["Reply-To"] = reply_to
msg.attach(MIMEText(text_body or html_body, "plain"))
msg.attach(MIMEText(html_body, "html"))
try:
await aiosmtplib.send(
msg,
hostname=settings.smtp_host,
port=settings.smtp_port,
username=settings.smtp_user,
password=settings.smtp_password,
start_tls=True,
)
return True
except Exception as exc:
logger.error("Email send failed to %s: %s", to_addr, exc)
return False
async def send_contact_email(self, contact_data: dict[str, Any]) -> bool:
"""Send contact form data to info@hms-licht-ton.de."""
html = f"""
<h2>Neue Kontaktanfrage</h2>
<p><strong>Name:</strong> {contact_data.get('name', '')}</p>
<p><strong>Email:</strong> {contact_data.get('email', '')}</p>
<p><strong>Telefon:</strong> {contact_data.get('phone', '')}</p>
<p><strong>Nachricht:</strong></p>
<p>{contact_data.get('message', '')}</p>
"""
text = (
f"Neue Kontaktanfrage\n"
f"Name: {contact_data.get('name', '')}\n"
f"Email: {contact_data.get('email', '')}\n"
f"Telefon: {contact_data.get('phone', '')}\n"
f"Nachricht: {contact_data.get('message', '')}\n"
)
return await self.send_email(
to_addr=settings.smtp_from,
subject="Neue Kontaktanfrage ueber hms-licht-ton.de",
html_body=html,
text_body=text,
reply_to=contact_data.get("email"),
)
async def send_rental_confirmation(
self,
to_addr: str,
reference_number: str,
event_name: str,
items: list[dict[str, Any]],
) -> bool:
"""Send rental request confirmation to customer."""
items_html = "".join(
f"<li>{item.get('equipment_name', 'Unbekannt')} - Menge: {item.get('quantity', 1)}</li>"
for item in items
)
html = f"""
<h2>Mietanfrage bestätigt {reference_number}</h2>
<p>Vielen Dank für Ihre Mietanfrage bei HMS Licht &amp; Ton.</p>
<p><strong>Veranstaltung:</strong> {event_name}</p>
<p><strong>Referenznummer:</strong> {reference_number}</p>
<h3>Artikel:</h3>
<ul>{items_html}</ul>
<p>Wir melden uns in Kürze bei Ihnen.</p>
<p>Ihr HMS Licht &amp; Ton Team</p>
"""
text = (
f"Mietanfrage bestaetigt - {reference_number}\n"
f"Veranstaltung: {event_name}\n"
f"Referenznummer: {reference_number}\n"
f"Artikel:\n"
+ "\n".join(
f"- {item.get('equipment_name', 'Unbekannt')}: {item.get('quantity', 1)}"
for item in items
)
)
return await self.send_email(
to_addr=to_addr,
subject=f"Mietanfrage bestaetigt {reference_number}",
html_body=html,
text_body=text,
)