diff --git a/.gitignore b/.gitignore index ab3e8ce..805c248 100644 --- a/.gitignore +++ b/.gitignore @@ -162,3 +162,4 @@ cython_debug/ # option (not recommended) you can uncomment the following to ignore the entire idea folder. #.idea/ +config.json diff --git a/config-example.json b/config-example.json new file mode 100644 index 0000000..bf773e5 --- /dev/null +++ b/config-example.json @@ -0,0 +1,17 @@ +{ + "ntfy_base_url": "https://ntfy.example.com", + "ntfy_topic": "nextcloud", + "ntfy_token": "default_token", + + "nextcloud_base_url": "https://nextcloud.example.com", + "nextcloud_notification_path": "/ocs/v2.php/apps/notifications/api/v1/notifications", + + "nextcloud_username": "user", + "nextcloud_password": "application_password", + + "nextcloud_poll_interval_seconds": 60, + "nextcloud_error_sleep_seconds": 600, + "nextcloud_204_sleep_seconds": 3600, + + "rate_limit_sleep_seconds": 600 +} diff --git a/main.py b/main.py new file mode 100644 index 0000000..1cd2fe4 --- /dev/null +++ b/main.py @@ -0,0 +1,208 @@ +import logging as log +import argparse + +import requests +import json +import base64 + +from time import sleep +from datetime import datetime + +log_levels = { + "DEBUG": log.DEBUG, + "INFO": log.INFO, + "WARNING": log.WARNING, + "ERROR": log.ERROR, + "CRITICAL": log.CRITICAL +} + +# Converts Nextcloud's notification buttons to ntfy.sh notification actions +def parse_actions(actions: list) -> list: + parsed_actions = [] + + for action in actions: + action_parsed = { + "action": "http", + "label": f"{action['label']}", + "url": f"{action['link']}", + "method": f"{action['type']}" + } + + # The `action['type']` is documented to be a HTTP request. + # But a value of `WEB` is also used + # which is used for opening links in the browser. + if action_parsed["method"] == "WEB": + del action_parsed["method"] + action_parsed["action"] = "view" + + parsed_actions.append(action_parsed) + + return parsed_actions + +def push_to_ntfy(url: str, token: str, topic: str, title: str, click = "", message = "", actions = []) -> requests.Response: + jsonData = { + "topic": f"{topic}", + "title": f"{title}", + "message": f"{message}", + "click": f"{click}", + "actions": actions + } + + response = requests.post(url, + data=json.dumps(jsonData), + headers={ + "Authorization": f"Bearer {token}" + }) + + return response + +# 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): + if app == "spreed": + return "Talk" + else: + return app + +def arg_parser(): + parser = argparse.ArgumentParser(description ='Nextcloud to ntfy.sh notification bridge.') + + parser.add_argument( + "--log_level", + type=str, + default="INFO", + choices=["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"], + help="Set the logging level (default: INFO)" + ) + parser.add_argument( + "-c", + "--config-file", + type=str, + default="./config.json", + required=True, + help="Path to the configuration file" + ) + + return parser.parse_args() + +def load_config(config_file: str) -> dict: + # Default values for the configuration + default_config = { + "ntfy_base_url": "https://ntfy.example.com", + "ntfy_topic": "nextcloud", + "ntfy_token": "default_token", + + "nextcloud_base_url": "https://nextcloud.example.com", + "nextcloud_notification_path": "/ocs/v2.php/apps/notifications/api/v1/notifications", + + "nextcloud_username": "user", + "nextcloud_password": "application_password", + + "nextcloud_poll_interval_seconds": 10, + "nextcloud_error_sleep_seconds": 300, + "nextcloud_204_sleep_seconds": 3600, + + "rate_limit_sleep_seconds": 600 + } + + try: + # Attempt to load the JSON config file + with open(config_file, "r") as file: + config_data = json.load(file) + + # Check and fill missing values with defaults + for key, value in default_config.items(): + if key not in config_data: + config_data[key] = value + + return config_data + + except FileNotFoundError: + print(f"Configuration file {config_file} not found. Using default values.") + return default_config + + except json.JSONDecodeError: + print(f"Error decoding JSON from {config_file}. Using default values.") + return default_config + +def main(): + args = arg_parser() + config = load_config(args.config_file) + + log.basicConfig( + format="{asctime} - {levelname} - {message}", + style="{", + datefmt="%d-%m-%Y %H:%M:%S", + level=log_levels[args.log_level] + ) + log.info("Started Nextcloud to ntfy.sh notification bridge.") + + 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_request_headers = { + "Authorization": f"{nextcloud_auth_header}", + "OCS-APIREQUEST": "true", + "Accept": "application/json" + } + while True: + log.info("Fetching notifications.") + response = requests.get(f"{config["nextcloud_base_url"]}{config["nextcloud_notification_path"]}", headers = nextcloud_request_headers) + if not response.ok: + log.error(f"Error while fetching notifications. Response code: {response.status_code}.") + log.warning(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).") + + log.debug(f"Got resonse code: {response.status_code}") + + log.debug(f"Received data:\n{response.text}") + data = json.loads(response.text) + + for notification in reversed(data["ocs"]["data"]): + + if datetime.fromisoformat(notification["datetime"]) <= last_datetime: + log.info("No new notifications.") + continue + last_datetime = datetime.fromisoformat(notification["datetime"]) + log.info("New notifications received.") + + title = f"Nextcloud - {translate_app_name(data["ocs"]["data"][0]["app"])}: {data["ocs"]["data"][0]["subject"]}" + log.debug(f"Notification title: {title}") + + message = notification["message"] + log.debug(f"Notification message:\n{message}") + + actions = parse_actions(notification["actions"]) + actions.append({ + "action": "http", + "label": "Dismiss", + "url": f"{config["nextcloud_base_url"]}{config["nextcloud_notification_path"]}/{notification["notification_id"]}", + "method": "DELETE", + "headers": { + "Authorization": f"{nextcloud_auth_header}", + "OCS-APIREQUEST": "true" + }, + "clear": True + }) + log.debug(f"Notification actions:\n{actions}") + + log.info(f"Pushing notification to ntfy.") + + response = push_to_ntfy(config["ntfy_base_url"], config["ntfy_token"], config["ntfy_topic"], title, notification["link"], message, actions) + if response.status_code == 429: + log.error(f"Error pushing notification to {config["ntfy_base_url"]}: Too Many Requests.") + log.warning(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}.") + log.critical(f"Response: {response.text}") + log.error("Stopping.") + exit(1) + + sleep(config["nextcloud_poll_interval_seconds"]) + +if __name__ == "__main__": + main()