Python多源CSV数据处理模式技术咨询

核心思路：分层处理，各司其职 #

把整个流程拆成数据源适配层和数据验证层，让Pydantic专注做它擅长的类型校验、数据清洗与标准化，把CSV的异构格式、特殊结构处理放在适配层，既不浪费Pydantic的能力，也不会让模型逻辑变得臃肿混乱。

具体实现步骤 #

1. 定义与数据源无关的Pydantic核心模型 #

先写出最终需要的统一数据模型，这部分只负责字段类型、验证规则（比如金额范围、日期格式）、默认值处理，完全隔离数据源的特殊性。

from pydantic import BaseModel, Field, validator
from datetime import date

class UnifiedFinancialRecord(BaseModel):
    record_id: str
    trade_date: date
    amount: float = Field(gt=0, description="交易金额必须大于0")
    instrument_code: str
    counterparty: str | None = None

    @validator('trade_date')
    def standardize_date(cls, v):
        # 统一处理各种日期字符串格式
        if isinstance(v, str):
            # 兼容斜杠、横杠分隔的日期格式
            return date.fromisoformat(v.replace('/', '-'))
        return v

2. 为每个CSV数据源编写专属适配器 #

每个适配器负责处理对应CSV的表头映射、结构转换（单行/多行转统一结构），输出符合核心模型字段的字典，再传给Pydantic做验证。

• 单行CSV适配器：直接用你已有的表头映射字典，将Pandas行数据转成对应字段的字典

class SingleRowCSVAdapter:
    def __init__(self, header_mapping: dict):
        self.header_mapping = header_mapping

    def adapt(self, row: dict) -> dict:
        adapted = {}
        for csv_col, model_field in self.header_mapping.items():
            adapted[model_field] = row.get(csv_col)
        return adapted

• 多行CSV适配器：处理多行对应一条记录的场景，先按标识字段分组聚合，再提取对应字段

class MultiRowCSVAdapter:
    def __init__(self, group_key: str, field_mapping: dict):
        self.group_key = group_key
        self.field_mapping = field_mapping

    def adapt(self, group_df) -> dict:
        adapted = {self.group_key: group_df.iloc[0][self.group_key]}
        for csv_col, model_field in self.field_mapping.items():
            # 根据数据源规则处理，比如取非空首值、求和或拼接
            adapted[model_field] = group_df[csv_col].dropna().iloc[0]
        return adapted

3. 串联流程：加载→适配→验证 #

用Pandas加载CSV后，根据数据源类型选择对应适配器，将处理后的字典传入Pydantic模型，此时Pydantic会自动完成类型转换、规则校验，并抛出明确的错误信息。

import pandas as pd

# 示例：处理单行结构CSV
single_row_mapping = {"交易ID": "record_id", "成交日期": "trade_date", "交易金额": "amount"}
adapter = SingleRowCSVAdapter(single_row_mapping)
df = pd.read_csv("single_row_finance_data.csv")

for _, row in df.iterrows():
    adapted_data = adapter.adapt(row.to_dict())
    try:
        record = UnifiedFinancialRecord(**adapted_data)
        # 处理验证通过的标准化数据
        print(record.dict())
    except Exception as e:
        # 记录错误日志或标记异常数据
        print(f"数据验证失败: {e}")

# 示例：处理多行结构CSV
multi_row_mapping = {"成交日期": "trade_date", "累计金额": "amount"}
adapter = MultiRowCSVAdapter(group_key="record_id", field_mapping=multi_row_mapping)
df = pd.read_csv("multi_row_finance_data.csv")

for group_id, group_df in df.groupby("record_id"):
    adapted_data = adapter.adapt(group_df)
    try:
        record = UnifiedFinancialRecord(**adapted_data)
        print(record.dict())
    except Exception as e:
        print(f"分组[{group_id}]验证失败: {e}")

4. 优化：适配器标准化（可选） #

可以定义适配器基类，强制所有适配器实现adapt方法，再写一个工厂函数根据数据源类型返回对应适配器，让代码扩展性更强。

方案优势 #

• Pydantic完全聚焦于数据最终形态的验证与标准化，充分发挥其类型检查、自动转换、错误提示的能力
• 适配器层单独处理每个数据源的特殊规则，逻辑清晰，新增数据源只需添加新适配器，无需修改核心模型
• 错误定位精准：能快速区分是数据源适配出错，还是数据本身不符合业务规则

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