fix(nix): Remove unnecessary package group, use python 3.11

Closes: #8

config-example.json

@@ -1,7 +1,7 @@
 {
   "ntfy_base_url": "https://ntfy.sh",
   "ntfy_topic": "nextcloud",
-  "ntfy_auth": "false",
+  "ntfy_auth": false,
   "ntfy_token": "authentication_token",
   
   "nextcloud_base_url": "https://nextcloud.example.com",
@@ -9,6 +9,10 @@
   "nextcloud_username": "user",
   "nextcloud_password": "application_password",
 
+  "heartbeat": false,
+  "heartbeat_url": "url",
+  "heartbeat_interval": 30,
+  
   "nextcloud_poll_interval_seconds": 60,
   "nextcloud_error_sleep_seconds": 600,
   "nextcloud_204_sleep_seconds": 3600,

default.nix

@@ -5,12 +5,13 @@ let
   pkgs = import (fetchTarball "https://github.com/NixOS/nixpkgs/archive/cf8cc1201be8bc71b7cbbbdaf349b22f4f99c7ae.tar.gz") {};
 in pkgs.mkShell {
   packages = [
-    (pkgs.python3.withPackages (python-pkgs: with python-pkgs; [
+    (pkgs.python311.withPackages (python-pkgs: with python-pkgs; [
       # select Python packages here
-      python-pkgs.python-lsp-server
+      python-lsp-server
-      python-pkgs.python-lsp-ruff
+      python-lsp-ruff
-      python-pkgs.pylsp-mypy
+      pylsp-mypy
-      python-pkgs.pylsp-rope
+      pylsp-rope
+      requests
     ]))
   ];
 }

nextcloud2ntfy.py

@@ -1,6 +1,7 @@
 import logging as log
 import argparse
 
+import threading
 import requests
 import json
 import base64
@@ -12,6 +13,7 @@ from datetime import datetime
 #   - 1: Response from 'ntfy' was < 400 while pushing notification
 #   - 2: No ntfy authentication token was provided
 #   - 3: Provided 'ntfy' authentication token is invalid
+#   - 4: No heartbeat url was given
 
 log_levels = {
     "DEBUG": log.DEBUG,
@@ -76,7 +78,7 @@ def push_to_ntfy(
 # Nextcloud apps have quite often internal names differing from the UI names.
 #  - Eg. `spreed` is `Talk`
 # This is the place to track the differences
-def translate_app_name(app=str) -> str:
+def translate_app_name(app: str) -> str:
     if app == "spreed":
         return "Talk"
     elif app == "event_update_notification":
@@ -116,12 +118,15 @@ def load_config(config_file: str) -> dict:
     default_config = {
         "ntfy_base_url": "https://ntfy.sh",
         "ntfy_topic": "nextcloud",
-        "ntfy_auth": "false",
+        "ntfy_auth": False,
         "ntfy_token": "authentication_token",
         "nextcloud_base_url": "https://nextcloud.example.com",
         "nextcloud_notification_path": "/ocs/v2.php/apps/notifications/api/v2/notifications",
         "nextcloud_username": "user",
         "nextcloud_password": "application_password",
+        "heartbeat": False,
+        "heartbeat_url": "url",
+        "heartbeat_interval": 30,
         "nextcloud_poll_interval_seconds": 60,
         "nextcloud_error_sleep_seconds": 600,
         "nextcloud_204_sleep_seconds": 3600,
@@ -138,22 +143,25 @@ def load_config(config_file: str) -> dict:
             if key not in config_data:
                 config_data[key] = value
 
-        if config_data["ntfy_auth"] == "false":
+        if config_data["ntfy_auth"] == False:
             config_data["ntfy_token"] == ""
-        elif config_data["ntfy_auth"] == "true" and (
+        elif config_data["ntfy_auth"] == True and (
             config_data["ntfy_token"] == ""
-            or config_data["ntfy_token"] == "authentication_token"
+            or config_data["ntfy_token"] == "authentication_token"):
-        ):
             print(
                 "Error: Option 'ntfy_auth' is set to 'true' but not 'ntfy_token' was set!"
             )
             exit(2)
-        elif config_data["ntfy_auth"] == "true" and not config_data[
+        elif config_data["ntfy_auth"] == True and not config_data[
-            "ntfy_token"
+             "ntfy_token" ].startswith("tk_"):
-        ].startswith("tk_"):
             print("Error: Authentication token set in 'ntfy_token' is invalid!")
             exit(3)
 
+        if config_data["heartbeat"] == True:
+            if config_data["heartbeat_url"] == "url" or config_data["heartbeat_url"] == "":
+                print("Error: 'heartbeat' is set to 'true' but no url was given.")
+                exit(4)
+
         return config_data
 
     except FileNotFoundError:
@@ -165,6 +173,13 @@ def load_config(config_file: str) -> dict:
         return default_config
 
 
+def monitoring_heartbeat(url, interval):
+    while True:
+        response = requests.get(url)
+        log.info("Sent heartbeat.")
+        log.debug(f"Response: {response}")
+        sleep(interval)
+
 def main():
     args = arg_parser()
     config = load_config(args.config_file)
@@ -177,8 +192,15 @@ def main():
     )
     log.info("Started Nextcloud to ntfy.sh notification bridge.")
 
+    if config["heartbeat"] == True:
+        heartbeat = threading.Thread(target=monitoring_heartbeat,
+            daemon=True,
+            args=("https://uptime.stfka.eu/api/push/pRCW5ARYxn?status=up&msg=OK&ping=",
+            config["heartbeat_interval"]))
+        heartbeat.start()
+
     last_datetime = datetime.fromisoformat("1970-01-01T00:00:00Z")
-    nextcloud_auth_header = f"Basic {base64.b64encode(f"{config["nextcloud_username"]}:{config["nextcloud_password"]}".encode("utf-8")).decode("utf-8")}"
+    nextcloud_auth_header = f"Basic {base64.b64encode('{}:{}'.format(config['nextcloud_username'], config['nextcloud_password']).encode('utf-8')).decode('utf-8')}"
     nextcloud_request_headers = {
         "Authorization": f"{nextcloud_auth_header}",
         "OCS-APIREQUEST": "true",
@@ -188,7 +210,7 @@ def main():
         log.debug("Fetching notifications.")
         try:
             response = requests.get(
-                f"{config["nextcloud_base_url"]}{config["nextcloud_notification_path"]}",
+                f"{config['nextcloud_base_url']}{config['nextcloud_notification_path']}",
                 headers=nextcloud_request_headers,
             )
         except requests.exceptions.SSLError as e:
@@ -198,14 +220,14 @@ def main():
                 f"Error while fetching notifications. Response code: {response.status_code}."
             )
             log.warning(
-                f"Sleeping for {config["nextcloud_error_sleep_seconds"]} seconds."
+                f"Sleeping for {config['nextcloud_error_sleep_seconds']} seconds."
             )
             sleep(config["nextcloud_error_sleep_seconds"])
             continue
 
         elif response.status_code == 204:
             log.debug(
-                f"Got code 204 while fetching notifications. Sleeping for {config["nextcloud_204_sleep_seconds"]/60/60} hour(s)."
+                f"Got code 204 while fetching notifications. Sleeping for {config['nextcloud_204_sleep_seconds']/60/60} hour(s)."
             )
 
         log.debug(f"Got resonse code: {response.status_code}")
@@ -229,9 +251,9 @@ def main():
 
             title = ""
             if notification["app"] == "admin_notifications":
-                title = f"Nextcloud: {notification["subject"]}"
+                title = f"Nextcloud: {notification['subject']}"
             else:
-                title = f"Nextcloud - {translate_app_name(notification["app"])}: {notification["subject"]}"
+                title = f"Nextcloud - {translate_app_name(notification['app'])}: {notification['subject']}"
             log.debug(f"Notification title: {title}")
 
             message = notification["message"]
@@ -242,7 +264,7 @@ def main():
                 {
                     "action": "http",
                     "label": "Dismiss",
-                    "url": f"{config["nextcloud_base_url"]}{config["nextcloud_notification_path"]}/{notification["notification_id"]}",
+                    "url": f"{config['nextcloud_base_url']}{config['nextcloud_notification_path']}/{notification['notification_id']}",
                     "method": "DELETE",
                     "headers": {
                         "Authorization": f"{nextcloud_auth_header}",
@@ -266,14 +288,14 @@ def main():
             )
             if response.status_code == 429:
                 log.error(
-                    f"Error pushing notification to {config["ntfy_base_url"]}: Too Many Requests."
+                    f"Error pushing notification to {config['ntfy_base_url']}: Too Many Requests."
                 )
                 log.warning(
-                    f"Sleeping for {config["rate_limit_sleep_seconds"]} seconds."
+                    f"Sleeping for {config['rate_limit_sleep_seconds']} seconds."
                 )
             elif not response.ok:
                 log.critical(
-                    f"Unknown erroro while pushing notification to {config["ntfy_base_url"]}. Error code: {response.status_code}."
+                    f"Unknown erroro while pushing notification to {config['ntfy_base_url']}. Error code: {response.status_code}."
                 )
                 log.critical(f"Response: {response.text}")
                 log.error("Stopping.")
@@ -281,6 +303,5 @@ def main():
 
         sleep(config["nextcloud_poll_interval_seconds"])
 
-
 if __name__ == "__main__":
     main()