mirror of
https://github.com/unixtensor/proxmox-ntfy.git
synced 2025-06-28 08:28:05 +00:00
Use local machine time instead of unix for messages and have better timestamp formatting
This commit is contained in:
@ -1,12 +1,5 @@
|
|||||||
import argparse
|
import argparse
|
||||||
|
|
||||||
_ntfy_configure_prompt = """\033[4mPlease configure an ntfy url before starting.\033[0m
|
|
||||||
Examples:
|
|
||||||
\033[32mpython3 main.py 10.0.13.37:42069
|
|
||||||
python3 main.py ntfy.domain.com\033[0m
|
|
||||||
|
|
||||||
Use \033[32m-h\033[0m or \033[32m--help\033[0m for a full list of options."""
|
|
||||||
|
|
||||||
def Interface():
|
def Interface():
|
||||||
parser = argparse.ArgumentParser()
|
parser = argparse.ArgumentParser()
|
||||||
parser.add_argument("server_address", help="The ntfy server address.")
|
parser.add_argument("server_address", help="The ntfy server address.")
|
||||||
|
@ -7,9 +7,8 @@ from typing import Optional
|
|||||||
from ntfy import Ntfy
|
from ntfy import Ntfy
|
||||||
|
|
||||||
_time_now = time.time()
|
_time_now = time.time()
|
||||||
last_cpu_check_warning: float = _time_now
|
last_cpu_check_warning: float = _time_now
|
||||||
|
last_check_debounce: int = 120 # Seconds
|
||||||
last_check_debounce: int = 120 # Seconds
|
|
||||||
|
|
||||||
class Tempature:
|
class Tempature:
|
||||||
cpu_temp_warning_message: str = "🌡️ CPU is at a high tempature."
|
cpu_temp_warning_message: str = "🌡️ CPU is at a high tempature."
|
||||||
|
20
src/main.py
20
src/main.py
@ -8,24 +8,28 @@ from datetime import datetime
|
|||||||
from typing import TypedDict
|
from typing import TypedDict
|
||||||
from ntfy import Ntfy
|
from ntfy import Ntfy
|
||||||
|
|
||||||
_monitoring_prompt = f"""{datetime.fromtimestamp(time.time())}
|
|
||||||
Ntfy monitoring software is now listening.
|
|
||||||
|
|
||||||
Source code available at:
|
|
||||||
<https://github.com/unixtensor/proxmox-ntfy>
|
|
||||||
<https://git.rhpidfyre.io/rhpidfyre/proxmox-ntfy>"""
|
|
||||||
|
|
||||||
class Config(TypedDict):
|
class Config(TypedDict):
|
||||||
cpu_temp_check_disabled: bool
|
cpu_temp_check_disabled: bool
|
||||||
cpu_warning_temp: int
|
cpu_warning_temp: int
|
||||||
update_interval: int
|
update_interval: int
|
||||||
ntfy_server_url: str
|
ntfy_server_url: str
|
||||||
|
|
||||||
|
def start_prompt(server_url: str) -> str:
|
||||||
|
return f"""{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
|
||||||
|
Listening and sending notifications to: \033[32m{server_url}\033[0m.
|
||||||
|
|
||||||
|
Source code available at:
|
||||||
|
<https://github.com/unixtensor/proxmox-ntfy>
|
||||||
|
<https://git.rhpidfyre.io/rhpidfyre/proxmox-ntfy>
|
||||||
|
------"""
|
||||||
|
|
||||||
def start(config: Config):
|
def start(config: Config):
|
||||||
ntfy = Ntfy(config["ntfy_server_url"])
|
ntfy = Ntfy(config["ntfy_server_url"])
|
||||||
ntfy_cpu_temp_monitor = cpu.Tempature(ntfy, config["cpu_warning_temp"])
|
ntfy_cpu_temp_monitor = cpu.Tempature(ntfy, config["cpu_warning_temp"])
|
||||||
|
|
||||||
print(_monitoring_prompt)
|
print(config)
|
||||||
|
print(start_prompt(config["ntfy_server_url"]))
|
||||||
|
|
||||||
while True:
|
while True:
|
||||||
if not config["cpu_temp_check_disabled"]:
|
if not config["cpu_temp_check_disabled"]:
|
||||||
ntfy_cpu_temp_monitor.ntfy_check()
|
ntfy_cpu_temp_monitor.ntfy_check()
|
||||||
|
@ -1,15 +1,18 @@
|
|||||||
import requests
|
import requests
|
||||||
import time
|
|
||||||
|
|
||||||
from datetime import datetime
|
from datetime import datetime
|
||||||
|
|
||||||
|
def print_t(out: str):
|
||||||
|
t = datetime.now()
|
||||||
|
print(f"({t.strftime('%Y-%m-%d')})[{t.strftime('%H:%M:%S')}]: " + out)
|
||||||
|
|
||||||
class Ntfy:
|
class Ntfy:
|
||||||
def __init__(self, server: str):
|
def __init__(self, server: str):
|
||||||
self.server = server
|
self.server = server
|
||||||
|
|
||||||
def send(self, message: str):
|
def send(self, message: str):
|
||||||
print(f"[{datetime.fromtimestamp(time.time())}]: {message}")
|
print_t(message)
|
||||||
try:
|
try:
|
||||||
requests.post(self.server, data=message)
|
requests.post(self.server, data=message)
|
||||||
except Exception as err:
|
except Exception as err:
|
||||||
print(f"Ntfy failed. {err}")
|
print_t(f"Ntfy failed. \033[31m{err}\033[0m")
|
||||||
|
Reference in New Issue
Block a user