From 086c59cc03ac1d4b463760a722c5c4294ae8e09c Mon Sep 17 00:00:00 2001 From: Misaka_Company Date: Thu, 15 Jan 2026 08:52:20 +0800 Subject: [PATCH] Refactor Uptime Kuma heartbeat implementation to use UptimeKumaMonitor class, enhancing maintainability and logging capabilities --- excel_sync_to_sql.py | 43 +++---------- run_incremental_sync.py | 44 +++---------- uptime_kuma_utils.py | 138 ++++++++++++++++++++++++++++++++++++++++ 3 files changed, 156 insertions(+), 69 deletions(-) create mode 100644 uptime_kuma_utils.py diff --git a/excel_sync_to_sql.py b/excel_sync_to_sql.py index aeb7ea1..33eb661 100644 --- a/excel_sync_to_sql.py +++ b/excel_sync_to_sql.py @@ -7,7 +7,6 @@ import datetime import urllib.parse import warnings import time -import requests import pandas as pd import numpy as np from sqlalchemy import create_engine, text @@ -19,6 +18,11 @@ from log_utils import (log_error, log_warning, log_info, log_processing, log_fil log_start, log_complete, log_stop, LoggerManager) from config import (DB_CONFIG, CACHE_DIR, EXCEL_CONFIGS, BATCH_SIZE, TABLE_SCHEMA, EXCEL_SYNC_INTERVAL, EXCEL_SYNC_UPTIME_KUMA_CONFIG) +from uptime_kuma_utils import UptimeKumaMonitor + +# 初始化 Uptime Kuma 监控器 +excel_uptime_monitor = UptimeKumaMonitor(EXCEL_SYNC_UPTIME_KUMA_CONFIG) +excel_uptime_monitor.set_logger(log_warning) # ================= 抑制 openpyxl 的数据验证警告 ================= warnings.filterwarnings('ignore', category=UserWarning, module='openpyxl') @@ -295,28 +299,7 @@ class DataSynchronizer: log_error(f"生成 ContractData 失败: {e}", exc_info=True) # ================= Uptime Kuma 心跳 ================= - -_last_heartbeat_time = 0 - -def send_heartbeat(): - """发送心跳信号到 Uptime Kuma""" - global _last_heartbeat_time - - if not EXCEL_SYNC_UPTIME_KUMA_CONFIG.get('enabled', False): - return - - try: - url = EXCEL_SYNC_UPTIME_KUMA_CONFIG['push_url'] - params = { - 'status': 'up', - 'msg': 'OK', - 'ping': '' - } - response = requests.get(url, params=params, timeout=5) - response.raise_for_status() - _last_heartbeat_time = time.time() - except Exception as e: - log_warning(f"心跳发送失败: {e}") +# 使用 uptime_kuma_utils.UptimeKumaMonitor 替代原有实现 def main(): # 初始化日志管理器 @@ -347,7 +330,7 @@ def main(): log_info("=" * 70) # 启动时发送第一次心跳 - send_heartbeat() + excel_uptime_monitor.send_heartbeat() try: while True: @@ -365,9 +348,7 @@ def main(): # 等待下次同步，期间持续发送心跳 while time.time() < next_sync_time: # 检查是否需要发送心跳 - time_since_last_heartbeat = time.time() - _last_heartbeat_time - if time_since_last_heartbeat >= EXCEL_SYNC_UPTIME_KUMA_CONFIG['heartbeat_interval']: - send_heartbeat() + excel_uptime_monitor.check_and_send_heartbeat() # 短暂休眠 time.sleep(1) @@ -382,13 +363,7 @@ def main(): time.sleep(EXCEL_SYNC_INTERVAL) finally: # 停止时发送心跳停止信号 - if EXCEL_SYNC_UPTIME_KUMA_CONFIG.get('enabled', False): - try: - url = EXCEL_SYNC_UPTIME_KUMA_CONFIG['push_url'] - params = {'status': 'down', 'msg': 'Service stopped'} - requests.get(url, params=params, timeout=5) - except: - pass + excel_uptime_monitor.send_stop_signal() if __name__ == "__main__": main() \ No newline at end of file diff --git a/run_incremental_sync.py b/run_incremental_sync.py index 39463fa..c500c0a 100644 --- a/run_incremental_sync.py +++ b/run_incremental_sync.py @@ -2,15 +2,19 @@ import time import os import sys import pyodbc -import requests 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 uptime_kuma_utils import UptimeKumaMonitor # 初始化日志管理器 LoggerManager("run_incremental_sync", log_prefix="incremental") +# 初始化 Uptime Kuma 监控器 +uptime_monitor = UptimeKumaMonitor(UPTIME_KUMA_CONFIG) +uptime_monitor.set_logger(log_warning) + # ================= 主逻辑 ================= def process_sync_task(): @@ -185,29 +189,7 @@ def process_sync_task(): except: pass # ================= Uptime Kuma 心跳 ================= - -_last_heartbeat_time = 0 - -def send_heartbeat(): - """发送心跳信号到 Uptime Kuma""" - global _last_heartbeat_time - - if not UPTIME_KUMA_CONFIG.get('enabled', False): - return - - try: - url = UPTIME_KUMA_CONFIG['push_url'] - params = { - 'status': 'up', - 'msg': 'OK', - 'ping': '' - } - response = requests.get(url, params=params, timeout=5) - response.raise_for_status() - _last_heartbeat_time = time.time() - except Exception as e: - # 心跳失败不影响主逻辑，仅记录日志 - log_warning(f"心跳发送失败: {e}") +# 使用 uptime_kuma_utils.UptimeKumaMonitor 替代原有实现 if __name__ == "__main__": log_start("增量同步服务已启动 (配置驱动模式)") @@ -218,7 +200,7 @@ if __name__ == "__main__": log_info("=" * 70) # 启动时发送第一次心跳 - send_heartbeat() + uptime_monitor.send_heartbeat() try: while True: @@ -226,9 +208,7 @@ if __name__ == "__main__": has_work = process_sync_task() # 检查是否需要发送心跳 - time_since_last_heartbeat = time.time() - _last_heartbeat_time - if time_since_last_heartbeat >= UPTIME_KUMA_CONFIG['heartbeat_interval']: - send_heartbeat() + uptime_monitor.check_and_send_heartbeat() # 如果有工作，说明可能还有积压，休息短一点(0.1s) # 如果没工作，休息标准间隔(5s) @@ -243,10 +223,4 @@ if __name__ == "__main__": time.sleep(5) finally: # 停止时发送心跳停止信号（可选） - if UPTIME_KUMA_CONFIG.get('enabled', False): - try: - url = UPTIME_KUMA_CONFIG['push_url'] - params = {'status': 'down', 'msg': 'Service stopped'} - requests.get(url, params=params, timeout=5) - except: - pass \ No newline at end of file + uptime_monitor.send_stop_signal() \ No newline at end of file diff --git a/uptime_kuma_utils.py b/uptime_kuma_utils.py new file mode 100644 index 0000000..4274f35 --- /dev/null +++ b/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