fn_registry/python/functions/infra/smtp_send.py

"""Envio de email via SMTP con smtplib — sin dependencias externas."""

import smtplib
from dataclasses import dataclass, field
from email.mime.base import MIMEBase
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email import encoders


@dataclass(frozen=True)
class EmailAttachmentPy:
    """Archivo adjunto a un email."""
    filename: str
    content_type: str  # MIME type, ej: "application/pdf"
    data: bytes


@dataclass(frozen=True)
class SMTPConfigPy:
    """Parametros de conexion SMTP.

    tls_mode puede ser:
        "tls"      — TLS directo (port 465)
        "starttls" — upgrade STARTTLS (port 587)
        ""         — sin cifrado (port 25)
    """
    host: str
    port: int
    username: str = ""
    password: str = ""
    tls_mode: str = "starttls"  # "tls", "starttls" o ""


def smtp_send(
    cfg: SMTPConfigPy,
    from_addr: str,
    to: list[str],
    subject: str,
    body_html: str = "",
    body_text: str = "",
    cc: list[str] | None = None,
    bcc: list[str] | None = None,
    attachments: list[EmailAttachmentPy] | None = None,
    headers: dict[str, str] | None = None,
) -> None:
    """Conecta al servidor SMTP, envia el email y cierra la conexion.

    Construye el mensaje MIME segun el contenido:
    - Solo texto: text/plain
    - Solo HTML: text/html
    - Ambos: multipart/alternative
    - Con adjuntos: multipart/mixed

    Args:
        cfg: Configuracion SMTP (host, port, usuario, password, tls_mode).
        from_addr: Direccion del remitente.
        to: Lista de destinatarios principales.
        subject: Asunto del correo.
        body_html: Cuerpo HTML (opcional).
        body_text: Cuerpo texto plano (opcional).
        cc: Lista de destinatarios en copia (opcional).
        bcc: Lista de destinatarios en copia oculta (opcional).
        attachments: Lista de adjuntos EmailAttachmentPy (opcional).
        headers: Headers MIME adicionales como diccionario (opcional).

    Raises:
        RuntimeError: Si la conexion SMTP falla, la autenticacion es incorrecta
                      o el envio no se puede completar.
    """
    cc = cc or []
    bcc = bcc or []
    attachments = attachments or []
    headers = headers or {}

    msg = _build_mime(from_addr, to, cc, subject, body_html, body_text, attachments, headers)

    all_recipients = list(to) + list(cc) + list(bcc)

    try:
        if cfg.tls_mode == "tls":
            smtp = smtplib.SMTP_SSL(cfg.host, cfg.port)
        else:
            smtp = smtplib.SMTP(cfg.host, cfg.port)

        with smtp:
            if cfg.tls_mode == "starttls":
                smtp.starttls()
            if cfg.username:
                smtp.login(cfg.username, cfg.password)
            smtp.sendmail(from_addr, all_recipients, msg.as_string())

    except smtplib.SMTPException as e:
        raise RuntimeError(f"smtp_send: SMTP error: {e}") from e
    except OSError as e:
        raise RuntimeError(f"smtp_send: connection error to {cfg.host}:{cfg.port}: {e}") from e


def _build_mime(
    from_addr: str,
    to: list[str],
    cc: list[str],
    subject: str,
    body_html: str,
    body_text: str,
    attachments: list[EmailAttachmentPy],
    headers: dict[str, str],
) -> MIMEMultipart | MIMEText:
    """Construye la estructura MIME del mensaje."""
    has_text = bool(body_text)
    has_html = bool(body_html)
    has_atts = bool(attachments)

    if has_atts:
        outer = MIMEMultipart("mixed")
        _set_headers(outer, from_addr, to, cc, subject, headers)
        body_part = _build_body_part(body_html, body_text, has_html, has_text)
        outer.attach(body_part)
        for att in attachments:
            outer.attach(_build_attachment(att))
        return outer

    if has_html and has_text:
        alt = MIMEMultipart("alternative")
        _set_headers(alt, from_addr, to, cc, subject, headers)
        alt.attach(MIMEText(body_text, "plain", "utf-8"))
        alt.attach(MIMEText(body_html, "html", "utf-8"))
        return alt

    if has_html:
        msg = MIMEText(body_html, "html", "utf-8")
        _set_headers(msg, from_addr, to, cc, subject, headers)
        return msg

    msg = MIMEText(body_text or "", "plain", "utf-8")
    _set_headers(msg, from_addr, to, cc, subject, headers)
    return msg


def _build_body_part(
    body_html: str, body_text: str, has_html: bool, has_text: bool
) -> MIMEBase:
    if has_html and has_text:
        alt = MIMEMultipart("alternative")
        alt.attach(MIMEText(body_text, "plain", "utf-8"))
        alt.attach(MIMEText(body_html, "html", "utf-8"))
        return alt
    if has_html:
        return MIMEText(body_html, "html", "utf-8")
    return MIMEText(body_text, "plain", "utf-8")


def _set_headers(msg: MIMEBase, from_addr: str, to: list[str], cc: list[str], subject: str, extra: dict[str, str]) -> None:
    msg["From"] = from_addr
    msg["To"] = ", ".join(to)
    if cc:
        msg["Cc"] = ", ".join(cc)
    msg["Subject"] = subject
    for k, v in extra.items():
        msg[k] = v


def _build_attachment(att: EmailAttachmentPy) -> MIMEBase:
    maintype, subtype = att.content_type.split("/", 1) if "/" in att.content_type else ("application", "octet-stream")
    part = MIMEBase(maintype, subtype)
    part.set_payload(att.data)
    encoders.encode_base64(part)
    part.add_header("Content-Disposition", "attachment", filename=att.filename)
    return part