Python新手求助：Django中拆分多段表头CSV并同步至数据库

1. 先拆分CSV文件 #

你的CSV是用空白行分隔多个独立数据集的，row.get('title')肯定行不通——因为分段标题是单独的行，并非数据字段。可以用下面的函数把上传文件按空行拆成一个个小数据集（每个数据集包含自身表头和数据行）：

def split_csv(file):
    # 读取所有行，过滤空行和纯换行的无效行
    lines = [line.strip() for line in file if line.strip()]
    datasets = []
    current_group = []
    
    for line in lines:
        # 分段标题（比如About、Address）不含逗号，以此判断并跳过
        if ',' not in line:
            # 若当前已有攒好的数据集，先存入列表再重置
            if current_group:
                datasets.append(current_group)
                current_group = []
            continue
        current_group.append(line)
    
    # 把最后一组数据补充进列表
    if current_group:
        datasets.append(current_group)
    
    return datasets

这个函数返回的列表中，每个元素就是一个分段的完整数据（表头+所有数据行）。

2. 绑定数据集与Django模型 #

提前定义好每个分段的表头和Django模型字段的映射关系，示例如下：

# 假设你的Django模型是AboutModel和AddressModel，剩余7个分段按此格式补充
MODEL_MAPPINGS = {
    # 用表头首字段+字段总数匹配不同分段（也可直接用完整表头字符串匹配）
    ('no', 4): {
        'model': AboutModel,
        'field_map': {'no': 'id', 'name': 'name', 'shape': 'shape', 'type': 'material'}
    },
    ('no', 5): {
        'model': AddressModel,
        'field_map': {'no': 'id', 'street': 'street', 'postcode': 'postcode', 'city': 'city', 'County': 'county'}
    },
    # 剩余7个分段的映射规则按上述格式添加
}

3. 批量处理新增/更新 #

用Django的update_or_create处理每条数据，同时用事务保证数据一致性（出错时全部回滚）：

import csv
from django.db import transaction

def import_datasets(datasets):
    with transaction.atomic():
        for dataset in datasets:
            # 用csv.DictReader解析当前数据集，自动以第一行为表头
            reader = csv.DictReader(dataset)
            header = reader.fieldnames
            # 匹配对应的模型配置
            match_key = (header[0], len(header))
            config = MODEL_MAPPINGS.get(match_key)
            if not config:
                # 跳过无法匹配的分段，也可添加日志记录异常
                continue
            
            target_model = config['model']
            field_map = config['field_map']
            
            for row in reader:
                # 转换CSV字段名为模型字段名，同时清理数据空格
                clean_data = {field_map[key]: value.strip() for key, value in row.items() if key in field_map}
                # 用唯一标识（比如no对应模型的id）判断是新增还是更新
                lookup = {field_map['no']: clean_data[field_map['no']]}
                # 执行新增或更新操作
                target_model.objects.update_or_create(
                    defaults=clean_data,
                    **lookup
                )

4. 整合到上传视图 #

在Django上传视图中调用上述两个函数：

from django.http import HttpResponseRedirect
from django.views import View

class CSVUploadView(View):
    def post(self, request):
        csv_file = request.FILES.get('csv_file')
        if not csv_file:
            # 处理未上传文件的情况，比如跳转回上传页提示
            return HttpResponseRedirect('/upload/?error=no_file')
        
        # 拆分CSV
        datasets = split_csv(csv_file)
        # 导入数据
        import_datasets(datasets)
        
        return HttpResponseRedirect('/upload/?success=1')

实用提醒 #

• 数据校验：一定要加数据验证！比如检查no是否为整数、必填字段是否为空，避免脏数据入库。
• 日志记录：可以添加日志模块，记录哪些分段成功导入、哪些失败，方便排查问题。
• 大文件优化：如果CSV文件过大，不要一次性读取所有行，改成逐行迭代处理，避免内存溢出。

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