本文介绍如何用 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”混合格式的推荐实践。
