rss-ntfy/app/ntfy.py

"""Publishing notifications to an ntfy server."""
from __future__ import annotations

from urllib.parse import quote

import httpx


def _topic_url(server: str, topic: str) -> str:
    server = (server or "https://ntfy.sh").rstrip("/")
    return f"{server}/{quote(topic.strip('/'))}"


def _auth_headers(token: str, username: str, password: str) -> dict[str, str]:
    """Build an Authorization header for private ntfy servers."""
    if token.strip():
        return {"Authorization": f"Bearer {token.strip()}"}
    if username.strip():
        import base64

        raw = f"{username}:{password}".encode()
        return {"Authorization": "Basic " + base64.b64encode(raw).decode()}
    return {}


async def publish(
    *,
    server: str,
    topic: str,
    title: str,
    message: str,
    click: str = "",
    tags: str = "",
    priority: int = 3,
    attach: str = "",
    token: str = "",
    username: str = "",
    password: str = "",
    markdown: bool = False,
    actions: list[dict] | None = None,
) -> None:
    """Send one notification to ntfy. Raises httpx.HTTPStatusError on failure.

    Title and click URLs must be ASCII (ntfy header limitation), so non-ASCII
    titles are pushed into the body and the title is best-effort stripped.
    """
    url = _topic_url(server, topic)
    headers: dict[str, str] = {"Priority": str(priority)}
    headers.update(_auth_headers(token, username, password))

    ascii_title = title.encode("ascii", "ignore").decode().strip()
    if ascii_title:
        headers["Title"] = ascii_title
    elif title.strip():
        # Title had only non-ASCII chars — prepend it to the body instead.
        message = f"{title}\n\n{message}"

    if click:
        try:
            click.encode("ascii")
            headers["Click"] = click
        except UnicodeEncodeError:
            pass
    if tags.strip():
        clean = ",".join(t.strip() for t in tags.split(",") if t.strip())
        if clean:
            headers["Tags"] = clean
    if attach:
        try:
            attach.encode("ascii")
            headers["Attach"] = attach
        except UnicodeEncodeError:
            pass
    if markdown:
        headers["Content-Type"] = "text/markdown"
    if actions:
        parts = []
        for a in actions:
            act = a.get("action", "view")
            label = a.get("label", "")
            url = a.get("url", "")
            clear = ", clear=true" if a.get("clear") else ""
            parts.append(f"{act}, {label}, {url}{clear}")
        headers["Actions"] = "; ".join(parts)

    async with httpx.AsyncClient(timeout=20) as client:
        resp = await client.post(url, content=message.encode("utf-8"), headers=headers)
        resp.raise_for_status()