home_control/lib/_notify.py

#!/usr/bin/env python3
"""Send text notifications to a configured ntfy topic."""

import argparse
import logging
import os
import sys
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen


LOG = logging.getLogger("home_control.notify")

def send_notification(message, topic_url=None, timeout=10):
    """Send *message* to ntfy and return the server response body."""
    if not isinstance(message, str):
        raise TypeError("message must be a string")
    if not message:
        raise ValueError("message must not be empty")
    url = topic_url or os.environ.get("NTFY_TOPIC_URL")
    if not url:
        raise ValueError("topic_url is required (or set NTFY_TOPIC_URL)")
    request = Request(
        url,
        data=message.encode("utf-8"),
        headers={"Content-Type": "text/plain; charset=utf-8"},
        method="POST",
    )
    with urlopen(request, timeout=timeout) as response:
        return response.read().decode("utf-8")


def _filter_values(config: dict, mode: str, field: str) -> set[str]:
    value = config.get("filter", {}).get(mode, {}).get(field, [])
    if not isinstance(value, list) or any(not isinstance(item, str) for item in value):
        raise ValueError(f"notify.filter.{mode}.{field} must be a list of strings")
    return set(value)


def accepts_event(config: dict, message: dict) -> bool:
    """Return whether an event passes configured event and ID filters."""
    accepted_events = _filter_values(config, "accept", "events")
    accepted_ids = _filter_values(config, "accept", "ids")
    ignored_events = _filter_values(config, "ignore", "events")
    ignored_ids = _filter_values(config, "ignore", "ids")

    if message["event"] in ignored_events or message["id"] in ignored_ids:
        return False
    if accepted_events and message["event"] not in accepted_events:
        return False
    if accepted_ids and message["id"] not in accepted_ids:
        return False
    return True


def on_event(config: dict, message: dict) -> None:
    """Generic controller action for event dictionaries."""
    if not config.get("enabled", True):
        return
    if not accepts_event(config, message):
        LOG.debug(
            "notification ignored by filter: event=%s id=%s",
            message["event"],
            message["id"],
        )
        return
    messages = config.get("messages", {})
    template = messages.get(message["event"], config.get("message", "{sender}: {event} ({id})"))
    text = message["text"] or str(template).format(**message)
    try:
        send_notification(
            text,
            topic_url=config.get("topic_url") or None,
            timeout=float(config.get("timeout", 10)),
        )
    except (OSError, RuntimeError, ValueError) as error:
        LOG.error(
            "notification failed for %s/%s/%s: %s",
            message["sender"],
            message["event"],
            message["id"],
            error,
        )


def parse_arguments():
    parser = argparse.ArgumentParser(description="Send a text message to an ntfy topic.")
    parser.add_argument("message", nargs="+", help="notification text")
    parser.add_argument("--topic-url", help="ntfy topic URL (or use NTFY_TOPIC_URL)")
    parser.add_argument("--timeout", type=float, default=10, help="request timeout in seconds")
    return parser.parse_args()


def main():
    args = parse_arguments()
    try:
        print(
            send_notification(
                " ".join(args.message),
                topic_url=args.topic_url,
                timeout=args.timeout,
            )
        )
        return 0
    except HTTPError as error:
        print(f"ntfy returned HTTP {error.code}: {error.reason}", file=sys.stderr)
    except URLError as error:
        print(f"Could not reach ntfy: {error.reason}", file=sys.stderr)
    except (TypeError, ValueError) as error:
        print(f"Invalid notification: {error}", file=sys.stderr)
    return 1


if __name__ == "__main__":
    raise SystemExit(main())