Python如何将CSV读取得到的字典转换为指定格式的JSON结构

下面针对你提到的从CSV读取的原始字典转换为目标嵌套结构的需求，提供3种不同复杂度的实现方式，可根据场景选择：

1. 原生Python硬编码实现 #

适合转换规则固定、一次性脚本场景，无额外依赖，实现最直接：

def convert_raw_to_target(raw: dict) -> list:
    source_id = str(raw["id"])
    return [
        {
            "metadata": {
                "type": "pole_model",
                "version": "0.1"
            },
            "party": [
                {
                    "source_system": "A",
                    "source_id": source_id,
                    "person": {
                        "surname": raw["surname"],
                        "given_names": raw["givennames"],
                        "date_of_birth": raw["dateofbirth"]
                    }
                }
            ],
            "location": [
                {
                    "source_system": "A",
                    "source_id": source_id,
                    "address": {
                        "line_1": raw["address"],
                        "postcode": raw["postcode"]
                    }
                },
                {
                    "contact": {
                        "source_system": "A",
                        "source_id": source_id,
                        "phone": {
                            "mobile": raw["mobile"]
                        }
                    }
                }
            ]
        }
    ]

# 调用示例
raw_dict = {
   "id":1,
   "surname":"Einstein",
   "givennames":"Albert",
   "dateofbirth":"27/08/2007",
   "address":"11 Willow Road,BOSTON MANOR",
   "postcode":"AXT 5JA",
   "mobile":"078 1453 6934"
}
target = convert_raw_to_target(raw_dict)

2. 映射配置化实现 #

适合转换规则可能频繁调整的场景，将字段映射关系和结构逻辑分离，无需修改转换核心代码就能调整规则：

# 字段映射规则：原始字段名 -> 目标结构路径（点分隔）
FIELD_MAPPING = {
    "surname": "party.person.surname",
    "givennames": "party.person.given_names",
    "dateofbirth": "party.person.date_of_birth",
    "address": "location.address.line_1",
    "postcode": "location.address.postcode",
    "mobile": "location.contact.phone.mobile"
}

# 固定公共参数
COMMON_PARAMS = {
    "source_system": "A",
    "metadata": {"type": "pole_model", "version": "0.1"}
}

def set_nested_value(obj: dict, path: str, value):
    keys = path.split(".")
    current = obj
    for key in keys[:-1]:
        current = current.setdefault(key, {})
    current[keys[-1]] = value

def convert_with_mapping(raw: dict) -> list:
    base_item = {
        "metadata": COMMON_PARAMS["metadata"],
        "party": [{"source_system": COMMON_PARAMS["source_system"], "source_id": str(raw["id"])}],
        "location": [
            {"source_system": COMMON_PARAMS["source_system"], "source_id": str(raw["id"])},
            {"contact": {"source_system": COMMON_PARAMS["source_system"], "source_id": str(raw["id"])}}
        ]
    }
    for raw_key, target_path in FIELD_MAPPING.items():
        set_nested_value(base_item, target_path, raw[raw_key])
    return [base_item]

3. 生产级序列化库实现（Pydantic） #

适合企业级业务项目，自带数据类型校验、异常捕获、自动类型转换功能，是目前Python生态最常用的数据转换方案：
首先安装依赖：pip install pydantic

from pydantic import BaseModel, Field
from typing import List

class Metadata(BaseModel):
    type: str = "pole_model"
    version: str = "0.1"

class Person(BaseModel):
    surname: str
    given_names: str = Field(alias="givennames")
    date_of_birth: str = Field(alias="dateofbirth")

class Party(BaseModel):
    source_system: str = "A"
    source_id: str = Field(alias="id")
    person: Person

class Address(BaseModel):
    line_1: str = Field(alias="address")
    postcode: str

class LocationAddressItem(BaseModel):
    source_system: str = "A"
    source_id: str = Field(alias="id")
    address: Address

class ContactPhone(BaseModel):
    mobile: str

class Contact(BaseModel):
    source_system: str = "A"
    source_id: str = Field(alias="id")
    phone: ContactPhone

class LocationContactItem(BaseModel):
    contact: Contact

class RootItem(BaseModel):
    metadata: Metadata = Metadata()
    party: List[Party]
    location: List[LocationAddressItem | LocationContactItem]

    @classmethod
    def from_raw(cls, raw: dict) -> "RootItem":
        raw_with_str_id = raw.copy()
        raw_with_str_id["id"] = str(raw["id"])
        return cls(
            party=[Party(**raw_with_str_id)],
            location=[
                LocationAddressItem(**raw_with_str_id),
                LocationContactItem(contact=Contact(**raw_with_str_id, phone=ContactPhone(**raw_with_str_id)))
            ]
        )

# 调用示例
target_obj = RootItem.from_raw(raw_dict)
# 转字典用 target_obj.model_dump()

选型推荐 #

• 临时脚本、规则固定：选原生硬编码，无依赖、开发最快
• 规则迭代频繁、同结构多数据源转换：选配置化映射方案，可维护性更高
• 生产环境、需要数据校验：选Pydantic/marshmallow等成熟序列化库，稳定性更高

内容的提问来源于stack exchange，提问作者michal-ko