262 lines
9.8 KiB
Python
262 lines
9.8 KiB
Python
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 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:
|
||
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():
|
||
logger = setup_logger()
|
||
|
||
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
|
||
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
|
||
log_database(logger, "SQL Server 连接成功")
|
||
except Exception as 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):
|
||
log_skip(logger, f"文件不存在: {acc_path}")
|
||
continue
|
||
|
||
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()
|
||
log_database(logger, 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(logger, 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(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}")
|
||
log_info(logger, 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(logger, f" 已启用 IDENTITY_INSERT")
|
||
except Exception as id_err:
|
||
log_error(logger, f" 无法启用 IDENTITY_INSERT: {id_err}")
|
||
raise
|
||
|
||
# 5. 传输数据
|
||
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)
|
||
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(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")
|
||
log_info(logger, f" 已关闭 IDENTITY_INSERT")
|
||
except Exception as 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
|
||
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
|
||
file_failed += 1
|
||
log_error(logger, f"表 [{acc_table}] 同步失败: {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()
|
||
|
||
# 文件级别汇总
|
||
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:
|
||
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" + "=" * 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:
|
||
log_info(logger, f"整体平均速率: {total_rows/sync_time:,.0f} 行/秒")
|
||
logger.info("=" * 70)
|
||
|
||
if __name__ == "__main__":
|
||
run_full_sync() |