Sanitized mirror from private repository - 2026-04-20 01:32:01 UTC
This commit is contained in:
Gitea Mirror Bot
196
scripts/lib/notify.py
Normal file
196
scripts/lib/notify.py
Normal file
@@ -0,0 +1,196 @@
|
||||
"""Notification helpers — ntfy and IMAP via Proton Bridge."""
|
||||
|
||||
import imaplib
|
||||
import logging
|
||||
import re
|
||||
import ssl
|
||||
import urllib.request
|
||||
from datetime import datetime
|
||||
from email.mime.multipart import MIMEMultipart
|
||||
from email.mime.text import MIMEText
|
||||
from html import escape
|
||||
from zoneinfo import ZoneInfo
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
SMTP_USER = "admin@thevish.io"
|
||||
SMTP_PASS = "REDACTED_PASSWORD" # pragma: allowlist secret
|
||||
DEFAULT_TO = "admin@thevish.io"
|
||||
|
||||
IMAP_HOST = "127.0.0.1"
|
||||
IMAP_PORT = 1143
|
||||
DIGEST_FOLDER = "Folders/Digests"
|
||||
|
||||
# Map subject prefixes to source labels and emoji indicators
|
||||
_SOURCE_MAP = {
|
||||
"Backup": ("Backup Validator", "#e74c3c", "\u2601\ufe0f"),
|
||||
"Disk Predictor": ("Disk Predictor", "#e67e22", "\U0001f4be"),
|
||||
"Config Drift": ("Config Drift", "#9b59b6", "\u2699\ufe0f"),
|
||||
"[Homelab]": ("Stack Monitor", "#3498db", "\U0001f433"),
|
||||
"Receipt Tracker": ("Receipt Tracker", "#27ae60", "\U0001f9fe"),
|
||||
"Subscription": ("Subscription Auditor", "#f39c12", "\U0001f4b3"),
|
||||
"Email Digest": ("Email Digest", "#2980b9", "\U0001f4e8"),
|
||||
}
|
||||
|
||||
|
||||
def _detect_source(subject: str) -> tuple[str, str, str]:
|
||||
"""Detect which script sent this based on the subject line."""
|
||||
for prefix, (label, color, icon) in _SOURCE_MAP.items():
|
||||
if prefix in subject:
|
||||
return label, color, icon
|
||||
return "Homelab Automation", "#7f8c8d", "\U0001f916"
|
||||
|
||||
|
||||
def _detect_status(subject: str) -> tuple[str, str]:
|
||||
"""Detect status from subject keywords."""
|
||||
subject_lower = subject.lower()
|
||||
if any(w in subject_lower for w in ["error", "fail", "issues found", "unsafe", "warning"]):
|
||||
return "Issue Detected", "#e74c3c"
|
||||
if any(w in subject_lower for w in ["ok", "restarted", "new"]):
|
||||
return "OK", "#27ae60"
|
||||
return "Report", "#7f8c8d"
|
||||
|
||||
|
||||
def _wrap_html(subject: str, body_html: str) -> str:
|
||||
"""Wrap content in a styled HTML email template."""
|
||||
source_label, source_color, icon = _detect_source(subject)
|
||||
status_label, status_color = _detect_status(subject)
|
||||
now = datetime.now(tz=ZoneInfo("America/Los_Angeles"))
|
||||
timestamp = now.strftime("%b %d, %Y at %I:%M %p %Z")
|
||||
|
||||
return f"""\
|
||||
<html>
|
||||
<body style="margin:0;padding:0;background:#f5f5f5;font-family:-apple-system,BlinkMacSystemFont,'Segoe UI',Roboto,sans-serif;">
|
||||
<table width="100%" cellpadding="0" cellspacing="0" style="background:#f5f5f5;padding:20px 0;">
|
||||
<tr><td align="center">
|
||||
<table width="600" cellpadding="0" cellspacing="0" style="background:#fff;border-radius:8px;overflow:hidden;box-shadow:0 2px 8px rgba(0,0,0,0.08);">
|
||||
|
||||
<!-- Header -->
|
||||
<tr><td style="background:{source_color};padding:16px 24px;">
|
||||
<span style="font-size:20px;margin-right:8px;">{icon}</span>
|
||||
<span style="color:#fff;font-size:16px;font-weight:600;">{source_label}</span>
|
||||
<span style="float:right;color:rgba(255,255,255,0.8);font-size:12px;line-height:28px;">{timestamp}</span>
|
||||
</td></tr>
|
||||
|
||||
<!-- Status bar -->
|
||||
<tr><td style="padding:12px 24px;background:{status_color}15;border-bottom:1px solid #eee;">
|
||||
<span style="color:{status_color};font-weight:600;font-size:13px;">\u25cf {status_label}</span>
|
||||
<span style="color:#666;font-size:13px;margin-left:12px;">{escape(subject)}</span>
|
||||
</td></tr>
|
||||
|
||||
<!-- Body -->
|
||||
<tr><td style="padding:20px 24px;color:#333;font-size:14px;line-height:1.6;">
|
||||
{body_html}
|
||||
</td></tr>
|
||||
|
||||
<!-- Footer -->
|
||||
<tr><td style="padding:12px 24px;background:#fafafa;border-top:1px solid #eee;text-align:center;">
|
||||
<span style="color:#999;font-size:11px;">Homelab Automation \u2022 homelab-vm \u2022 {timestamp}</span>
|
||||
</td></tr>
|
||||
|
||||
</table>
|
||||
</td></tr>
|
||||
</table>
|
||||
</body>
|
||||
</html>"""
|
||||
|
||||
|
||||
def _text_to_html(text: str) -> str:
|
||||
"""Convert plain text email body to formatted HTML."""
|
||||
escaped = escape(text)
|
||||
|
||||
# Bold section headers (lines ending with colon or lines of dashes)
|
||||
escaped = re.sub(
|
||||
r'^(.*?:)\s*$',
|
||||
r'<strong>\1</strong>',
|
||||
escaped,
|
||||
flags=re.MULTILINE,
|
||||
)
|
||||
|
||||
# Style lines starting with " -" or " *" as list items
|
||||
escaped = re.sub(
|
||||
r'^(\s+[-*])\s+(.+)$',
|
||||
r'<span style="color:#555;">\1</span> \2',
|
||||
escaped,
|
||||
flags=re.MULTILINE,
|
||||
)
|
||||
|
||||
# Highlight WARNING/ERROR/FAIL keywords
|
||||
escaped = re.sub(
|
||||
r'\b(ERROR|FAIL(?:ED)?|WARNING)\b',
|
||||
r'<span style="color:#e74c3c;font-weight:600;">\1</span>',
|
||||
escaped,
|
||||
)
|
||||
|
||||
# Highlight OK/SUCCESS/PASS keywords
|
||||
escaped = re.sub(
|
||||
r'\b(OK|SUCCESS|PASS(?:ED)?)\b',
|
||||
r'<span style="color:#27ae60;font-weight:600;">\1</span>',
|
||||
escaped,
|
||||
)
|
||||
|
||||
return f'<pre style="white-space:pre-wrap;word-wrap:break-word;font-family:\'SF Mono\',Monaco,Consolas,monospace;font-size:13px;margin:0;">{escaped}</pre>'
|
||||
|
||||
|
||||
def send_ntfy(topic: str, title: str, message: str, priority: str = "default",
|
||||
base_url: str = "https://ntfy.sh"):
|
||||
"""Send a push notification via ntfy."""
|
||||
url = f"{base_url.rstrip('/')}/{topic}"
|
||||
try:
|
||||
req = urllib.request.Request(url, data=message.encode(), headers={
|
||||
"Title": title,
|
||||
"Priority": priority,
|
||||
"Content-Type": "text/plain",
|
||||
})
|
||||
with urllib.request.urlopen(req, timeout=10):
|
||||
pass
|
||||
log.info("ntfy sent: %s", title)
|
||||
except Exception as e:
|
||||
log.warning("ntfy failed: %s", e)
|
||||
|
||||
|
||||
def send_email(subject: str, html_body: str = "", text_body: str = "",
|
||||
to: str = DEFAULT_TO, from_addr: str = SMTP_USER):
|
||||
"""Place email directly into Digests IMAP folder via Proton Bridge."""
|
||||
msg = MIMEMultipart("alternative")
|
||||
msg["Subject"] = subject
|
||||
msg["From"] = from_addr
|
||||
msg["To"] = to
|
||||
msg["Date"] = datetime.now(tz=ZoneInfo("America/Los_Angeles")).strftime(
|
||||
"%a, %d %b %Y %H:%M:%S %z"
|
||||
)
|
||||
|
||||
if text_body:
|
||||
msg.attach(MIMEText(text_body, "plain"))
|
||||
|
||||
# Build enhanced HTML: wrap provided HTML or convert plain text
|
||||
if html_body:
|
||||
wrapped = _wrap_html(subject, html_body)
|
||||
elif text_body:
|
||||
wrapped = _wrap_html(subject, _text_to_html(text_body))
|
||||
else:
|
||||
wrapped = ""
|
||||
|
||||
if wrapped:
|
||||
msg.attach(MIMEText(wrapped, "html"))
|
||||
|
||||
ctx = ssl.create_default_context()
|
||||
ctx.check_hostname = False
|
||||
ctx.verify_mode = ssl.CERT_NONE
|
||||
|
||||
imap = imaplib.IMAP4(IMAP_HOST, IMAP_PORT)
|
||||
imap.starttls(ctx)
|
||||
imap.login(SMTP_USER, SMTP_PASS)
|
||||
|
||||
# Create folder if it doesn't exist
|
||||
status, folders = imap.list()
|
||||
folder_exists = any(DIGEST_FOLDER.encode() in f for f in (folders or []))
|
||||
if not folder_exists:
|
||||
imap.create(DIGEST_FOLDER)
|
||||
log.info("Created IMAP folder: %s", DIGEST_FOLDER)
|
||||
|
||||
now = imaplib.Time2Internaldate(datetime.now(tz=ZoneInfo("UTC")))
|
||||
imap.append(DIGEST_FOLDER, "(\\Seen)", now, msg.as_bytes())
|
||||
imap.logout()
|
||||
|
||||
log.info("Email filed to Digests: %s", subject)
|
||||
Reference in New Issue
Block a user