68 lines
2.1 KiB
Python
68 lines
2.1 KiB
Python
# api/notify.py
|
|
import os, smtplib
|
|
from email.mime.text import MIMEText
|
|
from email.utils import formataddr
|
|
from dotenv import load_dotenv
|
|
|
|
# Load .env before reading anything
|
|
load_dotenv(os.getenv("ENV_FILE", ".env"))
|
|
|
|
_TRUES = {"1", "true", "yes", "on"}
|
|
|
|
def _b(s: str | None, default=False) -> bool:
|
|
if s is None:
|
|
return default
|
|
return str(s).strip().lower() in _TRUES
|
|
|
|
def _first(*names: str, default: str | None = None) -> str | None:
|
|
for n in names:
|
|
v = os.getenv(n)
|
|
if v is not None and str(v).strip() != "":
|
|
return v
|
|
return default
|
|
|
|
def _cfg():
|
|
"""Merge MAIL_* and SMTP_* envs; MAIL_* wins if both present."""
|
|
enabled = _b(_first("MAIL_ENABLED"), default=False)
|
|
host = _first("MAIL_HOST", "SMTP_HOST")
|
|
port = int(_first("MAIL_PORT", "SMTP_PORT", default="0") or "0")
|
|
user = _first("MAIL_USER", "SMTP_USER")
|
|
pwd = _first("MAIL_PASSWORD", "SMTP_PASS")
|
|
mail_from = _first("MAIL_FROM", default=user)
|
|
mail_to = _first("MAIL_TO_DEFAULT", "MAIL_TO")
|
|
tls = _b(_first("MAIL_TLS"), default=False)
|
|
return {
|
|
"enabled": enabled,
|
|
"host": host, "port": port, "user": user, "pwd": pwd,
|
|
"from": mail_from, "to": mail_to, "tls": tls
|
|
}
|
|
|
|
def send_email(subject: str, body_text: str, to_addr: str | None = None):
|
|
cfg = _cfg()
|
|
if not cfg["enabled"]:
|
|
print("[notify] MAIL_ENABLED is not true; skipping email")
|
|
return
|
|
|
|
host, port, user, pwd = cfg["host"], cfg["port"], cfg["user"], cfg["pwd"]
|
|
mail_from = cfg["from"]
|
|
mail_to = to_addr or cfg["to"]
|
|
|
|
if not (host and port and user and pwd and mail_to):
|
|
print("[notify] missing SMTP config; skipping email")
|
|
return
|
|
|
|
msg = MIMEText(body_text, "plain", "utf-8")
|
|
msg["Subject"] = subject
|
|
msg["From"] = formataddr(("Intesa Logs Agent", mail_from))
|
|
msg["To"] = mail_to
|
|
|
|
with smtplib.SMTP(host, port, timeout=20) as s:
|
|
if cfg["tls"]:
|
|
try:
|
|
s.starttls()
|
|
except smtplib.SMTPException:
|
|
pass
|
|
s.login(user, pwd)
|
|
s.send_message(msg)
|
|
print(f"[notify] sent email to {mail_to}")
|