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

作者：林焱

---

什么情况用

调用API、读取配置文件、解析网页接口返回——现代数据交换几乎全是JSON格式。如果你只会影刀的内置文本处理节点，遇到嵌套JSON基本束手无策。

典型场景：

• 调用企业微信API，返回的部门树是嵌套JSON，需要展平成列表
• 爬取接口返回了100条数据，需要按某个字段分组统计
• 从配置JSON里读取参数，需要校验字段是否存在
• 

核心场景：在影刀的Python节点中，高效解析和加工JSON数据。

---

怎么做

第一步：JSON操作速查——5个最常用的函数

import json

# 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. 从文件读取JSON
with open("config.json", "r", encoding="utf-8") as f:
    config = json.load(f)

# 5. 写入JSON文件
with open("output.json", "w", encoding="utf-8") as f:
    json.dump(data, f, ensure_ascii=False, indent=2)

第二步：安全解析——JSON可能不标准

def safe_json_parse(text):
    """
    安全解析JSON，处理各种不标准的情况
    """
    # 尝试1：直接解析
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        pass
    
    # 尝试2：去掉BOM头
    if text.startswith('\ufeff'):
        try:
            return json.loads(text[1:])
        except:
            pass
    
    # 尝试3：JSON可能被包在markdown代码块里
    if '```' in text:
        import re
        match = re.search(r'```(?:json)?\s*\n?(.*?)```', text, re.DOTALL)
        if match:
            try:
                return json.loads(match.group(1))
            except:
                pass
    
    # 尝试4：可能是单引号的伪JSON（Python dict的repr）
    try:
        import ast
        return ast.literal_eval(text)
    except:
        pass
    
    return None

第三步：嵌套JSON展平——最常用的加工

def flatten_json(nested_obj, parent_key="", sep="_"):
    """
    将嵌套JSON展平为一级字典
    例：{"user": {"name": "张三", "addr": {"city": "北京"}}}
     →  {"user_name": "张三", "user_addr_city": "北京"}
    """
    items = {}
    
    if isinstance(nested_obj, dict):
        for k, v in nested_obj.items():
            new_key = f"{parent_key}{sep}{k}" if parent_key else k
            if isinstance(v, (dict, list)):
                items.update(flatten_json(v, new_key, sep))
            else:
                items[new_key] = v
    
    elif isinstance(nested_obj, list):
        for i, v in enumerate(nested_obj):
            new_key = f"{parent_key}{sep}{i}"
            if isinstance(v, (dict, list)):
                items.update(flatten_json(v, new_key, sep))
            else:
                items[new_key] = v
    
    return items

# 测试
nested = {
    "order_id": "ORD001",
    "customer": {
        "name": "张三",
        "contact": {
            "phone": "13812345678",
            "email": "zhangsan@example.com"
        }
    },
    "items": [
        {"product": "笔记本", "price": 5999},
        {"product": "鼠标", "price": 199}
    ]
}

flat = flatten_json(nested)
for k, v in flat.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）。

import jmespath

# 示例数据：企业微信部门树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)
![在这里插入图片描述](https://i-blog.csdnimg.cn/direct/544147316d5c4bd7b8963c0cf30f376e.png#pic_center)

print(names)  # ['总公司']

# 递归提取所有层级的部门名称
all_names = jmespath.search("department[*].children[*].name", data)
print(all_names)  # ['技术部', '销售部']

# 更优雅的递归写法
# jmespath没有直接递归，但可以组合查询

# 不用jmespath的原生写法也能搞，就是啰嗦点：
def extract_all_departments(nodes, result=None):
    """递归提取所有部门"""
    if result is None:
        result = []
    for node in nodes:
        result.append({"id": node["id"], "name": node["name"]})
        if "children" in node:
            extract_all_departments(node["children"], result)
    return result

all_depts = extract_all_departments(data["department"])
print(f"共 {len(all_depts)} 个部门")
for d in all_depts:
    print(f"  {d['id']}: {d['name']}")

第五步：JSON到DataFrame——数据分析的前置步骤

import pandas as pd

def json_to_dataframe(json_data, record_path=None):
    """
    将JSON数据转为DataFrame
    自动处理嵌套结构
    """
    if isinstance(json_data, str):
        json_data = json.loads(json_data)
    
    # 如果是列表
    if isinstance(json_data, list):
        # 先展平
        flat_list = [flatten_json(item) for item in json_data]
        return pd.DataFrame(flat_list)
    
    # 如果是字典，提取record_path指定的列表
    if isinstance(json_data, dict):
        if record_path and record_path in json_data:
            records = json_data[record_path]
            flat_list = [flatten_json(item) for item in records]
            return pd.DataFrame(flat_list)
        else:
            return pd.DataFrame([flatten_json(json_data)])
    
    return pd.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() 会直接把内存撑爆。

import ijson  # pip install ijson

def stream_large_json(file_path, item_path="item"):
    """
    流式读取大JSON，逐个处理
    适用于 {"items": [巨大的数组]} 这种结构
    """
    count = 0
    with open(file_path, "r", encoding="utf-8") as f:
        # 逐个读取数组中的元素
        ![在这里插入图片描述](https://i-blog.csdnimg.cn/direct/0bb94ea2f49142c1bc5912d292dc4f73.png#pic_center)

        for item in ijson.items(f, f"{item_path}"):
            process_one(item)  # 处理单个元素
            count += 1
            if count % 1000 == 0:
                print(f"已处理 {count} 条...")
    
    print(f"共处理 {count} 条")
    return count

---

有什么坑

坑1：NaN/Infinity 不是合法JSON

pandas的DataFrame转JSON时，如果有NaN，json.dumps() 会报错。

import math
import json

# ❌ 这会报错
# data = {"value": float('nan')}
# json.dumps(data)  # ValueError: Out of range float values are not JSON compliant

# ✅ 处理NaN
def json_safe(obj):
    if isinstance(obj, float):
    
[video(video-zqOhJoM2-1783576560041)(type-csdn)(url-https://live.csdn.net/v/embed/524993)(image-https://v-
![在这里插入图片描述](https://i-blog.csdnimg.cn/direct/49a417d516ab4294894e39e5d9c8b606.png#pic_center)
blog.csdnimg.cn/asset/a547123d88ad712dccba346c9217e237/cover/Cover0.jpg)(title-TEMU店群如何管理运营？)]

        if math.isnan(obj):
            return None
        if math.isinf(obj):
            return None
    return obj

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数据处理场景都能搞定。