Refactor logging functions to enhance clarity and consistency in log messages

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("=" * 70)
-    logger.info("注意:将基于 config.SYNC_MAPPING 中的文件路径进行处理")
+    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)
                         
-                        # 定期记录进度到日志
+                        # 每5秒或每10000行记录一次进度
-                        if table_rows % (config.BATCH_SIZE * 10) == 0:
+                        current_time = time.time()
-                            elapsed = time.time() - start_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("\n" + "=" * 70)
-    logger.info("=== 全量同步结束 ===")
+    log_complete(logger, "全量同步任务结束")
-    logger.info(f"总表数: {total_tables} | 成功: {success_tables} | 失败: {failed_tables}")
+    logger.info("-" * 70)
-    logger.info(f"总行数: {total_rows:,} 行")
+    log_info(logger, f"总表数: {total_tables} 张 | 成功: {success_tables} 张 | 失败: {failed_tables} 张")
-    logger.info(f"总用时: {format_duration(sync_time)}")
+    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} 行/秒")
+        log_info(logger, f"整体平均速率: {total_rows/sync_time:,.0f} 行/秒")
-    logger.info("="*60)
+    logger.info("=" * 70)
 
 if __name__ == "__main__":
     run_full_sync()

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("=== 增量同步服务启动 (配置驱动模式) ===")
+    log_start("增量同步服务已启动 (配置驱动模式)")
-    logger.info("正在按 config.SYNC_MAPPING 轮询...")
+    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)