如何用含54字段、约5000行的大型CSV文件更新Django模型？

Alright, let's walk through how to tackle this CSV import workflow for your data center equipment Django model—since you're dealing with 5000-row files and need to filter out 20% invalid entries, efficiency and clarity are key.

1. Lock Down Validation Rules at the Model Level #

First, formalize your predefined equipment types and mandatory barcode requirement directly in your Django model to enforce data integrity from the start:

from django.db import models

# Predefined equipment type list (matches your requirements)
EQUIPMENT_TYPES = [
    ('rack', 'Rack'),
    ('pdu', 'PDU'),
    ('patch_panel', '配线架'),
    ('switch', '交换机'),
    ('server', '服务器'),
    ('blade_server', '刀片服务器'),
    # Add other predefined types here
]

class DataCenterEquipment(models.Model):
    equipment_type = models.CharField(max_length=50, choices=EQUIPMENT_TYPES)
    barcode = models.CharField(max_length=100, unique=True, verbose_name="内部资产编号")
    # Include all 52 remaining device fields here (e.g., rack_location, pdu_port_count, etc.)

    def __str__(self):
        return f"{self.get_equipment_type_display()} - {self.barcode}"

2. Efficient CSV Import with Bulk Processing #

For 5000-row datasets, avoid slow per-row ORM saves—use bulk_create instead. We'll first filter valid records, then batch-insert them safely:

import csv
from django.db import transaction
from django.core.exceptions import ValidationError
from .models import DataCenterEquipment
import logging

logger = logging.getLogger(__name__)

def import_equipment_csv(file_path):
    valid_records = []
    invalid_records = []
    allowed_types = dict(DataCenterEquipment.EQUIPMENT_TYPES).keys()
    # Pre-fetch existing barcodes to avoid duplicates (if needed for incremental imports)
    existing_barcodes = set(DataCenterEquipment.objects.values_list('barcode', flat=True))

    with open(file_path, 'r', encoding='utf-8') as csvfile:
        reader = csv.DictReader(csvfile)
        # Start row number at 2 to skip the header row in error messages
        for row_num, row in enumerate(reader, start=2):
            # Extract and clean core validation fields
            equipment_type = row.get('equipment_type', '').strip()
            barcode = row.get('barcode', '').strip()

            # Validate core conditions first
            if equipment_type not in allowed_types:
                invalid_records.append((row_num, "设备类型不在预定义列表中"))
                continue
            if not barcode:
                invalid_records.append((row_num, "缺少内部资产编号（条形码）"))
                continue
            if barcode in existing_barcodes:
                invalid_records.append((row_num, "该资产编号已存在于系统中"))
                continue

            # Map CSV fields to model fields and validate
            try:
                equipment = DataCenterEquipment(
                    equipment_type=equipment_type,
                    barcode=barcode,
                    # Map your remaining 52 fields here, e.g.:
                    # rack_number=row.get('rack_number'),
                    # manufacturer=row.get('manufacturer'),
                    # ...
                )
                # Run full model validation to catch any field-specific errors
                equipment.full_clean()
                valid_records.append(equipment)
                existing_barcodes.add(barcode)  # Update for subsequent rows
            except ValidationError as e:
                invalid_records.append((row_num, f"字段验证失败: {str(e)}"))
            except Exception as e:
                invalid_records.append((row_num, f"未知错误: {str(e)}"))

    # Batch insert valid records atomically (all or nothing)
    if valid_records:
        with transaction.atomic():
            # Adjust batch_size based on your database's performance (100-200 is safe for 5k rows)
            DataCenterEquipment.objects.bulk_create(valid_records, batch_size=150)
        logger.info(f"成功导入 {len(valid_records)} 条有效设备记录")

    # Log and document invalid records for follow-up
    if invalid_records:
        logger.warning(f"发现 {len(invalid_records)} 条无效记录，已跳过")
        # Write invalid rows to a separate CSV for review
        with open('invalid_equipment_import.csv', 'w', encoding='utf-8', newline='') as f:
            writer = csv.writer(f)
            writer.writerow(["行号", "错误原因"])
            writer.writerows(invalid_records)

    return invalid_records

3. Optional Optimizations for Scalability #

• Asynchronous Processing: If your CSV size grows beyond 10k rows, offload the import task to a background worker (like Celery) so it doesn't block your web app.
• Configurable Field Mapping: Store CSV-to-model field mappings in a JSON file instead of hardcoding—this makes it easier to adjust if CSV columns change:

  {
      "equipment_type": "设备类型",
      "barcode": "内部资产编号",
      "rack_number": "机柜编号"
  }
  

• Dry Run Mode: Add a flag to the import function that skips database insertion and only returns valid/invalid records, so you can preview changes before committing.

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