Sanitized mirror from private repository - 2026-04-16 07:19:56 UTC

scripts/lib/notify.py

@@ -0,0 +1,95 @@
+"""Notification helpers — ntfy and SMTP via Proton Bridge."""
+
+import imaplib
+import logging
+import smtplib
+import ssl
+import urllib.request
+from datetime import datetime
+from email.mime.multipart import MIMEMultipart
+from email.mime.text import MIMEText
+from zoneinfo import ZoneInfo
+
+log = logging.getLogger(__name__)
+
+SMTP_HOST = "127.0.0.1"
+SMTP_PORT = 1025
+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"
+
+
+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 _file_to_digests(msg_bytes: bytes):
+    """File a copy of the message into the Digests folder via Proton Bridge IMAP."""
+    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)
+
+    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, None, now, msg_bytes)
+    imap.logout()
+    log.info("Filed message to %s folder", DIGEST_FOLDER)
+
+
+def send_email(subject: str, html_body: str = "", text_body: str = "",
+               to: str = DEFAULT_TO, from_addr: str = SMTP_USER):
+    """Send email via Proton Bridge SMTP on localhost, then file into Digests folder."""
+    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"))
+    if html_body:
+        msg.attach(MIMEText(html_body, "html"))
+
+    ctx = ssl.create_default_context()
+    ctx.check_hostname = False
+    ctx.verify_mode = ssl.CERT_NONE
+
+    with smtplib.SMTP(SMTP_HOST, SMTP_PORT) as server:
+        server.starttls(context=ctx)
+        server.login(SMTP_USER, SMTP_PASS)
+        server.send_message(msg)
+
+    log.info("Email sent: %s -> %s", subject, to)
+
+    try:
+        _file_to_digests(msg.as_bytes())
+    except Exception as e:
+        log.warning("Failed to file to Digests folder: %s", e)