feat: add Uptime Kuma heartbeat monitoring

- Add heartbeat configuration to config.yaml
- Implement HeartbeatMonitor class in watch.py
- Send heartbeat every 40 seconds to Uptime Kuma push endpoint
- Update config.yaml.example with heartbeat settings
- Add heartbeat logging at INFO level

Co-Authored-By: Claude <noreply@anthropic.com>

.gitignore

@@ -21,6 +21,7 @@ Desktop.ini
 .DS_Store
 
 # Logs
+logs/
 *.log
 watch.log
 
@@ -32,3 +33,6 @@ config.yaml
 # Excel temp files
 Excel/
 *.tmp
+
+.claude/
+.agents/

CLAUDE.md

@@ -0,0 +1,40 @@
+# Excel to SQL Server Migration
+
+## 用途
+
+将局域网共享目录中的 Excel (.xlsm) 生产执行卡数据同步并迁移到 SQL Server。
+
+## 部署
+
+```bash
+python -m venv .venv
+.venv\Scripts\pip install -r requirements.txt
+cp config.yaml.example config.yaml  # 编辑数据库连接和 LAN 路径
+```
+
+## 命令
+
+```bash
+# 单次同步 LAN 文件并迁移变更的表
+.venv\Scripts\python watch.py --once
+
+# 守护进程模式（定时同步+迁移）
+.venv\Scripts\pythonw watch.py
+
+# 全量迁移（所有表，首次使用）
+.venv\Scripts\python migrate.py
+
+# 全量迁移（仅预览，不写库）
+.venv\Scripts\python migrate.py --dry-run
+
+# 清空后重写指定表
+.venv\Scripts\python migrate.py --truncate --table PG_2025 TM_2026
+```
+
+## 配置 (config.yaml)
+
+- `sql_server`: 数据库连接信息
+- `lan_sources`: 本地相对路径 → LAN UNC 绝对路径
+- `tables`: 目标表名与 Excel 源文件/sheet 的映射
+- `check_interval_minutes`: watch 守护进程同步间隔（分钟）
+- `logging`: 日志级别、目录、保留天数

config.yaml.example

@@ -1,5 +1,27 @@
 # Excel to SQL Server Migration Configuration
 
+# ================= 定时检查参数 =================
+check_interval_minutes: 5   # 每隔多少分钟扫描一次源文件
+
+# ================= 日志配置 =================
+logging:
+  level: "INFO"              # DEBUG / INFO / WARNING
+  log_dir: "logs"            # 日志目录（相对于项目根目录）
+  max_keep_days: 7          # 日志文件保留天数
+
+# ================= Uptime Kuma 心跳检测 =================
+heartbeat:
+  enabled: false             # 是否启用心跳检测
+  interval_seconds: 40       # 心跳发送间隔（秒）
+  url: "YOUR_UPTIME_KUMA_PUSH_URL"  # Uptime Kuma 推送端点 URL
+
+# ================= LAN 源文件路径 =================
+# key = 本地相对路径 (对应 tables.sources[].file)
+# value = 局域网 UNC 绝对路径
+lan_sources:
+  "PG/生产执行卡2022.xlsm": "\\\\192.168.110.113\\生产执行卡\\往年生产执行卡\\生产执行卡2022.xlsm"
+  # ... 其他 LAN 源文件路径
+
 # ================= SQL Server 连接 =================
 sql_server:
   driver: "ODBC Driver 18 for SQL Server"

logger.py

@@ -0,0 +1,65 @@
+"""
+Unified logging module.
+Provides a pre-configured logger with console + rotating file handlers.
+"""
+
+import logging
+import os
+from logging.handlers import TimedRotatingFileHandler
+
+
+def setup_logger(name='app', log_dir=None, level='INFO', max_keep_days=30):
+    """Create and configure a logger.
+
+    Args:
+        name: Logger name (used as basename for log file).
+        log_dir: Directory for log files. Defaults to <project_root>/logs.
+        level: Logging level string (DEBUG, INFO, WARNING, ERROR).
+        max_keep_days: Number of days to retain rotated log files.
+
+    Returns:
+        Configured logging.Logger instance.
+    """
+    if log_dir is None:
+        log_dir = 'logs'
+    # Resolve relative paths against project root
+    project_root = os.path.dirname(os.path.abspath(__file__))
+    if not os.path.isabs(log_dir):
+        log_dir = os.path.join(project_root, log_dir)
+    os.makedirs(log_dir, exist_ok=True)
+
+    logger = logging.getLogger(name)
+    logger.setLevel(getattr(logging, level.upper(), logging.INFO))
+
+    # Avoid adding duplicate handlers on repeated calls
+    if logger.handlers:
+        return logger
+
+    fmt = logging.Formatter(
+        '%(asctime)s [%(levelname)s] %(message)s',
+        datefmt='%Y-%m-%d %H:%M:%S'
+    )
+
+    # Console handler
+    ch = logging.StreamHandler()
+    ch.setLevel(getattr(logging, level.upper(), logging.INFO))
+    ch.setFormatter(fmt)
+    logger.addHandler(ch)
+
+    # File handler — rotate daily, keep max_keep_days backups
+    log_file = os.path.join(log_dir, f'{name}.log')
+    fh = TimedRotatingFileHandler(
+        log_file, when='midnight', interval=1,
+        backupCount=max_keep_days, encoding='utf-8'
+    )
+    fh.setLevel(getattr(logging, level.upper(), logging.INFO))
+    fh.setFormatter(fmt)
+    fh.suffix = '%Y-%m-%d'
+    logger.addHandler(fh)
+
+    return logger
+
+
+def get_logger(name='app'):
+    """Get an existing logger by name (does not reconfigure)."""
+    return logging.getLogger(name)

migrate.py

@@ -5,6 +5,7 @@ Configuration is loaded from config.yaml.
 """
 
 import argparse
+import logging
 import os
 import sys
 from collections import OrderedDict
@@ -14,6 +15,8 @@ import openpyxl
 import pyodbc
 import yaml
 
+logger = logging.getLogger('app')
+
 
 # ============================================================
 #  配置加载
@@ -126,12 +129,13 @@ def merge_sources(sources, excel_dir, primary_key, cfg):
     """合并多个工作表的数据。
     - 列名取并集（按出现顺序）
     - 数据按主键去重，后读覆盖先读
-    返回 (unified_headers, actual_pk_col, merged_data_ordered_dict)
+    返回 (unified_headers, actual_pk_col, merged_data_ordered_dict, source_row_counts)
     """
     all_headers = []       # 保持顺序的列名并集
     header_set = set()
     merged = OrderedDict()  # key=主键值, value=dict
     resolved_pk = primary_key  # 实际使用的主键列名
+    source_row_counts = []  # 各源的行数 [(file, sheet, count), ...]
 
     for src in sources:
         file_path = os.path.join(excel_dir, src['file'])
@@ -139,6 +143,7 @@ def merge_sources(sources, excel_dir, primary_key, cfg):
 
         headers, actual_pk_col, data = read_sheet(file_path, sheet_name, primary_key, cfg)
         resolved_pk = actual_pk_col
+        source_row_counts.append((src['file'], src['sheet'], len(data)))
 
         # 扩展统一列名集
         for h in headers:
@@ -156,7 +161,7 @@ def merge_sources(sources, excel_dir, primary_key, cfg):
             else:
                 merged[pk_val] = record
 
-    return all_headers, resolved_pk, merged
+    return all_headers, resolved_pk, merged, source_row_counts
 
 
 # ============================================================
@@ -166,7 +171,8 @@ def merge_sources(sources, excel_dir, primary_key, cfg):
 def get_connection(cfg):
     """根据配置建立 SQL Server 连接"""
     conn_str = get_connection_string(cfg)
-    return pyodbc.connect(conn_str, autocommit=False)
+    conn = pyodbc.connect(conn_str, autocommit=False)
+    return conn
 
 
 def create_schema(cursor, schema_name):
@@ -380,7 +386,7 @@ TYPE_CONVERTERS = {
 def batch_insert(cursor, table_name, headers, data_dict, cfg, actual_pk):
     """批量插入数据到 SQL Server。"""
     schema = cfg['schema']
-    batch_size = cfg.get('batch_size', 500)
+    batch_size = cfg.get('batch_size', 5000)
     type_rules = cfg['type_rules']
     full_name = f'[{schema}].[{table_name}]'
 
@@ -390,31 +396,30 @@ def batch_insert(cursor, table_name, headers, data_dict, cfg, actual_pk):
     placeholders = ', '.join(['?'] * len(headers))
     insert_sql = f"INSERT INTO {full_name} ({col_list}) VALUES ({placeholders})"
 
-    batch = []
-    total = 0
+    # 预构建所有行
+    all_rows = []
     for pk, record in data_dict.items():
         row = []
         for h in headers:
             val = record.get(h)
             if isinstance(val, str) and val.strip() == '':
                 val = None
-            # 强类型列清洗
-            if h in typed_cols and val is not None:
+            elif h in typed_cols and val is not None:
                 converter = TYPE_CONVERTERS[typed_cols[h]]
                 val = converter(val)
             row.append(val)
-        batch.append(row)
-        total += 1
+        all_rows.append(row)
 
-        if len(batch) >= batch_size:
-            cursor.executemany(insert_sql, batch)
-            cursor.commit()
-            batch = []
+    total = len(all_rows)
 
-    if batch:
-        cursor.executemany(insert_sql, batch)
-        cursor.commit()
+    # 启用 fast_executemany（pyodbc 5.x 在 cursor 上设置）
+    cursor.fast_executemany = True
 
+    # 分批 executemany，最后统一 commit
+    for i in range(0, total, batch_size):
+        cursor.executemany(insert_sql, all_rows[i:i + batch_size])
+
+    cursor.commit()
     return total
 
 
@@ -429,40 +434,40 @@ def migrate_table(cursor, table_config, cfg, dry_run=False, use_truncate=False):
     primary_key = cfg['primary_key']
     excel_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'Excel')
 
-    print(f"\n  迁移 {table_name} ...")
+    logger.info(f"迁移 {table_name} ...")
 
     # 1. 读取并合并数据
-    headers, actual_pk, merged = merge_sources(sources, excel_dir, primary_key, cfg)
+    headers, actual_pk, merged, source_row_counts = merge_sources(
+        sources, excel_dir, primary_key, cfg
+    )
 
     # 打印各源行数
-    for src in sources:
-        fpath = os.path.join(excel_dir, src['file'])
-        _, _, data = read_sheet(fpath, src['sheet'], primary_key, cfg)
-        print(f"    读取: {src['file']} > {src['sheet']} → {len(data):,} 行")
+    for file_name, sheet_name, count in source_row_counts:
+        logger.info(f"  读取: {file_name} > {sheet_name} → {count:,} 行")
 
-    print(f"    合并去重后: {len(merged):,} 行, {len(headers)} 列 (主键列: {actual_pk})")
+    logger.info(f"  合并去重后: {len(merged):,} 行, {len(headers)} 列 (主键列: {actual_pk})")
 
     if dry_run:
-        print(f"    [DRY-RUN] 建表 SQL:")
+        logger.info(f"  [DRY-RUN] 建表 SQL:")
         type_rules = cfg['type_rules']
         col_defs = ['[id] INT IDENTITY(1,1)']
         for h in headers:
             sql_type = infer_sql_type(h, type_rules, actual_pk)
             col_defs.append(f'  [{h}] {sql_type}')
         col_defs.append(f'  CONSTRAINT [PK_{table_name}] PRIMARY KEY ([{actual_pk}])')
-        print(f"    CREATE TABLE [{cfg['schema']}].[{table_name}] (")
-        print(',\n'.join(col_defs))
-        print("    );")
+        logger.info(f"  CREATE TABLE [{cfg['schema']}].[{table_name}] (")
+        logger.info(',\n'.join(col_defs))
+        logger.info("  );")
         return len(merged)
 
     # 2. 建表
     create_sql = ensure_table(cursor, table_name, headers, cfg, actual_pk, use_truncate=use_truncate)
     action = "TRUNCATE" if use_truncate and "TRUNCATE" in (create_sql or "") else "建表"
-    print(f"    {action}: [{cfg['schema']}].[{table_name}] ({len(headers)} 列)")
+    logger.info(f"  {action}: [{cfg['schema']}].[{table_name}] ({len(headers)} 列)")
 
     # 3. 插入数据
     total = batch_insert(cursor, table_name, headers, merged, cfg, actual_pk)
-    print(f"    插入: {total:,} 行 done")
+    logger.info(f"  插入: {total:,} 行 done")
 
     return total
 
@@ -480,15 +485,25 @@ def main():
     # 加载配置
     cfg = load_config(args.config)
 
+    # 初始化日志
+    from logger import setup_logger
+    log_cfg = cfg.get('logging', {})
+    setup_logger(
+        'app',
+        log_dir=log_cfg.get('log_dir'),
+        level=log_cfg.get('level', 'INFO'),
+        max_keep_days=log_cfg.get('max_keep_days', 30),
+    )
+
     # 按参数筛选表
     if args.table:
         specified = set(args.table)
         tables = [t for t in cfg['tables'] if t['table'] in specified]
         missing = specified - {t['table'] for t in tables}
         if missing:
-            print(f"  错误: 未找到表 {', '.join(missing)}")
+            logger.error(f"未找到表 {', '.join(missing)}")
             available = ', '.join(t['table'] for t in cfg['tables'])
-            print(f"  可用表: {available}")
+            logger.info(f"可用表: {available}")
             sys.exit(1)
     else:
         tables = cfg['tables']
@@ -496,11 +511,11 @@ def main():
     total_tables = len(tables)
     primary_key = cfg['primary_key']
 
-    print(f"{'=' * 60}")
-    print(f"  Excel → SQL Server 迁移工具")
-    print(f"  目标: [{cfg['schema']}], {total_tables} 张表")
-    print(f"  模式: {'DRY-RUN' if args.dry_run else 'LIVE'}")
-    print(f"{'=' * 60}")
+    logger.info("=" * 50)
+    logger.info(f"Excel → SQL Server 迁移工具")
+    logger.info(f"  目标: [{cfg['schema']}], {total_tables} 张表")
+    logger.info(f"  模式: {'DRY-RUN' if args.dry_run else 'LIVE'}")
+    logger.info("=" * 50)
 
     conn = None
     cursor = None
@@ -509,11 +524,11 @@ def main():
         conn = get_connection(cfg)
         cursor = conn.cursor()
         create_schema(cursor, cfg['schema'])
-        print(f"\n  Schema [{cfg['schema']}] 已就绪")
+        logger.info(f"Schema [{cfg['schema']}] 已就绪")
 
     total_rows = 0
     for i, table_cfg in enumerate(tables, 1):
-        print(f"\n[{i}/{total_tables}]", end='')
+        logger.info(f"[{i}/{total_tables}]")
         rows = migrate_table(cursor, table_cfg, cfg, dry_run=args.dry_run,
                              use_truncate=args.truncate)
         total_rows += rows
@@ -522,9 +537,9 @@ def main():
         cursor.close()
         conn.close()
 
-    print(f"\n{'=' * 60}")
-    print(f"  完成! 共迁移 {total_tables} 张表, {total_rows:,} 行")
-    print(f"{'=' * 60}")
+    logger.info("=" * 50)
+    logger.info(f"完成! 共迁移 {total_tables} 张表, {total_rows:,} 行")
+    logger.info("=" * 50)
 
 
 if __name__ == '__main__':

sync.py

@@ -3,9 +3,12 @@ LAN file sync module.
 Detects file changes, copies from LAN UNC paths to local, and determines which tables need migration.
 """
 
+import logging
 import os
 import shutil
 
+logger = logging.getLogger('app')
+
 
 def sync_file(local_path, lan_path):
     """Sync a single file from LAN to local.
@@ -16,6 +19,7 @@ def sync_file(local_path, lan_path):
     """
     try:
         if not os.path.exists(lan_path):
+            logger.error(f"LAN 路径不可达: {lan_path}")
             return 'error', f'LAN path not accessible: {lan_path}'
 
         lan_mtime = os.path.getmtime(lan_path)
@@ -24,6 +28,7 @@ def sync_file(local_path, lan_path):
         if local_existed:
             local_mtime = os.path.getmtime(local_path)
             if local_mtime == lan_mtime:
+                logger.debug(f"跳过 (mtime一致): {os.path.basename(local_path)}")
                 return 'skipped', None
 
         # Ensure parent directory exists
@@ -41,11 +46,14 @@ def sync_file(local_path, lan_path):
             raise
 
         if not local_existed:
+            logger.info(f"copied: {os.path.basename(local_path)}")
             return 'copied', None
         else:
+            logger.info(f"updated: {os.path.basename(local_path)}")
             return 'updated', None
 
     except Exception as e:
+        logger.error(f"同步失败: {local_path} - {e}")
         return 'error', str(e)
 
 

watch.py

@@ -5,41 +5,90 @@ Watch daemon - periodically syncs LAN Excel files and migrates changed tables to
 import argparse
 import logging
 import os
-import sys
 import time
+import threading
+import urllib.request
+import urllib.error
 
-import yaml
-
+from logger import setup_logger
 from migrate import load_config, get_connection, create_schema, migrate_table
 from sync import sync_all_files, determine_tables_to_migrate
 
-
-def setup_logging(log_file):
-    """Configure logging to both console and file."""
-    logger = logging.getLogger()
-    logger.setLevel(logging.INFO)
-
-    fmt = logging.Formatter(
-        '%(asctime)s [%(levelname)s] %(message)s',
-        datefmt='%Y-%m-%d %H:%M:%S'
-    )
-
-    # Console handler
-    ch = logging.StreamHandler(sys.stdout)
-    ch.setLevel(logging.INFO)
-    ch.setFormatter(fmt)
-    logger.addHandler(ch)
-
-    # File handler
-    fh = logging.FileHandler(log_file, encoding='utf-8')
-    fh.setLevel(logging.INFO)
-    fh.setFormatter(fmt)
-    logger.addHandler(fh)
-
-    return logger
+logger = logging.getLogger('app')
 
 
-def run_once(cfg, logger):
+class HeartbeatMonitor:
+    """Uptime Kuma heartbeat monitor running in a separate thread."""
+
+    def __init__(self, url, interval_seconds):
+        self.url = url
+        self.interval_seconds = interval_seconds
+        self.running = False
+        self.thread = None
+        logger.info(f"心跳监控初始化: 间隔 {interval_seconds} 秒")
+
+    def _send_heartbeat(self):
+        """Send a heartbeat request to Uptime Kuma."""
+        try:
+            # Add current timestamp to ping parameter
+            ping = str(int(time.time()))
+            full_url = f"{self.url}{ping}"
+
+            request = urllib.request.Request(full_url, method='GET')
+            request.add_header('User-Agent', 'Excel-to-SQL-Server-WatchDaemon/1.0')
+
+            with urllib.request.urlopen(request, timeout=10) as response:
+                if response.status == 200:
+                    logger.info(f"心跳发送成功")
+                else:
+                    logger.warning(f"心跳返回异常状态码: {response.status}")
+        except urllib.error.URLError as e:
+            logger.error(f"心跳发送失败 (网络错误): {e}")
+        except Exception as e:
+            logger.error(f"心跳发送失败 (未知错误): {e}")
+
+    def _run_loop(self):
+        """Main heartbeat loop running in the thread."""
+        logger.info("心跳监控线程已启动")
+
+        while self.running:
+            self._send_heartbeat()
+
+            # Wait for the interval or until stopped
+            for _ in range(self.interval_seconds):
+                if not self.running:
+                    break
+                time.sleep(1)
+
+        logger.info("心跳监控线程已停止")
+
+    def start(self):
+        """Start the heartbeat monitor thread."""
+        if self.running:
+            logger.warning("心跳监控已在运行")
+            return
+
+        self.running = True
+        self.thread = threading.Thread(target=self._run_loop, daemon=True)
+        self.thread.start()
+        logger.info("心跳监控已启动")
+
+    def stop(self):
+        """Stop the heartbeat monitor thread."""
+        if not self.running:
+            return
+
+        logger.info("正在停止心跳监控...")
+        self.running = False
+
+        # Wait for thread to finish
+        if self.thread and self.thread.is_alive():
+            self.thread.join(timeout=5)
+
+        logger.info("心跳监控已停止")
+
+
+def run_once(cfg):
     """Execute one sync + migrate cycle."""
     excel_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'Excel')
 
@@ -47,13 +96,13 @@ def run_once(cfg, logger):
     logger.info("开始同步 LAN 文件...")
     sync_results = sync_all_files(cfg, excel_dir)
 
-    for rel_path, (status, error_msg) in sync_results.items():
-        if status == 'error':
-            logger.error(f"  同步失败: {rel_path} - {error_msg}")
-        elif status in ('copied', 'updated'):
-            logger.info(f"  {status}: {rel_path}")
-        else:
-            logger.debug(f"  跳过: {rel_path}")
+    # 如果全部同步失败，大概率是网络中断
+    all_failed = sync_results and all(s == 'error' for s, _ in sync_results.values())
+    if all_failed:
+        logger.warning(
+            f"所有 LAN 源文件不可达 ({len(sync_results)} 个)，可能网络中断，跳过本轮同步"
+        )
+        return
 
     # 2. Determine which tables need migration
     tables_to_migrate = determine_tables_to_migrate(cfg, sync_results)
@@ -78,9 +127,9 @@ def run_once(cfg, logger):
 
             try:
                 rows = migrate_table(cursor, table_cfg, cfg, use_truncate=True)
-                logger.info(f"  迁移完成: {table_name} ({rows:,} 行)")
+                logger.info(f"迁移完成: {table_name} ({rows:,} 行)")
             except Exception as e:
-                logger.error(f"  迁移失败: {table_name} - {e}", exc_info=True)
+                logger.error(f"迁移失败: {table_name} - {e}", exc_info=True)
                 # Rollback any uncommitted transaction for this table
                 try:
                     conn.rollback()
@@ -106,12 +155,18 @@ def main():
                         help='覆盖配置文件中的 check_interval_minutes')
     args = parser.parse_args()
 
-    # Setup logging
-    log_file = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'watch.log')
-    logger = setup_logging(log_file)
-
     # Load config
     cfg = load_config(args.config)
+
+    # Setup logging
+    log_cfg = cfg.get('logging', {})
+    setup_logger(
+        'app',
+        log_dir=log_cfg.get('log_dir'),
+        level=log_cfg.get('level', 'INFO'),
+        max_keep_days=log_cfg.get('max_keep_days', 30),
+    )
+
     interval = args.interval or cfg.get('check_interval_minutes', 30)
 
     logger.info("=" * 50)
@@ -119,67 +174,36 @@ def main():
     logger.info(f"  同步间隔: {interval} 分钟")
     logger.info("=" * 50)
 
-    first_run = True
+    # Initialize heartbeat monitor if configured
+    heartbeat_monitor = None
+    heartbeat_cfg = cfg.get('heartbeat', {})
+    if heartbeat_cfg.get('enabled', False):
+        url = heartbeat_cfg.get('url', '').strip()
+        interval_seconds = heartbeat_cfg.get('interval_seconds', 40)
 
-    while True:
-        try:
-            if first_run:
-                logger.info("首次运行，强制同步并迁移所有表...")
-                # On first run, force all tables to migrate by marking all files as 'updated'
-                excel_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'Excel')
-                sync_results = sync_all_files(cfg, excel_dir)
+        if url:
+            heartbeat_monitor = HeartbeatMonitor(url, interval_seconds)
+            heartbeat_monitor.start()
+        else:
+            logger.warning("心跳已启用但 URL 未配置，跳过心跳监控")
 
-                # Log sync results
-                for rel_path, (status, error_msg) in sync_results.items():
-                    if status == 'error':
-                        logger.error(f"  同步失败: {rel_path} - {error_msg}")
-                    else:
-                        logger.info(f"  {status}: {rel_path}")
+    try:
+        while True:
+            try:
+                run_once(cfg)
+            except Exception as e:
+                logger.error(f"运行周期异常: {e}", exc_info=True)
 
-                # Force all tables to migrate on first run
-                all_tables = {t['table'] for t in cfg['tables']}
+            if args.once:
+                logger.info("--once 模式，退出")
+                break
 
-                conn = None
-                try:
-                    conn = get_connection(cfg)
-                    cursor = conn.cursor()
-                    create_schema(cursor, cfg['schema'])
-
-                    for table_cfg in cfg['tables']:
-                        table_name = table_cfg['table']
-                        try:
-                            rows = migrate_table(cursor, table_cfg, cfg, use_truncate=True)
-                            logger.info(f"  迁移完成: {table_name} ({rows:,} 行)")
-                        except Exception as e:
-                            logger.error(f"  迁移失败: {table_name} - {e}", exc_info=True)
-                            try:
-                                conn.rollback()
-                            except Exception:
-                                pass
-
-                    cursor.close()
-                except Exception as e:
-                    logger.error(f"数据库连接失败: {e}", exc_info=True)
-                finally:
-                    if conn:
-                        try:
-                            conn.close()
-                        except Exception:
-                            pass
-
-                first_run = False
-            else:
-                run_once(cfg, logger)
-
-        except Exception as e:
-            logger.error(f"运行周期异常: {e}", exc_info=True)
-
-        if args.once:
-            logger.info("--once 模式，退出")
-            break
-
-        logger.info(f"下次检查: {interval} 分钟后")
-        time.sleep(interval * 60)
+            logger.info(f"下次检查: {interval} 分钟后")
+            time.sleep(interval * 60)
+    finally:
+        # Stop heartbeat monitor on exit
+        if heartbeat_monitor:
+            heartbeat_monitor.stop()
 
 
 if __name__ == '__main__':