在 Python 中使用数组将 CSV 转换为嵌套的 json 对象答案

【问题标题】：Converting CSV to nested json objects with arrays in Python在 Python 中使用数组将 CSV 转换为嵌套的 json 对象
【发布时间】：2022-08-19 02:00:38
【问题描述】：

我正在尝试使用嵌套对象和数组将 csv 转换为 json。我正在尝试使其动态化，以便如果我向 csv 添加字段，它会在不修改转换器的情况下更新 json

输入.csv

id,input.a,input.b.x.input.b.y,input.c
1,2,a,b,\"3,4\"
1,2,c,d,\"3,4\"
2,3,p,q,\"5\"

输出.json

{
  \"1\":{
    \"a\": 2,
    \"b\": [
      {
        \"x\":a
        \"y\":b
      },
      {
        \"x\":c
        \"y\":d
      }
    ],
    \"c\" : [3,4]
  },
  
  \"2\":{
    \"a\": 3,
    \"b\": [
      {
        \"x\":p
        \"y\":q
      }
    ],
    \"c\" : [5]
  }
}

添加一些代码来创建一个最小的可重现示例将帮助您得到这个问题的答案。 stackoverflow.com/help/minimal-reproducible-example
看起来您可能需要编写自己的代码来执行此转换。对于\'id\'，您可能希望使用字典之类的数据结构。
如果给定id 的c 值不同，会发生什么情况？这些值是附加到数组还是嵌套数组？
输入非常严格，不会有不同的值
input.c 将始终是一个整数（或数字）数组，而 input.b.* 将始终是字符串？

标签： python-3.x pandas csv

【解决方案1】：

由于所有的打字和 TypedDicts，这个解决方案看起来更大，如果你真的想要，你可以删除它们。

当我对您的示例 input.csv 运行它时，我得到了您的示例 output.json：

#!/usr/bin/env python3
import csv
import json

from collections.abc import Iterator
from typing import TypedDict

# Build up to the final structure, JSON_Data
class B_Dict(TypedDict):
    x: str
    y: str


class ID_Dict(TypedDict):
    a: int
    b: list[B_Dict]
    c: list[int]


JSON_Data = dict[int, ID_Dict]

CSV_Row = list[str]
ID_idx = 0
A_idx = 1
Bx_idx = 2
By_idx = 3
C_idx = 4


def main():
    with open("input.csv", newline="") as f_in, open("output.json", "w") as f_out:
        reader = csv.reader(f_in)

        data = csv_to_json(reader)

        json.dump(data, f_out, indent=4)


def csv_to_json(csv_reader: Iterator[CSV_Row]) -> JSON_Data:
    header = next(csv_reader)
    assert header[ID_idx] == "id"
    assert header[A_idx] == "input.a"
    assert header[Bx_idx] == "input.b.x"
    assert header[By_idx] == "input.b.y"
    assert header[C_idx] == "input.c"

    data: JSON_Data = {}

    for row in csv_reader:
        id_ = int(row[ID_idx])
        a = int(row[A_idx])
        bx = row[Bx_idx]
        by = row[By_idx]
        c = [int(x) for x in row[C_idx].split(",")]

        # Since every row in the CSV is a self-contained ID_Dict, this
        # is the only logic we need: to create the ID_Dict once...
        if id_ not in data:
            data[id_] = ID_Dict(a=a, b=[B_Dict(x=bx, y=by)], c=c)
            continue

        # ...then update it as other rows with the same ID are encountered
        data[id_]["b"].append(B_Dict(x=bx, y=by))

    return data


if __name__ == "__main__":
    main()

如果您需要添加另一个键/字段，例如 b，这是一个列表，并且为每一行附加值，例如d:

| id | input.a | input.b.x | input.b.y | input.c | input.d.m | input.d.n |
|----|---------|-----------|-----------|---------|-----------|-----------|
| 1  | 2       | a         | b         | 3,4     | 10.0      | 11.1      |
| 1  | 2       | c         | d         | 3,4     | 12.2      | 13.3      |
| 2  | 3       | p         | q         | 5       | 98.8      | 99.9      |

为其添加一个 TypedDict，然后将其添加到 ID_Dict：

class D_Dict(TypedDict):
    m: float
    n: float

class ID_Dict(TypedDict):
    a: int
    b: list[B_Dict]
    c: list[int]
    d: list[D_Dict]  # ← add here

更新您的 CSV 标头索引和断言：

Dm_idx = 5
Dn_idx = 6

...

assert header[Dm_idx] == "input.d.m"
assert header[Dn_idx] == "input.d.n"

最后：

dm = float(row[Dm_idx])
dn = float(row[Dn_idx])

if id_ not in data:
    data[id_] = ID_Dict(
        a=a,
        b=[B_Dict(x=bx, y=by)],
        c=c,
        d=[D_Dict(m=dm, n=dn)],  # ← add here
    )
    continue

data[id_]["b"].append(B_Dict(x=bx, y=by))
data[id_]["d"].append(D_Dict(m=dm, n=dn))

【讨论】：