Add ntfy notification for full sync summary and enhance logging for success and failure results

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)
 
@@ -21,7 +22,7 @@ def format_duration(seconds):
 def has_identity_column(sql_cursor, schema, table):
     """检查表是否包含标识列"""
     query = """
-        SELECT COUNT(*) 
+        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
@@ -30,18 +31,69 @@ 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( "全量初始化同步")
     log_info( f"配置文件数: {len(SYNC_MAPPING)} 个")
     log_info( f"批次大小: {BATCH_SIZE} 行")
     log_info("=" * 70)
-    
+
     sync_start_time = time.time()
     total_tables = 0
     success_tables = 0
     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()
@@ -137,12 +189,21 @@ def run_full_sync():
                             log_warning( f"  关闭 IDENTITY_INSERT 时警告: {id_err}")
                     
                     sql_conn.commit()
-                    
+
                     # 显示最终统计
                     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
                     total_rows += table_rows
@@ -151,9 +212,20 @@ 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 状态
                     try:
                         sql_cursor.execute(f"SET IDENTITY_INSERT {full_target_name} OFF")
@@ -175,11 +247,11 @@ def run_full_sync():
             log_error( f"文件 [{os.path.basename(acc_path)}] 处理失败: {file_err}", exc_info=True)
 
     sql_conn.close()
-    
+
     # 总结统计
     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 +260,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()