"""Envía correos mediante el servidor SMTP de Brevo usando variables desde .env.""" from __future__ import annotations import argparse import os import smtplib from email.message import EmailMessage from pathlib import Path from typing import Iterable, Optional, Dict from LokiLogger import LokiLogger BASE_DIR = Path(__file__).resolve().parent ENV_FILE = BASE_DIR / ".env" def load_env_file(path: Path) -> None: """Carga pares KEY=VALUE desde un archivo .env a os.environ si no existen.""" if not path.exists(): return for raw_line in path.read_text(encoding="utf-8").splitlines(): line = raw_line.strip() if not line or line.startswith("#"): continue if "=" not in line: continue key, _, value = line.partition("=") key = key.strip() value = value.strip().strip('"').strip("'") os.environ.setdefault(key, value) def must_env(key: str) -> str: value = os.getenv(key) if value: return value raise SystemExit(f"Falta la variable de entorno requerida: {key}") def parse_recipients(raw: str) -> list[str]: recipients = [item.strip() for item in raw.split(",") if item.strip()] if not recipients: raise SystemExit("Lista de destinatarios vacía después de procesar MAIL_TO/--to") return recipients def resolve_path(path_str: str) -> Path: path = Path(path_str) if not path.is_absolute(): path = (BASE_DIR / path).resolve() return path def read_html_body(html_file: str | None, logger: Optional[LokiLogger] = None) -> str: if not html_file: # Plantilla mínima en caso de no disponer de archivo return ( "

Este correo fue enviado automáticamente en Brevo.

" "

Puedes modificar HTML_BODY_FILE en el .env para usar tu plantilla.

" ) path = resolve_path(html_file) if logger: logger.debug("Leyendo archivo HTML para cuerpo", add_fields={"html_path": str(path)}) try: return path.read_text(encoding="utf-8") except FileNotFoundError as exc: if logger: logger.error("Archivo HTML no encontrado", add_fields={"html_path": str(path)}) raise SystemExit(f"No se encontró el archivo HTML indicado: {path}") from exc def build_message( subject: str, html_body: str, sender: str, recipients: Iterable[str], logger: Optional[LokiLogger] = None, ) -> EmailMessage: message = EmailMessage() message["Subject"] = subject message["From"] = sender message["To"] = ", ".join(recipients) message.set_content("Este correo requiere un visor HTML compatible.") message.add_alternative(html_body, subtype="html") if logger: logger.debug( "Mensaje construido", add_fields={"subject": subject, "sender": sender, "recipients": list(recipients)}, ) return message def send_via_brevo( message: EmailMessage, user: str, password: str, server: str, port: int, logger: Optional[LokiLogger] = None, ) -> None: if logger: logger.info( "Conectando a servidor SMTP", add_fields={"smtp_server": server, "smtp_port": port, "sender": user}, ) with smtplib.SMTP(server, port) as smtp: smtp.starttls() smtp.login(user, password) smtp.send_message(message) if logger: logger.info("Correo enviado vía Brevo", add_fields={"smtp_server": server}) def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description="Envía un correo usando el SMTP de Brevo") parser.add_argument("--to", help="Direcciones destino separadas por coma. Si se omite se usa MAIL_TO del .env") parser.add_argument("--subject", help="Asunto a enviar. Si se omite se usa DEFAULT_SUBJECT del .env") parser.add_argument( "--html-file", help="Ruta al archivo HTML que se enviará. Si se omite se usa HTML_BODY_FILE del .env", ) return parser def collect_loki_labels(env: Dict[str, str]) -> Dict[str, str]: labels: Dict[str, str] = {} for key, value in env.items(): if key.startswith("LOKI_LABEL_"): labels[key.replace("LOKI_LABEL_", "").lower()] = value return labels def configure_logger() -> LokiLogger: endpoint = os.getenv("LOKI_ENDPOINT", "http://127.0.0.1:3101/loki/api/v1/push") min_level = os.getenv("LOKI_MIN_LEVEL", "INFO") service_name = os.getenv("LOKI_SERVICE_NAME", "brevo-mailer") labels = collect_loki_labels(os.environ) logger = LokiLogger( endpoint=endpoint, min_level=min_level, service_name=service_name, add_labels=labels or None, ) logger.debug( "LokiLogger inicializado", add_fields={ "endpoint": endpoint, "min_level": min_level, "service_name": service_name, "labels": labels, }, ) return logger def main() -> None: load_env_file(ENV_FILE) logger: Optional[LokiLogger] = None try: logger = configure_logger() except Exception as exc: print(f"No se pudo inicializar LokiLogger: {exc}") parser = build_parser() args = parser.parse_args() smtp_server = os.getenv("SMTP_SERVER", "smtp-relay.brevo.com") smtp_port = int(os.getenv("SMTP_PORT", "587")) smtp_user = must_env("SMTP_USER") smtp_pass = must_env("SMTP_PASS") sender = os.getenv("MAIL_FROM", smtp_user) if logger: logger.info( "Variables SMTP cargadas", add_fields={"smtp_server": smtp_server, "smtp_port": smtp_port, "sender": sender}, ) recipients_raw = args.to or os.getenv("MAIL_TO") if not recipients_raw: if logger: logger.error("No se definieron destinatarios para el envío") raise SystemExit("Configura MAIL_TO en el .env o usa --to para indicar destinatarios.") recipients = parse_recipients(recipients_raw) if logger: logger.info("Destinatarios preparados", add_fields={"recipients": recipients}) subject = args.subject or os.getenv("DEFAULT_SUBJECT") or "Envío automático con Brevo" html_file = args.html_file or os.getenv("HTML_BODY_FILE") if logger: logger.info( "Detalle del contenido a enviar", add_fields={"subject": subject, "html_file": html_file}, ) html_body = read_html_body(html_file, logger=logger) try: message = build_message(subject, html_body, sender, recipients, logger=logger) send_via_brevo( message, smtp_user, smtp_pass, smtp_server, smtp_port, logger=logger, ) except Exception as exc: if logger: logger.exception(exc, add_fields={"subject": subject}) raise print(f"Correo enviado a {', '.join(recipients)}") if logger: logger.info("Proceso completado correctamente", add_fields={"recipients": recipients}) if __name__ == "__main__": main()