Django中处理大文件（最大10GB）上传的最优方案

我一直在查阅Django文档和StackOverflow相关帖子，但还是搞不定大文件上传的处理。目前的代码如下，现在想改成分片上传来避免网站过载——网站直接上传到DigitalOcean Spaces，不存服务器，最大允许10GB文件。

当前代码 #

forms.py #

def validate_file(file):
    # Validate if no file submitted
    if not file:
        raise ValidationError("Error")

    # Check file size (5 GB limit)
    max_size = 5 * 1024 * 1024 * 1024  # 5 GB in bytes
    if file.size > max_size:
        raise ValidationError("Error")
    
    # Define allowed file types and their corresponding MIME types
    allowed_types = {
        # Audio formats
        'wav': ['audio/wav', 'audio/x-wav'],
        'mp3': ['audio/mpeg', 'audio/mp3'],
        'mpga': ['audio/mpeg'],
        'aac': ['audio/aac'],
        # 其他类型省略
    }

class UploadFileForm(forms.Form):
    file = forms.FileField(validators=[validate_file])

views.py #

def transcribe_Submit(request):
    if request.method == 'POST':
        form = UploadFileForm(request.POST, request.FILES)
        if form.is_valid():
            uploaded_file = request.FILES['file']

            session_id = str(uuid.uuid4())
            request.session['session_id'] = session_id

            try:
                transcribed_doc, created = TranscribedDocument.objects.get_or_create(id=session_id)
                transcribed_doc.audio_file = uploaded_file
                transcribed_doc.save()
            # 后续逻辑省略

models.py #

class TranscribedDocument(models.Model):
    id = models.CharField(primary_key=True, max_length = 40)
    audio_file = models.FileField(max_length = 140, upload_to='example_upload_url/', null= True)
    # 其他字段省略

分片上传实现方案 #

核心思路：前端分片+后端生成上传签名+文件直传Spaces，服务器仅处理签名和分片合并，完全规避大文件对服务器的资源占用。

1. 前端分片上传逻辑（JavaScript） #

将文件切割为固定大小分片（示例为50MB），逐个请求后端获取上传签名后直传Spaces，最后触发合并操作：

const chunkSize = 50 * 1024 * 1024; // 50MB分片
const fileInput = document.getElementById('file-input');
const file = fileInput.files[0];
const totalChunks = Math.ceil(file.size / chunkSize);
const chunkFilenames = [];

// 遍历上传所有分片
for (let i = 0; i < totalChunks; i++) {
    const start = i * chunkSize;
    const end = Math.min(start + chunkSize, file.size);
    const chunk = file.slice(start, end);

    // 请求后端获取分片上传签名
    const signRes = await fetch('/api/get-upload-sign/', {
        method: 'POST',
        headers: {'Content-Type': 'application/json'},
        body: JSON.stringify({
            filename: file.name,
            chunkIndex: i,
            fileSize: file.size,
            fileType: file.type
        })
    });
    const signData = await signRes.json();
    chunkFilenames.push(signData.chunk_filename);

    // 直传分片到Spaces
    await fetch(signData.upload_url, {
        method: 'PUT',
        body: chunk,
        headers: signData.headers
    });
}

// 所有分片上传完成后请求合并
await fetch('/api/merge-chunks/', {
    method: 'POST',
    headers: {'Content-Type': 'application/json'},
    body: JSON.stringify({
        filename: file.name,
        chunkFilenames: chunkFilenames,
        totalChunks: totalChunks
    })
});

2. 后端接口实现 #

生成上传签名接口（views.py）

import json
import uuid
import boto3
from django.http import JsonResponse
from django.conf import settings
from django.core.exceptions import ValidationError

def validate_file_metadata(file_size, file_type):
    # 校验文件大小（10GB上限）
    max_size = 10 * 1024 * 1024 * 1024
    if file_size > max_size:
        raise ValidationError("文件大小不能超过10GB")
    
    # 校验文件类型
    allowed_types = {
        'wav': ['audio/wav', 'audio/x-wav'],
        'mp3': ['audio/mpeg', 'audio/mp3'],
        'mpga': ['audio/mpeg'],
        'aac': ['audio/aac'],
    }
    ext = file_type.split('/')[-1].lower()
    if ext not in allowed_types.values() and ext not in allowed_types:
        raise ValidationError("不支持的文件类型")

def get_upload_sign(request):
    if request.method != 'POST':
        return JsonResponse({'error': '仅支持POST请求'}, status=400)
    
    try:
        data = json.loads(request.body)
        validate_file_metadata(data['fileSize'], data['fileType'])
        
        # 生成唯一分片文件名
        chunk_filename = f"temp_chunks/{uuid.uuid4()}_{data['chunkIndex']}_{data['filename']}"
        
        # 初始化Spaces客户端
        s3 = boto3.client(
            's3',
            aws_access_key_id=settings.DO_SPACES_ACCESS_KEY,
            aws_secret_access_key=settings.DO_SPACES_SECRET_KEY,
            endpoint_url=settings.DO_SPACES_ENDPOINT
        )
        
        # 生成预签名PUT URL（有效期1小时）
        upload_url = s3.generate_presigned_url(
            'put_object',
            Params={
                'Bucket': settings.DO_SPACES_BUCKET,
                'Key': chunk_filename,
                'ContentType': 'application/octet-stream'
            },
            ExpiresIn=3600
        )
        
        return JsonResponse({
            'upload_url': upload_url,
            'headers': {'Content-Type': 'application/octet-stream'},
            'chunk_filename': chunk_filename
        })
    except ValidationError as e:
        return JsonResponse({'error': str(e)}, status=400)

分片合并接口（views.py）

def merge_chunks(request):
    if request.method != 'POST':
        return JsonResponse({'error': '仅支持POST请求'}, status=400)
    
    try:
        data = json.loads(request.body)
        final_key = f"audio_uploads/{uuid.uuid4()}_{data['filename']}"
        
        s3 = boto3.client(
            's3',
            aws_access_key_id=settings.DO_SPACES_ACCESS_KEY,
            aws_secret_access_key=settings.DO_SPACES_SECRET_KEY,
            endpoint_url=settings.DO_SPACES_ENDPOINT
        )
        
        # 初始化分片上传
        upload_id = s3.create_multipart_upload(
            Bucket=settings.DO_SPACES_BUCKET,
            Key=final_key
        )['UploadId']
        
        # 收集分片信息
        parts = []
        for idx, chunk_key in enumerate(data['chunkFilenames']):
            # 获取分片ETag
            chunk_resp = s3.head_object(Bucket=settings.DO_SPACES_BUCKET, Key=chunk_key)
            parts.append({
                'PartNumber': idx + 1,
                'ETag': chunk_resp['ETag']
            })
        
        # 完成分片合并
        s3.complete_multipart_upload(
            Bucket=settings.DO_SPACES_BUCKET,
            Key=final_key,
            MultipartUpload={'Parts': parts},
            UploadId=upload_id
        )
        
        # 保存到数据库
        session_id = str(uuid.uuid4())
        TranscribedDocument.objects.create(
            id=session_id,
            audio_file=final_key
        )
        
        # 删除临时分片
        for chunk_key in data['chunkFilenames']:
            s3.delete_object(Bucket=settings.DO_SPACES_BUCKET, Key=chunk_key)
        
        return JsonResponse({'status': 'success', 'document_id': session_id})
    except Exception as e:
        return JsonResponse({'error': str(e)}, status=500)

3. 模型与表单调整 #

• models.py保持原有结构即可，FileField会自动映射到Spaces的文件路径
• 原有的UploadFileForm可弃用，改为前端做基础校验，后端通过validate_file_metadata做最终校验

关键注意事项 #

• 前端需添加分片上传重试逻辑，应对网络波动
• 在DigitalOcean Spaces控制台设置生命周期规则，自动删除24小时前的临时分片
• 可添加分片MD5校验，确保上传完整性
• 调整Web服务器（如Nginx）的超时时间，避免长请求被中断

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