[bin] Add email checker for hungryroot

bin/.bin/hungryroot-notifications

@@ -0,0 +1,224 @@
+#!/usr/bin/env -S uv run --script
+# /// script
+# requires-python = ">=3.11"
+# dependencies = [
+#   "requests>=2.31.0",
+# ]
+# ///
+
+import email
+import imaplib
+import os
+import ssl
+import time
+import re
+from email.header import decode_header, make_header
+
+import requests
+
+FORWARDED_HEADER_RE = re.compile(
+    r"^(From:.*?\n(?:.*\n)*?\n)",
+    re.IGNORECASE
+)
+
+TITLE_PREFIX_RE = re.compile(
+    r"""
+    ^\s*
+    (?:fwd:|fw:)\s*                 # optional forward prefix
+    (?:\[hungryroot/[^\]]+\]\s*)*   # one or more [hungryroot/*] blocks
+    """,
+    re.IGNORECASE | re.VERBOSE,
+)
+
+GITHUB_URL_RE = re.compile(r"https://github\.com/[^\s>,]+")
+
+FOOTER_START_MARKERS = (
+    "reply to this email directly",
+    "you are receiving this because",
+)
+
+IMAP_HOST = os.environ.get("IMAP_HOST", "box.unbl.ink")
+IMAP_USER = os.environ.get("IMAP_USER", "hungryroot@unbl.ink")
+IMAP_PASS = os.environ.get("IMAP_PASS", "Mainr0ot")
+
+NTFY_BASE = os.environ.get("NTFY_BASE", "https://ntfy.unbl.ink")
+NTFY_TOPIC = os.environ.get("NTFY_TOPIC", "hroot-notifications")
+POLL_SECONDS = int(os.environ.get("POLL_SECONDS", "20"))
+
+# Optional: if your ntfy needs auth
+NTFY_USER = os.environ.get("NTFY_USER")
+NTFY_PASS = os.environ.get("NTFY_PASS")
+NTFY_TOKEN = os.environ.get("NTFY_TOKEN")    # alternative to user/pass
+
+MAX_BODY_CHARS = int(os.environ.get("MAX_BODY_CHARS", "500"))
+
+
+def _decode(value: str) -> str:
+    if not value:
+        return ""
+    try:
+        return str(make_header(decode_header(value)))
+    except Exception:
+        return value
+
+
+def _extract_text(msg: email.message.Message) -> str:
+    """Best-effort plain text extraction."""
+    if msg.is_multipart():
+        for part in msg.walk():
+            ctype = part.get_content_type()
+            disp = (part.get("Content-Disposition") or "").lower()
+            if ctype == "text/plain" and "attachment" not in disp:
+                payload = part.get_payload(decode=True) or b""
+                charset = part.get_content_charset() or "utf-8"
+                return payload.decode(charset, errors="replace")
+        return ""
+    payload = msg.get_payload(decode=True) or b""
+    charset = msg.get_content_charset() or "utf-8"
+    return payload.decode(charset, errors="replace")
+
+
+def _first_github_link(text: str) -> str | None:
+    # crude but effective; improve later if you want
+    for token in text.split():
+        if token.startswith("https://github.com/"):
+            return token.strip(")>.,;\"'")
+    return None
+
+def _sanitize_header(value: str) -> str:
+    # Collapse CR/LF and excessive whitespace into single spaces
+    return " ".join(value.replace("\r", " ").replace("\n", " ").split())
+
+def normalize_title(title: str) -> str:
+    # Remove CR/LF and collapse whitespace first (safe for headers)
+    title = " ".join(title.replace("\r", " ").replace("\n", " ").split())
+    return TITLE_PREFIX_RE.sub("", title).strip()
+
+def strip_github_footer(text: str) -> tuple[str, str | None]:
+    """
+    Strip GitHub's email footer and return (clean_text, github_url).
+    """
+    lines = text.splitlines()
+    clean_lines = []
+    github_url = None
+    in_footer = False
+
+    for line in lines:
+        lower = line.lower()
+
+        if not in_footer and any(m in lower for m in FOOTER_START_MARKERS):
+            in_footer = True
+            continue
+
+        if in_footer:
+            if github_url is None:
+                match = GITHUB_URL_RE.search(line)
+                if match:
+                    github_url = match.group(0)
+            continue
+
+        clean_lines.append(line)
+
+    clean_text = "\n".join(clean_lines).rstrip()
+    return clean_text, github_url
+
+
+def strip_forwarded_headers(text: str) -> str:
+    """
+    Strip a forwarded-message header block, including folded headers.
+    """
+    lines = text.splitlines()
+    out = []
+    in_headers = True
+
+    for line in lines:
+        if in_headers:
+            # End of headers = first completely empty line
+            if line.strip() == "":
+                in_headers = False
+            # Still in headers: skip
+            continue
+        out.append(line)
+
+    return "\n".join(out).lstrip()
+
+def notify_ntfy(title: str, message: str, url: str | None = None):
+    endpoint = f"{NTFY_BASE.rstrip('/')}/{NTFY_TOPIC}"
+
+    safe_title = normalize_title(title)
+
+    headers = {
+        "Title": safe_title,
+        "Tags": "github",
+        "Priority": "4",
+    }
+    if url:
+        headers["Click"] = url
+
+    auth = None
+    if NTFY_TOKEN:
+        headers["Authorization"] = f"Bearer {NTFY_TOKEN}"
+    elif NTFY_USER and NTFY_PASS:
+        auth = (NTFY_USER, NTFY_PASS)
+
+    r = requests.post(
+        endpoint,
+        data=message.encode("utf-8"),
+        headers=headers,
+        auth=auth,
+        timeout=10,
+    )
+    r.raise_for_status()
+
+
+def fetch_unseen():
+    context = ssl.create_default_context()
+    with imaplib.IMAP4_SSL(IMAP_HOST, ssl_context=context) as M:
+        M.login(IMAP_USER, IMAP_PASS)
+        M.select("INBOX")
+
+        typ, data = M.search(None, "UNSEEN")
+        if typ != "OK":
+            return []
+
+        uids = data[0].split()
+        out = []
+
+        for uid in uids:
+            typ, msg_data = M.fetch(uid, "(RFC822)")
+            if typ != "OK" or not msg_data or not msg_data[0]:
+                continue
+
+            raw = msg_data[0][1]
+            msg = email.message_from_bytes(raw)
+
+            subject = _decode(msg.get("Subject", "GitHub notification"))
+            from_ = _decode(msg.get("From", ""))
+
+
+            body = strip_forwarded_headers(_extract_text(msg))
+            body, github_url = strip_github_footer(body)
+
+            out.append((uid, subject, from_, body, github_url))
+
+            # mark seen so we don't resend
+            M.store(uid, "+FLAGS", "\\Seen")
+
+        return out
+
+
+def main():
+    while True:
+        try:
+            items = fetch_unseen()
+            for _uid, subject, from_, snippet, github_url in items:
+                title = subject
+                msg = f"{snippet}\n\nFrom: {from_}"
+                notify_ntfy(title=title, message=msg, url=github_url)
+        except Exception as e:
+            print(f"[hungryroot-notifications] error: {e}")
+        time.sleep(POLL_SECONDS)
+
+
+if __name__ == "__main__":
+    main()

bin/.bin/hungryroot-notifications-wrapper

@@ -0,0 +1,14 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+# Load secret from pass
+export IMAP_PASS="$(pass work/hungryroot@unbl.ink)"
+
+# Optional sanity check
+if [[ -z "${IMAP_PASS}" ]]; then
+  echo "IMAP_PASS is empty" >&2
+  exit 1
+fi
+
+exec "$HOME/.bin/hungryroot-notifications"
+

systemd/.config/systemd/user/hungryroot-notifications.service

@@ -0,0 +1,24 @@
+[Unit]
+Description=Hungryroot GitHub email -> ntfy notifications
+After=network-online.target
+Wants=network-online.target
+
+[Service]
+Type=simple
+ExecStart=%h/.bin/hungryroot-notifications-wrapper
+Restart=always
+RestartSec=5
+
+Environment=IMAP_HOST=imap.unbl.ink
+Environment=IMAP_USER=hungryroot@unbl.ink
+
+Environment=NTFY_BASE=https://ntfy.unbl.ink
+Environment=NTFY_TOPIC=hroot-notifications
+Environment=POLL_SECONDS=20
+
+StandardOutput=journal
+StandardError=journal
+
+[Install]
+WantedBy=default.target
+