From 5defe7b80d1fc04b7f211f585717fee3441656e5 Mon Sep 17 00:00:00 2001 From: Misaka_Company Date: Wed, 7 Jan 2026 15:17:20 +0800 Subject: [PATCH] Refactor logging functions to enhance clarity and consistency in log messages --- init_full_sync.py | 130 ++++++++++++++++++++++++++++++---------- run_incremental_sync.py | 92 ++++++++++++++++++++++++---- 2 files changed, 180 insertions(+), 42 deletions(-) diff --git a/init_full_sync.py b/init_full_sync.py index 192a176..b84f4d5 100644 --- a/init_full_sync.py +++ b/init_full_sync.py @@ -29,9 +29,54 @@ def setup_logger(): ) logger = logging.getLogger(__name__) - logger.info(f"日志文件: {log_file}") + 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: @@ -56,8 +101,11 @@ def has_identity_column(sql_cursor, schema, table): def run_full_sync(): logger = setup_logger() - logger.info("=== 开始全量初始化同步 ===") - logger.info("注意:将基于 config.SYNC_MAPPING 中的文件路径进行处理") + logger.info("=" * 70) + log_start(logger, "全量初始化同步") + log_info(logger, f"配置文件数: {len(config.SYNC_MAPPING)} 个") + log_info(logger, f"批次大小: {config.BATCH_SIZE} 行") + logger.info("=" * 70) sync_start_time = time.time() total_tables = 0 @@ -69,23 +117,27 @@ def run_full_sync(): sql_conn = db_utils.get_sql_conn() sql_cursor = sql_conn.cursor() sql_cursor.fast_executemany = True - logger.info("✓ SQL Server 连接成功") + log_database(logger, "SQL Server 连接成功") except Exception as e: - logger.error(f"❌ SQL Server 连接失败: {e}") + log_error(logger, f"SQL Server 连接失败: {e}") return # 第一层循环:遍历配置文件中的所有文件路径 for acc_path, tables_map in config.SYNC_MAPPING.items(): if not os.path.exists(acc_path): - logger.warning(f"⚠️ [跳过] 文件不存在: {acc_path}") + log_skip(logger, f"文件不存在: {acc_path}") continue - logger.info(f"\n📂 正在处理文件: {acc_path}") + 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() - logger.info(f" ✓ Access 文件连接成功") + log_database(logger, f"Access 文件连接成功: {os.path.basename(acc_path)}") # 第二层循环:遍历该文件下的所有表映射 for acc_table, target_config in tables_map.items(): @@ -94,22 +146,21 @@ def run_full_sync(): target_table = target_config['target_table'] full_target_name = db_utils.fmt_table(target_schema, target_table) - logger.info(f" 👉 开始同步表: [{acc_table}] -> {full_target_name}") + 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}]") - logger.info(f" ✓ Access 表 [{acc_table}] 存在") # 2. 获取列结构 columns = [col[0] for col in acc_cursor.description] - logger.info(f" ✓ 获取到 {len(columns)} 个列: {', '.join(columns)}") + 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}") - logger.info(f" ✓ 已清空目标表") + log_info(logger, f" 已清空目标表") # 4. 检查并启用 IDENTITY_INSERT has_identity = has_identity_column(sql_cursor, target_schema, target_table) @@ -119,17 +170,18 @@ def run_full_sync(): try: sql_cursor.execute(f"SET IDENTITY_INSERT {full_target_name} ON") identity_enabled = True - logger.info(f" ✓ 已启用 IDENTITY_INSERT") + log_info(logger, f" 已启用 IDENTITY_INSERT") except Exception as id_err: - logger.error(f" ⚠️ 无法启用 IDENTITY_INSERT: {id_err}") + log_error(logger, f" 无法启用 IDENTITY_INSERT: {id_err}") raise # 5. 传输数据 - logger.info(f" 正在读取并写入数据...") + 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(config.BATCH_SIZE) @@ -138,33 +190,38 @@ def run_full_sync(): sql_cursor.executemany(insert_sql, rows) table_rows += len(rows) - # 定期记录进度到日志 - if table_rows % (config.BATCH_SIZE * 10) == 0: - elapsed = time.time() - start_time + # 每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 - logger.info(f" 进度: 已写入 {table_rows:,} 行 | 速率: {rate:,.0f} 行/秒") + 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") - logger.info(f" ✓ 已关闭 IDENTITY_INSERT") + log_info(logger, f" 已关闭 IDENTITY_INSERT") except Exception as id_err: - logger.warning(f" ⚠️ 关闭 IDENTITY_INSERT 时警告: {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 - logger.info(f" ✅ 完成。共 {table_rows:,} 行 | 平均速率: {final_rate:,.0f} 行/秒 | 总用时: {format_duration(table_time)}") + 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 - logger.error(f" ❌ 表级错误: {tbl_err}", exc_info=True) + file_failed += 1 + log_error(logger, f"表 [{acc_table}] 同步失败: {tbl_err}", exc_info=True) sql_conn.rollback() # 确保清理 IDENTITY_INSERT 状态 @@ -174,23 +231,32 @@ def run_full_sync(): pass acc_conn.close() - logger.info(f" ✓ Access 文件处理完成") + + # 文件级别汇总 + 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: - logger.error(f" ❌ 文件级错误: {file_err}", exc_info=True) + 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" + "="*60) - logger.info("=== 全量同步结束 ===") - logger.info(f"总表数: {total_tables} | 成功: {success_tables} | 失败: {failed_tables}") - logger.info(f"总行数: {total_rows:,} 行") - logger.info(f"总用时: {format_duration(sync_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: - logger.info(f"整体平均速率: {total_rows/sync_time:,.0f} 行/秒") - logger.info("="*60) + log_info(logger, f"整体平均速率: {total_rows/sync_time:,.0f} 行/秒") + logger.info("=" * 70) if __name__ == "__main__": run_full_sync() \ No newline at end of file diff --git a/run_incremental_sync.py b/run_incremental_sync.py index 0a8b2df..46beef7 100644 --- a/run_incremental_sync.py +++ b/run_incremental_sync.py @@ -35,6 +35,55 @@ def setup_logger(): logger = setup_logger() +# ================= 日志辅助函数 ================= +def log_success(message): + """成功消息 - 绿色感觉""" + logger.info(f"✅ [成功] {message}") + +def log_error(message): + """错误消息 - 红色感觉""" + logger.error(f"❌ [错误] {message}") + +def log_warning(message): + """警告消息 - 黄色感觉""" + logger.warning(f"⚠️ [警告] {message}") + +def log_info(message): + """信息消息""" + logger.info(f"ℹ️ {message}") + +def log_processing(message): + """处理中消息""" + logger.info(f"🔄 [处理] {message}") + +def log_skip(message): + """跳过消息""" + logger.info(f"⏭️ [跳过] {message}") + +def log_critical(message): + """严重错误""" + logger.critical(f"🔥 [严重] {message}") + +def log_start(message): + """启动消息""" + logger.info(f"🚀 [启动] {message}") + +def log_stop(message): + """停止消息""" + logger.info(f"🛑 [停止] {message}") + +def log_file(message): + """文件操作消息""" + logger.info(f"📂 [文件] {message}") + +def log_database(message): + """数据库操作消息""" + logger.info(f"💾 [数据库] {message}") + +def log_sync(message): + """同步操作消息""" + logger.info(f"🔃 [同步] {message}") + # ================= 主逻辑 ================= def process_sync_task(): @@ -102,7 +151,7 @@ def process_sync_task(): continue # 这个文件没有需要同步的记录，检查下一个文件 work_done = True # 标记有工作 - logger.info(f"📂 文件 {os.path.basename(clean_path)} 发现 {len(logs)} 条变更...") + log_file(f"{os.path.basename(clean_path)} 发现 {len(logs)} 条待同步变更") # 3. 本地分组 (按表名) # 结构: table_tasks[TableName] = { ids: {}, log_ids: [] } @@ -116,16 +165,22 @@ def process_sync_task(): # 4. 执行同步 (连接一次 Access，处理多张表) if not os.path.exists(clean_path): - logger.error(f"无法访问文件: {clean_path}") + log_error(f"无法访问文件: {clean_path}") continue try: acc_conn = db_utils.get_access_conn(clean_path) acc_cursor = acc_conn.cursor() + log_database(f"已连接 Access 文件: {os.path.basename(clean_path)}") except Exception as conn_err: - logger.error(f"连接 Access 失败: {conn_err}") + log_error(f"连接 Access 失败 [{os.path.basename(clean_path)}]: {conn_err}") continue + # 统计每个文件的同步情况 + file_success_count = 0 + 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'] @@ -137,6 +192,8 @@ def process_sync_task(): 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)} 条记录)") + try: # --- A. Access 查新数据 --- ids_placeholders = ','.join(['?'] * len(record_ids)) @@ -171,26 +228,40 @@ def process_sync_task(): sql_cursor.execute(update_log_sql, log_ids) sql_conn.commit() - logger.info(f" -> {target_table} 同步成功: {len(record_ids)} 条") + log_success(f"表 [{target_table}] 同步完成: {len(record_ids)} 条记录") + + file_success_count += 1 + file_total_records += len(record_ids) except Exception as tbl_err: - logger.error(f" -> 处理表 {acc_table} 异常: {tbl_err}") + log_error(f"表 [{acc_table}] 同步失败: {tbl_err}") sql_conn.rollback() + file_error_count += 1 acc_conn.close() # 关闭 Access 连接 + + # 输出文件级别的汇总 + if file_success_count > 0 or file_error_count > 0: + summary = f"文件 [{os.path.basename(clean_path)}] 同步汇总: " + summary += f"成功 {file_success_count} 张表 ({file_total_records} 条记录)" + if file_error_count > 0: + summary += f" | 失败 {file_error_count} 张表" + log_sync(summary) return work_done except Exception as e: - logger.critical(f"全局严重异常: {e}") + log_critical(f"全局异常: {e}") return False finally: try: sql_conn.close() except: pass if __name__ == "__main__": - logger.info("=== 增量同步服务启动 (配置驱动模式) ===") - logger.info("正在按 config.SYNC_MAPPING 轮询...") + log_start("增量同步服务已启动 (配置驱动模式)") + log_info(f"轮询间隔: {config.POLL_INTERVAL} 秒") + log_info(f"监控配置: {len(config.SYNC_MAPPING)} 个文件") + logger.info("=" * 70) while True: try: @@ -199,8 +270,9 @@ if __name__ == "__main__": # 如果没工作，休息标准间隔(5s) time.sleep(0.1 if has_work else config.POLL_INTERVAL) except KeyboardInterrupt: - logger.info("服务停止") + logger.info("=" * 70) + log_stop("收到停止信号，服务正在关闭...") break except Exception as e: - logger.critical(f"主循环崩溃: {e}") + log_critical(f"主循环崩溃: {e}") time.sleep(5) \ No newline at end of file