import pymysql import json from collections import defaultdict from tqdm import tqdm # 配置信息 SOURCE_DB = { 'host': 'localhost', 'user': 'root', 'password': 'root', 'database': '', 'charset': 'utf8mb4' } TARGET_DB = { 'host': 'localhost', 'user': 'root', 'password': 'root', 'database': '', 'charset': 'utf8mb4' } SOURCE_TENANT = '' TARGET_TENANT = '' # 表迁移顺序（按依赖关系排序） TABLE_ORDER = [ 'basics_factory', 'basics_equip_type', 'basics_factory_workshop', 'basics_factory_line', 'basics_factory_workshop_section', 'basics_factory_station', 'basics_wms_warehouse', 'basics_wms_area', 'basics_wms_location', 'basics_wms_shelf', 'basics_wms_position', 'basics_manpowersys_post', 'basics_manpowersys_staff', 'basics_hr_classes', 'basics_hr_workgroup', 'basics_hr_workgroup_staff', 'basics_material_group', 'basics_craft_process', 'basics_craft_route', 'basics_craft_material', 'basics_craft_bom', 'basics_craft_drawing_page', 'basics_equip_spare_part', 'basics_equip_component', 'basics_equip_ledger', 'basics_equip_ledger_spare', 'basics_quality_method', 'basics_quality_tool', 'basics_quality_sample', 'basics_quality_sample_details', 'basics_quality_category', 'basics_quality_plan', 'basics_quality_plan_item', 'basics_production_craftl_materiel', 'basics_production_craftl_route', 'basics_wms_bar_code', 'basics_wms_kn_rqgl', 'basics_file', 'conf_craft_route_material', 'conf_craft_route_process', 'conf_line_production', 'conf_line_station', 'conf_station' ] # 外键关系映射 FOREIGN_KEYS = { 'basics_factory_workshop': {'factory_id': 'basics_factory'}, 'basics_factory_line': { 'factory_id': 'basics_factory', 'workshop_id': 'basics_factory_workshop', 'section_id': 'basics_factory_workshop_section' }, 'basics_factory_workshop_section': { 'factory_id': 'basics_factory', 'workshop_id': 'basics_factory_workshop', 'line_id': 'basics_factory_line' }, 'basics_factory_station': {'factory_id': 'basics_factory'}, 'basics_equip_ledger': { 'factory_id': 'basics_factory', 'workshop_id': 'basics_factory_workshop', 'line_id': 'basics_factory_line', 'workshop_section_id': 'basics_factory_workshop_section', 'equip_type_id': 'basics_equip_type' }, 'basics_equip_ledger_spare': { 'equip_ledger_id': 'basics_equip_ledger', 'spare_part_id': 'basics_equip_spare_part' }, 'basics_wms_warehouse': { 'factory_id': 'basics_factory' }, 'basics_wms_area': { 'factory_id': 'basics_factory', 'warehouse_id': 'basics_wms_warehouse' }, 'basics_wms_location': { 'factory_id': 'basics_factory', 'warehouse_id': 'basics_wms_warehouse', 'area_id': 'basics_wms_area' }, 'basics_wms_shelf': { 'factory_id': 'basics_factory', 'warehouse_id': 'basics_wms_warehouse', 'area_id': 'basics_wms_area', 'location_id': 'basics_wms_location' }, 'basics_wms_position': { 'factory_id': 'basics_factory', 'warehouse_id': 'basics_wms_warehouse', 'area_id': 'basics_wms_area', 'location_id': 'basics_wms_location', 'shelf_id': 'basics_wms_shelf' }, 'basics_hr_classes': {'line_id': 'basics_factory_line'}, 'basics_hr_workgroup': {'line_id': 'basics_factory_line'}, 'basics_hr_workgroup_staff': { 'workgroup_id': 'basics_hr_workgroup', 'staff_id': 'basics_manpowersys_staff' }, 'basics_material_group': {'factory_id': 'basics_factory'}, 'basics_craft_drawing_page': {'craft_id': 'basics_craft_route'}, 'basics_craft_bom': {'parent_id': 'basics_craft_bom'}, 'basics_quality_category': {'parent_id': 'basics_quality_category'}, 'basics_quality_plan_item': {'plan_code': 'basics_quality_plan'}, 'basics_wms_bar_code': {'type_id': None} # 动态处理 } def migrate_data(): # 连接数据库 source_conn = pymysql.connect(**SOURCE_DB) target_conn = pymysql.connect(**TARGET_DB) source_cursor = source_conn.cursor(pymysql.cursors.DictCursor) target_cursor = target_conn.cursor(pymysql.cursors.DictCursor) # ID映射字典 {table: {old_id: new_id}} id_map = defaultdict(dict) # 自关联表待处理队列 {table: [(new_id, old_parent_id)]} self_ref_queue = defaultdict(list) try: for table in tqdm(TABLE_ORDER, desc="迁移表进度"): # 查询源数据 source_cursor.execute( f"SELECT * FROM {table} WHERE tenant_uuid = %s AND deleted = 0", (SOURCE_TENANT,) ) rows = source_cursor.fetchall() if not rows: continue # 获取目标表所有字段（排除自增ID） target_cursor.execute(f"DESCRIBE {table}") columns = [col['Field'] for col in target_cursor.fetchall() if col['Field'] != 'id'] # 处理每行数据 for row in tqdm(rows, desc=f"迁移 {table}"): old_id = row['id'] del row['id'] # 更新租户UUID row['tenant_uuid'] = TARGET_TENANT # 处理外键映射 if table in FOREIGN_KEYS: for fk_col, ref_table in FOREIGN_KEYS[table].items(): if fk_col in row and row[fk_col]: # 动态类型表处理 if ref_table is None: ref_table = row.get('type', 'unknown') if ref_table in id_map: row[fk_col] = id_map[ref_table].get(row[fk_col], row[fk_col]) # 处理JSON字段 if 'extra' in row and isinstance(row['extra'], str): try: row['extra'] = json.dumps(json.loads(row['extra'])) except: pass # 准备插入数据 values = [row.get(col) for col in columns] placeholders = ', '.join(['%s'] * len(values)) # 执行插入 target_cursor.execute( f"INSERT INTO {table} ({', '.join(columns)}) " f"VALUES ({placeholders})", values ) new_id = target_cursor.lastrowid # 保存ID映射 id_map[table][old_id] = new_id # 处理自关联字段 if table in ['basics_craft_bom', 'basics_quality_category']: parent_id = row.get('parent_id') if parent_id: self_ref_queue[table].append((new_id, parent_id)) # 处理自关联更新 if table in self_ref_queue: for new_id, old_parent_id in self_ref_queue[table]: new_parent_id = id_map[table].get(old_parent_id) if new_parent_id: target_cursor.execute( f"UPDATE {table} SET parent_id = %s WHERE id = %s", (new_parent_id, new_id) ) # 所有表都成功后才提交 target_conn.commit() print("数据迁移完成！") except Exception as e: print(f"迁移失败: {str(e)}") target_conn.rollback() finally: source_cursor.close() target_cursor.close() source_conn.close() target_conn.close() if __name__ == "__main__": migrate_data()