139 lines
3.7 KiB
Python
139 lines
3.7 KiB
Python
"""
|
|
Uptime Kuma 心跳监控工具
|
|
|
|
用于向 Uptime Kuma 服务发送心跳信号，监控服务运行状态。
|
|
"""
|
|
|
|
import time
|
|
import requests
|
|
|
|
|
|
class UptimeKumaMonitor:
|
|
"""
|
|
Uptime Kuma 心跳监控器
|
|
|
|
示例:
|
|
from uptime_kuma_utils import UptimeKumaMonitor
|
|
|
|
# 初始化监控器
|
|
monitor = UptimeKumaMonitor({
|
|
'enabled': True,
|
|
'push_url': 'https://uptimekuma.example.com/api/push/xxx',
|
|
'heartbeat_interval': 59
|
|
})
|
|
|
|
# 启动时发送首次心跳
|
|
monitor.send_heartbeat()
|
|
|
|
# 主循环中定期发送心跳
|
|
while True:
|
|
monitor.check_and_send_heartbeat()
|
|
# ... 执行任务 ...
|
|
|
|
# 停止时发送停止信号
|
|
monitor.send_stop_signal()
|
|
"""
|
|
|
|
def __init__(self, config):
|
|
"""
|
|
初始化监控器
|
|
|
|
Args:
|
|
config (dict): 配置字典，包含:
|
|
- enabled (bool): 是否启用心跳
|
|
- push_url (str): Uptime Kuma 推送 URL
|
|
- heartbeat_interval (int): 心跳间隔(秒)
|
|
"""
|
|
self.config = config or {}
|
|
self.enabled = self.config.get('enabled', False)
|
|
self.push_url = self.config.get('push_url')
|
|
self.heartbeat_interval = self.config.get('heartbeat_interval', 60)
|
|
self._last_heartbeat_time = 0
|
|
self._logger = None
|
|
|
|
def set_logger(self, logger_func):
|
|
"""
|
|
设置日志记录函数
|
|
|
|
Args:
|
|
logger_func: 日志记录函数，如 log_warning, log_info 等
|
|
"""
|
|
self._logger = logger_func
|
|
|
|
def _log(self, func_name, message):
|
|
"""内部日志记录方法"""
|
|
if self._logger:
|
|
self._logger(message)
|
|
|
|
def send_heartbeat(self):
|
|
"""
|
|
发送心跳信号到 Uptime Kuma
|
|
|
|
Returns:
|
|
bool: 是否成功发送
|
|
"""
|
|
if not self.enabled or not self.push_url:
|
|
return False
|
|
|
|
try:
|
|
params = {
|
|
'status': 'up',
|
|
'msg': 'OK',
|
|
'ping': ''
|
|
}
|
|
response = requests.get(self.push_url, params=params, timeout=5)
|
|
response.raise_for_status()
|
|
self._last_heartbeat_time = time.time()
|
|
return True
|
|
except Exception as e:
|
|
self._log('warning', f"心跳发送失败: {e}")
|
|
return False
|
|
|
|
def send_stop_signal(self):
|
|
"""
|
|
发送停止信号到 Uptime Kuma
|
|
|
|
Returns:
|
|
bool: 是否成功发送
|
|
"""
|
|
if not self.enabled or not self.push_url:
|
|
return False
|
|
|
|
try:
|
|
params = {'status': 'down', 'msg': 'Service stopped'}
|
|
response = requests.get(self.push_url, params=params, timeout=5)
|
|
response.raise_for_status()
|
|
return True
|
|
except Exception as e:
|
|
# 停止信号失败不影响主逻辑
|
|
return False
|
|
|
|
def check_and_send_heartbeat(self):
|
|
"""
|
|
检查是否需要发送心跳，如果需要则发送
|
|
|
|
Returns:
|
|
bool: 是否发送了心跳
|
|
"""
|
|
if not self.enabled:
|
|
return False
|
|
|
|
time_since_last = time.time() - self._last_heartbeat_time
|
|
if time_since_last >= self.heartbeat_interval:
|
|
return self.send_heartbeat()
|
|
return False
|
|
|
|
def get_time_since_last_heartbeat(self):
|
|
"""
|
|
获取距离上次心跳的时间（秒）
|
|
|
|
Returns:
|
|
float: 距离上次心跳的秒数
|
|
"""
|
|
return time.time() - self._last_heartbeat_time
|
|
|
|
@property
|
|
def last_heartbeat_time(self):
|
|
"""获取上次心跳时间戳"""
|
|
return self._last_heartbeat_time
|