影刀RPA JSON文件处理：读写与嵌套解析
影刀RPA JSON文件处理：读写与嵌套解析
署名：林焱
什么情况用什么
RPA流程中大量使用JSON——API返回数据、配置文件、网页数据交互。掌握JSON的读写和嵌套解析是处理API数据的必备技能。
| 场景 | 推荐函数 | 特点 |
|---|---|---|
| 读取JSON文件 | json.load | 文件读取 |
| JSON字符串转字典 | json.loads | 字符串解析 |
| 字典转JSON字符串 | json.dumps | 序列化 |
| 写入JSON文件 | json.dump | 文件写入 |
怎么做
一：基础读写
importjson# 写入JSON文件config={"site_url":"https://example.com","max_retries":3,"timeout":30,"headers":{"User-Agent":"Mozilla/5.0","Accept":"application/json"},"targets":["首页","产品页","关于我们"]}withopen(r"D:\config\settings.json","w",encoding="utf-8")asf:# ensure_ascii=False: 中文不转义为\uXXXX# indent=2: 格式化缩进json.dump(config,f,ensure_ascii=False,indent=2)# 读取JSON文件withopen(r"D:\config\settings.json","r",encoding="utf-8")asf:config=json.load(f)print(config["site_url"])print(config["headers"]["User-Agent"])print(config["targets"][0])二：字符串与JSON互转
拼多多店群自动化上架方案
importjson# JSON字符串 → Python字典（解析API响应）api_response='{"code": 0, "data": {"name": "张三", "age": 28}, "message": "success"}'result=json.loads(api_response)print(result["code"])# 0print(result["data"]["name"])# 张三print(result["data"]["age"])# 28# Python字典 → JSON字符串（构造API请求体）request_data={"action":"search","filters":{"category":"电子产品","price_min":100,"price_max":5000}}json_str=json.dumps(request_data,ensure_ascii=False)print(json_str)三：嵌套JSON解析
importjson# 复杂嵌套JSONapi_data={"status":"ok","results":[{"id":1,"name":"商品A","price":99.9,"tags":["热销","新品"],"seller":{"name":"XX旗舰店","rating":4.8,"address":{"city":"杭州","district":"西湖区"}}},{"id":2,"name":"商品B","price":199.0,"tags":["推荐"],"seller":{"name":"YY专营店","rating":4.5,"address":{"city":"上海","district":"浦东新区"}}}]}# 安全取值函数：避免KeyErrordefsafe_get(data,*keys,default=None):"""安全获取嵌套字典的值"""forkeyinkeys:ifisinstance(data,dict):data=data.get(key,default)else:returndefaultreturndata# 提取所有商品信息foriteminapi_data["results"]:name=safe_get(item,"name")price=safe_get(item,"price")seller=safe_get(item,"seller","name")city=safe_get(item,"seller","address","city")tags=safe_get(item,"tags",default=[])print(f"商品:{name}, 价格:{price}, 卖家:{seller}, 城市:{city}, 标签:{tags}")四：JSON配置管理
importjsonimportosfromdatetimeimportdatetimeclassJsonConfig:"""JSON配置文件管理器"""def__init__(self,config_path):self.path=config_path self.data={}self.load()defload(self):"""加载配置"""ifos.path.exists(self.path):withopen(self.path,"r",encoding="utf-8")asf:self.data=json.load(f)else:self.data={}returnself.datadefsave(self):"""保存配置"""os.makedirs(os.path.dirname(self.path),exist_ok=True)withopen(self.path,"w",encoding="utf-8")asf:json.dump(self.data,f,ensure_ascii=False,indent=2)defget(self,key,default=None):"""获取值（支持点分隔的路径：a.b.c）"""keys=key.split(".")value=self.dataforkinkeys:ifisinstance(value,dict):value=value.get(k,default)else:returndefaultreturnvalueifvalueisnotNoneelsedefaultdefset(self,key,value):"""设置值（支持点分隔的路径）"""keys=key.split(".")data=self.dataforkinkeys[:-1]:ifknotindata:data[k]={}data=data[k]data[keys[-1]]=value self.save()defupdate(self,dict_data):"""批量更新"""self.data.update(dict_data)self.save()# 使用config=JsonConfig(r"D:\config\flow_config.json")# 读取（支持点路径）url=config.get("api.base_url","https://default.com")timeout=config.get("api.timeout",30)# 写入config.set("api.base_url","https://api.example.com")config.set("api.timeout",15)config.set("last_run",datetime.now().strftime("%Y-%m-%d %H:%M"))完整流程：API数据解析保存
importjsonimportosfromdatetimeimportdatetime# yd_input: api_response (JSON字符串), output_dirapi_response=yd_input.get("api_response","{}")output_dir=yd_input.get("output_dir",r"D:\数据\API结果")# 解析API响应try:data=json.loads(api_response)exceptjson.JSONDecodeErrorase:yd_output={"status":"error","message":f"JSON解析失败:{e}"}else:# 提取数据results=[]ifisinstance(data,dict):# 单个对象if"data"indata:items=data["data"]ifisinstance(items,list):foriteminitems:results.append({"id":item.get("id"),"name":item.get("name",""),"price":item.get("price",0),"stock":item.get("stock",0),"status":item.get("status","unknown")})elifisinstance(items,dict):results.append({"id":items.get("id"),"name":items.get("name",""),"price":items.get("price",0),})elifisinstance(data,list):# 直接是数组foritemindata:ifisinstance(item,dict):results.append({"id":item.get("id"),"name":item.get("name",""),"price":item.get("price",0),})# 保存结果os.makedirs(output_dir,exist_ok=True)timestamp=datetime.now().strftime("%Y%m%d_%H%M%S")output_path=os.path.join(output_dir,f"api_result_{timestamp}.json")withopen(output_path,"w",encoding="utf-8")asf:json.dump({"fetch_time":timestamp,"count":len(results),"data":results},f,ensure_ascii=False,indent=2)yd_output={"status":"ok","count":len(results),"output":output_path}有什么坑
坑一：JSON中的中文被转义成\uXXXX
现象：写入JSON文件后，中文变成了\u5f20\u4e09这种形式。
原因：json.dumps默认ensure_ascii=True，会把非ASCII字符转义。
解决：设置ensure_ascii=False：
importjson data={"name":"张三","city":"北京"}# 正确：中文正常显示json_str=json.dumps(data,ensure_ascii=False)print(json_str)# {"name": "张三", "city": "北京"}# 错误：中文被转义# json_str = json.dumps(data) ❌# {"name": "\u5f20\u4e09", "city": "\u5317\u4eac"}# 写入文件时也要加withopen("data.json","w",encoding="utf-8")asf:json.dump(data,f,ensure_ascii=False,indent=2)坑二：JSON解析失败（格式错误）
现象：json.loads报JSONDecodeError: Expecting property name enclosed in double quotes。
原因：JSON规范要求键名和字符串值必须用双引号，不能用单引号。Python字典可以用单引号，但JSON不行。
解决：
TEMU店群如何管理运营？
importjson# 正确的JSON：双引号data=json.loads('{"name": "张三", "age": 28}')# 错误：单引号不是合法JSON# data = json.loads("{'name': '张三', 'age': 28}") ❌# 如果数据源用了单引号（比如Python repr输出），需要先替换bad_json="{'name': '张三', 'age': 28}"fixed_json=bad_json.replace("'",'"')try:data=json.loads(fixed_json)except:# 更安全的方式：用ast.literal_eval解析Python字面量importast data=ast.literal_eval(bad_json)坑三：JSON中数值精度丢失
现象：API返回的金额是99.90，解析后变成了99.9。或者大整数丢失精度。
原因：JSON的数字类型在Python中被解析为float或int，float会有精度问题。
解决：金额用字符串传输，或用Decimal：
importjsonfromdecimalimportDecimal# 方案1：金额作为字符串data={"price":"99.90"}# JSON中是字符串price_str=json.loads('{"price": "99.90"}')["price"]print(price_str)# "99.90" 保持原样# 方案2：用parse_float参数解析为Decimaldata=json.loads('{"price": 99.90}',parse_float=Decimal)print(data["price"])# Decimal('99.90')# 方案3：大整数用parse_int# JavaScript的JSON.parse对超大整数会丢精度# Python的json模块默认不会，但序列化为字符串时要注意big_num=123456789012345678901234567890data={"id":big_num}# json.dumps会正确序列化大整数json_str=json.dumps(data)# {"id": 123456789012345678901234567890}坑四：嵌套JSON取值时KeyError
现象：取data["user"]["name"]时报KeyError: "user"，因为"user"键不存在。
解决：用安全取值：
# 方案1：逐层getname=data.get("user",{}).get("name","未知")# 方案2：try-excepttry:name=data["user"]["name"]except(KeyError,TypeError):name="未知"# 方案3：封装安全取值函数（见前面的 safe_get）defsafe_get(data,*keys,default=None):forkeyinkeys:ifisinstance(data,dict):data=data.get(key,default)elifisinstance(data,list)andisinstance(key,int):if0<=key<len(data):data=data[key]else:returndefaultelse:returndefaultreturndata# 使用name=safe_get(data,"user","name",default="未知")city=safe_get(data,"user","address","city",default="未知")first_tag=safe_get(data,"tags",0,default="")