Implement Gotify notifcations

README.md

@@ -49,11 +49,13 @@ Every time an upgrade is performed, the pacman output is stored into /var/log/pa
 
 ## Notification
 When configuring your notification system, please note that pacroller will not send any notification if stdin is a tty (can be overridden by the `--interactive` switch).
-Notification will be sent through all configured methods when it requires manual inspection. Currently, two notification methods are supported: SMTP and telegram
+Notification will be sent through all configured methods when it requires manual inspection. Currently, two notification methods are supported: SMTP, telegram, and Gotify
 ### SMTP
 Configure `/etc/pacroller/smtp.json` to receive email notifications.
 ### Telegram
 Configure `/etc/pacroller/telegram.json` to receive telegram notifications.
+### Gotify
+Configure `/etc/pacroller/gotify.json` to receive Gotify notifications.
 
 ## Limitations
 - Your favourite package may not be supported, however it's easy to add another set of rules.

src/pacroller/config.py

@@ -9,6 +9,7 @@
 CONFIG_FILE = 'config.json'
 CONFIG_FILE_SMTP = 'smtp.json'
 CONFIG_FILE_TG = 'telegram.json'
+CONFIG_FILE_GOTIFY = 'gotify.json'
 F_KNOWN_OUTPUT_OVERRIDE = 'known_output_override.py'
 LIB_DIR = Path('/var/lib/pacroller')
 DB_FILE = 'db'
@@ -46,6 +47,17 @@
 else:
     _tg_config = dict()
 
+if (gotify_cfg := (CONFIG_DIR / CONFIG_FILE_GOTIFY)).exists():
+    try:
+        _gotify_cfg_text = gotify_cfg.read_text()
+    except PermissionError:
+        _gotify_config = dict()
+    else:
+        _gotify_config: dict = json.loads(_gotify_cfg_text)
+else:
+    _gotify_config = dict()
+
+
 def _import_module(fpath: Path) -> Any:
     spec = importlib.util.spec_from_file_location(str(fpath).removesuffix('.py').replace('/', '.'), fpath)
     mod = importlib.util.module_from_spec(spec)
@@ -115,3 +127,8 @@ def _import_module(fpath: Path) -> Any:
 TG_BOT_TOKEN = _tg_config.get('bot_token', "")
 TG_API_HOST = _tg_config.get('api_host', 'api.telegram.org')
 TG_RECIPIENT = _tg_config.get('recipient', "")
+
+GOTIFY_ENABLED = bool(_gotify_config.get('enabled', False))
+GOTIFY_TOKEN = _gotify_config.get('token', "")
+GOTIFY_HOST = _gotify_config.get('host_url', '')
+GOTIFY_PRIORITY = _gotify_config.get('priority', 5)

src/pacroller/gotify.json

@@ -0,0 +1,6 @@
+{
+    "enabled": false,
+    "host_url": "gotify.example.com",
+    "token": "123456789",
+    "priority": 8
+}

src/pacroller/mailer.py

@@ -5,12 +5,11 @@
 import logging
 import urllib.request, urllib.parse
 from platform import node
-from pacroller.config import NETWORK_RETRY, SMTP_ENABLED, SMTP_SSL, SMTP_HOST, SMTP_PORT, SMTP_FROM, SMTP_TO, SMTP_AUTH, TG_ENABLED, TG_BOT_TOKEN, TG_API_HOST, TG_RECIPIENT, DEF_HTTP_HDRS
+from pacroller.config import NETWORK_RETRY, SMTP_ENABLED, SMTP_SSL, SMTP_HOST, SMTP_PORT, SMTP_FROM, SMTP_TO, SMTP_AUTH, TG_ENABLED, TG_BOT_TOKEN, TG_API_HOST, TG_RECIPIENT, GOTIFY_ENABLED, GOTIFY_TOKEN, GOTIFY_HOST, GOTIFY_PRIORITY, DEF_HTTP_HDRS
 
 logger = logging.getLogger()
 hostname = node() or "unknown-host"
 
-
 def send_email(text: str, subject: str, mailto: List[str]) -> bool:
     for _ in range(NETWORK_RETRY):
         try:
@@ -60,19 +59,41 @@ def send_tg_message(text: str, subject: str, mailto: List[str]) -> bool:
         return False
     return all_succeeded
 
+def send_gotify_message(text: str, subject: str) -> bool:
+    for _ in range(1):
+        all_succeeded = True
+        try:
+            url = f'https://{GOTIFY_HOST}/message?token={GOTIFY_TOKEN}'
+            headers = {'Content-Type': 'application/json'}
+            data = json.dumps({"message": f"{text[:4000]}", "title": f"{subject}", "priority": GOTIFY_PRIORITY})
+            req = urllib.request.Request(url, data=data.encode('utf-8'), headers={**DEF_HTTP_HDRS, **headers})
+            resp = urllib.request.urlopen(req).read().decode('utf-8')
+        except Exception:
+            logger.exception("error while send_gotify_message")
+        else:
+            logger.debug(f"Gotify message sent {text=}")
+            break
+    else:
+        logger.error(f"unable to send Gotify message after 5 attempts {text=}")
+        return False
+    return all_succeeded
+
 
 class MailSender:
     def __init__(self) -> None:
         pass
-    def send_text_plain(elsf, text: str, subject: str = f"pacroller on {hostname}") -> bool:
+    def send_text_plain(self, text: str, subject: str = f"pacroller on {hostname}") -> bool:
         if_failed = False
         if SMTP_ENABLED:
             if not send_email(text, subject, SMTP_TO.split()):
                 if_failed = True
         if TG_ENABLED:
             if not send_tg_message(text, subject, TG_RECIPIENT.split()):
                 if_failed = True
-        if not SMTP_ENABLED and not TG_ENABLED:
+        if GOTIFY_ENABLED:
+            if not send_gotify_message(text, subject):
+                f_failed = True
+        if not SMTP_ENABLED and not TG_ENABLED and not GOTIFY_ENABLED:
             return None
         return not if_failed