影刀RPA 数据迁移自动化：跨系统跨数据库数据搬运

影刀RPA 数据迁移自动化跨系统跨数据库数据搬运作者林焱什么情况用什么这是最常见的RPA场景之一把A系统的数据搬到B系统。但不是简单的CTRLC/CTRLV——A系统是老旧的CS架构客户端只支持导出TXTB系统是SaaS网页只支持Excel导入中间还需要做字段映射、数据清洗、去重校验。影刀RPA做数据迁移从源系统提取数据 → 清洗转换 → 加载到目标系统 → 验证数据一致性。俗称ETLExtract-Transform-Load只不过用RPA的方式实现。适用场景旧系统下线数据迁移到新系统CRM系统切换Salesforce → 国产CRM财务系统升级金蝶旧版 → 用友新版多店铺数据统一到ERP怎么做拼多多店群自动化报活动上架第一步从源系统提取数据老旧CS系统常见的提取方式方式1导出文件1. 【打开桌面应用】打开旧系统客户端 2. 【登录】输入账号密码 3. 【点击】菜单数据管理 → 导出数据 4. 【选择】导出范围、导出字段 5. 【点击】导出 6. 【等待文件生成】 7. 【读取文件】TXT/CSV/DBF格式读入影刀变量方式2数据库直连如果能拿到连接信息importpyodbc# 连接古老的SQL Server 2000connpyodbc.connect(DRIVER{SQL Server};SERVER192.168.1.50;DATABASEOldCRM;UIDsa;PWDxxx)dfpd.read_sql(SELECT * FROM customers,conn)conn.close()方式3如果连数据库都连不上最后一招——模拟人工复制1. 【打开桌面应用】打开旧系统 2. 【循环翻页】逐页读取客户列表 3. 【获取文本】逐行读取数据 4. 【写入Excel】追加到本地文件这招最慢但有时候是唯一的办法。第二步数据清洗与映射A系统的字段名和B系统可能完全不一样。# 字段映射表field_mapping{CUST_NAME:客户名称,CUST_TEL:联系电话,CUST_ADDR:详细地址,CREATE_DT:创建日期,CUST_TYPE:客户类型,}# 值映射编码转换type_mapping{1:个人客户,2:企业客户,3:政府客户,}# 数据清洗df_cleandf.copy()df_clean.rename(columnsfield_mapping,inplaceTrue)df_clean[客户类型]df_clean[客户类型].map(type_mapping)# 去重按客户名称电话去重df_clean.drop_duplicates(subset[客户名称,联系电话],keepfirst,inplaceTrue)# 数据校验invalid_phonesdf_clean[~df_clean[联系电话].str.match(r^1[3-9]\d{9}$)]iflen(invalid_phones)0:print(f警告{len(invalid_phones)}条手机号格式异常)第三步加载到目标系统方式1Excel批量导入1. 【执行Python】生成目标系统要求的Excel模板 2. 【打开网页】登录目标系统 3. 【点击】数据导入 → 选择文件 4. 【上传文件】选择生成的Excel 5. 【点击】开始导入 6. 【等待】导入完成 7. 【检查】导入结果——成功N条失败M条方式2逐条录入没有批量导入功能时1. 【打开网页】登录目标系统 2. 【点击】新增客户 3. 【循环】遍历清洗后的数据 - 【填写】客户名称 - 【填写】联系电话 - 【填写】详细地址 -  - 【下拉选择】客户类型 - 【点击】保存 - 【等待】1秒 4. 每50条暂停5秒避免限流第四步数据一致性验证迁移完了必须验证。TEMU店群矩阵自动化运营核价报活动# 验证逻辑source_countlen(df_clean)target_countget_target_system_count()# 从目标系统查总数checks{源系统记录数:source_count,目标系统记录数:target_count,差异:source_count-target_count,一致性:通过ifsource_counttarget_countelse异常}ifchecks[差异]!0:# 具体找出哪些记录没迁移成功source_idsset(df_clean[原系统ID])target_idsset(get_target_system_ids())missingsource_ids-target_idsprint(f缺失{len(missing)}条记录ID:{list(missing)[:10]}...)有什么坑坑1导出文件的编码问题老旧系统导出的TXT很可能是GBK编码而目标系统要求UTF-8。导入前必须转码withopen(old_data.txt,r,encodinggbk)asf:contentf.read()withopen(new_data.txt,w,encodingutf-8)asf:f.write(content)坑2大数量逐条录入极其缓慢如果10万条数据逐条录入每秒1条也要27小时。必须找到批量导入的入口——哪怕要写代码绕过去。实在没有批量入口考虑直接写数据库如果目标系统允许。坑3迁移过程中源系统仍在新增数据如果迁移跑了两天这两天里源系统又有新数据了。迁移方案必须包含增量迁移——记录迁移截止时间点之后再跑一次只迁新增的。坑4关联数据顺序问题客户表和订单表有关联如果先迁了订单再迁客户订单里的客户ID找不到对应的客户导入失败。迁移顺序必须是基础数据客户、商品→ 业务数据订单、合同。总结数据迁移的关键在准备阶段——字段映射表做得越细后面越顺。建议先迁移100条测试数据验证全流程无误后再批量迁。