From a1c8a572bd11c466e67ffe70bc6f3a4b78dff570 Mon Sep 17 00:00:00 2001 From: Misaka_Company Date: Wed, 7 Jan 2026 15:01:27 +0800 Subject: [PATCH] Enhance logging and synchronization features across multiple scripts - Added detailed logging functionality to init_full_sync.py and run_incremental_sync.py for better tracking of synchronization processes. - Updated configuration mappings in config.py to include additional Access database tables. - Improved error handling and user feedback in vbareplace.py, including the ability to refresh linked tables in Access. - Adjusted polling intervals and batch sizes for performance optimization. --- config.py | 293 ++++++++++++++++++++++++++++++++++++++-- init_full_sync.py | 161 ++++++++++++++++++---- run_incremental_sync.py | 214 ++++++++++++++--------------- vbareplace.py | 42 +++++- 4 files changed, 555 insertions(+), 155 deletions(-) diff --git a/config.py b/config.py index 7737738..acccace 100644 --- a/config.py +++ b/config.py @@ -40,19 +40,290 @@ SYNC_MAPPING = { "target_schema": "productionContractData", "target_table": "26年温度计合同数据", "pk_col": "ID" + }, + "25年压力表合同数据": { + "target_schema": "productionContractData", + "target_table": "25年压力表合同数据", + "pk_col": "ID" + }, + "25年温度计合同数据": { + "target_schema": "productionContractData", + "target_table": "25年温度计合同数据", + "pk_col": "ID" + } + + }, + r"\\192.168.110.114\生产进度表\2026年数据\成品入库.accdb": { + "成品交检记录": { + "target_schema": "productWarehousing", + "target_table": "成品交检记录_YEAR2026", + "pk_col": "ID" + }, + "成品入库记录": { + "target_schema": "productWarehousing", + "target_table": "成品入库记录_YEAR2026", + "pk_col": "ID" } }, - - # # 第二个 Access 文件 (可能有同名的 UserInfo 表) - # r"D:\Data\ProjectB\backup.mdb": { - # "UserInfo": { - # "target_schema": "dbo", - # "target_table": "ProjB_UserInfo", # 映射到 B 表 - # "pk_col": "ID" - # } - # } + r"\\192.168.110.114\生产进度表\2026年数据\一车间.accdb": { + "一车间记录": { + "target_schema": "workshopOne", + "target_table": "一车间记录_YEAR2026", + "pk_col": "ID" + } + }, + r"\\192.168.110.114\生产进度表\2026年数据\二车间.accdb": { + "二车间记录": { + "target_schema": "workshopTwo", + "target_table": "二车间记录_YEAR2026", + "pk_col": "ID" + } + }, + r"\\192.168.110.114\生产进度表\2026年数据\三车间.accdb": { + "三车间记录": { + "target_schema": "workshopThree", + "target_table": "三车间记录_YEAR2026", + "pk_col": "ID" + }, + "温度计调校记录": { + "target_schema": "workshopThree", + "target_table": "温度计调校记录_YEAR2026", + "pk_col": "ID" + } + }, + r"\\192.168.110.114\生产进度表\2026年数据\机加工.accdb": { + "隔膜机加接收": { + "target_schema": "machining", + "target_table": "隔膜机加接收_YEAR2026", + "pk_col": "ID" + }, + "膜片焊接": { + "target_schema": "machining", + "target_table": "膜片焊接_YEAR2026", + "pk_col": "ID" + }, + "膜片接收": { + "target_schema": "machining", + "target_table": "膜片接收_YEAR2026", + "pk_col": "ID" + }, + "喷涂寄出": { + "target_schema": "machining", + "target_table": "喷涂寄出_YEAR2026", + "pk_col": "ID" + } + }, + r"\\192.168.110.114\生产进度表\2026年数据\计划.accdb": { + "报警器接单记录": { + "target_schema": "contractPlanning", + "target_table": "报警器接单记录_YEAR2026", + "pk_col": "ID" + }, + "二车间接单记录": { + "target_schema": "contractPlanning", + "target_table": "二车间接单记录_YEAR2026", + "pk_col": "ID" + }, + "隔膜接单记录": { + "target_schema": "contractPlanning", + "target_table": "隔膜接单记录_YEAR2026", + "pk_col": "ID" + }, + "过压保护器接单记录": { + "target_schema": "contractPlanning", + "target_table": "过压保护器接单记录_YEAR2026", + "pk_col": "ID" + }, + "精密表部件接收记录": { + "target_schema": "contractPlanning", + "target_table": "精密表部件接收记录_YEAR2026", + "pk_col": "ID" + }, + "精密表接单记录": { + "target_schema": "contractPlanning", + "target_table": "精密表接单记录_YEAR2026", + "pk_col": "ID" + }, + "库房接单记录": { + "target_schema": "contractPlanning", + "target_table": "库房接单记录_YEAR2026", + "pk_col": "ID" + }, + "软管接单记录": { + "target_schema": "contractPlanning", + "target_table": "软管接单记录_YEAR2026", + "pk_col": "ID" + }, + "氩弧焊接单记录": { + "target_schema": "contractPlanning", + "target_table": "氩弧焊接单记录_YEAR2026", + "pk_col": "ID" + }, + "一车间接单记录": { + "target_schema": "contractPlanning", + "target_table": "一车间接单记录_YEAR2026", + "pk_col": "ID" + }, + "阻尼器接单记录": { + "target_schema": "contractPlanning", + "target_table": "阻尼器接单记录_YEAR2026", + "pk_col": "ID" + } + }, + r"\\192.168.110.114\生产进度表\2026年数据\检验记录数据库.accdb": { + "检验合格记录表": { + "target_schema": "inspectionRecords", + "target_table": "检验合格记录表_YEAR2026", + "pk_col": "ID" + } + }, + r"\\192.168.110.114\生产进度表\2026年数据\零件库.accdb": { + "表盘缺件记录": { + "target_schema": "partsWarehouse", + "target_table": "表盘缺件记录_YEAR2026", + "pk_col": "ID" + }, + "表盘入库记录": { + "target_schema": "partsWarehouse", + "target_table": "表盘入库记录_YEAR2026", + "pk_col": "ID" + }, + "部件发放记录": { + "target_schema": "partsWarehouse", + "target_table": "部件发放记录_YEAR2026", + "pk_col": "ID" + }, + "出库单发出记录": { + "target_schema": "partsWarehouse", + "target_table": "出库单发出记录_YEAR2026", + "pk_col": "ID" + }, + "出库单记录": { + "target_schema": "partsWarehouse", + "target_table": "出库单记录_YEAR2026", + "pk_col": "ID" + }, + "法兰缺件记录": { + "target_schema": "partsWarehouse", + "target_table": "法兰缺件记录_YEAR2026", + "pk_col": "ID" + }, + "法兰入库记录": { + "target_schema": "partsWarehouse", + "target_table": "法兰入库记录_YEAR2026", + "pk_col": "ID" + }, + "卡到库房记录": { + "target_schema": "partsWarehouse", + "target_table": "卡到库房记录_YEAR2026", + "pk_col": "ID" + }, + "喷涂到厂": { + "target_schema": "partsWarehouse", + "target_table": "喷涂到厂_YEAR2026", + "pk_col": "ID" + }, + "缺件记录": { + "target_schema": "partsWarehouse", + "target_table": "缺件记录_YEAR2026", + "pk_col": "ID" + }, + "温度计法兰缺件记录": { + "target_schema": "partsWarehouse", + "target_table": "温度计法兰缺件记录_YEAR2026", + "pk_col": "ID" + }, + "温度计法兰入库记录": { + "target_schema": "partsWarehouse", + "target_table": "温度计法兰入库记录_YEAR2026", + "pk_col": "ID" + } + }, + r"\\192.168.110.114\生产进度表\2026年数据\温度计记录.accdb": { + "采 技 机记录": { + "target_schema": "thermometerRecord", + "target_table": "采 技 机记录_YEAR2026", + "pk_col": "ID" + }, + "温度计调校记录": { + "target_schema": "thermometerRecord", + "target_table": "温度计调校记录_YEAR2026", + "pk_col": "ID" + }, + "温度计检验记录": { + "target_schema": "thermometerRecord", + "target_table": "温度计检验记录_YEAR2026", + "pk_col": "ID" + }, + "温度计组装记录": { + "target_schema": "thermometerRecord", + "target_table": "温度计组装记录_YEAR2026", + "pk_col": "ID" + } + }, + r"\\192.168.110.114\生产进度表\2026年数据\锡焊数据.accdb": { + "操作者1完成记录": { + "target_schema": "solderingData", + "target_table": "操作者1完成记录_YEAR2026", + "pk_col": "ID" + }, + "操作者2完成记录": { + "target_schema": "solderingData", + "target_table": "操作者2完成记录_YEAR2026", + "pk_col": "ID" + }, + "零件到车间记录": { + "target_schema": "solderingData", + "target_table": "零件到车间记录_YEAR2026", + "pk_col": "ID" + } + }, + r"\\192.168.110.114\生产进度表\2026年数据\氩弧焊.accdb": { + "表壳焊接记录": { + "target_schema": "TIGWelding", + "target_table": "表壳焊接记录_YEAR2026", + "pk_col": "ID" + }, + "超压": { + "target_schema": "TIGWelding", + "target_table": "超压_YEAR2026", + "pk_col": "ID" + }, + "氦测": { + "target_schema": "TIGWelding", + "target_table": "氦测_YEAR2026", + "pk_col": "ID" + }, + "接收": { + "target_schema": "TIGWelding", + "target_table": "接收_YEAR2026", + "pk_col": "ID" + }, + "退火": { + "target_schema": "TIGWelding", + "target_table": "退火_YEAR2026", + "pk_col": "ID" + }, + "氩弧焊记录": { + "target_schema": "TIGWelding", + "target_table": "氩弧焊记录_YEAR2026", + "pk_col": "ID" + }, + "氩弧焊领料": { + "target_schema": "TIGWelding", + "target_table": "氩弧焊领料_YEAR2026", + "pk_col": "ID" + } + }, + r"\\192.168.110.114\生产进度表\2026年数据\执行卡下发记录.accdb": { + "执行卡下发记录": { + "target_schema": "executionCardIssuanceRecord", + "target_table": "执行卡下发记录_YEAR2026", + "pk_col": "ID" + } + } } # ================= 运行参数 ================= -POLL_INTERVAL = 60 # 轮询间隔(秒) -BATCH_SIZE = 1000 # 批量处理大小 \ No newline at end of file +POLL_INTERVAL = 5 # 轮询间隔(秒) +BATCH_SIZE = 10000 # 批量处理大小 \ No newline at end of file diff --git a/init_full_sync.py b/init_full_sync.py index 490653b..192a176 100644 --- a/init_full_sync.py +++ b/init_full_sync.py @@ -1,83 +1,196 @@ -# init_full_sync.py import pyodbc import config 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 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(): - print("=== 开始全量初始化同步 ===") - print("注意：将基于 config.SYNC_MAPPING 中的文件路径进行处理") + logger = setup_logger() - sql_conn = db_utils.get_sql_conn() - sql_cursor = sql_conn.cursor() - sql_cursor.fast_executemany = True + logger.info("=== 开始全量初始化同步 ===") + logger.info("注意:将基于 config.SYNC_MAPPING 中的文件路径进行处理") + + 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 + logger.info("✓ SQL Server 连接成功") + except Exception as e: + logger.error(f"❌ SQL Server 连接失败: {e}") + return - # 第一层循环：遍历配置文件中的所有文件路径 + # 第一层循环:遍历配置文件中的所有文件路径 for acc_path, tables_map in config.SYNC_MAPPING.items(): if not os.path.exists(acc_path): - print(f"⚠️ [跳过] 文件不存在: {acc_path}") + logger.warning(f"⚠️ [跳过] 文件不存在: {acc_path}") continue - print(f"\n📂 正在处理文件: {acc_path}") + logger.info(f"\n📂 正在处理文件: {acc_path}") try: acc_conn = db_utils.get_access_conn(acc_path) acc_cursor = acc_conn.cursor() + logger.info(f" ✓ Access 文件连接成功") - # 第二层循环：遍历该文件下的所有表映射 + # 第二层循环:遍历该文件下的所有表映射 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) - print(f" 👉 同步表: [{acc_table}] -> {full_target_name}") + logger.info(f" 👉 开始同步表: [{acc_table}] -> {full_target_name}") + 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)}") 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" ✓ 已清空目标表") - # 4. 开启 IDENTITY_INSERT (尝试) - try: sql_cursor.execute(f"SET IDENTITY_INSERT {full_target_name} ON") - except: pass + # 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 + logger.info(f" ✓ 已启用 IDENTITY_INSERT") + except Exception as id_err: + logger.error(f" ⚠️ 无法启用 IDENTITY_INSERT: {id_err}") + raise # 5. 传输数据 - print(f" 正在读取并写入数据...") + logger.info(f" 正在读取并写入数据...") acc_cursor.execute(f"SELECT * FROM [{acc_table}]") - total_rows = 0 + table_rows = 0 + start_time = time.time() + while True: rows = acc_cursor.fetchmany(config.BATCH_SIZE) if not rows: break + sql_cursor.executemany(insert_sql, rows) - total_rows += len(rows) - print(f" 已写入 {total_rows} 行...", end='\r') + table_rows += len(rows) + + # 定期记录进度到日志 + if table_rows % (config.BATCH_SIZE * 10) == 0: + elapsed = time.time() - start_time + rate = table_rows / elapsed if elapsed > 0 else 0 + logger.info(f" 进度: 已写入 {table_rows:,} 行 | 速率: {rate:,.0f} 行/秒") # 6. 关闭 IDENTITY_INSERT - try: sql_cursor.execute(f"SET IDENTITY_INSERT {full_target_name} OFF") - except: pass + if identity_enabled: + try: + sql_cursor.execute(f"SET IDENTITY_INSERT {full_target_name} OFF") + logger.info(f" ✓ 已关闭 IDENTITY_INSERT") + except Exception as id_err: + logger.warning(f" ⚠️ 关闭 IDENTITY_INSERT 时警告: {id_err}") sql_conn.commit() - print(f"\n ✅ 完成。") + + # 显示最终统计 + 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)}") + + success_tables += 1 + total_rows += table_rows except Exception as tbl_err: - print(f" ❌ 表级错误: {tbl_err}") + failed_tables += 1 + logger.error(f" ❌ 表级错误: {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() + logger.info(f" ✓ Access 文件处理完成") except Exception as file_err: - print(f" ❌ 文件级错误: {file_err}") + logger.error(f" ❌ 文件级错误: {file_err}", exc_info=True) sql_conn.close() - print("\n=== 全量同步结束 ===") + + # 总结统计 + 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)}") + if total_rows > 0 and sync_time > 0: + logger.info(f"整体平均速率: {total_rows/sync_time:,.0f} 行/秒") + logger.info("="*60) 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 cc0bea0..0a8b2df 100644 --- a/run_incremental_sync.py +++ b/run_incremental_sync.py @@ -6,182 +6,162 @@ import config import db_utils import logging from logging.handlers import TimedRotatingFileHandler -from collections import defaultdict - -# ================= 日志系统配置 (新增) ================= +# ================= 日志系统配置 ================= def setup_logger(): - """ - 配置双向日志系统：控制台输出 + 按天轮转的文件日志 - """ - # 1. 确保日志目录存在 log_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'log') if not os.path.exists(log_dir): os.makedirs(log_dir) - # 2. 创建 Logger logger = logging.getLogger("SyncService") logger.setLevel(logging.INFO) - logger.handlers = [] # 清除旧句柄，防止重复打印 + logger.handlers = [] - # 3. 文件处理器 (按天轮转) - # filename: 正在写入的日志名 - # when='midnight': 每天午夜轮转 - # interval=1: 间隔1天 - # backupCount=30: 保留最近30个文件 log_file_path = os.path.join(log_dir, 'sync_service.log') file_handler = TimedRotatingFileHandler( log_file_path, when='midnight', interval=1, backupCount=30, encoding='utf-8' ) - file_handler.suffix = "%Y-%m-%d" # 轮转后的文件名后缀 - # 文件日志格式：[时间] [级别] 消息 + file_handler.suffix = "%Y-%m-%d" file_fmt = logging.Formatter('%(asctime)s - [%(levelname)s] - %(message)s') file_handler.setFormatter(file_fmt) - # 4. 控制台处理器 (屏幕输出) console_handler = logging.StreamHandler(sys.stdout) - # 控制台日志格式：[时间] 消息 (保持简洁) console_fmt = logging.Formatter('%(asctime)s - %(message)s', datefmt='%H:%M:%S') console_handler.setFormatter(console_fmt) - # 5. 添加处理器 logger.addHandler(file_handler) logger.addHandler(console_handler) - return logger -# 初始化全局 Logger logger = setup_logger() -# ================= 辅助函数 ================= - -def clean_access_path(raw_addr): - """清洗 TableAddress 字段""" - if not raw_addr: return "" - s = str(raw_addr).strip() - if s.upper().startswith(";DATABASE="): return s[10:] - if s.upper().startswith("LOCAL:"): return s[6:] - return s - -def get_norm_path(path): - """获取标准化路径""" - if not path: return "" - return os.path.normpath(path).lower() - # ================= 主逻辑 ================= -def process_logs(): +def process_sync_task(): + """ + 逻辑重构： + 遍历 config.SYNC_MAPPING 中的每一个文件 -> 去日志表查询该文件下特定表的未同步记录。 + """ sql_conn = db_utils.get_sql_conn() sql_cursor = sql_conn.cursor() - sql_cursor.fast_executemany = True - - cols = config.LOG_TABLE_CONFIG + sql_cursor.fast_executemany = True # 高性能开关 - # --- 0. 预处理配置映射 --- - normalized_mapping = {} - for cfg_path, cfg_tables in config.SYNC_MAPPING.items(): - norm_key = get_norm_path(cfg_path) - normalized_mapping[norm_key] = { - 'original_path': cfg_path, - 'tables': cfg_tables - } + # 标记是否有工作被处理（用于控制轮询休眠时间） + work_done = False + + cols = config.LOG_TABLE_CONFIG + log_full_name = db_utils.fmt_table(cols['schema'], cols['table_name']) try: - log_full_name = db_utils.fmt_table(cols['schema'], cols['table_name']) - - # --- 1. 读取日志 --- - query_log = f""" - SELECT TOP 1000 - {cols['col_log_id']}, - {cols['col_table_name']}, - {cols['col_record_id']}, - {cols['col_address']} - FROM {log_full_name} - WHERE {cols['col_synced']} = 0 - AND TableType = 'LINKED_ACCESS' - ORDER BY {cols['col_log_id']} ASC - """ - sql_cursor.execute(query_log) - logs = sql_cursor.fetchall() - - if not logs: - return False - - logger.info(f"📥 收到 {len(logs)} 条链接表变更日志...") - - # --- 2. 任务分组 --- - tasks = defaultdict(lambda: defaultdict(lambda: {'record_ids': set(), 'log_ids': []})) - - for row in logs: - log_id, acc_table, record_id, raw_address = row + # === 核心循环：以配置文件为驱动 === + for clean_path, tables_map in config.SYNC_MAPPING.items(): - clean_path = clean_access_path(raw_address) - norm_lookup_key = get_norm_path(clean_path) - - # 校验配置 - if norm_lookup_key not in normalized_mapping: - # 记录一条警告日志，但不刷屏 - logger.warning(f"忽略未配置的文件路径: {clean_path}") - continue - - mapping_node = normalized_mapping[norm_lookup_key] - - if acc_table not in mapping_node['tables']: - logger.warning(f"忽略未配置的表: {acc_table} (在文件 {os.path.basename(clean_path)} 中)") + # 1. 准备查询条件 + # 获取该文件下所有需要同步的表名列表 + target_tables = list(tables_map.keys()) + if not target_tables: continue - tasks[clean_path][acc_table]['record_ids'].add(record_id) - tasks[clean_path][acc_table]['log_ids'].append(log_id) - - # --- 3. 执行同步循环 --- - for file_path, tables_data in tasks.items(): + # 构造 TableAddress 的精确匹配条件 + # VBA 逻辑：网络路径带 ";DATABASE=", 本地路径带 "LOCAL:" + # 我们直接构造这两个字符串，让 SQL Server 做精确匹配，效率极高 + # 注意：将路径转为 Windows 标准反斜杠 + win_path = os.path.normpath(clean_path) + addr_candidates = [ + f";DATABASE={win_path}", # 情况1 + f"LOCAL={win_path}", # 情况2 (注意 VBA 代码里可能是 LOCAL: 或 LOCAL=，请核对) + f"LOCAL:{win_path}", # 情况3 + win_path # 情况4 (兼容没有前缀的情况) + ] - if not os.path.exists(file_path): - logger.error(f"无法访问文件: {file_path}") + # 2. 构造动态 SQL 查询 + # WHERE Synced=0 AND Address IN (...) AND TableName IN (...) + placeholders_addr = ','.join(['?'] * len(addr_candidates)) + placeholders_tbl = ','.join(['?'] * len(target_tables)) + + query_log = f""" + SELECT TOP 1000 + {cols['col_log_id']}, + {cols['col_table_name']}, + {cols['col_record_id']} + FROM {log_full_name} + WHERE {cols['col_synced']} = 0 + AND TableType = 'LINKED_ACCESS' + AND {cols['col_address']} IN ({placeholders_addr}) + AND {cols['col_table_name']} IN ({placeholders_tbl}) + ORDER BY {cols['col_log_id']} ASC + """ + + # 参数列表：先放地址，再放表名 + params = addr_candidates + target_tables + + sql_cursor.execute(query_log, params) + logs = sql_cursor.fetchall() + + if not logs: + continue # 这个文件没有需要同步的记录，检查下一个文件 + + work_done = True # 标记有工作 + logger.info(f"📂 文件 {os.path.basename(clean_path)} 发现 {len(logs)} 条变更...") + + # 3. 本地分组 (按表名) + # 结构: table_tasks[TableName] = { ids: {}, log_ids: [] } + table_tasks = {} + for row in logs: + log_id, acc_table, record_id = row + if acc_table not in table_tasks: + table_tasks[acc_table] = {'record_ids': set(), 'log_ids': []} + table_tasks[acc_table]['record_ids'].add(record_id) + table_tasks[acc_table]['log_ids'].append(log_id) + + # 4. 执行同步 (连接一次 Access，处理多张表) + if not os.path.exists(clean_path): + logger.error(f"无法访问文件: {clean_path}") continue try: - acc_conn = db_utils.get_access_conn(file_path) + acc_conn = db_utils.get_access_conn(clean_path) acc_cursor = acc_conn.cursor() except Exception as conn_err: - logger.error(f"连接 Access 失败 [{file_path}]: {conn_err}") + logger.error(f"连接 Access 失败: {conn_err}") continue - norm_key = get_norm_path(file_path) - config_node = normalized_mapping[norm_key] - - for acc_table, data in tables_data.items(): + for acc_table, data in table_tasks.items(): record_ids = list(data['record_ids']) log_ids = data['log_ids'] - target_conf = config_node['tables'][acc_table] + # 读取目标配置 + target_conf = tables_map[acc_table] target_schema = target_conf['target_schema'] target_table = target_conf['target_table'] pk_col = target_conf['pk_col'] target_full_name = db_utils.fmt_table(target_schema, target_table) try: + # --- A. Access 查新数据 --- ids_placeholders = ','.join(['?'] * len(record_ids)) - - # A. Access 读取 acc_sql = f"SELECT * FROM [{acc_table}] WHERE [{pk_col}] IN ({ids_placeholders})" acc_cursor.execute(acc_sql, record_ids) new_rows = acc_cursor.fetchall() acc_cols = [col[0] for col in acc_cursor.description] - # B. SQL Server 写入 + # --- B. SQL Server 删旧插新 (事务) --- + # 1. 删除 del_sql = f"DELETE FROM {target_full_name} WHERE [{pk_col}] IN ({ids_placeholders})" sql_cursor.execute(del_sql, record_ids) + # 2. 插入 if new_rows: insert_sql = db_utils.generate_insert_sql(target_schema, target_table, acc_cols) try: sql_cursor.execute(f"SET IDENTITY_INSERT {target_full_name} ON") except: pass + sql_cursor.executemany(insert_sql, new_rows) + try: sql_cursor.execute(f"SET IDENTITY_INSERT {target_full_name} OFF") except: pass - # C. 更新日志状态 + # 3. 标记日志 Synced = 1 log_placeholders = ','.join(['?'] * len(log_ids)) update_log_sql = f""" UPDATE {log_full_name} @@ -191,15 +171,15 @@ def process_logs(): sql_cursor.execute(update_log_sql, log_ids) sql_conn.commit() - logger.info(f"✅ 同步成功: {target_table} 更新 {len(record_ids)} 条 (来源: {os.path.basename(file_path)})") + logger.info(f" -> {target_table} 同步成功: {len(record_ids)} 条") except Exception as tbl_err: - logger.error(f"处理表 {acc_table} 异常: {tbl_err}") + logger.error(f" -> 处理表 {acc_table} 异常: {tbl_err}") sql_conn.rollback() - acc_conn.close() + acc_conn.close() # 关闭 Access 连接 - return True + return work_done except Exception as e: logger.critical(f"全局严重异常: {e}") @@ -209,14 +189,18 @@ def process_logs(): except: pass if __name__ == "__main__": - logger.info("=== 增量同步服务已启动 (日志模式: log/sync_service.log) ===") + logger.info("=== 增量同步服务启动 (配置驱动模式) ===") + logger.info("正在按 config.SYNC_MAPPING 轮询...") + while True: try: - has_work = process_logs() - time.sleep(0.5 if has_work else config.POLL_INTERVAL) + has_work = process_sync_task() + # 如果有工作，说明可能还有积压，休息短一点(0.1s) + # 如果没工作，休息标准间隔(5s) + time.sleep(0.1 if has_work else config.POLL_INTERVAL) except KeyboardInterrupt: - logger.info("收到停止指令，服务正在关闭...") + logger.info("服务停止") break except Exception as e: logger.critical(f"主循环崩溃: {e}") - time.sleep(5) # 防止死循环刷屏 \ No newline at end of file + time.sleep(5) \ No newline at end of file diff --git a/vbareplace.py b/vbareplace.py index db79582..2c87c2f 100644 --- a/vbareplace.py +++ b/vbareplace.py @@ -29,7 +29,7 @@ def main(): if getattr(sys, 'frozen', False): script_dir = os.path.dirname(sys.executable) else: - script_dir = os.path.dirname(os.path.abspath(__file)) + script_dir = os.path.dirname(os.path.abspath(__file__)) vba_txt_file = os.path.join(script_dir, "vba.txt") @@ -46,8 +46,8 @@ def main(): return try: - # 启动 Access - access = win32com.client.Dispatch("Access.Application") + # 启动 Access（使用 DispatchEx 强制创建新实例，避免与已打开的 Access 冲突） + access = win32com.client.DispatchEx("Access.Application") access.Visible = False # 后台运行，不显示窗口 # 打开数据库 @@ -84,13 +84,45 @@ def main(): access.DoCmd.Close(0, form_name, 0) # 尝试关闭，避免卡住 print(f"处理窗体 {form_name} 时出错: {e}") + # 刷新所有链接表 + refreshed_tables = 0 + failed_tables = 0 + try: + db = access.CurrentDb() + table_count = db.TableDefs.Count + + for i in range(table_count): + table_def = db.TableDefs(i) + # 检查是否是链接表（Connect属性不为空） + if table_def.Connect: + try: + table_def.RefreshLink() + print(f"已刷新链接表: {table_def.Name}") + refreshed_tables += 1 + except Exception as e: + print(f"刷新链接表 {table_def.Name} 失败: {e}") + failed_tables += 1 + except Exception as e: + print(f"刷新链接表时出错: {e}") + access.CloseCurrentDatabase() access.Quit() - messagebox.showinfo("成功", f"操作完成！\n\n已处理 {form_count} 个窗体，成功替换 {replaced} 个。\n\n文件已保存: {os.path.basename(access_file)}") + # 构建结果消息 + result_msg = f"操作完成！\n\n" + result_msg += f"已处理 {form_count} 个窗体，成功替换 {replaced} 个。\n" + if refreshed_tables > 0 or failed_tables > 0: + result_msg += f"已刷新 {refreshed_tables} 个链接表" + if failed_tables > 0: + result_msg += f"，{failed_tables} 个失败" + result_msg += "。\n" + result_msg += f"\n文件已保存: {os.path.basename(access_file)}" + + messagebox.showinfo("成功", result_msg) except Exception as e: messagebox.showerror("运行错误", f"操作失败:\n{e}\n\n请检查是否已启用 VBA 项目访问信任。") if __name__ == "__main__": - main() \ No newline at end of file + main() +# pyinstaller --onefile --noconsole --name "VBA代码批量替换" vbareplace.py \ No newline at end of file