222 lines
7.0 KiB
Python
222 lines
7.0 KiB
Python
"""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 (
|
|
"<html><body><p>Este correo fue enviado automáticamente en Brevo.</p>"
|
|
"<p>Puedes modificar HTML_BODY_FILE en el .env para usar tu plantilla.</p></body></html>"
|
|
)
|
|
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()
|