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

Compare commits

base: jiriks74:main
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:v1.0.2
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. "main" and "v1.0.2" have entirely different histories.
main ... v1.0.2

3 changed files with 26 additions and 52 deletions
Show stats Expand all files Collapse all files

6
config-example.json
View file

@ -1,17 +1,13 @@
{
"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_username": "user",
"nextcloud_password": "application_password",
"heartbeat": false,
"heartbeat_url": "url",
"heartbeat_interval": 30,
"nextcloud_poll_interval_seconds": 60,
"nextcloud_error_sleep_seconds": 600,

11
default.nix
View file

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

61
nextcloud2ntfy.py
View file

@ -1,7 +1,6 @@
import logging as log
import argparse
import threading
import requests
import json
import base64
@ -13,7 +12,6 @@ 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,
@ -78,7 +76,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":
@ -118,15 +116,12 @@ 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,
@ -143,25 +138,22 @@ 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[
"ntfy_token" ].startswith("tk_"):
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)
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:
@ -173,13 +165,6 @@ 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)
@ -192,15 +177,8 @@ 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('{}:{}'.format(config['nextcloud_username'], config['nextcloud_password']).encode('utf-8')).decode('utf-8')}"
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",
@ -210,7 +188,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:
@ -220,14 +198,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}")
@ -251,9 +229,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"]
@ -264,7 +242,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}",
@ -288,14 +266,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.")
@ -303,5 +281,6 @@ def main():
sleep(config["nextcloud_poll_interval_seconds"])
if __name__ == "__main__":
main()