From dd27cbaf9791a9ad161be37f813f06ffb4379e76 Mon Sep 17 00:00:00 2001 From: Misaka_Company Date: Mon, 15 Jun 2026 13:06:53 +0800 Subject: [PATCH] fix: remove redundant ntfy messages and harden Uptime Kuma heartbeat run_incremental_sync.py: - Replace ntfy-pushing log_start/log_success with log_info. Liveness is owned by Uptime Kuma's heartbeat, so the startup and per-table success messages no longer duplicate Uptime's notifications. uptime_kuma_utils.py (shared util, backward compatible): - Run the heartbeat in a background daemon thread (new start()/stop()), decoupled from the sync loop so heartbeat network time/retries never delay syncing. - Add retry (up to 3 attempts) and raise request timeout 5s -> 10s to tolerate Uptime Kuma's intermittent slow responses / transient 4xx. - Keep send_heartbeat/check_and_send_heartbeat/send_stop_signal as backward-compatible wrappers; excel_sync_to_sql.py still uses them and now benefits from the longer timeout + retry. Co-Authored-By: Claude --- run_incremental_sync.py | 22 ++--- uptime_kuma_utils.py | 172 +++++++++++++++++++++++----------------- 2 files changed, 112 insertions(+), 82 deletions(-) diff --git a/run_incremental_sync.py b/run_incremental_sync.py index 65876fe..46e32ff 100644 --- a/run_incremental_sync.py +++ b/run_incremental_sync.py @@ -4,8 +4,8 @@ import sys import pyodbc from config import SQL_SERVER_CONN, ACCESS_DRIVER, SYNC_MAPPING, POLL_INTERVAL, LOG_TABLE_CONFIG, UPTIME_KUMA_CONFIG import db_utils -from log_utils import (LoggerManager, log_success, log_error, log_warning, log_info, log_processing, - log_skip, log_critical, log_start, log_stop, log_file, log_database, log_sync) +from log_utils import (LoggerManager, log_error, log_warning, log_info, log_processing, + log_skip, log_critical, log_stop, log_file, log_database, log_sync) from uptime_kuma_utils import UptimeKumaMonitor # 初始化日志管理器 @@ -313,11 +313,11 @@ def process_sync_task(): file_error_count += 1 continue - # 同步成功 + # 同步成功 (仅本地日志记录, 不推送 ntfy; 服务存活状态由 Uptime Kuma 心跳负责) msg = f"表 [{target_table}] 同步并校验通过: {len(inserted_pk_set)} 条入库" if removed_count > 0: msg += f", {removed_count} 条随源删除" - log_success(msg) + log_info(msg) file_success_count += 1 file_total_records += len(record_ids) @@ -368,7 +368,8 @@ def process_sync_task(): # 使用 uptime_kuma_utils.UptimeKumaMonitor 替代原有实现 if __name__ == "__main__": - log_start("增量同步服务已启动 (配置驱动模式 + 逐主键落库校验)") + # 仅本地日志记录, 不推送 ntfy 启动消息 (服务存活状态由 Uptime Kuma 心跳负责, 避免重复通知) + log_info("增量同步服务已启动 (配置驱动模式 + 逐主键落库校验)") log_info(f"轮询间隔: {POLL_INTERVAL} 秒") log_info(f"监控配置: {len(SYNC_MAPPING)} 个文件") log_info(f"提交后复核: {'开启' if ENABLE_POST_COMMIT_VERIFY else '关闭'}") @@ -376,16 +377,15 @@ if __name__ == "__main__": log_info(f"心跳间隔: {UPTIME_KUMA_CONFIG['heartbeat_interval']} 秒") log_info("=" * 70) - # 启动时发送第一次心跳 - uptime_monitor.send_heartbeat() + # 启动后台心跳线程: 立即发送首跳, 之后按间隔周期发送 (与同步主循环解耦) + uptime_monitor.start() try: while True: try: has_work = process_sync_task() - # 检查是否需要发送心跳 - uptime_monitor.check_and_send_heartbeat() + # 心跳由后台线程独立周期发送, 此处无需再调用 # 如果有工作，说明可能还有积压，休息短一点(0.1s) # 如果没工作，休息标准间隔 @@ -399,5 +399,5 @@ if __name__ == "__main__": log_critical(f"主循环崩溃: {e}") time.sleep(5) finally: - # 停止时发送心跳停止信号（可选） - uptime_monitor.send_stop_signal() \ No newline at end of file + # 停止后台心跳线程, 并向 Uptime Kuma 发送 down 信号 + uptime_monitor.stop() \ No newline at end of file diff --git a/uptime_kuma_utils.py b/uptime_kuma_utils.py index 4274f35..9f79c78 100644 --- a/uptime_kuma_utils.py +++ b/uptime_kuma_utils.py @@ -2,44 +2,58 @@ 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 心跳监控器 - 示例: - from uptime_kuma_utils import UptimeKumaMonitor + 推荐 (后台线程模式, 与业务循环解耦):: - # 初始化监控器 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() + 向后兼容 (同步模式, 不推荐新代码使用):: - # 主循环中定期发送心跳 - while True: - monitor.check_and_send_heartbeat() - # ... 执行任务 ... - - # 停止时发送停止信号 + monitor.send_heartbeat() # 同步发一次 (带重试) + monitor.check_and_send_heartbeat() # 按间隔节流后同步发送 monitor.send_stop_signal() """ def __init__(self, config): """ - 初始化监控器 - Args: - config (dict): 配置字典，包含: + config (dict): 配置字典, 包含: - enabled (bool): 是否启用心跳 - push_url (str): Uptime Kuma 推送 URL - heartbeat_interval (int): 心跳间隔(秒) @@ -50,89 +64,105 @@ class UptimeKumaMonitor: 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): - """ - 设置日志记录函数 - - Args: - logger_func: 日志记录函数，如 log_warning, log_info 等 - """ + """设置日志记录函数 (如 log_warning)。""" self._logger = logger_func - def _log(self, func_name, message): - """内部日志记录方法""" + def _log(self, message): if self._logger: self._logger(message) - def send_heartbeat(self): - """ - 发送心跳信号到 Uptime Kuma + # ================= 单次请求 + 重试 ================= - Returns: - bool: 是否成功发送 + 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 - 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 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): - """ - 发送停止信号到 Uptime Kuma - - Returns: - bool: 是否成功发送 - """ + """发送停止(down)信号。失败不影响主逻辑。""" 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 + return self._do_request('down', 'Service stopped') 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: + if time.time() - self._last_heartbeat_time >= 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