diff --git a/README.md b/README.md index 62993b5..b4fb7d4 100644 --- a/README.md +++ b/README.md @@ -869,6 +869,26 @@ Add: There is no need to run `source`, activate the virtual environment, or use a wrapper script. +## Sending an ntfy notification + +`ntfy_notify.py` provides both a command-line interface and a reusable Python +function. Send a message from the command line with: + +```bash +/opt/ups/.venv/bin/python /opt/ups/ntfy_notify.py "Some text" +``` + +Use it as a Python library with: + +```python +from ntfy_notify import send_notification + +send_notification("Some text") +``` + +The default topic can be overridden with `--topic-url` or the +`NTFY_TOPIC_URL` environment variable. + After a few minutes: ```bash diff --git a/ntfy_notify.py b/ntfy_notify.py new file mode 100755 index 0000000..8c18172 --- /dev/null +++ b/ntfy_notify.py @@ -0,0 +1,95 @@ +#!/usr/bin/env python3 + +"""Send a text notification to the Home Alert ntfy topic.""" + +import argparse +import os +import sys +from urllib.error import HTTPError, URLError +from urllib.request import Request, urlopen + + +DEFAULT_TOPIC_URL = ( + "https://ntfy.sh/VK20_Home_Alert_______4671938674123049873459" +) + + +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", + DEFAULT_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 parse_arguments(): + """Parse command-line arguments.""" + + parser = argparse.ArgumentParser( + description="Send a text message to the Home Alert ntfy topic.", + ) + parser.add_argument( + "message", + nargs="+", + help="notification text", + ) + parser.add_argument( + "--topic-url", + help="override the ntfy topic URL", + ) + parser.add_argument( + "--timeout", + type=float, + default=10, + help="request timeout in seconds (default: 10)", + ) + return parser.parse_args() + + +def main(): + args = parse_arguments() + + try: + response = send_notification( + " ".join(args.message), + topic_url=args.topic_url, + timeout=args.timeout, + ) + print(response) + return 0 + + except HTTPError as exc: + print( + f"ntfy returned HTTP {exc.code}: {exc.reason}", + file=sys.stderr, + ) + except URLError as exc: + print(f"Could not reach ntfy: {exc.reason}", file=sys.stderr) + except (TypeError, ValueError) as exc: + print(f"Invalid notification: {exc}", file=sys.stderr) + + return 1 + + +if __name__ == "__main__": + raise SystemExit(main())