Refactor Uptime Kuma heartbeat implementation to use UptimeKumaMonitor class, enhancing maintainability and logging capabilities

uptime_kuma_utils.py

@@ -0,0 +1,138 @@
+"""
+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