如何用Python将表结构CSV转换为指定层级JSON？

需要把包含Oracle表结构的CSV文件转换为指定的层级JSON格式，现有CSV数据如下：

TABLE_NAME,TIPO_CARGA,COLUMN_NAME,DATA_TYPE
CUENTAS,Full load,PERFIL,NVARCHAR2
CUENTAS,Full load,LLAMADA_INI,DATE
CUENTAS,Full load,LLAMADA_END,DATE
CUENTAS_PLAN,Delta,REGISTRO,NVARCHAR2
CUENTAS_PLAN,Delta,ULTIMA_VISITA,DATE
CUENTAS_PLAN,Delta,FECHA,DATE
CUENTAS_TACTICAS,Delta,CIUDAD,NVARCHAR2
CUENTAS_TACTICAS,Delta,SALIDA,NVARCHAR2
CUENTAS_TACTICAS,Delta,COMPLETADO,DATE
CUENTAS_MIEMBROS,Full load,NEGOCIO,NVARCHAR2
CUENTAS_MIEMBROS,Full load,CORE,NVARCHAR2
CUENTAS_MIEMBROS,Full load,DESC,NVARCHAR2

原使用pandas和json编写的代码：

import pandas as pd
import json

df = pd.read_csv("list_tablas_short.csv")

print(df.head())
thisisjson = df.to_json(orient='records')

print('CSV to JSON:\n', thisisjson)

thisisjson_dict = json.loads(thisisjson)
with open('data.json', 'w') as json_file:
json.dump(thisisjson_dict, json_file)

当前生成的JSON不符合预期，输出示例：

[
  {
    "TABLE_NAME;\"COLUMN_NAME\";\"DATA_TYPE\"": "CUENTAS;PERFIL;NVARCHAR2"
  },
  {
    "TABLE_NAME;\"COLUMN_NAME\";\"DATA_TYPE\"": "CUENTAS;LLAMADA_INI;DATE"
  },
  {...}
]

期望生成的层级JSON格式：

{
  "tablas": [
    {
      "nombre": "CUENTAS",
      "load_type": "Full load",
      "fields": [
        {
          "campo": "PERFIL",
          "tipo": "NVARCHAR2"
        },
        {
          "campo": "LLAMADA_INI",
          "tipo": "DATE"
        },
        {
          "campo": "LLAMADA_END",
          "tipo": "DATE"
        }
      ]
    },
    {
      "nombre": "CUENTAS_PLAN",
      "load_type": "Delta",
      "fields": [
        {
          "campo": "REGISTRO",
          "tipo": "NVARCHAR2"
        },
        {
          "campo": "ULTIMA_VISITA",
          "tipo": "DATE"
        },
        {
          "campo": "FECHA",
          "tipo": "DATE"
        }
      ]
    }
  ]
}

修正代码实现 #

import pandas as pd
import json

# 读取CSV文件，明确指定分隔符避免解析错误
df = pd.read_csv("list_tablas_short.csv", sep=",")

# 按表名和加载类型分组，整理字段为目标格式
grouped_data = df.groupby(['TABLE_NAME', 'TIPO_CARGA']).apply(
    lambda group: group[['COLUMN_NAME', 'DATA_TYPE']].rename(
        columns={'COLUMN_NAME': 'campo', 'DATA_TYPE': 'tipo'}
    ).to_dict('records')
).reset_index()

# 构建最终层级结构
final_result = {"tablas": []}
for _, row in grouped_data.iterrows():
    final_result["tablas"].append({
        "nombre": row['TABLE_NAME'],
        "load_type": row['TIPO_CARGA'],
        "fields": row[0]
    })

# 写入格式化的JSON文件
with open('data.json', 'w', encoding='utf-8') as json_file:
    json.dump(final_result, json_file, ensure_ascii=False, indent=2)

关键说明 #

1. 修复CSV解析问题：原代码输出异常是因为CSV列解析错误，明确指定sep=","确保列正确拆分。
2. 分组整理字段：通过groupby按表名和加载类型聚合，将每组的列信息转换为campo和tipo格式的字典列表。
3. 构建层级结构：遍历分组结果，组装成tablas数组包含的表对象，完全匹配预期格式。
4. 优化JSON输出：用ensure_ascii=False保留特殊字符，indent=2生成可读性更强的格式化JSON。

内容的提问来源于stack exchange，提问作者Julio