jiriks74/nextcloud2ntfy
1
0
Fork 0
Code Issues 5 Pull requests Projects Releases 5 Packages Wiki Activity Actions

Compare commits

base: jiriks74:904c712d4491d60b9bb3f844f7a1c52c42b06d9b
Branches Tags
jiriks74:main
jiriks74:v1.1.1
jiriks74:v1.1.0
jiriks74:v1.0.2
jiriks74:v1.0.1
jiriks74:v1.0.0
..
compare: jiriks74:12dc13237306eb5a39c728e234ce76c6c8d5cab6
Branches Tags
jiriks74:main
jiriks74:v1.1.1
jiriks74:v1.1.0
jiriks74:v1.0.2
jiriks74:v1.0.1
jiriks74:v1.0.0

No commits in common. "904c712d4491d60b9bb3f844f7a1c52c42b06d9b" and "12dc13237306eb5a39c728e234ce76c6c8d5cab6" have entirely different histories.
904c712d44 ... 12dc132373

1 changed files with 58 additions and 91 deletions
Show stats Expand all files Collapse all files

149
nextcloud2ntfy.py
View file

@ -18,10 +18,9 @@ log_levels = {
"INFO": log.INFO, "INFO": log.INFO,
"WARNING": log.WARNING, "WARNING": log.WARNING,
"ERROR": log.ERROR, "ERROR": log.ERROR,
"CRITICAL": log.CRITICAL, "CRITICAL": log.CRITICAL
} }
# Converts Nextcloud's notification buttons to ntfy.sh notification actions # Converts Nextcloud's notification buttons to ntfy.sh notification actions
def parse_actions(actions: list) -> list: def parse_actions(actions: list) -> list:
parsed_actions = [] parsed_actions = []
@ -32,7 +31,7 @@ def parse_actions(actions: list) -> list:
"label": f"{action['label']}", "label": f"{action['label']}",
"url": f"{action['link']}", "url": f"{action['link']}",
"method": f"{action['type']}", "method": f"{action['type']}",
"clear": True, "clear": True
} }
# The `action['type']` is documented to be a HTTP request. # The `action['type']` is documented to be a HTTP request.
@ -46,32 +45,31 @@ def parse_actions(actions: list) -> list:
return parsed_actions return parsed_actions
def push_to_ntfy(url: str, token: str, topic: str, title: str, click = "", message = "", actions = []) -> requests.Response:
def push_to_ntfy(
url: str, token: str, topic: str, title: str, click="", message="", actions=[]
) -> requests.Response:
jsonData = { jsonData = {
"topic": f"{topic}", "topic": f"{topic}",
"title": f"{title}", "title": f"{title}",
"message": f"{message}", "message": f"{message}",
"click": f"{click}", "click": f"{click}",
"actions": actions, "actions": actions
} }
if token != "": if token != "":
response = requests.post( response = requests.post(url,
url, data=json.dumps(jsonData), headers={"Authorization": f"Bearer {token}"} data=json.dumps(jsonData),
) headers={
"Authorization": f"Bearer {token}"
})
else: else:
response = requests.post(url, data=json.dumps(jsonData)) response = requests.post(url,
data=json.dumps(jsonData))
return response return response
# Nextcloud apps have quite often internal names differing from the UI names. # Nextcloud apps have quite often internal names differing from the UI names.
# - Eg. `spreed` is `Talk` # - Eg. `spreed` is `Talk`
# This is the place to track the differences # 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": if app == "spreed":
return "Talk" return "Talk"
elif app == "event_update_notification": elif app == "event_update_notification":
@ -79,31 +77,27 @@ def translate_app_name(app=str) -> str:
else: else:
return app return app
def arg_parser() -> argparse.Namespace: def arg_parser() -> argparse.Namespace:
parser = argparse.ArgumentParser( parser = argparse.ArgumentParser(description ='Nextcloud to ntfy.sh notification bridge.')
description="Nextcloud to ntfy.sh notification bridge."
)
parser.add_argument( parser.add_argument(
"--log_level", "--log_level",
type=str, type=str,
default="INFO", default="INFO",
choices=["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"], choices=["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"],
help="Set the logging level (default: INFO)", help="Set the logging level (default: INFO)"
) )
parser.add_argument( parser.add_argument(
"-c", "-c",
"--config-file", "--config-file",
type=str, type=str,
default="./config.json", default="./config.json",
required=True, required=True,
help="Path to the configuration file", help="Path to the configuration file"
) )
return parser.parse_args() return parser.parse_args()
def load_config(config_file: str) -> dict: def load_config(config_file: str) -> dict:
# Default values for the configuration # Default values for the configuration
default_config = { default_config = {
@ -111,21 +105,25 @@ def load_config(config_file: str) -> dict:
"ntfy_topic": "nextcloud", "ntfy_topic": "nextcloud",
"ntfy_auth": "false", "ntfy_auth": "false",
"ntfy_token": "authentication_token", "ntfy_token": "authentication_token",
"nextcloud_base_url": "https://nextcloud.example.com", "nextcloud_base_url": "https://nextcloud.example.com",
"nextcloud_notification_path": "/ocs/v2.php/apps/notifications/api/v2/notifications", "nextcloud_notification_path": "/ocs/v2.php/apps/notifications/api/v2/notifications",
"nextcloud_username": "user", "nextcloud_username": "user",
"nextcloud_password": "application_password", "nextcloud_password": "application_password",
"nextcloud_poll_interval_seconds": 60, "nextcloud_poll_interval_seconds": 60,
"nextcloud_error_sleep_seconds": 600, "nextcloud_error_sleep_seconds": 600,
"nextcloud_204_sleep_seconds": 3600, "nextcloud_204_sleep_seconds": 3600,
"rate_limit_sleep_seconds": 600,
"rate_limit_sleep_seconds": 600
} }
try: try:
# Attempt to load the JSON config file # Attempt to load the JSON config file
with open(config_file, "r") as file: with open(config_file, "r") as file:
config_data = json.load(file) config_data = json.load(file)
# Check and fill missing values with defaults # Check and fill missing values with defaults
for key, value in default_config.items(): for key, value in default_config.items():
if key not in config_data: if key not in config_data:
@ -134,19 +132,13 @@ def load_config(config_file: str) -> dict:
if config_data["ntfy_auth"] == "false": if config_data["ntfy_auth"] == "false":
config_data["ntfy_token"] == "" config_data["ntfy_token"] == ""
elif config_data["ntfy_auth"] == "true" and ( elif config_data["ntfy_auth"] == "true" and (
config_data["ntfy_token"] == "" 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!")
):
print(
"Error: Option 'ntfy_auth' is set to 'true' but not 'ntfy_token' was set!"
)
exit(2) exit(2)
elif config_data["ntfy_auth"] == "true" and not config_data[ elif config_data["ntfy_auth"] == "true" and not config_data["ntfy_token"].startswith("tk_"):
"ntfy_token"
].startswith("tk_"):
print("Error: Authentication token set in 'ntfy_token' is invalid!") print("Error: Authentication token set in 'ntfy_token' is invalid!")
exit(3) exit(3)
return config_data return config_data
except FileNotFoundError: except FileNotFoundError:
@ -157,7 +149,6 @@ def load_config(config_file: str) -> dict:
print(f"Error decoding JSON from {config_file}. Using default values.") print(f"Error decoding JSON from {config_file}. Using default values.")
return default_config return default_config
def main(): def main():
args = arg_parser() args = arg_parser()
config = load_config(args.config_file) config = load_config(args.config_file)
@ -166,7 +157,7 @@ def main():
format="{asctime} - {levelname} - {message}", format="{asctime} - {levelname} - {message}",
style="{", style="{",
datefmt="%d-%m-%Y %H:%M:%S", datefmt="%d-%m-%Y %H:%M:%S",
level=log_levels[args.log_level], level=log_levels[args.log_level]
) )
log.info("Started Nextcloud to ntfy.sh notification bridge.") log.info("Started Nextcloud to ntfy.sh notification bridge.")
@ -175,28 +166,19 @@ def main():
nextcloud_request_headers = { nextcloud_request_headers = {
"Authorization": f"{nextcloud_auth_header}", "Authorization": f"{nextcloud_auth_header}",
"OCS-APIREQUEST": "true", "OCS-APIREQUEST": "true",
"Accept": "application/json", "Accept": "application/json"
} }
while True: while True:
log.debug("Fetching notifications.") log.debug("Fetching notifications.")
response = requests.get( response = requests.get(f"{config["nextcloud_base_url"]}{config["nextcloud_notification_path"]}", headers = nextcloud_request_headers)
f"{config["nextcloud_base_url"]}{config["nextcloud_notification_path"]}",
headers=nextcloud_request_headers,
)
if not response.ok: if not response.ok:
log.error( log.error(f"Error while fetching notifications. Response code: {response.status_code}.")
f"Error while fetching notifications. Response code: {response.status_code}." log.warning(f"Sleeping for {config["nextcloud_error_sleep_seconds"]} seconds.")
)
log.warning(
f"Sleeping for {config["nextcloud_error_sleep_seconds"]} seconds."
)
sleep(config["nextcloud_error_sleep_seconds"]) sleep(config["nextcloud_error_sleep_seconds"])
continue continue
elif response.status_code == 204: elif response.status_code == 204:
log.debug( 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}") log.debug(f"Got resonse code: {response.status_code}")
@ -207,10 +189,12 @@ def main():
log.error("Error parsing response from Nextcloud!") log.error("Error parsing response from Nextcloud!")
log.error(f"Response code: {response.status_code}") log.error(f"Response code: {response.status_code}")
log.error(f"Response body:\n{response.text}") log.error(f"Response body:\n{response.text}")
log.error("=====================================") log.error(f"=====================================")
log.error(f"Exception:\n{e}") log.error(f"Exception:\n{e}")
for notification in reversed(data["ocs"]["data"]): for notification in reversed(data["ocs"]["data"]):
if datetime.fromisoformat(notification["datetime"]) <= last_datetime: if datetime.fromisoformat(notification["datetime"]) <= last_datetime:
log.debug("No new notifications.") log.debug("No new notifications.")
continue continue
@ -228,49 +212,32 @@ def main():
log.debug(f"Notification message:\n{message}") log.debug(f"Notification message:\n{message}")
actions = parse_actions(notification["actions"]) actions = parse_actions(notification["actions"])
actions.append( actions.append({
{ "action": "http",
"action": "http", "label": "Dismiss",
"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",
"method": "DELETE", "headers": {
"headers": { "Authorization": f"{nextcloud_auth_header}",
"Authorization": f"{nextcloud_auth_header}", "OCS-APIREQUEST": "true"
"OCS-APIREQUEST": "true", },
}, "clear": True
"clear": True, })
}
)
log.debug(f"Notification actions:\n{actions}") log.debug(f"Notification actions:\n{actions}")
log.info(f"Pushing notification to ntfy.")
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)
response = push_to_ntfy(
config["ntfy_base_url"],
config["ntfy_token"],
config["ntfy_topic"],
title,
notification["link"],
message,
actions,
)
if response.status_code == 429: if response.status_code == 429:
log.error( 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.")
)
log.warning(
f"Sleeping for {config["rate_limit_sleep_seconds"]} seconds."
)
elif not response.ok: elif not response.ok:
log.critical( 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.critical(f"Response: {response.text}")
log.error("Stopping.") log.error("Stopping.")
exit(1) exit(1)
sleep(config["nextcloud_poll_interval_seconds"]) sleep(config["nextcloud_poll_interval_seconds"])
if __name__ == "__main__": if __name__ == "__main__":
main() main()