Django多文件上传：实现单文件完成即异步分析推向前端

核心思路 #

把原来的「批量上传后统一分析」拆成「单个文件独立上传+实时异步分析+结果推送」，分前端、后端、部署三个部分改造：

一、前端改造：拆分多文件为逐个上传 #

放弃原来的ClearableFileInput批量提交逻辑，用JS遍历选中的文件，逐个发起上传请求：

// 监听拖拽区域的文件落下事件
document.getElementById('file-drop-area').addEventListener('drop', async (e) => {
  e.preventDefault();
  const files = Array.from(e.dataTransfer.files);
  
  // 逐个处理文件上传
  for (const file of files) {
    const formData = new FormData();
    formData.append('file', file);
    
    try {
      const response = await fetch('/api/upload-single/', {
        method: 'POST',
        body: formData,
        headers: {
          'X-CSRFToken': getCookie('csrftoken') // 必须处理CSRF验证
        }
      });
      
      const resData = await response.json();
      if (resData.success) {
        // 先在页面标记该文件已上传，等待分析结果
        addUploadedFileMark(file.name);
      }
    } catch (err) {
      console.error(`${file.name} 上传失败:`, err);
    }
  }
});

// 辅助函数：获取页面CSRF Token
function getCookie(name) {
  let cookieValue = null;
  if (document.cookie && document.cookie !== '') {
    const cookies = document.cookie.split(';');
    for (let i = 0; i < cookies.length; i++) {
      const cookie = cookies[i].trim();
      if (cookie.substring(0, name.length + 1) === (name + '=')) {
        cookieValue = decodeURIComponent(cookie.substring(name.length + 1));
        break;
      }
    }
  }
  return cookieValue;
}

// 更新页面：标记已上传文件
function addUploadedFileMark(fileName) {
  const item = document.createElement('div');
  item.className = 'uploaded-item';
  item.innerHTML = `${fileName} - 上传完成，分析中...`;
  document.getElementById('upload-list').appendChild(item);
}

二、后端改造：接收单文件+触发异步分析 #

新建单个文件上传的API视图，接收到文件后直接丢入线程池异步分析，同时通过WebSocket推送结果：

from django.http import JsonResponse
from django.views.decorators.csrf import csrf_exempt
from concurrent.futures import ThreadPoolExecutor
import os
from myapp.utils import push_result_via_websocket  # 自定义的WebSocket推送函数

# 线程池大小根据服务器CPU核心数调整，比如8核服务器设为8
executor = ThreadPoolExecutor(max_workers=8)

def analyze_single_file(file_path, file_name):
    """文件分析逻辑：生成计算值列表"""
    analysis_result = []
    try:
        with open(file_path, 'rb') as f:
            content = f.read()
            # 这里替换成你的实际计算逻辑
            analysis_result = [len(content), content[:10].hex()]
        # 分析完成后推送到前端
        push_result_via_websocket({
            'file_name': file_name,
            'result': analysis_result
        })
    except Exception as e:
        # 记录错误日志
        print(f"分析 {file_name} 失败: {str(e)}")
        push_result_via_websocket({
            'file_name': file_name,
            'result': '分析失败',
            'error': str(e)
        })
    finally:
        # 清理临时文件
        if os.path.exists(file_path):
            os.remove(file_path)

@csrf_exempt
def upload_single_file(request):
    if request.method == 'POST' and request.FILES.get('file'):
        uploaded_file = request.FILES['file']
        # 临时保存文件（如果文件不大也可以直接在内存处理）
        temp_path = f'/tmp/upload_{uploaded_file.name}'
        with open(temp_path, 'wb+') as dest:
            for chunk in uploaded_file.chunks():
                dest.write(chunk)
        # 提交到线程池异步分析
        executor.submit(analyze_single_file, temp_path, uploaded_file.name)
        return JsonResponse({
            'success': True,
            'file_name': uploaded_file.name
        })
    return JsonResponse({'success': False, 'msg': '无效请求'})

三、实时结果推送：用WebSocket实现 #

Django需要用channels库支持WebSocket，配合ASGI服务器（Daphne）部署：

前端WebSocket代码 #

// 建立WebSocket连接
const ws = new WebSocket(`ws://${window.location.host}/ws/analysis/`);

ws.onmessage = function(event) {
    const data = JSON.parse(event.data);
    // 更新页面显示分析结果
    updateAnalysisResult(data.file_name, data.result);
};

ws.onerror = function(err) {
    console.error('WebSocket连接出错:', err);
    // 断开后自动重连
    setTimeout(() => window.location.reload(), 3000);
};

// 更新页面的分析结果展示
function updateAnalysisResult(fileName, result) {
    const items = document.querySelectorAll('.uploaded-item');
    for (const item of items) {
        if (item.textContent.includes(fileName)) {
            item.innerHTML = `${fileName} - 分析完成: ${result.join(', ')}`;
            break;
        }
    }
}

后端WebSocket配置（channels） #

1. 安装依赖：pip install channels channels-redis
2. 在Django配置中添加ASGI应用和通道层：

# settings.py
INSTALLED_APPS = [
    # ...其他应用
    'channels',
]

ASGI_APPLICATION = 'myapp.asgi.application'
CHANNEL_LAYERS = {
    'default': {
        'BACKEND': 'channels_redis.core.RedisChannelLayer',
        'CONFIG': {
            "hosts": [('127.0.0.1', 6379)],
        },
    },
}

3. 编写WebSocket消费者处理连接：

# consumers.py
from channels.generic.websocket import AsyncWebsocketConsumer
import json

class AnalysisConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        await self.channel_layer.group_add(
            'analysis_group',
            self.channel_name
        )
        await self.accept()

    async def disconnect(self, close_code):
        await self.channel_layer.group_discard(
            'analysis_group',
            self.channel_name
        )

    async def send_result(self, event):
        result = event['result']
        await self.send(text_data=json.dumps(result))

四、部署调整 #

原来的Nginx+Gunicorn架构需要新增Daphne处理WebSocket请求：

Nginx配置片段 #

# HTTP请求转发到Gunicorn
location / {
    proxy_pass http://127.0.0.1:8000;
    proxy_set_header Host $host;
    proxy_set_header X-Real-IP $remote_addr;
}

# WebSocket请求转发到Daphne
location /ws/ {
    proxy_pass http://127.0.0.1:8001;
    proxy_http_version 1.1;
    proxy_set_header Upgrade $http_upgrade;
    proxy_set_header Connection "upgrade";
    proxy_set_header Host $host;
}

启动服务 #

• 启动Gunicorn处理HTTP：gunicorn myapp.wsgi:application --bind 127.0.0.1:8000
• 启动Daphne处理WebSocket：daphne myapp.asgi:application --bind 127.0.0.1:8001

关键注意事项 #

• 调整Django和Nginx的文件大小限制：Django设置FILE_UPLOAD_MAX_MEMORY_SIZE，Nginx设置client_max_body_size
• 线程池大小不要超过服务器CPU核心数，避免资源耗尽
• 临时文件要及时清理，防止磁盘占满
• 做好异常处理：上传失败、分析报错、WebSocket断开重连

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