import logging as log
import argparse
import requests
import json
import base64
from time import sleep
from datetime import datetime
# Exit codes
# - 1: Response from 'ntfy' was < 400 while pushing notification
# - 2: No ntfy authentication token was provided
# - 3: Provided 'ntfy' authentication token is invalid
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, nextcloud_auth_header) -> list:
parsed_actions = []
for action in actions:
action_parsed = {
"action": "http",
"label": f"{action['label']}",
"url": f"{action['link']}",
"method": f"{action['type']}",
"clear": True,
}
# 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"
else:
action_parsed["headers"] = {
"Authorization": f"{nextcloud_auth_header}",
"OCS-APIREQUEST": "true",
}
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,
}
if token != "":
response = requests.post(
url, data=json.dumps(jsonData), headers={"Authorization": f"Bearer {token}"}
)
else:
response = requests.post(url, data=json.dumps(jsonData))
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) -> str:
if app == "spreed":
return "Talk"
elif app == "event_update_notification":
return "Calendar"
elif app == "twofactor_nextcloud_notification":
return "2FA"
else:
return app
def arg_parser() -> argparse.Namespace:
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.sh",
"ntfy_topic": "nextcloud",
"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",
"nextcloud_poll_interval_seconds": 60,
"nextcloud_error_sleep_seconds": 600,
"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
if config_data["ntfy_auth"] == "false":
config_data["ntfy_token"] == ""
elif config_data["ntfy_auth"] == "true" and (
config_data["ntfy_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[
"ntfy_token"
].startswith("tk_"):
print("Error: Authentication token set in 'ntfy_token' is invalid!")
exit(3)
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.debug("Fetching notifications.")
try:
response = requests.get(
f"{config["nextcloud_base_url"]}{config["nextcloud_notification_path"]}",
headers=nextcloud_request_headers,
)
except requests.exceptions.SSLError as e:
log.error(f"SSL error fetching notifications. Maybe the server/proxy is down?\n{e}")
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}")
try:
data = json.loads(response.text)
except Exception as e:
log.error("Error parsing response from Nextcloud!")
log.error(f"Response code: {response.status_code}")
log.error(f"Response body:\n{response.text}")
log.error("=====================================")
log.error(f"Exception:\n{e}")
for notification in reversed(data["ocs"]["data"]):
if datetime.fromisoformat(notification["datetime"]) <= last_datetime:
log.debug("No new notifications.")
continue
last_datetime = datetime.fromisoformat(notification["datetime"])
log.info("New notifications received.")
title = ""
if notification["app"] == "admin_notifications":
title = f"Nextcloud: {notification["subject"]}"
else:
title = f"Nextcloud - {translate_app_name(notification["app"])}: {notification["subject"]}"
log.debug(f"Notification title: {title}")
message = notification["message"]
log.debug(f"Notification message:\n{message}")
actions = parse_actions(notification["actions"], nextcloud_auth_header)
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("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()