import pyodbc from config import SQL_SERVER_CONN, ACCESS_DRIVER, SYNC_MAPPING, BATCH_SIZE import db_utils import time import os import logging from datetime import datetime def setup_logger(): """初始化日志配置，每次运行生成独立的日志文件""" # 创建log目录 log_dir = "log" if not os.path.exists(log_dir): os.makedirs(log_dir) # 生成带时间戳的日志文件名 timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") log_file = os.path.join(log_dir, f"full_sync_{timestamp}.log") # 配置日志格式 logging.basicConfig( level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s', datefmt='%Y-%m-%d %H:%M:%S', handlers=[ logging.FileHandler(log_file, encoding='utf-8'), logging.StreamHandler() # 同时输出到控制台 ] ) logger = logging.getLogger(__name__) logger.info(f"📝 日志文件: {log_file}") return logger # ================= 日志辅助函数 ================= def log_success(logger, message): """成功消息""" logger.info(f"✅ [成功] {message}") def log_error(logger, message, exc_info=False): """错误消息""" logger.error(f"❌ [错误] {message}", exc_info=exc_info) def log_warning(logger, message): """警告消息""" logger.warning(f"⚠️ [警告] {message}") def log_info(logger, message): """信息消息""" logger.info(f"ℹ️ {message}") def log_processing(logger, message): """处理中消息""" logger.info(f"🔄 [处理] {message}") def log_skip(logger, message): """跳过消息""" logger.info(f"⏭️ [跳过] {message}") def log_start(logger, message): """启动消息""" logger.info(f"🚀 [启动] {message}") def log_file(logger, message): """文件操作消息""" logger.info(f"📂 [文件] {message}") def log_database(logger, message): """数据库操作消息""" logger.info(f"💾 [数据库] {message}") def log_sync(logger, message): """同步操作消息""" logger.info(f"🔃 [同步] {message}") def log_complete(logger, message): """完成消息""" logger.info(f"🎯 [完成] {message}") 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 run_full_sync(): logger = setup_logger() logger.info("=" * 70) log_start(logger, "全量初始化同步") log_info(logger, f"配置文件数: {len(SYNC_MAPPING)} 个") log_info(logger, f"批次大小: {BATCH_SIZE} 行") logger.info("=" * 70) sync_start_time = time.time() total_tables = 0 success_tables = 0 failed_tables = 0 total_rows = 0 try: sql_conn = db_utils.get_sql_conn() sql_cursor = sql_conn.cursor() sql_cursor.fast_executemany = True log_database(logger, "SQL Server 连接成功") except Exception as e: log_error(logger, f"SQL Server 连接失败: {e}") return # 第一层循环:遍历配置文件中的所有文件路径 for acc_path, tables_map in SYNC_MAPPING.items(): if not os.path.exists(acc_path): log_skip(logger, f"文件不存在: {acc_path}") continue log_file(logger, 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(logger, 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(logger, 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(logger, 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(logger, 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(logger, f" 已启用 IDENTITY_INSERT") except Exception as id_err: log_error(logger, f" 无法启用 IDENTITY_INSERT: {id_err}") raise # 5. 传输数据 log_info(logger, 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(logger, 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(logger, f" 已关闭 IDENTITY_INSERT") except Exception as id_err: log_warning(logger, 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(logger, f"表 [{target_table}] 完成: {table_rows:,} 行 | 速率: {final_rate:,.0f} 行/秒 | 用时: {format_duration(table_time)}") 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 log_error(logger, f"表 [{acc_table}] 同步失败: {tbl_err}", exc_info=True) 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(logger, summary) except Exception as file_err: log_error(logger, f"文件 [{os.path.basename(acc_path)}] 处理失败: {file_err}", exc_info=True) sql_conn.close() # 总结统计 sync_time = time.time() - sync_start_time logger.info("\n" + "=" * 70) log_complete(logger, "全量同步任务结束") logger.info("-" * 70) log_info(logger, f"总表数: {total_tables} 张 | 成功: {success_tables} 张 | 失败: {failed_tables} 张") log_info(logger, f"总行数: {total_rows:,} 行") log_info(logger, f"总用时: {format_duration(sync_time)}") if total_rows > 0 and sync_time > 0: log_info(logger, f"整体平均速率: {total_rows/sync_time:,.0f} 行/秒") logger.info("=" * 70) if __name__ == "__main__": run_full_sync()