diff --git a/run_incremental_sync.py b/run_incremental_sync.py index 759fbf2..4f8ef79 100644 --- a/run_incremental_sync.py +++ b/run_incremental_sync.py @@ -113,6 +113,7 @@ def process_sync_task(): log_error(f"无法访问文件: {clean_path}") continue + acc_conn = None try: acc_conn = db_utils.get_access_conn(clean_path) acc_cursor = acc_conn.cursor() @@ -129,7 +130,7 @@ def process_sync_task(): for acc_table, data in table_tasks.items(): record_ids = list(data['record_ids']) log_ids = data['log_ids'] - + # 读取目标配置 target_conf = tables_map[acc_table] target_schema = target_conf['target_schema'] @@ -139,7 +140,25 @@ def process_sync_task(): log_processing(f"正在同步表 [{acc_table}] → [{target_table}] ({len(record_ids)} 条记录)") + # 检查目标表是否存在,不存在则自动创建 + if not db_utils.table_exists(sql_cursor, target_schema, target_table): + db_utils.ensure_schema(sql_cursor, target_schema) + acc_cursor.execute(f"SELECT TOP 1 * FROM [{acc_table}]") + db_utils.create_table_from_access( + sql_cursor, target_schema, target_table, + acc_cursor.description, pk_col + ) + log_info(f"目标表 [{target_table}] 不存在,已自动创建") + + # 事务状态追踪 + identity_enabled = False + transaction_begun = False + try: + # --- 开始事务 --- + sql_cursor.execute("BEGIN TRANSACTION") + transaction_begun = True + # --- A. Access 查新数据 --- ids_placeholders = ','.join(['?'] * len(record_ids)) acc_sql = f"SELECT * FROM [{acc_table}] WHERE [{pk_col}] IN ({ids_placeholders})" @@ -147,62 +166,70 @@ def process_sync_task(): new_rows = acc_cursor.fetchall() acc_cols = [col[0] for col in acc_cursor.description] - # --- B. SQL Server 删旧插新 (事务) --- - # 1. 删除 + # --- 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. 插入 + # 2. 插入新数据 if new_rows: insert_sql = db_utils.generate_insert_sql(target_schema, target_table, acc_cols) # 检查并启用 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 {target_full_name} ON") - identity_enabled = True - except Exception as id_err: - log_error(f"无法启用 IDENTITY_INSERT: {id_err}") - raise + sql_cursor.execute(f"SET IDENTITY_INSERT {target_full_name} ON") + identity_enabled = True sql_cursor.executemany(insert_sql, new_rows) # 关闭 IDENTITY_INSERT if identity_enabled: - try: - sql_cursor.execute(f"SET IDENTITY_INSERT {target_full_name} OFF") - except Exception as id_err: - log_warning(f"关闭 IDENTITY_INSERT 时警告: {id_err}") + sql_cursor.execute(f"SET IDENTITY_INSERT {target_full_name} OFF") + identity_enabled = False # 3. 标记日志 Synced = 1 log_placeholders = ','.join(['?'] * len(log_ids)) update_log_sql = f""" - UPDATE {log_full_name} - SET {cols['col_synced']} = 1 + UPDATE {log_full_name} + SET {cols['col_synced']} = 1 WHERE {cols['col_log_id']} IN ({log_placeholders}) """ sql_cursor.execute(update_log_sql, log_ids) + # 提交事务 sql_conn.commit() + transaction_begun = False log_info(f"表 [{target_table}] 同步完成: {len(record_ids)} 条记录") file_success_count += 1 file_total_records += len(record_ids) except Exception as tbl_err: - # 确保清理 IDENTITY_INSERT 状态 - try: - sql_cursor.execute(f"SET IDENTITY_INSERT {target_full_name} OFF") - except: - pass + # 回滚事务 + if transaction_begun: + sql_conn.rollback() + transaction_begun = False + + # 清理 IDENTITY_INSERT 状态 + if identity_enabled: + try: + sql_cursor.execute(f"SET IDENTITY_INSERT {target_full_name} OFF") + except: + pass + log_error(f"表 [{acc_table}] 同步失败: {tbl_err}") - sql_conn.rollback() file_error_count += 1 - acc_conn.close() # 关闭 Access 连接 + finally: + # 确保 Access 连接关闭 + try: + if acc_conn: + acc_conn.close() + log_info(f"已关闭 Access 连接: {os.path.basename(clean_path)}") + except Exception as close_err: + log_warning(f"关闭 Access 连接时出错: {close_err}") # 输出文件级别的汇总 if file_success_count > 0 or file_error_count > 0: