Python脚本实现Mysql数据迁移

import mysql.connector
import json

# 数据库连接配置
source_db_config = {
    'host': 'localhost',
    'port': 3306,
    'user': 'CallSystemProd',
    'password': '******',
    'database': ''******','  # 源数据库名称
}

target_db_config = {
    'host': 'localhost',
    'port': 3306,
    'user': 'CallSystemTest',
    'password': ''******',',
    'database': ''******','  # 目标数据库名称
}

# 连接到源数据库
try:
    source_conn = mysql.connector.connect(**source_db_config)
    source_cursor = source_conn.cursor(dictionary=True)
    print("成功连接到源数据库")
except mysql.connector.Error as err:
    print(f"连接源数据库失败: {err}")
    exit()

# 连接到目标数据库
try:
    target_conn = mysql.connector.connect(**target_db_config)
    target_cursor = target_conn.cursor()
    print("成功连接到目标数据库")
except mysql.connector.Error as err:
    print(f"连接目标数据库失败: {err}")
    exit()

# 获取源表数据（S_SMSSendLogs）
source_cursor.execute("""
    SELECT Id, SMSId, Phone, Status, IsSucceed, Trade_id, RequestStr, 
           ResponseStr, RobotTaskId, ExtendCode, CreateTime, UpdateTime, 
           CreateUserId, UpdateUserId, IsDelete, TaskCallResultId, ProviderType
    FROM S_SMSSendLogs
""")
columns = source_cursor.fetchall()

# 获取源表数据（S_SMSTemplates），用于RequestStr中的替换
source_cursor.execute("""
    SELECT SMSId, Content, Id
    FROM S_SMSTemplates
""")
templates = {template['SMSId']: template for template in source_cursor.fetchall()}

# 获取与TaskCallResultId相关的数据（t_taskcallresults和t_repeattaskcallresults）
source_cursor.execute("""
    SELECT id, CreateUserId, UpdateUserId, ResidentsId, CustomerName
    FROM T_TaskCallResults
""")
taskcallresults = {result['id']: result for result in source_cursor.fetchall()}

source_cursor.execute("""
    SELECT id, CreateUserId, UpdateUserId, ResidentsId, CustomerName
    FROM T_RepeatTaskCallResults
""")
repeattaskcallresults = {result['id']: result for result in source_cursor.fetchall()}

# 获取RobotTaskId和OrgId相关的表数据（t_taskcallresults, t_robotrepeattasks, sysuser）
source_cursor.execute("""
    SELECT id, CreateUserId
    FROM T_TaskCallResults
""")
taskcallresults_user_map = {result['id']: result['CreateUserId'] for result in source_cursor.fetchall()}

source_cursor.execute("""
    SELECT id, CreateUserId
    FROM T_RobotRepeatTasks
""")
robotrepeattasks_user_map = {result['id']: result['CreateUserId'] for result in source_cursor.fetchall()}

source_cursor.execute("""
    SELECT Id, OrgId
    FROM SysUser
""")
sysuser_map = {user['Id']: user['OrgId'] for user in source_cursor.fetchall()}

# 构造批量插入语句
insert_query = """
    INSERT INTO S_SMSSendLogs (Id, SMSId, Phone, Status, IsSucceed, Trade_id, RequestStr, 
                               ResponseStr, RobotTaskId, ExtendCode, CreateTime, UpdateTime, 
                               CreateUserId, UpdateUserId, IsDelete, TaskCallResultId, 
                               ProviderType, OrgId, Content, CharCount, BillingCount, ResidentsId, 
                               CustomerName, TemplateId)
    VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
"""

# 构造插入的值（批量数据），应用转换规则
insert_values = []
for column in columns:
    # Status 字段直接迁移，无需转换
    status = column['Status']

    # 将 IsSucceed 转换为 tinyint类型的 0
    is_succeed = 0 if column['IsSucceed'] == 1 else column['IsSucceed']

    # RequestStr 替换操作
    request_str = column['RequestStr']
    template = templates.get(column['SMSId'])
    if template:
        template_content = template['Content']
        try:
            # 获取 RequestStr 中的变量并替换
            request_values = json.loads(request_str).get('arguments', {})
            for key, value in request_values.items():
                template_content = template_content.replace(f"#{key}#", str(value) if value is not None else "")
        except json.JSONDecodeError:
            pass  # 如果 JSON 格式不正确，跳过替换
        request_str = template_content

    # 计算字符数
    char_count = len(request_str)
    
    # 计算计费条数
    billing_count = (char_count + 66) // 67  # 每条超过67个字算第二条

    # 获取相关的 UserId 和 OrgId
    create_user_id = taskcallresults.get(column['TaskCallResultId'], {}).get('CreateUserId') or \
                     repeattaskcallresults.get(column['TaskCallResultId'], {}).get('CreateUserId')

    update_user_id = taskcallresults.get(column['TaskCallResultId'], {}).get('UpdateUserId') or \
                     repeattaskcallresults.get(column['TaskCallResultId'], {}).get('UpdateUserId')

    # ProviderType 转换
    provider_type = column['ProviderType']
    if provider_type == 1:
        provider_type = 0
    elif provider_type == 2:
        provider_type = 3

    # OrgId 通过 CreateUserId 获取
    create_user_id_for_org = taskcallresults_user_map.get(column['RobotTaskId']) or \
                             robotrepeattasks_user_map.get(column['RobotTaskId'])
    org_id = sysuser_map.get(create_user_id_for_org, None)

    # 构造数据插入
    insert_values.append((
        column['Id'], column['SMSId'], column['Phone'], status, is_succeed, column['Trade_id'],
        request_str, column['ResponseStr'], column['RobotTaskId'], column['ExtendCode'],
        column['CreateTime'], column['UpdateTime'], create_user_id, update_user_id,
        column['IsDelete'], column['TaskCallResultId'], provider_type, org_id, request_str, char_count,
        billing_count, taskcallresults.get(column['TaskCallResultId'], {}).get('ResidentsId'),
        taskcallresults.get(column['TaskCallResultId'], {}).get('CustomerName'),
        template['Id'] if template else None
    ))

# 执行批量插入
target_cursor.executemany(insert_query, insert_values)
target_conn.commit()
print(f"成功迁移了 {len(insert_values)} 条数据")

# 关闭数据库连接
source_cursor.close()
source_conn.close()
target_cursor.close()
target_conn.close()