Unity游戏Lua性能优化实战：从编码习惯到架构设计
2026/7/10 2:49:03
影刀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"]}'data=json.loads(json_str)print(data["name"])# 张三# 2. Python对象 → JSON字符串obj={"status":"ok","count":100}text=json.dumps(obj,ensure_ascii=False)# ensure_ascii=False保证中文不转码print(text)# {"status": "ok", "count": 100}# 3. 美化输出pretty=json.dumps(obj,ensure_ascii=False,indent=2)# 4. 从文件读取JSONwithopen("config.json","r",encoding="utf-8")asf:config=json.load(f)# 5. 写入JSON文件withopen("output.json","w",encoding="utf-8")asf:json.dump(data,f,ensure_ascii=False,indent=2)第二步：安全解析——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# 尝试3：JSON可能被包在markdown代码块里if'```'intext:importrematch=re.search(r'```(?:json)?\s*\n?(.*?)```',text,re.DOTALL)ifmatch:try:returnjson.loads(match.group(1))except:pass# 尝试4：可能是单引号的伪JSON（Python dict的repr）try: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_key=f"{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_key=f"{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":"zhangsan@example.com"}},"items":[{"product":"笔记本","price":5999},{"product":"鼠标","price":199}]}flat=flatten_json(nested)fork,vinflat.items():print(f"{k}:{v}")# 输出：# order_id: ORD001# customer_name: 张三# customer_contact_phone: 13812345678# customer_contact_email: zhangsan@example.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查询——比写循环简洁很多# 提取所有部门名称names=jmespath.search("department[*].name",data)print(names)# ['总公司']# 递归提取所有层级的部门名称all_names=jmespath.search("department[*].children[*].name",data)print(all_names)# ['技术部', '销售部']# 更优雅的递归写法# jmespath没有直接递归，但可以组合查询# 不用jmespath的原生写法也能搞，就是啰嗦点：defextract_all_departments(nodes,result=None):"""递归提取所有部门"""ifresultisNone:result=[]fornodeinnodes:result.append({"id":node["id"],"name":node["name"]})if"children"innode:extract_all_departments(node["children"],result)returnresult all_depts=extract_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_path=None):""" 将JSON数据转为DataFrame 自动处理嵌套结构 """ifisinstance(json_data,str):json_data=json.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:records=json_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}}df=json_to_dataframe(api_response,record_path="data")# 从list里提取df2=json_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文件有几百MB，json.load()会直接把内存撑爆。
importijson# pip install ijsondefstream_large_json(file_path,item_path="item"):""" 流式读取大JSON，逐个处理 适用于 {"items": [巨大的数组]} 这种结构 """count=0withopen(file_path,"r",encoding="utf-8")asf:# 逐个读取数组中的元素foriteminijson.items(f,f"{item_path}"):process_one(item)# 处理单个元素count+=1ifcount%1000==0:print(f"已处理{count}条...")print(f"共处理{count}条")returncount有什么坑
坑1：NaN/Infinity 不是合法JSON
pandas的DataFrame转JSON时，如果有NaN，json.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')}clean=json.dumps(data,default=json_safe)坑2：中文编码——ensure_ascii=False忘了加
json.dumps()默认会把中文转成\uXXXX。
data={"城市":"北京"}print(json.dumps(data))# {"\u57ce\u5e02": "\u5317\u4eac"}print(json.dumps(data,ensure_ascii=False))# {"城市": "北京"}坑3：JSON的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数据处理场景都能搞定。