将 CSV 中嵌套 JSON 字符串字段自动展开为扁平化 JSON 对象

本文介绍如何用 Python 将含嵌套 jsON 字符串的类 csv 数据（如 _data 列存储转义 json）解析并扁平化为标准 JSON 结构，重点解决双引号转义、字段提取与类型转换问题，无需第三方库即可稳健实现。

本文介绍如何用 python 将含嵌套 json 字符串的类 csv 数据（如 `_data` 列存储转义 json）解析并扁平化为标准 json 结构，重点解决双引号转义、字段提取与类型转换问题，无需第三方库即可稳健实现。

在实际数据管道中，常遇到“伪 CSV”格式：表面是 CSV 表头与逗号分隔，但某列（如 _data）实际存储的是经过双引号转义的 JSON 字符串（例如 “{“”property_id””:””xx””}”）。这类数据无法直接用 csv.DictReader 或 pandas.read_csv 正确解析——因为标准 CSV 解析器会将整个 {…} 视为普通字符串，而不会递归解析其内部结构。

核心挑战在于三点：

• CSV 行中 JSON 字段被包裹在双引号内，且内部双引号被转义为 “”（符合 RFC 4180 的 CSV 双引号转义规则）；
• 需将 _data 列的 JSON 字符串反序列化，并将其键值“提升”至顶层对象；
• 同时保留原始 CSV 的其他字段（如 row_id, _emitted_at），并做必要类型转换（如 _emitted_at 转为整数）。

以下是一个轻量、健壮、不依赖第三方库（仅用标准库 re 和 json）的解决方案：

import re import json  # 匹配模式：捕获 row_id（非逗号前缀）、_emitted_at（逗号后至下一个双引号前）、_data（双引号包裹的 JSON 字符串） PATTERN = re.compile(r'^([^,]*?),([^,]*?)"(.*)"$')  def parse_csv_line(line: str) -> dict | None:     line = line.strip()     if not line or line.startswith('"row_id"'):  # 跳过表头行         return None      matched = PATTERN.match(line)     if not matched:         raise ValueError(f"Unparseable line: {line}")      row_id, emitted_at, data_json_str = matched.groups()      # 修复 CSV 转义：将 "" 替换为 "，使字符串可被 json.loads 解析     data_json_str = data_json_str.replace('""', '"')      try:         nested_data = json.loads(data_json_str)     except json.JSONDecodeError as e:         raise ValueError(f"Invalid JSON in _data field: {data_json_str!r}") from e      # 合并字段：将原始 CSV 字段与解析后的 JSON 字段合并     result = {         'row_id': row_id.strip('"'),  # 去除可能的外层引号（如 "rid_1" → rid_1）         '_emitted': int(emitted_at.strip()),         **nested_data  # 展开嵌套对象，实现扁平化     }     return result  # 示例输入（注意：真实场景中请从文件读取） raw_input = '''"row_id",_emitted_at,_data  rid_1,124,"{""property_id"":""xx"",""date"":""20220221"",""active1DayUsers"":3}" rid_2,124,"{""property_id"":""xx"",""date"":""20220222"",""active1DayUsers"":2}"'''  lines = [line for line in raw_input.splitlines() if line.strip()] # 跳过首行表头 parsed_records = [parse_csv_line(line) for line in lines[1:] if line.strip()]  print(json.dumps(parsed_records, indent=2, ensure_ascii=False))

✅ 输出结果完全符合预期：

[   {     "row_id": "rid_1",     "_emitted": 124,     "property_id": "xx",     "date": "20220221",     "active1DayUsers": 3   },   {     "row_id": "rid_2",     "_emitted": 124,     "property_id": "xx",     "date": "20220222",     "active1DayUsers": 2   } ]

? 关键注意事项：

• 不要使用 csv.reader 直接解析：因 _data 列含逗号和嵌套引号，标准 CSV 解析器会错误切分行；正则匹配更可控。
• replace(‘””‘, ‘”‘) 不可省略：这是处理 excel/CSV 导出中 JSON 字段转义的核心步骤，否则 json.loads 必然失败。
• 字段名一致性：示例中将 _emitted_at 重命名为 _emitted 以简化键名，你可根据需求保留原名或统一命名规范。
• 生产环境增强建议：
  • 使用 with open(…) 逐行读取大文件，避免内存溢出；
  • 添加日志与异常上下文（如出错行号），便于调试；
  • 若 _data 可能为空或缺失，应在 json.loads 前加空值判断；
  • 如需支持更多嵌套层级（如 _data 内还有 JSON 字符串字段），可递归调用 parse_csv_line 或改用 jsonpath-ng 等库。

该方法平衡了简洁性与鲁棒性，适用于 etl 预处理、日志解析及数据清洗等典型场景，是处理“CSV+内嵌JSON”混合格式的推荐实践。