mirror of
https://github.com/unixtensor/proxmox-ntfy.git
synced 2025-06-28 08:28:05 +00:00
address.py
This commit is contained in:
14
src/address.py
Normal file
14
src/address.py
Normal file
@ -0,0 +1,14 @@
|
|||||||
|
import re
|
||||||
|
|
||||||
|
class Address:
|
||||||
|
def __init__(self, address: str):
|
||||||
|
self.address = address
|
||||||
|
|
||||||
|
def is_valid(self) -> bool:
|
||||||
|
return re.search(r"^\d+[.]\d+[.]\d+[.]\d+|^(https|http)://.+$", self.address) != None
|
||||||
|
|
||||||
|
def format(self, topic: str) -> str:
|
||||||
|
addr = self.address
|
||||||
|
if self.address[len(self.address)-1] != "/":
|
||||||
|
addr += "/"
|
||||||
|
return addr + topic
|
36
src/main.py
36
src/main.py
@ -1,38 +1,28 @@
|
|||||||
import time
|
import time
|
||||||
import re
|
|
||||||
|
|
||||||
import command
|
import command
|
||||||
import cli
|
import cli
|
||||||
import cpu
|
import cpu
|
||||||
|
|
||||||
from datetime import datetime
|
from datetime import datetime
|
||||||
|
from address import Address
|
||||||
from typing import TypedDict
|
from typing import TypedDict
|
||||||
from ntfy import Ntfy
|
from ntfy import Ntfy
|
||||||
|
|
||||||
def start_prompt(server_url: str) -> str:
|
class Prompt:
|
||||||
|
@staticmethod
|
||||||
|
def start(address: str) -> str:
|
||||||
return f"""{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
|
return f"""{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
|
||||||
Listening and sending notifications to: \033[32m{server_url}\033[0m.
|
Listening and sending notifications to: \033[32m{address}\033[0m.
|
||||||
|
|
||||||
Source code available at:
|
Source code available at:
|
||||||
* <https://github.com/unixtensor/proxmox-ntfy>
|
* <https://github.com/unixtensor/proxmox-ntfy>
|
||||||
<https://git.rhpidfyre.io/rhpidfyre/proxmox-ntfy>
|
<https://git.rhpidfyre.io/rhpidfyre/proxmox-ntfy>
|
||||||
------"""
|
------"""
|
||||||
|
|
||||||
class Address:
|
@staticmethod
|
||||||
def __init__(self, address: str):
|
def address_not_valid(address: str) -> str:
|
||||||
self.address = address
|
return f"""The address "{address}" is not valid.
|
||||||
|
|
||||||
def is_valid(self) -> bool:
|
|
||||||
return re.search(r"^\d+[.]\d+[.]\d+[.]\d+|^(https|http)://.+$", self.address) != None
|
|
||||||
|
|
||||||
def format(self, topic: str) -> str:
|
|
||||||
addr = self.address
|
|
||||||
if self.address[len(self.address)-1] != "/":
|
|
||||||
addr += "/"
|
|
||||||
return addr + topic
|
|
||||||
|
|
||||||
def not_valid_prompt(self) -> str:
|
|
||||||
return f"""The address "{self.address}" is not valid.
|
|
||||||
Accepted address types:
|
Accepted address types:
|
||||||
\033[32m10.0.0.69:42069
|
\033[32m10.0.0.69:42069
|
||||||
http://domain.com
|
http://domain.com
|
||||||
@ -82,9 +72,6 @@ class Init:
|
|||||||
if self.config["cpu_warning_temp"] >= self.config["cpu_temp_critical"]:
|
if self.config["cpu_warning_temp"] >= self.config["cpu_temp_critical"]:
|
||||||
print("CPU warning tempature cannot be greater than or equal to the crtitical tempature.")
|
print("CPU warning tempature cannot be greater than or equal to the crtitical tempature.")
|
||||||
return
|
return
|
||||||
|
|
||||||
print(f"{self.config}\n" + start_prompt(self.config["ntfy_server_url"]))
|
|
||||||
|
|
||||||
if not self.config["startup_notify_disabled"]:
|
if not self.config["startup_notify_disabled"]:
|
||||||
self.__start_notify()
|
self.__start_notify()
|
||||||
self.__listen()
|
self.__listen()
|
||||||
@ -93,6 +80,9 @@ def main():
|
|||||||
cli_args = cli.Interface()
|
cli_args = cli.Interface()
|
||||||
address = Address(cli_args.server_address_no_topic)
|
address = Address(cli_args.server_address_no_topic)
|
||||||
if address.is_valid():
|
if address.is_valid():
|
||||||
|
formatted_address = address.format(cli_args.topic)
|
||||||
|
print(Prompt.start(formatted_address))
|
||||||
|
|
||||||
Init({
|
Init({
|
||||||
"cpu_temp_critical_timeout": cli_args.cpu_temp_critical_timeout,
|
"cpu_temp_critical_timeout": cli_args.cpu_temp_critical_timeout,
|
||||||
"cpu_temp_critical_message": cli_args.cpu_temp_critical_message,
|
"cpu_temp_critical_message": cli_args.cpu_temp_critical_message,
|
||||||
@ -105,10 +95,10 @@ def main():
|
|||||||
"cpu_temp_critical": cli_args.cpu_temp_critical,
|
"cpu_temp_critical": cli_args.cpu_temp_critical,
|
||||||
"cpu_warning_temp": cli_args.cpu_temp_warning,
|
"cpu_warning_temp": cli_args.cpu_temp_warning,
|
||||||
"update_interval": cli_args.update_rate,
|
"update_interval": cli_args.update_rate,
|
||||||
"ntfy_server_url": address.format(cli_args.topic)
|
"ntfy_server_url": formatted_address
|
||||||
}).start()
|
}).start()
|
||||||
else:
|
else:
|
||||||
print(address.not_valid_prompt())
|
print(Prompt.address_not_valid(cli_args.server_address_no_topic))
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
if command.package_installed("lm-sensors"):
|
if command.package_installed("lm-sensors"):
|
||||||
|
Reference in New Issue
Block a user