影刀RPA 电商订单自动处理：从下单到发货的全链路自动化
什么情况用什么 → 怎么做 → 有什么坑
作者：林焱 | 飞行社出品
什么情况用什么
每天几百个订单，客服手动录单、拣货、发货、回填快递单号，重复劳动到怀疑人生？
这套方案适合：
- 淘宝/京东/拼多多/抖音小店多平台订单汇总
- 订单自动审核（黑名单、异常地址、重复下单检测）
- 拣货单自动生成并打印
- 发货后快递单号自动回填平台
核心工具：影刀RPA + 电商开放平台API（淘宝开放平台/京东宙斯/拼多多开放平台） + 打印机控制
怎么做
第一步：对接电商平台API，获取订单数据
拼多多店群自动化上架方案
以淘宝开放平台为例，订单数据通过taobao.trades.sold.get接口获取：
importtop.apiimportjsonfromdatetimeimportdatetime,timedeltadeffetch_taobao_orders(session,start_time,end_time):"""获取淘宝卖家已卖出的订单"""req=top.api.TradesSoldGetRequest()req.fields="tid,status,payment,receiver_name,receiver_phone,orders,created"req.start_created=start_time req.end_created=end_time req.status="WAIT_SELLER_SEND_GOODS"# 等待卖家发货req.page_no=1req.page_size=100try:resp=req.getResponse(session)trades=resp['trades_sold_get_response']['trades']['trade']returntradesexceptExceptionase:print(f"获取订单失败:{e}")return[]# 获取今天未发货的订单session=get_top_session()# 你的TOP会话管理函数today=datetime.now().strftime("%Y-%m-%d %H:%M:%S")yesterday=(datetime.now()-timedelta(days=1)).strftime("%Y-%m-%d %H:%M:%S")orders=fetch_taobao_orders(session,yesterday,today)print(f"获取到{len(orders)}个待发货订单")关键点：不同平台的API字段差异很大，需要写一个统一的订单标准化函数：
defnormalize_order(platform,raw_order):"""将不同平台的订单统一为标准格式"""ifplatform=="taobao":return{"platform":"淘宝","order_id":raw_order["tid"],"buyer_name":raw_order["receiver_name"],"phone":raw_order["receiver_phone"],"amount":float(raw_order["payment"]),"skus":[item["num_iid"]foriteminraw_order["orders"]["order"]],"created":raw_order["created"]}elifplatform=="jd":# 京东订单格式转换return{"platform":"京东","order_id":raw_order["order_id"],"buyer_name":raw_order["consignee_info"]["consignee"],# ...}# 其他平台类似处理第二步：订单自动审核，过滤异常订单
defaudit_order(order,blacklist,suspicious_rules):""" 订单审核：返回 (是否通过, 原因) """# 1. 黑名单检查iforder["buyer_name"]inblacklistororder["phone"]inblacklist:returnFalse,"买家在黑名单中"# 2. 异常地址检测address=order.get("address","")iflen(address)<10:returnFalse,"收货地址过短，疑似虚假"# 3. 重复下单检测（同手机号短时间多次下单）recent_orders=get_recent_orders_by_phone(order["phone"],hours=24)iflen(recent_orders)>=3:returnFalse,f"24小时内同手机号下单{len(recent_orders)}次，疑似刷单"# 4. 金额异常（超过平均客单价5倍）avg_order_value=get_avg_order_value(days=30)iforder["amount"]>avg_order_value*5:returnFalse,"订单金额异常偏高，需人工审核"returnTrue,"审核通过"# 批量审核fororderinorders:passed,reason=audit_order(order,blacklist,rules)ifpassed:mark_order_audited(order["order_id"])else:send_to_manual_review(order,reason)第三步：生成拣货单并打印
审核通过的订单，自动生成拣货单：
defgenerate_picking_list(audited_orders):"""生成拣货单：按仓库库位排序，优化拣货路径"""fromcollectionsimportdefaultdict# 按SKU汇总数量sku_summary=defaultdict(int)fororderinaudited_orders:forskuinorder["skus"]:sku_summary[sku]+=1# 查询每个SKU的库位信息picking_items=[]forsku,qtyinsku_summary.items():location=get_sku_location(sku)# 从ERP/WMS查询库位picking_items.append({"sku":sku,"sku_name":get_sku_name(sku),"quantity":qty,"location":location,"zone":location[:2]# A区/B区，用于路径优化})# 按库位排序（同区集中拣货）picking_items.sort(key=lambdax:(x["zone"],x["location"]))# 生成打印内容content=build_picking_list_print_content(picking_items,audited_orders)returncontentdefprint_picking_list(content):"""调用打印机打印拣货单"""importwin32printimportwin32api printer_name=win32print.GetDefaultPrinter()hPrinter=win32print.OpenPrinter(printer_name)try:hJob=win32print.StartDocPrinter(hPrinter,1,("拣货单",None,"RAW"))win32print.StartPagePrinter(hPrinter)win32print.WritePrinter(hPrinter,content.encode('gbk'))win32print.EndPagePrinter(hPrinter)win32print.EndDocPrinter(hPrinter)finally:win32print.ClosePrinter(hPrinter)第四步：发货后自动回填快递单号
defship_and_sync(order_id,platform,tracking_number,logistics_company):"""发货并回填快递单号到平台"""ifplatform=="taobao":req=top.api.LogisticsConsignReserveRequest()req.tid=order_id req.out_sid=tracking_number req.company_code=map_logistics_code(logistics_company)# 平台要求的物流公司编码resp=req.getResponse(session)elifplatform=="jd":# 京东物流接口url="https://api.jd.com/routerjson"params={"method":"jingdong.pop.omnichannel.uploadInvoice"# ...}resp=requests.post(url,json=params)# 记录发货状态到本地数据库update_order_shipping_status(order_id,tracking_number,logistics_company)print(f"订单{order_id}发货成功，快递单号：{tracking_number}")第五步：用影刀RPA编排完整流程
在影刀中创建「电商订单自动处理」主流程：
【定时触发】每30分钟运行一次 ↓ 【Python节点】fetch_taobao_orders() → 获取待发货订单 ↓ 【循环】遍历每个订单 ↓ 【Python节点】audit_order() → 审核订单 ↓ 【条件判断】审核通过？ ├─ 是 → 加入待拣货列表 └─ 否 → 写入人工审核队列，发送企微通知 ↓ 【Python节点】generate_picking_list() → 生成拣货单 ↓ 【Python节点】print_picking_list() → 打印拣货单 ↓ 【等待操作】拣货完成，扫描快递单号（人工触发或PDA扫描） ↓ 【Python节点】ship_and_sync() → 回填快递单号 ↓ 【企微通知】发送发货通知给买家有什么坑
坑1：电商平台API有调用频率限制
淘宝开放平台限制：每日最高调用量根据你的API套餐而定，一般是100万次/天，但单次查询不能超过40次/秒。
# 错误做法：快速连续调用fororderinorders:resp=api.get_order_detail(order["tid"])# 会被限流！# 正确做法：加限速控制importtimedefrate_limited_api_call(func,*args,max_per_second=10,**kwargs):"""限速调用API"""ifnothasattr(rate_limited_api_call,"last_call_time"):rate_limited_api_call.last_call_time=0elapsed=time.time()-rate_limited_api_call.last_call_timeifelapsed<1.0/max_per_second:time.sleep(1.0/max_per_second-elapsed)rate_limited_api_call.last_call_time=time.time()returnfunc(*args,**kwargs)坑2：打印机调用在Windows服务模式下失败
影刀RPA如果以Windows服务运行（开机自启、无用户登录），win32print.GetDefaultPrinter()会返回空，因为服务没有用户会话的打印机配置。
TEMU店群如何管理运营？
解决方案：使用指定打印机名称，而不是默认打印机：
# 修改为指定打印机PRINTER_NAME="Zebra ZT230"# 你的拣货单打印机型号defprint_picking_list(content):hPrinter=win32print.OpenPrinter(PRINTER_NAME)# 用指定名称# ...或者用影刀的「启动程序」指令调用一个.vbs脚本打印，绕过会话限制。
坑3：不同平台的物流公司编码不统一
淘宝要求的是STO（申通）、YTO（圆通）这种缩写，京东要求的是全名，拼多多又是一套。维护一个映射表：
LOGISTICS_CODE_MAP={"淘宝":{"申通快递":"STO","圆通速递":"YTO","中通快递":"ZTO","韵达快递":"YD","顺丰速运":"SF"},"京东":{"申通快递":"shentong","圆通速递":"yuantong",# ...}}坑4：缺货订单的处理
如果拣货时发现库存不足，需要自动拆分订单（有货部分先发）或者挂起订单。建议在生成拣货单时实时查询库存：
defcheck_stock_before_picking(sku):"""拣货前实时查库存"""stock=get_current_stock(sku)ifstock<=0:returnFalse,"缺货"returnTrue,f"库存充足（剩余{stock}）"总结
| 步骤 | 关键动作 | 预计节省时间 |
|---|---|---|
| 订单获取 | API自动拉取，无需手动导出 | 每天省1小时 |
| 订单审核 | 自动过滤异常，减少人工审核量80% | 每天省2小时 |
| 拣货单生成 | 按库位排序，优化拣货路径 | 拣货效率提升30% |
| 快递单回填 | 自动回填，无需手动逐个操作 | 每天省1.5小时 |
实际落地时，建议先接一个平台验证流程，跑通后再复制到其他平台。多平台订单汇总到一张表里，统一管理，这才是自动化的真正价值。
