#!/usr/bin/env python3
"""
notifier.py - Sends a single classified notification email.
Formats a human-readable HTML + text email based on:
- correlation outcome (SUCCESS / PARTIAL / REPO_FAILURE)
- per-module statuses
- repository check results (if run)
"""
import logging
import smtplib
from datetime import datetime, timezone
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from typing import Optional
log = logging.getLogger(__name__)
OUTCOME_EMOJI = {
"SUCCESS": "OK",
"PARTIAL": "WARNING",
"REPO_FAILURE": "CRITICAL",
}
OUTCOME_COLOR = {
"SUCCESS": "#2e7d32",
"PARTIAL": "#e65100",
"REPO_FAILURE": "#b71c1c",
}
def _build_text(correlation: dict, repo_status: Optional[dict]) -> str:
outcome = correlation["outcome"]
lines = [
f"NS8 Backup Monitor - {OUTCOME_EMOJI[outcome]}: {outcome}",
f"Time: {datetime.now(timezone.utc).isoformat()}",
f"Plans checked: {', '.join(correlation.get('backup_ids', []))}",
f"Modules: {correlation['succeeded']} OK / {correlation['failed']} FAILED / {correlation['total']} total",
"",
]
if correlation["failed_modules"]:
lines.append("Failed modules:")
for m in correlation["failed_modules"]:
lines.append(f" - [{m['module_id']}] backup_id={m['backup_id']}: {m.get('error', 'unknown error')}")
lines.append("")
if repo_status:
lines.append("Repository check:")
for dest in repo_status.get("destinations", []):
lines.append(f" - [{dest['repo_id']}] {dest['status']}: {dest.get('error', '')}")
if repo_status.get("note"):
lines.append(f" NOTE: {repo_status['note']}")
lines.append("")
if correlation.get("note"):
lines.append(f"Note: {correlation['note']}")
return "\n".join(lines)
def _build_html(correlation: dict, repo_status: Optional[dict]) -> str:
outcome = correlation["outcome"]
color = OUTCOME_COLOR[outcome]
label = OUTCOME_EMOJI[outcome]
ts = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC")
plan_ids = ", ".join(correlation.get("backup_ids", [])) or "N/A"
rows = ""
for m in correlation.get("modules", []):
bg = "#e8f5e9" if m["result"] == "success" else "#ffebee"
icon = "✓" if m["result"] == "success" else "✗"
rows += (
f''
f'| {icon} | '
f'{m["module_id"]} | '
f'{m["backup_id"]} | '
f'{m.get("timestamp", "")} | '
f'{m.get("error", "") or ""} | '
'
'
)
repo_section = ""
if repo_status:
repo_rows = ""
for dest in repo_status.get("destinations", []):
bg = "#e8f5e9" if dest["status"] == "OK" else "#ffebee"
repo_rows += (
f''
f'| {dest["repo_id"]} | '
f'{dest["status"]} | '
f'{dest.get("error", "")} | '
'
'
)
if repo_rows:
repo_section = f"""
Repository check
| Repo ID |
Status |
Detail |
{repo_rows}
"""
if repo_status.get("note"):
repo_section += f'{repo_status["note"]}
'
html = f"""
NS8 Backup Monitor — {label}: {outcome}
{ts} • Plans: {plan_ids}
{correlation['succeeded']} module(s) OK |
{correlation['failed']} FAILED |
{correlation['total']} total
|
Module |
Backup ID |
Timestamp |
Error |
{rows}
{repo_section}
"""
return html
def send_notification(
config: dict,
alerts: list,
correlation: dict,
repo_status: Optional[dict] = None
):
outcome = correlation["outcome"]
mail_cfg = config.get("mail", {})
smtp_cfg = config.get("smtp", {})
mail_from = mail_cfg.get("from", "ns8-backup-monitor@localhost")
mail_to = mail_cfg.get("to", [])
subject_prefix = mail_cfg.get("subject_prefix", "[NS8 Backup]")
subject = f"{subject_prefix} {OUTCOME_EMOJI[outcome]}: {outcome} - {datetime.now(timezone.utc).strftime('%Y-%m-%d')}"
text_body = _build_text(correlation, repo_status)
html_body = _build_html(correlation, repo_status)
msg = MIMEMultipart("alternative")
msg["Subject"] = subject
msg["From"] = mail_from
msg["To"] = ", ".join(mail_to)
msg.attach(MIMEText(text_body, "plain"))
msg.attach(MIMEText(html_body, "html"))
host = smtp_cfg.get("host", "localhost")
port = smtp_cfg.get("port", 25)
use_tls = smtp_cfg.get("use_tls", False)
use_starttls = smtp_cfg.get("use_starttls", False)
username = smtp_cfg.get("username", "")
password = smtp_cfg.get("password", "")
try:
if use_tls:
smtp = smtplib.SMTP_SSL(host, port)
else:
smtp = smtplib.SMTP(host, port)
if use_starttls:
smtp.starttls()
if username and password:
smtp.login(username, password)
smtp.sendmail(mail_from, mail_to, msg.as_string())
smtp.quit()
log.info(f"Notification sent: {subject} -> {mail_to}")
except Exception as e:
log.error(f"Failed to send notification: {e}")