mirror of
https://github.com/unixtensor/proxmox-ntfy.git
synced 2025-06-28 14:18:05 +00:00
--disable-ntfy-logs
and --topic TOPIC
This commit is contained in:
11
src/cli.py
11
src/cli.py
@ -6,14 +6,17 @@ def Interface():
|
||||
parser = argparse.ArgumentParser(description="Proxmox monitoring tool for phone notifications using ntfy.sh")
|
||||
parser.add_argument("server_address", help="The ntfy server address.")
|
||||
|
||||
parser.add_argument("--topic", default="proxmox", help="The ntfy topic name that notifications will be sent to. Default = proxmox")
|
||||
|
||||
parser.add_argument("--disable-uptime-notifys", action="store_true", help="Disable uptime notifications.")
|
||||
parser.add_argument("--disable-startup-ping", action="store_true", help="Disable the start up ping.")
|
||||
parser.add_argument("--disable-cpu-temp", action="store_true", help="Disable notifications for CPU tempature.")
|
||||
parser.add_argument("--disable-ntfy-logs", action="store_true", help="Disable logging ntfy activity to the output.")
|
||||
|
||||
parser.add_argument("--cpu-temp-warning", type=int, default=70, help="CPU tempature for the warning alert. default = 70")
|
||||
parser.add_argument("--update-rate", type=int, default=1, help="How often updates happen in seconds. default = 1")
|
||||
|
||||
parser.add_argument("--cpu-temp-warning", type=int, default=70, help="CPU tempature for the warning alert. default = 70")
|
||||
parser.add_argument("--cpu-temp-warning-message", default=cpu.Tempature.cpu_temp_warning_message, help="The notification message if the CPU is at a high tempature. (message) [TEMP] C")
|
||||
parser.add_argument("--startup-ping-message", default="🖥️ Ntfy proxmox monitoring started.", help="The notification message when the program is started.")
|
||||
|
||||
parser.add_argument("--update-rate", type=int, default=1, help="How often updates happen in seconds. default = 1")
|
||||
parser.add_argument("--startup-ping-message", default="🖥️ Ntfy proxmox monitoring started.", help="The notification message when the program is started.")
|
||||
|
||||
return parser.parse_args()
|
@ -7,12 +7,11 @@ from print_t import print_t
|
||||
from typing import Optional
|
||||
from ntfy import Ntfy
|
||||
|
||||
_time_now = time.time()
|
||||
last_cpu_check_warning: float = _time_now
|
||||
cpu_temp_queue_check: int = 120 # Seconds
|
||||
last_cpu_check_warning: float = time.time()
|
||||
|
||||
class Tempature:
|
||||
cpu_temp_warning_message: str = "🌡️ CPU is at a high tempature."
|
||||
cpu_temp_queue_check: int = 120 # Seconds
|
||||
|
||||
def __init__(self, ntfy_instance: Ntfy, cpu_warning_temp: int):
|
||||
self.cpu_warning_temp = cpu_warning_temp
|
||||
@ -28,7 +27,7 @@ class Tempature:
|
||||
return None
|
||||
|
||||
def __queue_time(self, last_check: float) -> bool:
|
||||
return (time.time() - last_check) > cpu_temp_queue_check * 1000
|
||||
return (time.time() - last_check) > Tempature.cpu_temp_queue_check * 1000
|
||||
|
||||
def ntfy_check(self):
|
||||
cpu_temp = self.get()
|
||||
|
@ -22,6 +22,7 @@ class Config(TypedDict):
|
||||
cpu_temp_check_disabled: bool
|
||||
startup_ping_disabled: bool
|
||||
startup_ping_message: str
|
||||
ntfy_logs_disabled: bool
|
||||
cpu_warning_temp: int
|
||||
update_interval: int
|
||||
ntfy_server_url: str
|
||||
@ -29,7 +30,7 @@ class Config(TypedDict):
|
||||
class Init:
|
||||
def __init__(self, config: Config):
|
||||
self.config = config
|
||||
self.ntfy = Ntfy(config["ntfy_server_url"])
|
||||
self.ntfy = Ntfy(config["ntfy_server_url"], config["ntfy_logs_disabled"])
|
||||
self.monitor_cpu_temp = cpu.Tempature(self.ntfy, config["cpu_warning_temp"])
|
||||
|
||||
cpu.Tempature.cpu_temp_warning_message = config["cpu_temp_warning_message"]
|
||||
@ -56,7 +57,8 @@ if __name__ == "__main__":
|
||||
"cpu_temp_check_disabled": cli_args.disable_cpu_temp,
|
||||
"startup_ping_disabled": cli_args.disable_startup_ping,
|
||||
"startup_ping_message": cli_args.startup_ping_message,
|
||||
"ntfy_logs_disabled": cli_args.disable_ntfy_logs,
|
||||
"cpu_warning_temp": cli_args.cpu_temp_warning,
|
||||
"update_interval": cli_args.update_rate,
|
||||
"ntfy_server_url": cli_args.server_address,
|
||||
"ntfy_server_url": cli_args.server_address + "/" + cli_args.topic, #Heh.
|
||||
}).start()
|
@ -3,12 +3,14 @@ import requests
|
||||
from print_t import print_t
|
||||
|
||||
class Ntfy:
|
||||
def __init__(self, server: str):
|
||||
def __init__(self, server: str, logging_disabled: bool):
|
||||
self.server = server
|
||||
self.logging_disabled = logging_disabled
|
||||
|
||||
def send(self, message: str):
|
||||
print_t("Ntfy OUT: " + message)
|
||||
if not self.logging_disabled:
|
||||
print_t("Ntfy OUT: " + message)
|
||||
try:
|
||||
requests.post(self.server, data=message)
|
||||
requests.post(self.server, data=message.encode(encoding="utf-8"))
|
||||
except Exception as err:
|
||||
print_t(f"\033[31m{err}\033[0m")
|
||||
|
Reference in New Issue
Block a user