Merge feat/auto-create-table-and-verify-sync into main

Merge auto table creation with per-PK sync verification, combining
both features: target tables are auto-created when missing, and
every sync batch is verified by primary key before marking synced.

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

.claude/settings.local.json

@@ -1,14 +0,0 @@
-{
-  "permissions": {
-    "allow": [
-      "Bash(python etl_manager.py --help)",
-      "Bash(python run_incremental_sync.py)",
-      "Bash(python -c \"import run_incremental_sync; print\\(''run_incremental_sync.py: import OK''\\)\")",
-      "Bash(python -c \"import etl_manager; print\\(''etl_manager.py: import OK''\\)\")",
-      "Bash(python -c \"import init_full_sync; print\\(''init_full_sync.py: import OK''\\)\")",
-      "Bash(python -c \"import migration; print\\(''migration.py: import OK''\\)\")",
-      "Bash(python migration.py)",
-      "Bash(python -c \"import log_utils; from log_utils import LoggerManager, log_start, log_complete; print\\(''log_utils: OK''\\)\")"
-    ]
-  }
-}

.gitignore

@@ -5,3 +5,7 @@ dist
 log
 *.spec
 temp
+
+# Claude 临时文件
+.claude/
+tmpclaude-*

config/__init__.py

@@ -12,8 +12,8 @@ from .field_mappings import TABLE_SCHEMA, CONTRACT_MAPPING
 
 # 应用设置
 from .app_settings import (
-    LOG_TABLE_CONFIG, NTFY_CONFIG,
-    POLL_INTERVAL, BATCH_SIZE, CACHE_DIR, TEMP_DIR,
+    LOG_TABLE_CONFIG, NTFY_CONFIG, UPTIME_KUMA_CONFIG, EXCEL_SYNC_UPTIME_KUMA_CONFIG,
+    POLL_INTERVAL, EXCEL_SYNC_INTERVAL, BATCH_SIZE, CACHE_DIR, TEMP_DIR,
     EXECUTION_CARD_FIELDS, CONTRACT_DATA_FIELDS, CONTRACT_DATA_MAPPING
 )
 
@@ -26,7 +26,7 @@ __all__ = [
     # Field Mappings
     'TABLE_SCHEMA', 'CONTRACT_MAPPING',
     # App Settings
-    'LOG_TABLE_CONFIG', 'NTFY_CONFIG',
-    'POLL_INTERVAL', 'BATCH_SIZE', 'CACHE_DIR', 'TEMP_DIR',
+    'LOG_TABLE_CONFIG', 'NTFY_CONFIG', 'UPTIME_KUMA_CONFIG', 'EXCEL_SYNC_UPTIME_KUMA_CONFIG',
+    'POLL_INTERVAL', 'EXCEL_SYNC_INTERVAL', 'BATCH_SIZE', 'CACHE_DIR', 'TEMP_DIR',
     'EXECUTION_CARD_FIELDS', 'CONTRACT_DATA_FIELDS', 'CONTRACT_DATA_MAPPING'
 ]

config/app_settings.py

@@ -28,9 +28,25 @@ NTFY_CONFIG = {
     }
 }
 
+# ================= Uptime Kuma 心跳配置 =================
+# 增量同步服务心跳
+UPTIME_KUMA_CONFIG = {
+    'enabled': True,
+    'push_url': 'https://uptimekuma.server10086.icu/api/push/clMSgOQs4CJeF7DJAOiHaFaaZzoL8eB9',
+    'heartbeat_interval': 59  # 心跳间隔(秒)，需要与 Uptime Kuma 设置一致
+}
+
+# Excel 同步服务心跳
+EXCEL_SYNC_UPTIME_KUMA_CONFIG = {
+    'enabled': True,
+    'push_url': 'https://uptimekuma.server10086.icu/api/push/MEXi5DYnC3nTMs2OfMqdelK3FawymtVY',
+    'heartbeat_interval': 59  # 心跳间隔(秒)，需要与 Uptime Kuma 设置一致
+}
+
 # ================= 运行参数 =================
 # 合并原 config.py 和 update_config.py 的配置
-POLL_INTERVAL = 5     # 轮询间隔(秒)
+POLL_INTERVAL = 30     # 轮询间隔(秒)
+EXCEL_SYNC_INTERVAL = 600  # Excel 同步周期(秒)，默认 10 分钟
 BATCH_SIZE = 10000    # 批量处理大小
 CACHE_DIR = os.path.join(os.getcwd(), "temp")  # Excel 缓存目录
 TEMP_DIR = os.path.join(os.getcwd(), "temp")           # 临时目录（兼容 migration.py）

config/file_sources.py

@@ -26,6 +26,16 @@ SYNC_MAPPING = {
             "target_schema": "productionContractData",
             "target_table": "25年温度计合同数据",
             "pk_col": "ID"
+        },
+        "25年变送器合同数据": {
+            "target_schema": "productionContractData",
+            "target_table": "25年变送器合同数据",
+            "pk_col": "ID"
+        },
+        "26年变送器合同数据": {
+            "target_schema": "productionContractData",
+            "target_table": "26年变送器合同数据",
+            "pk_col": "ID"
         }
     },
     r"\\192.168.110.114\生产进度表\2026年数据\成品入库.accdb": {
@@ -86,6 +96,11 @@ SYNC_MAPPING = {
             "target_schema": "machining",
             "target_table": "喷涂寄出_YEAR2026",
             "pk_col": "ID"
+        },
+        "车波纹": {
+            "target_schema": "machining",
+            "target_table": "车波纹_YEAR2026",
+            "pk_col": "ID"
         }
     },
     r"\\192.168.110.114\生产进度表\2026年数据\计划.accdb": {
@@ -296,6 +311,13 @@ SYNC_MAPPING = {
             "target_table": "执行卡下发记录_YEAR2026",
             "pk_col": "ID"
         }
+    },
+    r"\\192.168.110.114\生产进度表\2026年数据\弯管车间.accdb": {
+        "执行卡下发记录": {
+            "target_schema": "tubeBending",
+            "target_table": "烘洗_YEAR2026",
+            "pk_col": "ID"
+        }
     }
 }
 

db_utils.py

@@ -1,7 +1,82 @@
 # db_utils.py
 import pyodbc
+import datetime
+import decimal
 from config import SQL_SERVER_CONN, ACCESS_DRIVER
 
+
+# ================= 表存在性检查 & 自动建表 =================
+
+def table_exists(cursor, schema, table):
+    """检查 SQL Server 表是否存在"""
+    query = """
+        SELECT COUNT(*)
+        FROM sys.tables t
+        JOIN sys.schemas s ON t.schema_id = s.schema_id
+        WHERE s.name = ? AND t.name = ?
+    """
+    cursor.execute(query, (schema, table))
+    return cursor.fetchone()[0] > 0
+
+
+def ensure_schema(cursor, schema):
+    """确保 schema 存在，不存在则创建"""
+    cursor.execute("SELECT SCHEMA_ID(?)", (schema,))
+    if cursor.fetchone()[0] is None:
+        cursor.execute(f"CREATE SCHEMA [{schema}]")
+
+
+def _access_col_to_sql(col_info, pk_col):
+    """将 Access 列描述 (cursor.description 元组) 转为 SQL Server 列定义
+
+    Access ODBC 驱动的 type_code 是 Python 类型对象:
+      int              → INT
+      str              → NVARCHAR(size)  (size > 4000 时为 Memo 字段 → NVARCHAR(MAX))
+      datetime.datetime → DATETIME
+      float            → FLOAT
+      decimal.Decimal  → DECIMAL(p,s)
+      bool             → BIT
+    """
+    col_name, type_code, _, size, precision, scale, nullable = col_info
+    is_pk = (col_name == pk_col)
+
+    if type_code is int:
+        sql_type = "INT"
+        if is_pk:
+            sql_type += " IDENTITY(1,1) PRIMARY KEY"
+    elif type_code is float:
+        sql_type = "FLOAT"
+    elif type_code is bool:
+        sql_type = "BIT"
+    elif type_code is datetime.datetime:
+        sql_type = "DATETIME"
+    elif type_code is decimal.Decimal:
+        sql_type = f"DECIMAL({precision or 18}, {scale or 0})"
+    elif type_code is str:
+        # size > 4000 → Access Memo 字段，用 NVARCHAR(MAX)
+        if not size or size <= 0 or size > 4000:
+            sql_type = "NVARCHAR(MAX)"
+        else:
+            sql_type = f"NVARCHAR({size})"
+        if is_pk:
+            sql_type += " PRIMARY KEY"
+    else:
+        sql_type = "NVARCHAR(255)"
+        if is_pk:
+            sql_type += " PRIMARY KEY"
+
+    return f"[{col_name}] {sql_type}"
+
+
+def create_table_from_access(sql_cursor, target_schema, target_table,
+                             acc_description, pk_col):
+    """根据 Access cursor.description 在 SQL Server 自动建表"""
+    col_defs = [_access_col_to_sql(col, pk_col) for col in acc_description]
+    full_name = fmt_table(target_schema, target_table)
+    col_str = ",\n    ".join(col_defs)
+    create_sql = f"CREATE TABLE {full_name} (\n    {col_str}\n)"
+    sql_cursor.execute(create_sql)
+
 def get_sql_conn():
     """获取 SQL Server 连接"""
     # 显式添加 TrustServerCertificate=yes 以兼容 ODBC Driver 18+

excel_sync_to_sql.py

@@ -6,6 +6,7 @@ import argparse
 import datetime
 import urllib.parse
 import warnings
+import time
 import pandas as pd
 import numpy as np
 from sqlalchemy import create_engine, text
@@ -14,8 +15,14 @@ from sqlalchemy.types import NVARCHAR, Integer, Date
 
 # 导入配置
 from log_utils import (log_error, log_warning, log_info, log_processing, log_file, log_sync,
-                         log_start, log_complete, LoggerManager)
-from config import DB_CONFIG, CACHE_DIR, EXCEL_CONFIGS, BATCH_SIZE, TABLE_SCHEMA
+                         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')
@@ -291,25 +298,72 @@ class DataSynchronizer:
         except Exception as e:
             log_error(f"生成 ContractData 失败: {e}", exc_info=True)
 
+# ================= Uptime Kuma 心跳 =================
+# 使用 uptime_kuma_utils.UptimeKumaMonitor 替代原有实现
+
 def main():
     # 初始化日志管理器
-    LoggerManager("etl_manager", log_prefix="sync")
+    LoggerManager("excel_sync", log_prefix="excel_sync")
 
+    # 解析参数
     parser = argparse.ArgumentParser(description="Excel数据同步至SQL Server")
     parser.add_argument('--force', action='store_true', help='强制同步所有文件')
+    parser.add_argument('--once', action='store_true', help='只运行一次后退出')
     args = parser.parse_args()
 
     syncer = DataSynchronizer(force_sync=args.force)
 
-    if args.force:
-        log_start("Excel 同步任务 (强制模式)")
-    else:
-        log_start("Excel 同步任务 (增量模式)")
+    # 启动信息
+    mode = "强制模式" if args.force else "增量模式"
+    if args.once:
+        log_start(f"Excel 同步任务 ({mode}, 单次运行)")
+        syncer.process_excel_files()
+        syncer.generate_contract_data()
+        log_complete("Excel 同步任务已完成")
+        return
 
-    syncer.process_excel_files()
-    syncer.generate_contract_data()
+    # 周期性运行模式
+    log_start(f"Excel 同步服务已启动 ({mode})")
+    log_info(f"同步周期: {EXCEL_SYNC_INTERVAL} 秒 ({EXCEL_SYNC_INTERVAL//60} 分钟)")
+    if EXCEL_SYNC_UPTIME_KUMA_CONFIG.get('enabled', False):
+        log_info(f"心跳间隔: {EXCEL_SYNC_UPTIME_KUMA_CONFIG['heartbeat_interval']} 秒")
+    log_info("=" * 70)
 
-    log_complete("Excel 同步任务已完成")
+    # 启动时发送第一次心跳
+    excel_uptime_monitor.send_heartbeat()
+
+    try:
+        while True:
+            try:
+                # 执行同步任务
+                log_info(f"开始执行周期性同步检查...")
+                syncer.process_excel_files()
+                syncer.generate_contract_data()
+                log_info(f"周期性同步检查完成")
+
+                # 下次同步时间
+                next_sync_time = time.time() + EXCEL_SYNC_INTERVAL
+                log_info(f"下次同步将在 {EXCEL_SYNC_INTERVAL//60} 分钟后进行")
+
+                # 等待下次同步，期间持续发送心跳
+                while time.time() < next_sync_time:
+                    # 检查是否需要发送心跳
+                    excel_uptime_monitor.check_and_send_heartbeat()
+
+                    # 短暂休眠
+                    time.sleep(1)
+
+            except KeyboardInterrupt:
+                log_info("=" * 70)
+                log_stop("收到停止信号，服务正在关闭...")
+                break
+            except Exception as e:
+                log_error(f"同步任务异常: {e}", exc_info=True)
+                log_info(f"将在 {EXCEL_SYNC_INTERVAL//60} 分钟后重试...")
+                time.sleep(EXCEL_SYNC_INTERVAL)
+    finally:
+        # 停止时发送心跳停止信号
+        excel_uptime_monitor.send_stop_signal()
 
 if __name__ == "__main__":
     main()

init_full_sync.py

@@ -3,6 +3,7 @@ from config import SQL_SERVER_CONN, ACCESS_DRIVER, SYNC_MAPPING, BATCH_SIZE
 import db_utils
 import time
 import os
+import ntfy_utils
 from log_utils import (LoggerManager, log_success, log_error, log_warning, log_info, log_processing,
                         log_skip, log_start, log_file, log_database, log_sync, log_complete)
 
@@ -30,6 +31,54 @@ def has_identity_column(sql_cursor, schema, table):
     sql_cursor.execute(query, (schema, table))
     return sql_cursor.fetchone()[0] > 0
 
+def _send_summary_notification(sync_results, success_count, failed_count, total_rows, total_time):
+    """发送汇总 ntfy 通知"""
+    if not sync_results:
+        return
+
+    # 分类成功和失败的表
+    success_tables = [r for r in sync_results if r['status'] == 'success']
+    failed_tables = [r for r in sync_results if r['status'] == 'failed']
+
+    # 构建消息
+    lines = []
+    lines.append("🎯 全量同步完成\n")
+
+    # 成功的表
+    if success_tables:
+        lines.append(f"✅ 成功 {len(success_tables)} 张表:")
+        for r in success_tables:
+            lines.append(f"  • {r['table']}: {r['rows']:,} 行, {format_duration(r['duration'])}")
+        lines.append("")
+
+    # 失败的表
+    if failed_tables:
+        lines.append(f"❌ 失败 {failed_count} 张表:")
+        for r in failed_tables:
+            error_brief = r['error'][:50] + "..." if len(r['error']) > 50 else r['error']
+            lines.append(f"  • {r['table']}: {error_brief}")
+        lines.append("")
+
+    # 总计
+    lines.append(f"总计: {total_rows:,} 行 | 用时: {format_duration(total_time)}")
+
+    message = "\n".join(lines)
+
+    # 根据成功/失败情况决定优先级和标签
+    if failed_count > 0:
+        priority = "high"
+        tags = ["warning", "database"]
+    else:
+        priority = "default"
+        tags = ["white_check_mark", "database"]
+
+    ntfy_utils.send_ntfy(
+        message=message,
+        title="全量同步完成",
+        priority=priority,
+        tags=tags
+    )
+
 def run_full_sync():
     log_info("=" * 70)
     log_start( "全量初始化同步")
@@ -43,6 +92,9 @@ def run_full_sync():
     failed_tables = 0
     total_rows = 0
 
+    # 收集每张表的同步结果，用于最终汇总通知
+    sync_results = []  # 格式: {'table': str, 'status': 'success'|'failed', 'rows': int, 'duration': float, 'error': str}
+    
     try:
         sql_conn = db_utils.get_sql_conn()
         sql_cursor = sql_conn.cursor()
@@ -88,9 +140,18 @@ def run_full_sync():
                     log_info( f"  检测到 {len(columns)} 个列")
                     insert_sql = db_utils.generate_insert_sql(target_schema, target_table, columns)
 
-                    # 3. TRUNCATE 目标表
-                    sql_cursor.execute(f"TRUNCATE TABLE {full_target_name}")
-                    log_info( f"  已清空目标表")
+                    # 3. 检查目标表是否存在，不存在则自动创建
+                    if not db_utils.table_exists(sql_cursor, target_schema, target_table):
+                        db_utils.ensure_schema(sql_cursor, target_schema)
+                        db_utils.create_table_from_access(
+                            sql_cursor, target_schema, target_table,
+                            acc_cursor.description, target_config['pk_col']
+                        )
+                        sql_conn.commit()  # DDL 必须单独提交，否则 pyodbc 参数绑定无法获取正确的列元数据
+                        log_info( f"  目标表不存在，已自动创建")
+                    else:
+                        sql_cursor.execute(f"TRUNCATE TABLE {full_target_name}")
+                        log_info( f"  已清空目标表")
                     
                     # 4. 检查并启用 IDENTITY_INSERT
                     has_identity = has_identity_column(sql_cursor, target_schema, target_table)
@@ -141,7 +202,16 @@ def run_full_sync():
                     # 显示最终统计
                     table_time = time.time() - table_start_time
                     final_rate = table_rows / table_time if table_time > 0 else 0
-                    log_success( f"表 [{target_table}] 完成: {table_rows:,} 行 | 速率: {final_rate:,.0f} 行/秒 | 用时: {format_duration(table_time)}")
+                    log_info( f"表 [{target_table}] 完成: {table_rows:,} 行 | 速率: {final_rate:,.0f} 行/秒 | 用时: {format_duration(table_time)}")
+
+                    # 收集成功结果
+                    sync_results.append({
+                        'table': target_table,
+                        'status': 'success',
+                        'rows': table_rows,
+                        'duration': table_time,
+                        'error': None
+                    })
 
                     success_tables += 1
                     file_success += 1
@@ -151,7 +221,18 @@ def run_full_sync():
                 except Exception as tbl_err:
                     failed_tables += 1
                     file_failed += 1
-                    log_error( f"表 [{acc_table}] 同步失败: {tbl_err}", exc_info=True)
+                    error_msg = str(tbl_err)
+                    log_info( f"表 [{acc_table}] 同步失败: {error_msg}")
+
+                    # 收集失败结果
+                    sync_results.append({
+                        'table': target_table,
+                        'status': 'failed',
+                        'rows': 0,
+                        'duration': time.time() - table_start_time,
+                        'error': error_msg
+                    })
+
                     sql_conn.rollback()
 
                     # 确保清理 IDENTITY_INSERT 状态
@@ -179,7 +260,7 @@ def run_full_sync():
     # 总结统计
     sync_time = time.time() - sync_start_time
     log_info("\n" + "=" * 70)
-    log_complete( "全量同步任务结束")
+    log_info( "全量同步任务结束")
     log_info("-" * 70)
     log_info( f"总表数: {total_tables} 张 | 成功: {success_tables} 张 | 失败: {failed_tables} 张")
     log_info( f"总行数: {total_rows:,} 行")
@@ -188,5 +269,8 @@ def run_full_sync():
         log_info( f"整体平均速率: {total_rows/sync_time:,.0f} 行/秒")
     log_info("=" * 70)
 
+    # 发送汇总 ntfy 通知
+    _send_summary_notification(sync_results, success_tables, failed_tables, total_rows, sync_time)
+
 if __name__ == "__main__":
     run_full_sync()

run_incremental_sync.py

@@ -2,24 +2,133 @@ import time
 import os
 import sys
 import pyodbc
-from config import SQL_SERVER_CONN, ACCESS_DRIVER, SYNC_MAPPING, POLL_INTERVAL, LOG_TABLE_CONFIG
+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)
+
+# 提交后是否再做一次独立复核 (额外保险, 略增开销; 若不需要可设为 False)
+ENABLE_POST_COMMIT_VERIFY = True
+
+# IN 子句单批最大参数数 (SQL Server 上限约 2100, 留足余量)
+IN_CLAUSE_BATCH = 900
+
+
+class SyncVerificationError(Exception):
+    """数据落库校验未通过时抛出, 触发事务回滚并保留 Synced=0 以便下一轮重试"""
+    pass
+
+
+# ================= 辅助函数 =================
+
+def has_identity_column(sql_cursor, schema, table):
+    """检查表是否包含标识列"""
+    query = """
+        SELECT COUNT(*)
+        FROM sys.columns c
+        JOIN sys.tables t ON c.object_id = t.object_id
+        JOIN sys.schemas s ON t.schema_id = s.schema_id
+        WHERE s.name = ? AND t.name = ? AND c.is_identity = 1
+    """
+    sql_cursor.execute(query, (schema, table))
+    return sql_cursor.fetchone()[0] > 0
+
+
+def _chunked(values, size=IN_CLAUSE_BATCH):
+    """将集合/列表分批, 避免 IN 子句参数数量超过 SQL Server 上限"""
+    items = list(values)
+    for i in range(0, len(items), size):
+        yield items[i:i + size]
+
+
+def _norm_key(v):
+    """主键归一化, 用于跨来源(日志侧 / Access 侧 / SQL 侧)的集合比较, 规避类型差异"""
+    return str(v).strip() if v is not None else None
+
+
+def fetch_existing_pks(sql_cursor, full_name, pk_col, pk_values):
+    """
+    拿一批主键去目标表查询, 返回其中【实际存在】的主键集合 (分批查询)。
+    这是逐主键校验的基础: 用 "查到了哪些" 来精确判断存在 / 不存在,
+    而不是用数量相减 (数量法会被 "一边少插一边漏删" 互相抵消而误判)。
+    """
+    found = set()
+    for chunk in _chunked(pk_values):
+        if not chunk:
+            continue
+        placeholders = ','.join(['?'] * len(chunk))
+        sql = f"SELECT [{pk_col}] FROM {full_name} WHERE [{pk_col}] IN ({placeholders})"
+        sql_cursor.execute(sql, list(chunk))
+        for row in sql_cursor.fetchall():
+            found.add(row[0])
+    return found
+
+
+def verify_sync_result(sql_cursor, full_name, pk_col, record_ids, inserted_pk_set):
+    """
+    逐主键校验本批同步结果 (不依赖数量比较):
+
+      - 删除校验: 源端已读不到的记录 (record_ids 中不在 inserted_pk_set 的部分),
+                  删除后必须【全部不存在】于目标表; 只要还查得到任意一条即判失败。
+      - 插入校验: 从 Access 实际读到并重新插入的主键 (inserted_pk_set),
+                  必须【全部存在】于目标表; 只要有一条查不到即判失败。
+
+    校验不通过抛出 SyncVerificationError, 并给出具体出问题的主键样例。
+    返回值: 本批 "随源删除" 的记录条数 (供日志展示)。
+    """
+    inserted_keys = {_norm_key(pk) for pk in inserted_pk_set}
+
+    # 1) 删除校验: 应删除的记录, 删除后必须不在目标表
+    to_delete = [rid for rid in record_ids if _norm_key(rid) not in inserted_keys]
+    if to_delete:
+        leftover = fetch_existing_pks(sql_cursor, full_name, pk_col, to_delete)
+        if leftover:
+            sample = list(leftover)[:5]
+            raise SyncVerificationError(
+                f"删除校验失败: {len(leftover)} 条记录删除后仍存在于目标表, 例如 {sample}")
+
+    # 2) 插入校验: 应插入的记录, 插入后必须存在于目标表
+    if inserted_pk_set:
+        present_keys = {_norm_key(p) for p in fetch_existing_pks(sql_cursor, full_name, pk_col, inserted_pk_set)}
+        missing = [pk for pk in inserted_pk_set if _norm_key(pk) not in present_keys]
+        if missing:
+            raise SyncVerificationError(
+                f"插入校验失败: {len(missing)} 条记录插入后在目标表查不到, 例如 {missing[:5]}")
+
+    return len(to_delete)
+
+
+def find_pk_index(acc_cols, pk_col):
+    """在 Access 结果列中定位主键列下标 (大小写不敏感兜底)"""
+    if pk_col in acc_cols:
+        return acc_cols.index(pk_col)
+    lower_map = {c.lower(): i for i, c in enumerate(acc_cols)}
+    return lower_map.get(pk_col.lower())
+
+
 # ================= 主逻辑 =================
 
 def process_sync_task():
     """
-    逻辑重构：
+    逻辑重构:
     遍历 SYNC_MAPPING 中的每一个文件 -> 去日志表查询该文件下特定表的未同步记录。
+
+    校验机制 (逐主键确认):
+    删旧插新后, 在提交前对本批数据做删除校验 + 插入校验 ——
+    应删除的主键确认已不在目标表、应插入的主键确认已在目标表, 二者都通过才标记
+    Synced=1 并提交; 否则回滚整批, 保持 Synced=0, 下一轮自动重试。
     """
     sql_conn = db_utils.get_sql_conn()
     sql_cursor = sql_conn.cursor()
-    sql_cursor.fast_executemany = True # 高性能开关
+    sql_cursor.fast_executemany = True  # 高性能开关
 
     # 标记是否有工作被处理（用于控制轮询休眠时间）
     work_done = False
@@ -32,25 +141,21 @@ def process_sync_task():
         for clean_path, tables_map in SYNC_MAPPING.items():
 
             # 1. 准备查询条件
-            # 获取该文件下所有需要同步的表名列表
             target_tables = list(tables_map.keys())
             if not target_tables:
                 continue
 
             # 构造 TableAddress 的精确匹配条件
-            # VBA 逻辑：网络路径带 ";DATABASE=", 本地路径带 "LOCAL:"
-            # 我们直接构造这两个字符串，让 SQL Server 做精确匹配，效率极高
             # 注意：将路径转为 Windows 标准反斜杠
             win_path = os.path.normpath(clean_path)
             addr_candidates = [
-                f";DATABASE={win_path}", # 情况1
-                f"LOCAL={win_path}",      # 情况2 (注意 VBA 代码里可能是 LOCAL: 或 LOCAL=，请核对)
-                f"LOCAL:{win_path}",      # 情况3
-                win_path                  # 情况4 (兼容没有前缀的情况)
+                f";DATABASE={win_path}",  # 情况1
+                f"LOCAL={win_path}",       # 情况2
+                f"LOCAL:{win_path}",       # 情况3
+                win_path                   # 情况4 (兼容没有前缀的情况)
             ]
 
             # 2. 构造动态 SQL 查询
-            # WHERE Synced=0 AND Address IN (...) AND TableName IN (...)
             placeholders_addr = ','.join(['?'] * len(addr_candidates))
             placeholders_tbl = ','.join(['?'] * len(target_tables))
 
@@ -74,13 +179,13 @@ def process_sync_task():
             logs = sql_cursor.fetchall()
 
             if not logs:
-                continue # 这个文件没有需要同步的记录，检查下一个文件
+                continue  # 这个文件没有需要同步的记录，检查下一个文件
 
-            work_done = True # 标记有工作
+            work_done = True  # 标记有工作
             log_file(f"{os.path.basename(clean_path)} 发现 {len(logs)} 条待同步变更")
 
             # 3. 本地分组 (按表名)
-            # 结构: table_tasks[TableName] = { ids: {}, log_ids: [] }
+            # 结构: table_tasks[TableName] = { record_ids: set(), log_ids: [] }
             table_tasks = {}
             for row in logs:
                 log_id, acc_table, record_id = row
@@ -94,6 +199,7 @@ def process_sync_task():
                 log_error(f"无法访问文件: {clean_path}")
                 continue
 
+            acc_conn = None
             try:
                 acc_conn = db_utils.get_access_conn(clean_path)
                 acc_cursor = acc_conn.cursor()
@@ -107,65 +213,137 @@ def process_sync_task():
             file_error_count = 0
             file_total_records = 0
 
-            for acc_table, data in table_tasks.items():
-                record_ids = list(data['record_ids'])
-                log_ids = data['log_ids']
+            try:
+                for acc_table, data in table_tasks.items():
+                    record_ids = list(data['record_ids'])
+                    log_ids = data['log_ids']
 
-                # 读取目标配置
-                target_conf = tables_map[acc_table]
-                target_schema = target_conf['target_schema']
-                target_table = target_conf['target_table']
-                pk_col = target_conf['pk_col']
-                target_full_name = db_utils.fmt_table(target_schema, target_table)
+                    # 读取目标配置
+                    target_conf = tables_map[acc_table]
+                    target_schema = target_conf['target_schema']
+                    target_table = target_conf['target_table']
+                    pk_col = target_conf['pk_col']
+                    target_full_name = db_utils.fmt_table(target_schema, target_table)
 
-                log_processing(f"正在同步表 [{acc_table}] → [{target_table}] ({len(record_ids)} 条记录)")
+                    # 检查目标表是否存在，不存在则自动创建
+                    if not db_utils.table_exists(sql_cursor, target_schema, target_table):
+                        db_utils.ensure_schema(sql_cursor, target_schema)
+                        acc_cursor.execute(f"SELECT TOP 1 * FROM [{acc_table}]")
+                        db_utils.create_table_from_access(
+                            sql_cursor, target_schema, target_table,
+                            acc_cursor.description, pk_col
+                        )
+                        log_info(f"目标表 [{target_table}] 不存在，已自动创建")
 
+                    log_processing(f"正在同步表 [{acc_table}] → [{target_table}] ({len(record_ids)} 条记录)")
+
+                    identity_enabled = False
+                    try:
+                        # --- A. 从 Access 读取最新数据 ---
+                        ids_placeholders = ','.join(['?'] * len(record_ids))
+                        acc_sql = f"SELECT * FROM [{acc_table}] WHERE [{pk_col}] IN ({ids_placeholders})"
+                        acc_cursor.execute(acc_sql, record_ids)
+                        new_rows = acc_cursor.fetchall()
+                        acc_cols = [col[0] for col in acc_cursor.description]
+
+                        # 定位主键列, 取出 Access 实际读到的主键集合 (校验基准)
+                        pk_index = find_pk_index(acc_cols, pk_col)
+                        if new_rows and pk_index is None:
+                            raise SyncVerificationError(
+                                f"在 Access 表 [{acc_table}] 中找不到主键列 [{pk_col}]")
+                        inserted_pk_set = (set(row[pk_index] for row in new_rows)
+                                           if pk_index is not None else set())
+
+                        # --- B. SQL Server 删除旧记录 ---
+                        del_sql = f"DELETE FROM {target_full_name} WHERE [{pk_col}] IN ({ids_placeholders})"
+                        sql_cursor.execute(del_sql, record_ids)
+
+                        # --- C. 插入新记录 ---
+                        if new_rows:
+                            insert_sql = db_utils.generate_insert_sql(target_schema, target_table, acc_cols)
+
+                            # 检查并启用 IDENTITY_INSERT
+                            if has_identity_column(sql_cursor, target_schema, target_table):
+                                try:
+                                    sql_cursor.execute(f"SET IDENTITY_INSERT {target_full_name} ON")
+                                    identity_enabled = True
+                                except Exception as id_err:
+                                    log_error(f"无法启用 IDENTITY_INSERT: {id_err}")
+                                    raise
+
+                            sql_cursor.executemany(insert_sql, new_rows)
+
+                            # 立即关闭 IDENTITY_INSERT (会话级设置, 不随事务回滚)
+                            if identity_enabled:
+                                sql_cursor.execute(f"SET IDENTITY_INSERT {target_full_name} OFF")
+                                identity_enabled = False
+
+                        # --- D. 提交前校验 (逐主键确认: 应删的已不在 / 应插的已在) ---
+                        removed_count = verify_sync_result(
+                            sql_cursor, target_full_name, pk_col, record_ids, inserted_pk_set)
+
+                        # --- E. 校验通过 -> 标记日志 Synced = 1 ---
+                        log_placeholders = ','.join(['?'] * len(log_ids))
+                        update_log_sql = f"""
+                        UPDATE {log_full_name}
+                        SET {cols['col_synced']} = 1
+                        WHERE {cols['col_log_id']} IN ({log_placeholders})
+                        """
+                        sql_cursor.execute(update_log_sql, log_ids)
+
+                        # --- F. 提交事务 (删/插/标记 原子生效) ---
+                        sql_conn.commit()
+
+                        # --- G. 提交后独立复核 (可选, 防 "提交成功但未持久化" 等极端情况) ---
+                        if ENABLE_POST_COMMIT_VERIFY:
+                            try:
+                                verify_sync_result(
+                                    sql_cursor, target_full_name, pk_col, record_ids, inserted_pk_set)
+                            except SyncVerificationError as post_err:
+                                log_critical(
+                                    f"严重: 表 [{target_table}] 提交后复核失败! {post_err}; "
+                                    f"撤销同步标记以便重试")
+                                revert_sql = f"""
+                                UPDATE {log_full_name}
+                                SET {cols['col_synced']} = 0
+                                WHERE {cols['col_log_id']} IN ({log_placeholders})
+                                """
+                                sql_cursor.execute(revert_sql, log_ids)
+                                sql_conn.commit()
+                                file_error_count += 1
+                                continue
+
+                        # 同步成功
+                        msg = f"表 [{target_table}] 同步并校验通过: {len(inserted_pk_set)} 条入库"
+                        if removed_count > 0:
+                            msg += f", {removed_count} 条随源删除"
+                        log_success(msg)
+
+                        file_success_count += 1
+                        file_total_records += len(record_ids)
+
+                    except Exception as tbl_err:
+                        # 清理可能残留的 IDENTITY_INSERT 会话状态
+                        if identity_enabled:
+                            try:
+                                sql_cursor.execute(f"SET IDENTITY_INSERT {target_full_name} OFF")
+                            except:
+                                pass
+                        # 回滚本表的 删/插/标记, 保持 Synced=0, 下一轮自动重试
+                        try:
+                            sql_conn.rollback()
+                        except:
+                            pass
+                        log_error(f"表 [{acc_table}] 同步失败 (已回滚, 将重试): {tbl_err}")
+                        file_error_count += 1
+            finally:
+                # 确保 Access 连接关闭
                 try:
-                    # --- A. Access 查新数据 ---
-                    ids_placeholders = ','.join(['?'] * len(record_ids))
-                    acc_sql = f"SELECT * FROM [{acc_table}] WHERE [{pk_col}] IN ({ids_placeholders})"
-                    acc_cursor.execute(acc_sql, record_ids)
-                    new_rows = acc_cursor.fetchall()
-                    acc_cols = [col[0] for col in acc_cursor.description]
-
-                    # --- B. SQL Server 删旧插新 (事务) ---
-                    # 1. 删除
-                    del_sql = f"DELETE FROM {target_full_name} WHERE [{pk_col}] IN ({ids_placeholders})"
-                    sql_cursor.execute(del_sql, record_ids)
-
-                    # 2. 插入
-                    if new_rows:
-                        insert_sql = db_utils.generate_insert_sql(target_schema, target_table, acc_cols)
-                        try: sql_cursor.execute(f"SET IDENTITY_INSERT {target_full_name} ON")
-                        except: pass
-                        
-                        sql_cursor.executemany(insert_sql, new_rows)
-                        
-                        try: sql_cursor.execute(f"SET IDENTITY_INSERT {target_full_name} OFF")
-                        except: pass
-
-                    # 3. 标记日志 Synced = 1
-                    log_placeholders = ','.join(['?'] * len(log_ids))
-                    update_log_sql = f"""
-                    UPDATE {log_full_name} 
-                    SET {cols['col_synced']} = 1 
-                    WHERE {cols['col_log_id']} IN ({log_placeholders})
-                    """
-                    sql_cursor.execute(update_log_sql, log_ids)
-
-                    sql_conn.commit()
-                    log_success(f"表 [{target_table}] 同步完成: {len(record_ids)} 条记录")
-                    
-                    file_success_count += 1
-                    file_total_records += len(record_ids)
-
-                except Exception as tbl_err:
-                    log_error(f"表 [{acc_table}] 同步失败: {tbl_err}")
-                    sql_conn.rollback()
-                    file_error_count += 1
-
-            acc_conn.close() # 关闭 Access 连接
-            
+                    if acc_conn:
+                        acc_conn.close()
+                        log_info(f"已关闭 Access 连接: {os.path.basename(clean_path)}")
+                except Exception as close_err:
+                    log_warning(f"关闭 Access 连接时出错: {close_err}")
             # 输出文件级别的汇总
             if file_success_count > 0 or file_error_count > 0:
                 summary = f"文件 [{os.path.basename(clean_path)}] 同步汇总: "
@@ -180,25 +358,46 @@ def process_sync_task():
         log_critical(f"全局异常: {e}")
         return False
     finally:
-        try: sql_conn.close()
-        except: pass
+        try:
+            sql_conn.close()
+        except:
+            pass
+
+
+# ================= Uptime Kuma 心跳 =================
+# 使用 uptime_kuma_utils.UptimeKumaMonitor 替代原有实现
 
 if __name__ == "__main__":
-    log_start("增量同步服务已启动 (配置驱动模式)")
+    log_start("增量同步服务已启动 (配置驱动模式 + 逐主键落库校验)")
     log_info(f"轮询间隔: {POLL_INTERVAL} 秒")
     log_info(f"监控配置: {len(SYNC_MAPPING)} 个文件")
+    log_info(f"提交后复核: {'开启' if ENABLE_POST_COMMIT_VERIFY else '关闭'}")
+    if UPTIME_KUMA_CONFIG.get('enabled', False):
+        log_info(f"心跳间隔: {UPTIME_KUMA_CONFIG['heartbeat_interval']} 秒")
     log_info("=" * 70)
 
-    while True:
-        try:
-            has_work = process_sync_task()
-            # 如果有工作，说明可能还有积压，休息短一点(0.1s)
-            # 如果没工作，休息标准间隔(5s)
-            time.sleep(0.1 if has_work else POLL_INTERVAL)
-        except KeyboardInterrupt:
-            log_info("=" * 70)
-            log_stop("收到停止信号，服务正在关闭...")
-            break
-        except Exception as e:
-            log_critical(f"主循环崩溃: {e}")
-            time.sleep(5)
+    # 启动时发送第一次心跳
+    uptime_monitor.send_heartbeat()
+
+    try:
+        while True:
+            try:
+                has_work = process_sync_task()
+
+                # 检查是否需要发送心跳
+                uptime_monitor.check_and_send_heartbeat()
+
+                # 如果有工作，说明可能还有积压，休息短一点(0.1s)
+                # 如果没工作，休息标准间隔
+                time.sleep(0.1 if has_work else POLL_INTERVAL)
+
+            except KeyboardInterrupt:
+                log_info("=" * 70)
+                log_stop("收到停止信号，服务正在关闭...")
+                break
+            except Exception as e:
+                log_critical(f"主循环崩溃: {e}")
+                time.sleep(5)
+    finally:
+        # 停止时发送心跳停止信号（可选）
+        uptime_monitor.send_stop_signal()

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