如何在Django中通过模板结合runscript函数实现CSV上传至数据库

一、重构原有导入函数 #

原函数固定读取本地文件，先改造成支持接收上传文件对象的通用函数，同时优化冗余查询和异常处理：

import pandas as pd
from django.db.models import ObjectDoesNotExist
from .models import Employee, SubBudgetOwner, Subsidiary, Department, Account, SubBudget, Combination

def process_csv(file_obj):
    # 从上传的文件对象读取CSV
    df = pd.read_csv(file_obj, dtype=str)
    # 过滤id为空的行
    df = df[pd.isnull(df["id"])]

    def get_sub_budget_owner(v_BO, v_SBO):
        # 获取或创建Budget Owner，异常时调用默认值
        try:
            bo_obj, _ = Employee.objects.get_or_create(name=v_BO.strip())
        except ObjectDoesNotExist:
            bo_obj = Employee.objects.get(name="99999 No Budget Owner")
        
        # 获取或创建Sub Budget Owner，异常时调用默认值
        try:
            sbo_obj, _ = Employee.objects.get_or_create(name=v_SBO.strip())
        except ObjectDoesNotExist:
            sbo_obj = Employee.objects.get(name="99998 No Sub Budget Owner")
        
        # 获取或创建SubBudgetOwner关联记录
        sub_budget_owner_obj, _ = SubBudgetOwner.objects.get_or_create(
            budget_owner=bo_obj,
            sub_budget_owner=sbo_obj
        )
        return sub_budget_owner_obj

    for i, row in df.iterrows():
        # 提取行数据并去除首尾空格，避免空格导致重复数据
        v_subsidiary = row["Subsidiary"].strip()
        v_department = row["Department"].strip()
        v_account = row["Account"].strip()
        v_sub_budget = row["Sub Budget"].strip()
        v_budget_owner = row["Budget Owner"].strip()
        v_sub_budget_owner = row["Sub Budget Owner"].strip()

        # 批量获取或创建关联对象，省去重复查询
        subsidiary_obj, _ = Subsidiary.objects.get_or_create(name=v_subsidiary)
        department_obj, _ = Department.objects.get_or_create(name=v_department)
        account_obj, _ = Account.objects.get_or_create(name=v_account)
        sub_budget_obj, _ = SubBudget.objects.get_or_create(name=v_sub_budget)
        sub_budget_owner_obj = get_sub_budget_owner(v_budget_owner, v_sub_budget_owner)

        # 创建Combination记录
        Combination.objects.get_or_create(
            subsidiary=subsidiary_obj,
            department=department_obj,
            account=account_obj,
            sub_budget=sub_budget_obj,
            budget_owner=sub_budget_owner_obj,
        )
        print(f"处理第{i}行: {row}")

优化说明：

• 函数名改为process_csv更语义化，支持传入任意文件对象
• 用ObjectDoesNotExist替代裸except，避免捕获无关异常
• 利用get_or_create返回值直接获取对象，省去重复get查询
• 给字符串字段添加strip()处理，避免空格导致的重复数据

二、编写上传表单类 #

在forms.py中创建表单，用于接收CSV文件：

from django import forms

class CSVUploadForm(forms.Form):
    csv_file = forms.FileField(
        label="选择CSV文件",
        widget=forms.FileInput(attrs={"accept": ".csv"})
    )

三、编写类视图 #

使用FormView处理表单提交、文件上传和数据导入逻辑：

from django.views.generic.edit import FormView
from django.urls import reverse_lazy
from django.contrib import messages
from .forms import CSVUploadForm
from .utils import process_csv  # 假设重构后的函数放在utils.py中

class CSVUploadView(FormView):
    template_name = "csv_upload.html"
    form_class = CSVUploadForm
    success_url = reverse_lazy("csv_upload")  # 上传成功后返回当前页面

    def form_valid(self, form):
        # 获取上传的文件对象
        csv_file = form.cleaned_data["csv_file"]
        try:
            # 调用导入函数处理文件
            process_csv(csv_file)
            messages.success(self.request, "CSV数据导入成功！")
        except Exception as e:
            messages.error(self.request, f"导入失败: {str(e)}")
        return super().form_valid(form)

四、编写上传模板 #

创建templates/csv_upload.html，实现GUI上传界面：

<!DOCTYPE html>
<html>
<head>
    <title>CSV数据上传</title>
</head>
<body>
    <h1>上传CSV文件导入数据</h1>
    {% if messages %}
        {% for message in messages %}
            <div style="color: {% if message.tags == 'success' %}green{% else %}red{% endif %};">
                {{ message }}
            </div>
        {% endfor %}
    {% endif %}
    <form method="post" enctype="multipart/form-data">
        {% csrf_token %}
        {{ form.as_p }}
        <button type="submit">上传并导入</button>
    </form>
</body>
</html>

注意：表单必须设置enctype="multipart/form-data"才能正确解析上传的文件。

五、配置URL路由 #

在urls.py中添加视图路由：

from django.urls import path
from .views import CSVUploadView

urlpatterns = [
    # 其他路由...
    path("csv-upload/", CSVUploadView.as_view(), name="csv_upload"),
]

六、测试功能 #

1. 启动Django服务器，访问/csv-upload/页面
2. 选择符合格式的CSV文件上传
3. 查看数据库是否新增对应数据，页面会显示成功/失败提示

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