diff --git a/init_full_sync.py b/init_full_sync.py index 3eb16c8..13d513f 100644 --- a/init_full_sync.py +++ b/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() \ No newline at end of file