dotfiles/bin/.bin/hungryroot-notifications

#!/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
)

PREFIX_RE = re.compile(r"^\s*(?:(?:re|fw|fwd)\s*:\s*)+", re.IGNORECASE)
HUNGRYROOT_REPOS_RE = re.compile(r"^\s*(?:\[hungryroot/[^\]]+\]\s*)+", re.IGNORECASE)

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", "")

NTFY_BASE = os.environ.get("NTFY_BASE", "https://ntfy.unbl.ink")
NTFY_TOPIC = os.environ.get("NTFY_TOPIC", "hroot-notifications")
NTFY_ERROR_TOPIC = os.environ.get("NTFY_ERROR_TOPIC", "errors")
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:
    # Unfold/sanitize any CRLF + weird spacing from forwarded headers
    title = " ".join(title.replace("\r", " ").replace("\n", " ").split())

    # Strip Re:/Fwd: chains like "Re: Re: Fwd:"
    title = PREFIX_RE.sub("", title).strip()

    # Strip ONLY hungryroot repo prefix blocks, keep [BE-####] etc.
    title = HUNGRYROOT_REPOS_RE.sub("", title).strip()

    return title

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 notify_error(err: str):
    endpoint = f"{NTFY_BASE.rstrip('/')}/{NTFY_ERROR_TOPIC}"
    headers = {
        "Title": "hungryroot-notifications error",
        "Tags": "warning",
        "Priority": "5",
    }

    auth = None
    if NTFY_TOKEN:
        headers["Authorization"] = f"Bearer {NTFY_TOKEN}"
    elif NTFY_USER and NTFY_PASS:
        auth = (NTFY_USER, NTFY_PASS)

    # Keep it short so it doesn't spam
    body = err.strip()
    if len(body) > 1500:
        body = body[:1500] + "…"

    r = requests.post(endpoint, data=body.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:
            msg = f"[hungryroot-notifications] {type(e).__name__}: {e}"
            print(msg)
            try:
                notify_error(msg)
            except Exception as e2:
                print(f"[hungryroot-notifications] failed to notify error: {e2}")
        time.sleep(POLL_SECONDS)


if __name__ == "__main__":
    main()