影刀RPA JSON数据的深度处理:解析、生成、转换
影刀RPA JSON数据的深度处理解析、生成、转换作者林焱做RPA几乎每天都和JSON打交道——API返回的是JSON、配置文件用JSON、网页数据接口也是JSON。但很多人处理JSON时踩坑嵌套层级太深取不到值、中文变成unicode编码、写入文件后格式混乱、JSON和字典搞混。这篇文章把JSON在影刀RPA里的常见操作讲透。一、什么情况用什么场景推荐方法说明解析API返回的JSONjson.loads()字符串转字典把数据保存为JSON文件json.dump()写入文件读取JSON配置文件json.load()从文件读取把字典转为JSON字符串json.dumps()字典转字符串深层嵌套JSON取值逐层访问或递归防止KeyErrorJSON和Excel互转pandas中转数据格式转换JSON数据清洗遍历过滤去除空值/无效数据二、怎么做2.1 JSON基础操作importjson# JSON字符串 → Python字典json_str{name: 张三, age: 25, city: 北京}datajson.loads(json_str)print(data[name])# 张三print(type(data))# class dict# Python字典 → JSON字符串data{name:李四,age:30,city:上海}json_strjson.dumps(data,ensure_asciiFalse)# ensure_asciiFalse保留中文print(json_str)# {name: 李四, age: 30, city: 上海}[video(video-ZNnY73eg-1783576687085)(type-csdn)(url-https://live.csdn.net/v/embed/525010)(image-https://v-blog.csdnimg.cn/asset/f4faa587144cb7070f19e8b36813806b/cover/Cover0.jpg)(title-店群矩阵自动化突破运营极限)]# 写入JSON文件data{products:[{name:商品A,price:99},{name:商品B,price:199}]}withopen(rD:\data\products.json,w,encodingutf-8)asf:json.dump(data,f,ensure_asciiFalse,indent2)# indent2 美化格式# 读取JSON文件withopen(rD:\data\products.json,r,encodingutf-8)asf:datajson.load(f)print(data)2.2 处理嵌套JSONAPI返回的JSON经常嵌套很深importjson# 典型的API返回结构api_response { code: 200, message: success, data: { total: 50, page: 1, items: [ { id: 1, name: 商品A, price: 99.5, stock: 100, tags: [热销, 推荐], seller: { name: 旗舰店, rating: 4.8 } }, { id: 2, name: 商品B, price: 199.0, stock: 0, tags: [], seller: { name: 专营店, rating: 4.5 } } ] } } datajson.loads(api_response)# 逐层访问codedata[code]itemsdata[data][items]totaldata[data][total]print(f状态码:{code})print(f总数:{total})print(f商品数:{len(items)})# 遍历商品列表foriteminitems:nameitem[name]priceitem[price]stockitem[stock]tagsitem[tags]seller_nameitem[seller][name]seller_ratingitem[seller][rating]status在售ifstock0else缺货tag_str/.join(tags)iftagselse无标签print(f{name}| ¥{price}|{status}| 标签:{tag_str}| 店铺:{seller_name}({seller_rating}))2.3 安全取值防止KeyError嵌套JSON取值时如果某个key不存在会报KeyErrorimportjson data{user:{name:张三,address:{city:北京# street 不存在}}}# 危险写法直接取值key不存在就报错# street data[user][address][street] # KeyError!# 安全写法一用get()方法不存在返回Nonestreetdata.get(user,{}).get(address,{}).get(street,未知)print(f街道:{street})# 未知# 安全写法二用get()指定默认值citydata.get(user,{}).get(address,{}).get(city,未知)print(f城市:{city})# 北京# 安全写法三try-excepttry:streetdata[user][address][street]except(KeyError,TypeError):street未知print(f街道:{street})封装一个安全取值函数defsafe_get(data,*keys,defaultNone): 安全地从嵌套字典中取值 用法: safe_get(data, user, address, city, default未知) currentdataforkeyinkeys:ifisinstance(current,dict):currentcurrent.get(key)elifisinstance(current,list)andisinstance(key,int):if0keylen(current):currentcurrent[key]else:returndefaultelse:returndefaultifcurrentisNone:returndefaultreturncurrent# 使用示例data{user:{address:{city:北京}}}citysafe_get(data,user,address,city,default未知)streetsafe_get(data,user,address,street,default未知)print(f城市:{city})# 北京print(f街道:{street})# 未知2.4 JSON和Excel互转importjsonimportopenpyxldefjson_to_excel(json_data,excel_path,sheet_name数据):将JSON列表数据转为Excelifnotjson_dataornotisinstance(json_data,list):print(JSON数据不是列表格式)returnFalse# 获取所有可能的列名合并所有item的keyall_keysset()foriteminjson_data:ifisinstance(item,dict):all_keys.update(item.keys())headerssorted(all_keys)# 写入Excelwbopenpyxl.Workbook()wswb.active ws.titlesheet_name# 写表头forcol_idx,headerinenumerate(headers,1):ws.cell(row1,columncol_idx,valueheader)ws.cell(row1,columncol_idx).fontopenpyxl.styles.Font(boldTrue)# 写数据forrow_idx,iteminenumerate(json_data,2):forcol_idx,headerinenumerate(headers,1):valueitem.get(header,)# 处理嵌套的dict/listifisinstance(value,(dict,list)):valuejson.dumps(value,ensure_asciiFalse)ws.cell(rowrow_idx,columncol_idx,valuevalue)wb.save(excel_path)print(fJSON → Excel:{len(json_data)}行 →{excel_path})returnTruedefexcel_to_json(excel_path,json_path,sheet_nameNone):将Excel转为JSON列表wbopenpyxl.load_workbook(excel_path,read_onlyTrue)wswb[sheet_name]ifsheet_nameelsewb.active# 读取表头headers[]forcellinnext(ws.iter_rows(min_row1,max_row1)):headers.append(str(cell.value).strip()ifcell.valueelse)# 读取数据data[]forrowinws.iter_rows(min_row2,values_onlyTrue):item{}foridx,valueinenumerate(row):ifidxlen(headers)andheaders[idx]:# 尝试解析JSON类型的单元格ifisinstance(value,str)andvalue.startswith({):try:valuejson.loads(value)except:passitem[headers[idx]]valueifany(visnotNoneforvinitem.values()):data.append(item)wb.close()# 写入JSON文件withopen(json_path,w,encodingutf-8)asf:json.dump(data,f,ensure_asciiFalse,indent2)print(fExcel → JSON:{len(data)}行 →{json_path})returnTrue# 使用示例# json_data [# {name: 商品A, price: 99, stock: 100},# {name: 商品B, price: 199, stock: 50},# ]# json_to_excel(json_data, rD:\data\products.xlsx)# excel_to_json(rD:\data\products.xlsx, rD:\data\products.json)2.5 JSON数据清洗从API获取的JSON数据经常需要清洗importjsondefclean_json_data(data):清洗JSON数据ifisinstance(data,dict):cleaned{}forkey,valueindata.items():# 跳过空值ifvalueisNoneorvalue:continue# 递归清洗cleaned[key]clean_json_data(value)returncleanedelifisinstance(data,list):cleaned[]foritemindata:cleaned_itemclean_json_data(item)ifcleaned_item:# 跳过空结果cleaned.append(cleaned_item)returncleanedelse:# 基础类型直接返回ifisinstance(data,str):returndata.strip()returndata# 测试raw_data[{name: 商品A ,price:99,desc:,tags:[热销,None,推荐]},{name:商品B,price:None,desc: 描述 ,tags:[]},{name:,price:0,desc:None,tags:None},]cleanedclean_json_data(raw_data)print(json.dumps(cleaned,ensure_asciiFalse,indent2))2.6 JSON配置文件管理importjsonimportosfromdatetimeimportdatetimeclassJsonConfig:JSON配置文件管理器def__init__(self,config_path):self.config_pathconfig_path self.config{}self.load()defload(self):加载配置ifos.path.exists(self.config_path):withopen(self.config_path,r,encodingutf-8)asf:self.configjson.load(f)else:self.config{}defsave(self):保存配置os.makedirs(os.path.dirname(self.config_path),exist_okTrue)withopen(self.config_path,w,encodingutf-8)asf:json.dump(self.config,f,ensure_asciiFalse,indent2)defget(self,key,defaultNone):获取配置项returnself.config.get(key,default)defset(self,key,value):设置配置项self.config[key]value self.save()defupdate(self,data):批量更新self.config.update(data)self.save()defdelete(self,key):删除配置项ifkeyinself.config:delself.config[key]self.save()# 使用示例# config JsonConfig(rD:\config\rpa_config.json)# config.set(target_url, https://example.com)# config.set(max_retries, 3)# config.set(last_run, datetime.now().strftime(%Y-%m-%d %H:%M:%S))# print(config.get(target_url))三、有什么坑坑1中文变成\uXXXXimportjson data{name:张三}# 默认中文被转义print(json.dumps(data))# {name: \u5f20\u4e09}# 加ensure_asciiFalse保留中文print(json.dumps(data,ensure_asciiFalse))# {name: 张三}解决所有json.dumps和json.dump都加ensure_asciiFalse。坑2JSON里有单引号导致解析失败JSON标准要求字符串用双引号但Python字典用单引号。直接把字典str()当JSON用会报错temu店群自动化报活动案例importjson# 错误字典的str不是合法JSONdata_strstr({name:张三})# {name: 张三}# json.loads(data_str) # 报错单引号不合法# 正确用json.dumpsdata_strjson.dumps({name:张三},ensure_asciiFalse)json.loads(data_str)# 正常解析坑3JSON文件编码问题importjson# 错误不指定编码Windows默认GBK# with open(data.json, w) as f: # 可能乱码# json.dump(data, f)# 正确指定utf-8编码withopen(data.json,w,encodingutf-8)asf:json.dump(data,f,ensure_asciiFalse,indent2)# 读取也要指定编码withopen(data.json,r,encodingutf-8)asf:datajson.load(f)坑4JSON中的数字精度丢失importjson# 大整数精度丢失data{id:12345678901234567890}json_strjson.dumps(data)parsedjson.loads(json_str)print(parsed[id])# 12345678901234567000 ← 末尾精度丢失# 解决用字符串存储大数字data{id:12345678901234567890}坑5JSON文件被截断导致解析失败写入JSON文件时程序崩溃文件只写了一半下次读取时json.loads报错。解决先写临时文件写完再重命名importjsonimportosdefsafe_write_json(data,file_path):安全写入JSON文件temp_pathfile_path.tmp# 先写临时文件withopen(temp_path,w,encodingutf-8)asf:json.dump(data,f,ensure_asciiFalse,indent2)# 写完后重命名原子操作ifos.path.exists(file_path):os.remove(file_path)os.rename(temp_path,file_path)print(f安全写入:{file_path})defsafe_read_json(file_path,defaultNone):安全读取JSON文件try:withopen(file_path,r,encodingutf-8)asf:returnjson.load(f)except(json.JSONDecodeError,FileNotFoundError)ase:print(f读取失败:{e}返回默认值)returndefaultifdefaultisnotNoneelse{}总结JSON处理的核心要点永远加ensure_asciiFalse——否则中文变\uXXXX读写指定encoding‘utf-8’——防止编码问题嵌套取值用get()——防止KeyError大数字用字符串——防止精度丢失写入用临时文件模式——防止写入中断导致文件损坏