影刀RPA JSON数据处理：API返回数据的解析与加工

影刀RPA JSON数据处理API返回数据的解析与加工作者林焱什么情况用调用API、读取配置文件、解析网页接口返回——现代数据交换几乎全是JSON格式。如果你只会影刀的内置文本处理节点遇到嵌套JSON基本束手无策。典型场景调用企业微信API返回的部门树是嵌套JSON需要展平成列表爬取接口返回了100条数据需要按某个字段分组统计从配置JSON里读取参数需要校验字段是否存在核心场景在影刀的Python节点中高效解析和加工JSON数据。怎么做拼多多店群自动化上架方案第一步JSON操作速查——5个最常用的函数importjson# 1. 解析JSON字符串 → Python对象json_str{name: 张三, age: 28, skills: [Python, RPA]}datajson.loads(json_str)print(data[name])# 张三# 2. Python对象 → JSON字符串obj{status:ok,count:100}textjson.dumps(obj,ensure_asciiFalse)# ensure_asciiFalse保证中文不转码print(text)# {status: ok, count: 100}# 3. 美化输出prettyjson.dumps(obj,ensure_asciiFalse,indent2)# 4. 从文件读取JSONwithopen(config.json,r,encodingutf-8)asf:configjson.load(f)# 5. 写入JSON文件withopen(output.json,w,encodingutf-8)asf:json.dump(data,f,ensure_asciiFalse,indent2)第二步安全解析——JSON可能不标准defsafe_json_parse(text): 安全解析JSON处理各种不标准的情况 # 尝试1直接解析try:returnjson.loads(text)exceptjson.JSONDecodeError:pass# 尝试2去掉BOM头iftext.startswith(\ufeff):try:returnjson.loads(text[1:])except:pass# 尝试3JSON可能被包在markdown代码块里ifintext:importrematchre.search(r(?:json)?\s*\n?(.*?),text,re.DOTALL)ifmatch:try:returnjson.loads(match.group(1))except:pass# 尝试4可能是单引号的伪JSONPython dict的reprtry:importastreturnast.literal_eval(text)except:passreturnNone第三步嵌套JSON展平——最常用的加工defflatten_json(nested_obj,parent_key,sep_): 将嵌套JSON展平为一级字典 例{user: {name: 张三, addr: {city: 北京}}} → {user_name: 张三, user_addr_city: 北京} items{}ifisinstance(nested_obj,dict):fork,vinnested_obj.items():new_keyf{parent_key}{sep}{k}ifparent_keyelsekifisinstance(v,(dict,list)):items.update(flatten_json(v,new_key,sep))else:items[new_key]velifisinstance(nested_obj,list):fori,vinenumerate(nested_obj):new_keyf{parent_key}{sep}{i}ifisinstance(v,(dict,list)):items.update(flatten_json(v,new_key,sep))else:items[new_key]vreturnitems# 测试nested{order_id:ORD001,customer:{name:张三,contact:{phone:13812345678,email:zhangsanexample.com}},items:[{product:笔记本,price:5999},{product:鼠标,price:199}]}flatflatten_json(nested)fork,vinflat.items():print(f{k}:{v})# 输出# order_id: ORD001# customer_name: 张三# customer_contact_phone: 13812345678# customer_contact_email: zhangsanexample.com# items_0_product: 笔记本# items_0_price: 5999# items_1_product: 鼠标# items_1_price: 199第四步JSON数据提取与筛选用jmespath库做JSON查询语法比手写循环简洁很多需pip install jmespath。importjmespath# 示例数据企业微信部门树API返回data{errcode:0,department:[{id:1,name:总公司,children:[{id:2,name:技术部,children:[{id:5,name:前端组},{id:6,name:后端组}]},{id:3,name:销售部,children:[{id:7,name:华东区},{id:8,name:华南区}]}]}]}# jmespath查询——比写循环简洁很多# 提取所有部门名称namesjmespath.search(department[*].name,data)print(names)# [总公司]# 递归提取所有层级的部门名称all_namesjmespath.search(department[*].children[*].name,data)print(all_names)# [技术部, 销售部]# 更优雅的递归写法# jmespath没有直接递归但可以组合查询# 不用jmespath的原生写法也能搞就是啰嗦点defextract_all_departments(nodes,resultNone):递归提取所有部门ifresultisNone:result[]fornodeinnodes:result.append({id:node[id],name:node[name]})ifchildreninnode:extract_all_departments(node[children],result)returnresult all_deptsextract_all_departments(data[department])print(f共{len(all_depts)}个部门)fordinall_depts:print(f{d[id]}:{d[name]})第五步JSON到DataFrame——数据分析的前置步骤importpandasaspddefjson_to_dataframe(json_data,record_pathNone): 将JSON数据转为DataFrame 自动处理嵌套结构 ifisinstance(json_data,str):json_datajson.loads(json_data)# 如果是列表ifisinstance(json_data,list):# 先展平flat_list[flatten_json(item)foriteminjson_data]returnpd.DataFrame(flat_list)# 如果是字典提取record_path指定的列表ifisinstance(json_data,dict):ifrecord_pathandrecord_pathinjson_data:recordsjson_data[record_path]flat_list[flatten_json(item)foriteminrecords]returnpd.DataFrame(flat_list)else:returnpd.DataFrame([flatten_json(json_data)])returnpd.DataFrame()# 示例api_response{code:200,data:{list:[{name:张三,dept:{name:技术部,id:2},salary:15000},{name:李四,dept:{name:销售部,id:3},salary:12000},{name:王五,dept:{name:技术部,id:2},salary:18000},],total:3}}dfjson_to_dataframe(api_response,record_pathdata)# 从list里提取df2json_to_dataframe(api_response[data][list])print(df2)# name dept_name dept_id salary# 0 张三 技术部 2 15000# 1 李四 销售部 3 12000# 2 王五 技术部 2 18000# 直接分组统计print(df2.groupby(dept_name)[salary].mean())第六步处理超大JSON——流式读取如果JSON文件有几百MBjson.load()会直接把内存撑爆。importijson# pip install ijsondefstream_large_json(file_path,item_pathitem): 流式读取大JSON逐个处理 适用于 {items: [巨大的数组]} 这种结构 count0withopen(file_path,r,encodingutf-8)asf:# 逐个读取数组中的元素foriteminijson.items(f,f{item_path}):process_one(item)# 处理单个元素count1ifcount%10000:print(f已处理{count}条...)print(f共处理{count}条)returncount有什么坑坑1NaN/Infinity 不是合法JSONpandas的DataFrame转JSON时如果有NaNjson.dumps()会报错。importmathimportjson# ❌ 这会报错# data {value: float(nan)}# json.dumps(data) # ValueError: Out of range float values are not JSON compliant# ✅ 处理NaNdefjson_safe(obj):ifisinstance(obj,float):[video(video-zqOhJoM2-1783576560041)(type-csdn)(url-https://live.csdn.net/v/embed/524993)(image-https://v-blog.csdnimg.cn/asset/a547123d88ad712dccba346c9217e237/cover/Cover0.jpg)(title-TEMU店群如何管理运营)]ifmath.isnan(obj):returnNoneifmath.isinf(obj):returnNonereturnobj data{value:float(nan)}cleanjson.dumps(data,defaultjson_safe)坑2中文编码——ensure_asciiFalse忘了加json.dumps()默认会把中文转成\uXXXX。data{城市:北京}print(json.dumps(data))# {\u57ce\u5e02: \u5317\u4eac}print(json.dumps(data,ensure_asciiFalse))# {城市: 北京}坑3JSON的key可能和Python关键字冲突API返回的JSON有{class: A, type: B}class是Python关键字obj.class会报语法错误。解决方法用字典访问obj[class]而不是obj.class。坑4大JSON解析性能10万条记录的JSON用json.loads()全部加载再处理内存可能飙到几个G。解决方法用ijson流式处理或者只提取需要的字段再加载到DataFrame。总结JSON数据处理是RPA Python节点的基本功。核心是安全解析容错各种格式、嵌套展平flatten_json、分组统计pandas、大文件流式处理ijson。熟练掌握这些80%的API数据处理场景都能搞定。