一、问题场景几乎每个制造类项目第一步都卡在同一件事上客户有几年的物料台账躺在 Excel 里少则几百行多则上万行要迁进 ERP。手工录不现实直接导又一定带脏数据——全角半角混着、名称带空格、同一个件好几条记录、计量单位写什么的都有。结果就是导入报错一片或者更糟——导进去了但要花一个月去清理重复数据。这篇把流程拆成四步清洗 → 查重 → 比对已有数据 → 分批入库每一步都落成可复用脚本。这套做法在几十万行规模以内的台账上都够用。二、环境前置Python 3.8安装 pandas、openpyxl、requests目标系统为金蝶云星空V7.5已开通 WebAPI 并建好接口用户关键前提计量单位、物料分组这些引用类数据要先在系统里建好否则导入必失败先在测试账套跑通再动正式环境三、实现步骤读取与清洗清洗的目标只有一个让「同一个东西写成不同样子」的情况在进入比对前就消失。import reimport pandas as pddef normalize(s):s str(s).strip().replace( “, “) # 去首尾空白 全角空格s “”.join(chr(ord© - 0xfee0) if 0xff01 ord© 0xff5e else c for c in s) # 全角转半角return re.sub(r”\s”, , s).upper()def load_excel(path, cols):df pd.read_excel(path, dtypestr).fillna(“”)df.columns [c.strip() for c in df.columns]missing [c for c in cols if c not in df.columns]if missing:raise ValueError(“Excel 缺少列{}”.format(missing))for c in cols:df[c] df[c].map(normalize)return df[cols]COLS [“物料编码”, “物料名称”, “规格型号”, “计量单位”, “物料分组”]df load_excel(“物料台账.xlsx”, COLS)print(“读取 {} 行”.format(len(df)))全角转半角这一步很容易被忽略但它是同名不同码的主要来源之一。2. 查重先出清单不要自动合并def find_duplicates(df):key df[“物料名称”] “|” df[“规格型号”]dup df[key.duplicated(keepFalse)].copy()dup[“_key”] key[key.duplicated(keepFalse)]return dup.sort_values(“_key”).drop(columns“_key”)dups find_duplicates(df)dups.to_excel(“疑似重复清单.xlsx”, indexFalse)print(“疑似重复 {} 行已导出待人工确认”.format(len(dups)))这一步必须人工确认不能自动合并。名同规格同的可能是真重复也可能是不同供应商的通用件合并错了会影响后续成本核算。清单给业务确认完再回填到台账里。3. 与系统已有数据比对迁移最怕的是重复导入。以物料编码为准先拉出系统里已存在的编码集合。import json, requestsdef login(base, acct, user, pwd):s requests.Session()s.headers.update({“Content-Type”: “application/json”})res s.post(“{}/Kingdee.BOS.WebApi.ServicesStub.AuthService.ValidateUser.common.kdsvc”.format(base),datajson.dumps({“parameters”: [acct, user, pwd, 2052]}), timeout60).json()if res ! 1:raise RuntimeError(“登录失败{}”.format(res))return sdef query_bills(s, base, form_id, field_keys, filter_str“”, limit2000):param {“FormId”: form_id, “FieldKeys”: field_keys, “FilterString”: filter_str,“OrderString”: “FNumber asc”, “TopRowCount”: 0, “StartRow”: 0, “Limit”: limit}url “{}/Kingdee.BOS.WebApi.ServicesStub.DynamicFormService.ExecuteBillQuery.common.kdsvc”.format(base)return s.post(url, datajson.dumps({“parameters”: [json.dumps(param, ensure_asciiFalse)]}), timeout120).json()拉已有物料编码rows query_bills(s, BASE, “BD_MATERIAL”, “FNumber,FName”)exists {r[0] for r in rows}todo df[~df[“物料编码”].isin(exists)]print(“新品 {} 条已存在跳过 {} 条”.format(len(todo), len(df) - len(todo)))4. 分批入库失败落表一次推 100 条失败的行写到 CSV人工看完再重跑不要整个批次回滚重来。def save_material(s, base, row):model {“Creator”: “”, “NeedReturnFields”: [“FNumber”], “IsAutoSubmitAndAudit”: False,“Model”: {“FNumber”: row[“物料编码”],“FName”: row[“物料名称”],“FSpecification”: row[“规格型号”],“FBaseUnitId”: {“FNumber”: row[“计量单位”]},“FMaterialGroup”: {“FNumber”: row[“物料分组”]},},}url “{}/Kingdee.BOS.WebApi.ServicesStub.DynamicFormService.Save.common.kdsvc”.format(base)return s.post(url, datajson.dumps({“parameters”: [json.dumps({“FormId”: “BD_MATERIAL”, “data”: model}, ensure_asciiFalse)]},ensure_asciiFalse), timeout120).json()ok, fail 0, []for i in range(0, len(todo), 100):for _, row in todo.iloc[i:i 100].iterrows():try:res save_material(s, BASE, row)if res and res[0].get(“Status”, {}).get(“IsSuccess”):ok 1else:fail.append([row[“物料编码”], str(res)[:200]])except Exception as e:fail.append([row[“物料编码”], str(e)[:200]])pd.DataFrame(fail, columns[“物料编码”, “失败原因”]).to_excel(“导入失败清单.xlsx”, indexFalse)print(“成功 {} 条失败 {} 条”.format(ok, len(fail)))5. 收尾对账迁移完必须核对一次Excel 总行数 成功数 已存在跳过数 失败数。三个数对不上说明中间有丢行回头查读取环节。四、字段映射表Excel 列ERP 字段说明物料编码FNumber唯一键必填建议先定编码规则再迁物料名称FName必填规格型号FSpecification建议纳入查重键计量单位FBaseUnitId.FNumber引用字段系统里必须已存在物料分组FMaterialGroup.FNumber引用字段同上是否批次管理FIsBatchManage布尔值注意 Excel 里的「是/否」要转换创建组织FCreateOrgId.FNumber有多个公司主体时必填单一主体也要给五、常见报错与处理现象原因处理提示编码已存在系统里已有该编码迁移前先查已有数据改为跳过或走更新必填字段缺失计量单位或创建组织没给补 FBaseUnitId、FCreateOrgId计量单位不存在Excel 里的单位名与系统编码不一致先导系统计量单位表做一层映射名称含非法字符单引号、全角符号、换行在清洗阶段统一处理掉部分行失败单条校验不过分组不存在等失败清单落表人工修正后重跑请求超时单批条数过大每批控制在 100 条以内六、几个容易踩的坑先小批量试。 拿 10 条跑通全流程确认字段映射无误再放开整批。编码规则要在迁移前定。 把 Excel 里的乱编码原样搬进系统等于把历史问题一起迁走后面治理成本更高。引用类数据先建。 计量单位、物料分组、供应商分类这些不在系统里存在导入必报错。幂等靠编码查重不靠接口报错。 迁移脚本一定会重跑每次先拉已有编码做差集比等接口报已存在更省事。一定要做条数对账。 成功率再高行数对不上就说明有数据丢了这一步不能省。七、小结数据迁移的难点是数据本身接口只占很小一部分。把「清洗 → 查重 → 比对 → 分批入库 → 对账」拆成五步每步都有产出物重复清单、失败清单、对账表迁移就从一次性高风险动作变成了可回滚、可复盘的流程。