import pyodbc 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) # 初始化日志管理器 LoggerManager("init_full_sync", log_prefix="full_sync") def format_duration(seconds): """格式化时长显示""" if seconds < 60: return f"{seconds:.1f}秒" elif seconds < 3600: return f"{seconds/60:.1f}分钟" else: return f"{seconds/3600:.1f}小时" 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 _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() sql_cursor = sql_conn.cursor() sql_cursor.fast_executemany = True log_database( "SQL Server 连接成功") except Exception as e: log_error( f"SQL Server 连接失败: {e}") return # 第一层循环:遍历配置文件中的所有文件路径 for acc_path, tables_map in SYNC_MAPPING.items(): if not os.path.exists(acc_path): log_skip( f"文件不存在: {acc_path}") continue log_file( f"开始处理: {os.path.basename(acc_path)}") file_start_time = time.time() file_success = 0 file_failed = 0 file_rows = 0 try: acc_conn = db_utils.get_access_conn(acc_path) acc_cursor = acc_conn.cursor() log_database( f"Access 文件连接成功: {os.path.basename(acc_path)}") # 第二层循环:遍历该文件下的所有表映射 for acc_table, target_config in tables_map.items(): total_tables += 1 target_schema = target_config['target_schema'] target_table = target_config['target_table'] full_target_name = db_utils.fmt_table(target_schema, target_table) log_processing( f"表 [{acc_table}] → [{target_table}]") table_start_time = time.time() try: # 1. 检查 Access 表是否存在 acc_cursor.execute(f"SELECT TOP 1 * FROM [{acc_table}]") # 2. 获取列结构 columns = [col[0] for col in acc_cursor.description] 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" 已清空目标表") # 4. 检查并启用 IDENTITY_INSERT has_identity = has_identity_column(sql_cursor, target_schema, target_table) identity_enabled = False if has_identity: try: sql_cursor.execute(f"SET IDENTITY_INSERT {full_target_name} ON") identity_enabled = True log_info( f" 已启用 IDENTITY_INSERT") except Exception as id_err: log_error( f" 无法启用 IDENTITY_INSERT: {id_err}") raise # 5. 传输数据 log_info( f" 开始数据传输...") acc_cursor.execute(f"SELECT * FROM [{acc_table}]") table_rows = 0 start_time = time.time() last_log_time = start_time while True: rows = acc_cursor.fetchmany(BATCH_SIZE) if not rows: break sql_cursor.executemany(insert_sql, rows) table_rows += len(rows) # 每5秒或每10000行记录一次进度 current_time = time.time() if (current_time - last_log_time >= 5) or (table_rows % 10000 == 0): elapsed = current_time - start_time rate = table_rows / elapsed if elapsed > 0 else 0 log_info( f" 进度: {table_rows:,} 行 | 速率: {rate:,.0f} 行/秒") last_log_time = current_time # 6. 关闭 IDENTITY_INSERT if identity_enabled: try: sql_cursor.execute(f"SET IDENTITY_INSERT {full_target_name} OFF") log_info( f" 已关闭 IDENTITY_INSERT") except Exception as id_err: 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_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 file_rows += table_rows except Exception as tbl_err: failed_tables += 1 file_failed += 1 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") except: pass acc_conn.close() # 文件级别汇总 file_time = time.time() - file_start_time summary = f"文件 [{os.path.basename(acc_path)}] 完成: " summary += f"成功 {file_success} 张表" if file_failed > 0: summary += f" | 失败 {file_failed} 张表" summary += f" | 共 {file_rows:,} 行 | 用时: {format_duration(file_time)}" log_sync( summary) except Exception as file_err: 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_info( "全量同步任务结束") log_info("-" * 70) log_info( f"总表数: {total_tables} 张 | 成功: {success_tables} 张 | 失败: {failed_tables} 张") log_info( f"总行数: {total_rows:,} 行") log_info( f"总用时: {format_duration(sync_time)}") if total_rows > 0 and sync_time > 0: 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()