""" Email SMTP sender – invio notifiche via SMTP con TLS/SSL o STARTTLS. Config non sensibile (config): { "smtp_host": "smtp.example.com", "smtp_port": 465, "smtp_use_tls": true, # SSL/TLS diretto (porta 465) "smtp_use_starttls": false, # STARTTLS (porta 587) – alternativo a use_tls "from_email": "noreply@example.com", "from_name": "PEChub Notifiche", "to_email": "destinatario@example.com" } Config sensibile (config_enc → config_secret): { "smtp_password": "..." } Dipendenza: aiosmtplib (gia' in backend/pyproject.toml) """ from datetime import datetime from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText import aiosmtplib DEFAULT_TIMEOUT = 15.0 class EmailSMTPError(Exception): """Errore durante l'invio di un'email di notifica.""" def __init__(self, message: str, smtp_code: int | None = None): super().__init__(message) self.smtp_code = smtp_code async def send_email_notification( smtp_host: str, smtp_port: int, smtp_user: str, smtp_password: str, from_email: str, to_email: str, subject: str, body_text: str, body_html: str | None = None, from_name: str = "PEChub Notifiche", use_tls: bool = True, use_starttls: bool = False, timeout: float = DEFAULT_TIMEOUT, ) -> None: """ Invia un'email di notifica via SMTP. Args: smtp_host: host SMTP smtp_port: porta SMTP smtp_user: username autenticazione smtp_password: password autenticazione from_email: indirizzo mittente to_email: indirizzo destinatario subject: oggetto email body_text: testo plain body_html: testo HTML (opzionale) from_name: nome visualizzato mittente use_tls: usa SSL/TLS diretto (porta 465) use_starttls: usa STARTTLS (porta 587) – alternativo a use_tls timeout: timeout connessione in secondi Raises: EmailSMTPError: in caso di errori di autenticazione, connessione o invio """ msg = MIMEMultipart("alternative") msg["Subject"] = subject msg["From"] = f"{from_name} <{from_email}>" if from_name else from_email msg["To"] = to_email msg["Date"] = datetime.utcnow().strftime("%a, %d %b %Y %H:%M:%S +0000") msg["X-Mailer"] = "PEChub/1.0" msg.attach(MIMEText(body_text, "plain", "utf-8")) if body_html: msg.attach(MIMEText(body_html, "html", "utf-8")) try: await aiosmtplib.send( msg, hostname=smtp_host, port=smtp_port, username=smtp_user, password=smtp_password, use_tls=use_tls, start_tls=use_starttls, timeout=timeout, ) except aiosmtplib.SMTPAuthenticationError as exc: raise EmailSMTPError( f"Autenticazione SMTP fallita per {smtp_user}@{smtp_host}: {exc}", smtp_code=535, ) from exc except aiosmtplib.SMTPConnectError as exc: raise EmailSMTPError( f"Connessione SMTP fallita a {smtp_host}:{smtp_port}: {exc}" ) from exc except aiosmtplib.SMTPServerDisconnected as exc: raise EmailSMTPError( f"Server SMTP {smtp_host} ha chiuso la connessione: {exc}" ) from exc except aiosmtplib.SMTPException as exc: raise EmailSMTPError(f"Errore SMTP: {exc}") from exc except Exception as exc: raise EmailSMTPError(f"Errore invio email: {exc}") from exc async def send_test_email( smtp_host: str, smtp_port: int, smtp_user: str, smtp_password: str, from_email: str, to_email: str, channel_name: str = "PEChub", from_name: str = "PEChub Notifiche", use_tls: bool = True, use_starttls: bool = False, ) -> None: """ Invia un'email di test per verificare la configurazione del canale. Raises: EmailSMTPError: se la connessione o l'autenticazione falliscono """ ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S") subject = f"[PEChub] Test canale email: {channel_name}" body_text = ( f"PEChub – Test canale Email\n\n" f"Il canale '{channel_name}' e' configurato correttamente.\n\n" f"Data/ora: {ts}\n" f"Destinatario: {to_email}" ) body_html = ( f"
Il canale {channel_name} e' configurato correttamente.
" f"Data/ora: {ts}
"
f"Destinatario: {to_email}
Inviato da PEChub Notification Engine
" ) await send_email_notification( smtp_host=smtp_host, smtp_port=smtp_port, smtp_user=smtp_user, smtp_password=smtp_password, from_email=from_email, to_email=to_email, subject=subject, body_text=body_text, body_html=body_html, from_name=from_name, use_tls=use_tls, use_starttls=use_starttls, )