BLD_sync/uptime_kuma_utils.py

"""
Uptime Kuma 心跳监控工具

用于向 Uptime Kuma 服务发送心跳信号，监控服务运行状态。

设计要点:
- 推荐通过 start() 启动后台心跳线程, 与业务主循环解耦 ——
  心跳的网络耗时 / 重试不会阻塞业务逻辑。
- 单次心跳自带重试 (MAX_RETRIES), 容忍 Uptime Kuma 服务偶发的慢响应 / 超时 / 4xx。
- send_heartbeat() / check_and_send_heartbeat() / send_stop_signal() 保留为
  向后兼容的同步接口 (excel_sync_to_sql.py 等仍在使用)。
"""

import time
import threading
import requests


# 单次请求超时(秒) —— 适当放宽, 容忍 Uptime Kuma 偶发的慢响应
REQUEST_TIMEOUT = 10
# 单次心跳的最大尝试次数 (含首次)
MAX_RETRIES = 3
# 重试间隔(秒)
RETRY_BACKOFF = 2
# stop() 时等待后台线程退出的最长时间(秒); 线程为 daemon, 超时也不会阻塞进程退出
STOP_JOIN_TIMEOUT = 40


class UptimeKumaMonitor:
    """
    Uptime Kuma 心跳监控器

    推荐 (后台线程模式, 与业务循环解耦)::

        monitor = UptimeKumaMonitor({
            'enabled': True,
            'push_url': 'https://uptimekuma.example.com/api/push/xxx',
            'heartbeat_interval': 59
        })
        monitor.start()          # 启动后台线程: 立即发首跳, 之后按间隔周期发送
        try:
            ...                  # 业务主循环
        finally:
            monitor.stop()       # 停止线程并发送一次 down 信号

    向后兼容 (同步模式, 不推荐新代码使用)::

        monitor.send_heartbeat()              # 同步发一次 (带重试)
        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
        # 后台线程相关
        self._thread = None
        self._stop_event = threading.Event()

    def set_logger(self, logger_func):
        """设置日志记录函数 (如 log_warning)。"""
        self._logger = logger_func

    def _log(self, message):
        if self._logger:
            self._logger(message)

    # ================= 单次请求 + 重试 =================

    def _do_request(self, status, msg):
        """发送一次心跳请求, 返回是否成功 (HTTP 2xx)。失败时记录日志。"""
        try:
            params = {'status': status, 'msg': msg, 'ping': ''}
            response = requests.get(self.push_url, params=params, timeout=REQUEST_TIMEOUT)
            response.raise_for_status()
            return True
        except Exception as e:
            self._log(f"心跳发送失败 (status={status}): {e}")
            return False

    def _send_with_retry(self, status='up', msg='OK'):
        """
        带重试的心跳发送: 任一尝试成功即视为成功。
        成功发送 'up' 时更新最近心跳时间。返回是否最终成功。
        """
        if not self.enabled or not self.push_url:
            return False
        for attempt in range(1, MAX_RETRIES + 1):
            if self._do_request(status, msg):
                if status == 'up':
                    self._last_heartbeat_time = time.time()
                return True
            if attempt < MAX_RETRIES:
                time.sleep(RETRY_BACKOFF)
        return False

    # ================= 后台线程模式 (推荐) =================

    def start(self):
        """
        启动后台心跳线程: 立即发送一次, 之后按 heartbeat_interval 周期发送。
        与业务主循环完全解耦, 心跳的网络耗时 / 重试不会阻塞业务。
        重复调用安全 (已在运行则直接返回)。
        """
        if not self.enabled or not self.push_url:
            return
        if self._thread is not None and self._thread.is_alive():
            return
        self._stop_event.clear()
        self._thread = threading.Thread(
            target=self._heartbeat_loop, daemon=True, name='uptime-kuma-heartbeat')
        self._thread.start()

    def _heartbeat_loop(self):
        # 启动立即发一次
        self._send_with_retry()
        # 周期发送, 直到 stop() 触发 _stop_event
        # Event.wait(interval) 在超时返回 False (继续发), 被置位时返回 True (退出)
        while not self._stop_event.wait(self.heartbeat_interval):
            self._send_with_retry()

    def stop(self):
        """停止后台心跳线程, 并向 Uptime Kuma 发送一次 down 信号。"""
        self._stop_event.set()
        if self._thread is not None:
            self._thread.join(timeout=STOP_JOIN_TIMEOUT)
            self._thread = None
        self.send_stop_signal()

    # ================= 向后兼容的同步接口 =================

    def send_heartbeat(self):
        """同步发送一次心跳 (自带重试)。向后兼容用法。"""
        return self._send_with_retry()

    def send_stop_signal(self):
        """发送停止(down)信号。失败不影响主逻辑。"""
        if not self.enabled or not self.push_url:
            return False
        return self._do_request('down', 'Service stopped')

    def check_and_send_heartbeat(self):
        """按间隔节流后同步发送心跳。向后兼容用法。"""
        if not self.enabled:
            return False
        if time.time() - self._last_heartbeat_time >= self.heartbeat_interval:
            return self.send_heartbeat()
        return False

    def get_time_since_last_heartbeat(self):
        """获取距离上次心跳的时间(秒)。"""
        return time.time() - self._last_heartbeat_time

    @property
    def last_heartbeat_time(self):
        """获取上次心跳时间戳。"""
        return self._last_heartbeat_time