1ae901a040
- Add new Access table mappings for transmitter contracts (25/26年变送器) and tube bending workshop (弯管车间烘洗) - Add auto-create target table from Access schema when table does not exist (db_utils + init_full_sync) - Implement per-primary-key verification in incremental sync: verify deletes are gone and inserts are present before marking Synced=1 - Add post-commit re-verification with Synced rollback on failure for automatic retry - Batch IN clause parameters to stay under SQL Server limit - Adjust poll interval from 5s to 30s - Improve IDENTITY_INSERT cleanup in finally blocks Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
276 lines
11 KiB
Python
276 lines
11 KiB
Python
import pyodbc
|
|
from config import SQL_SERVER_CONN, ACCESS_DRIVER, SYNC_MAPPING, BATCH_SIZE
|
|
import db_utils
|
|
import time
|
|
import os
|
|
import ntfy_utils
|
|
from log_utils import (LoggerManager, log_success, log_error, log_warning, log_info, log_processing,
|
|
log_skip, log_start, log_file, log_database, log_sync, log_complete)
|
|
|
|
# 初始化日志管理器
|
|
LoggerManager("init_full_sync", log_prefix="full_sync")
|
|
|
|
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 _send_summary_notification(sync_results, success_count, failed_count, total_rows, total_time):
|
|
"""发送汇总 ntfy 通知"""
|
|
if not sync_results:
|
|
return
|
|
|
|
# 分类成功和失败的表
|
|
success_tables = [r for r in sync_results if r['status'] == 'success']
|
|
failed_tables = [r for r in sync_results if r['status'] == 'failed']
|
|
|
|
# 构建消息
|
|
lines = []
|
|
lines.append("🎯 全量同步完成\n")
|
|
|
|
# 成功的表
|
|
if success_tables:
|
|
lines.append(f"✅ 成功 {len(success_tables)} 张表:")
|
|
for r in success_tables:
|
|
lines.append(f" • {r['table']}: {r['rows']:,} 行, {format_duration(r['duration'])}")
|
|
lines.append("")
|
|
|
|
# 失败的表
|
|
if failed_tables:
|
|
lines.append(f"❌ 失败 {failed_count} 张表:")
|
|
for r in failed_tables:
|
|
error_brief = r['error'][:50] + "..." if len(r['error']) > 50 else r['error']
|
|
lines.append(f" • {r['table']}: {error_brief}")
|
|
lines.append("")
|
|
|
|
# 总计
|
|
lines.append(f"总计: {total_rows:,} 行 | 用时: {format_duration(total_time)}")
|
|
|
|
message = "\n".join(lines)
|
|
|
|
# 根据成功/失败情况决定优先级和标签
|
|
if failed_count > 0:
|
|
priority = "high"
|
|
tags = ["warning", "database"]
|
|
else:
|
|
priority = "default"
|
|
tags = ["white_check_mark", "database"]
|
|
|
|
ntfy_utils.send_ntfy(
|
|
message=message,
|
|
title="全量同步完成",
|
|
priority=priority,
|
|
tags=tags
|
|
)
|
|
|
|
def run_full_sync():
|
|
log_info("=" * 70)
|
|
log_start( "全量初始化同步")
|
|
log_info( f"配置文件数: {len(SYNC_MAPPING)} 个")
|
|
log_info( f"批次大小: {BATCH_SIZE} 行")
|
|
log_info("=" * 70)
|
|
|
|
sync_start_time = time.time()
|
|
total_tables = 0
|
|
success_tables = 0
|
|
failed_tables = 0
|
|
total_rows = 0
|
|
|
|
# 收集每张表的同步结果，用于最终汇总通知
|
|
sync_results = [] # 格式: {'table': str, 'status': 'success'|'failed', 'rows': int, 'duration': float, 'error': str}
|
|
|
|
try:
|
|
sql_conn = db_utils.get_sql_conn()
|
|
sql_cursor = sql_conn.cursor()
|
|
sql_cursor.fast_executemany = True
|
|
log_database( "SQL Server 连接成功")
|
|
except Exception as e:
|
|
log_error( f"SQL Server 连接失败: {e}")
|
|
return
|
|
|
|
# 第一层循环:遍历配置文件中的所有文件路径
|
|
for acc_path, tables_map in SYNC_MAPPING.items():
|
|
if not os.path.exists(acc_path):
|
|
log_skip( f"文件不存在: {acc_path}")
|
|
continue
|
|
|
|
log_file( 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( 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( 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( f" 检测到 {len(columns)} 个列")
|
|
insert_sql = db_utils.generate_insert_sql(target_schema, target_table, columns)
|
|
|
|
# 3. 检查目标表是否存在，不存在则自动创建
|
|
if not db_utils.table_exists(sql_cursor, target_schema, target_table):
|
|
db_utils.ensure_schema(sql_cursor, target_schema)
|
|
db_utils.create_table_from_access(
|
|
sql_cursor, target_schema, target_table,
|
|
acc_cursor.description, target_config['pk_col']
|
|
)
|
|
sql_conn.commit() # DDL 必须单独提交，否则 pyodbc 参数绑定无法获取正确的列元数据
|
|
log_info( f" 目标表不存在，已自动创建")
|
|
else:
|
|
sql_cursor.execute(f"TRUNCATE TABLE {full_target_name}")
|
|
log_info( 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( f" 已启用 IDENTITY_INSERT")
|
|
except Exception as id_err:
|
|
log_error( f" 无法启用 IDENTITY_INSERT: {id_err}")
|
|
raise
|
|
|
|
# 5. 传输数据
|
|
log_info( 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( 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( f" 已关闭 IDENTITY_INSERT")
|
|
except Exception as id_err:
|
|
log_warning( 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_info( f"表 [{target_table}] 完成: {table_rows:,} 行 | 速率: {final_rate:,.0f} 行/秒 | 用时: {format_duration(table_time)}")
|
|
|
|
# 收集成功结果
|
|
sync_results.append({
|
|
'table': target_table,
|
|
'status': 'success',
|
|
'rows': table_rows,
|
|
'duration': table_time,
|
|
'error': None
|
|
})
|
|
|
|
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
|
|
error_msg = str(tbl_err)
|
|
log_info( f"表 [{acc_table}] 同步失败: {error_msg}")
|
|
|
|
# 收集失败结果
|
|
sync_results.append({
|
|
'table': target_table,
|
|
'status': 'failed',
|
|
'rows': 0,
|
|
'duration': time.time() - table_start_time,
|
|
'error': error_msg
|
|
})
|
|
|
|
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( summary)
|
|
|
|
except Exception as file_err:
|
|
log_error( f"文件 [{os.path.basename(acc_path)}] 处理失败: {file_err}", exc_info=True)
|
|
|
|
sql_conn.close()
|
|
|
|
# 总结统计
|
|
sync_time = time.time() - sync_start_time
|
|
log_info("\n" + "=" * 70)
|
|
log_info( "全量同步任务结束")
|
|
log_info("-" * 70)
|
|
log_info( f"总表数: {total_tables} 张 | 成功: {success_tables} 张 | 失败: {failed_tables} 张")
|
|
log_info( f"总行数: {total_rows:,} 行")
|
|
log_info( f"总用时: {format_duration(sync_time)}")
|
|
if total_rows > 0 and sync_time > 0:
|
|
log_info( f"整体平均速率: {total_rows/sync_time:,.0f} 行/秒")
|
|
log_info("=" * 70)
|
|
|
|
# 发送汇总 ntfy 通知
|
|
_send_summary_notification(sync_results, success_tables, failed_tables, total_rows, sync_time)
|
|
|
|
if __name__ == "__main__":
|
|
run_full_sync() |