如何在Django应用中完成CSV/TSV翻译后返回文件且不阻塞用户

核心需求 #

基于Django开发的CSV/TSV跨语言翻译工具，当前同步处理耗时任务导致用户需等待页面响应，需替换原HttpResponseRedirect("/thanks")逻辑，实现异步处理翻译任务，完成后自动返回生成的翻译文件。

解决方案思路 #

1. 使用Celery实现异步任务处理，避免阻塞请求
2. 扩展模型记录任务状态与翻译后文件路径
3. 前端通过轮询检查任务状态，完成后触发文件下载
4. 新增视图用于返回翻译后的文件

步骤1：配置Celery（异步任务框架） #

首先安装依赖：

pip install celery redis

在项目根目录创建celery.py：

import os
from celery import Celery

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'your_project.settings')
app = Celery('your_project')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()

在settings.py中添加Celery配置：

CELERY_BROKER_URL = 'redis://localhost:6379/0'
CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'

步骤2：扩展模型记录任务信息 #

修改models.py，新增任务ID、状态和翻译后文件字段：

from django.db import models
from django.core.validators import FileExtensionValidator

file_type_choices = [
    ('CSV', 'Comma-Separated Values'),
    ('TSV', 'Tab-Separated Values')
]
valid_extensions = ['tsv', 'csv']

class FileModel(models.Model):
    file = models.FileField(upload_to="files", 
                            validators=[FileExtensionValidator(allowed_extensions=valid_extensions)])
    file_type = models.CharField("Choose your file type", choices=file_type_choices,max_length=3)
    source_language = models.CharField("Choose the language in which file is", max_length=30)
    target_language = models.CharField("Choose the language to which you want to translate the file", max_length=30)
    # 新增字段
    task_id = models.CharField(max_length=255, blank=True, null=True)
    status = models.CharField(max_length=20, default='PENDING')  # PENDING/STARTED/SUCCESS/FAILURE
    translated_file = models.FileField(upload_to="translated_files", blank=True, null=True)

执行迁移：

python manage.py makemigrations
python manage.py migrate

步骤3：将翻译任务转为异步任务 #

在app目录下创建tasks.py，封装翻译逻辑：

from celery import shared_task
from django.core.files import File
import os
from .models import FileModel
from csv_trans import translate

@shared_task(bind=True)
def translate_file_task(self, file_id):
    file_instance = FileModel.objects.get(id=file_id)
    self.update_state(state='STARTED')
    
    separator = "," if file_instance.file_type == "CSV" else "\t"
    # 假设translate函数返回翻译后的文件路径
    translated_file_path = translate(
        str(file_instance.file), 
        file_instance.source_language, 
        file_instance.target_language, 
        separator
    )
    
    # 保存翻译后的文件到模型
    with open(translated_file_path, 'rb') as f:
        file_instance.translated_file.save(
            f"{os.path.basename(file_instance.file.name).split('.')[0]}_translated.{file_instance.file_type.lower()}",
            File(f),
            save=False
        )
    file_instance.status = 'SUCCESS'
    file_instance.save()
    
    # 删除临时文件（如果需要）
    os.remove(translated_file_path)
    return "Translation completed"

步骤4：修改视图逻辑 #

更新views.py，替换同步处理为异步任务，并新增任务状态检查和文件下载视图：

from django.shortcuts import render, redirect
from django.http import JsonResponse, FileResponse
from django.views import View
from .forms import FileUploadForm
from .models import FileModel
from .tasks import translate_file_task
import os

class CreateProfileView(View):
    def get(self, request):
        form = FileUploadForm()
        return render(request, "translate/index.html", {"form": form})
    
    def post(self, request):
        submitted_form = FileUploadForm(request.POST, request.FILES)
        if submitted_form.is_valid():
            file_instance = FileModel(
                file = request.FILES['file'],
                file_type = request.POST['file_type'],
                source_language = request.POST['source_language'],
                target_language = request.POST['target_language']
            )
            file_instance.save()
            
            # 启动异步翻译任务
            task = translate_file_task.delay(file_instance.id)
            file_instance.task_id = task.id
            file_instance.save()
            
            # 跳转到等待页面，传递task_id
            return redirect('task_status', task_id=task.id)
        
        return render(request, "translate/index.html", {"form":submitted_form})

def task_status(request, task_id):
    task = translate_file_task.AsyncResult(task_id)
    file_instance = FileModel.objects.get(task_id=task_id)
    
    if task.state == 'SUCCESS':
        # 任务完成，返回文件下载响应
        file_path = file_instance.translated_file.path
        filename = os.path.basename(file_path)
        return FileResponse(open(file_path, 'rb'), as_attachment=True, filename=filename)
    elif task.state == 'FAILURE':
        return render(request, "translate/task_failure.html", {"error": task.info})
    else:
        # 返回任务状态，供前端轮询
        return render(request, "translate/task_waiting.html", {"task_id": task_id, "state": task.state})

def check_task_status(request, task_id):
    task = translate_file_task.AsyncResult(task_id)
    return JsonResponse({
        'state': task.state,
        'success': task.state == 'SUCCESS'
    })

步骤5：新增模板文件 #

task_waiting.html（等待页面）

<!DOCTYPE html>
<html>
<head>
    <meta charset="utf-8">
    <title>Processing...</title>
    <link href="https://cdn.jsdelivr.net/npm/bootstrap@5.3.0-alpha1/dist/css/bootstrap.min.css" rel="stylesheet">
</head>
<body>
    <div class="col-md-6 offset-md-3 mt-5 text-center">
        <h3>Your file is being translated. Please wait...</h3>
        <div class="spinner-border mt-3" role="status">
            <span class="visually-hidden">Loading...</span>
        </div>
    </div>
    <script>
        // 每隔2秒轮询任务状态
        setInterval(() => {
            fetch(`/check-task-status/{{ task_id }}/`)
                .then(response => response.json())
                .then(data => {
                    if (data.success) {
                        window.location.reload();
                    }
                });
        }, 2000);
    </script>
</body>
</html>

task_failure.html（任务失败页面）

<!DOCTYPE html>
<html>
<head>
    <meta charset="utf-8">
    <title>Translation Failed</title>
    <link href="https://cdn.jsdelivr.net/npm/bootstrap@5.3.0-alpha1/dist/css/bootstrap.min.css" rel="stylesheet">
</head>
<body>
    <div class="col-md-6 offset-md-3 mt-5 text-center">
        <h3>Translation failed!</h3>
        <p>{{ error }}</p>
        <a href="/" class="btn btn-primary">Try Again</a>
    </div>
</body>
</html>

步骤6：配置URL路由 #

在app的urls.py中添加路由：

from django.urls import path
from .views import CreateProfileView, task_status, check_task_status

urlpatterns = [
    path('', CreateProfileView.as_view(), name='upload'),
    path('task-status/<str:task_id>/', task_status, name='task_status'),
    path('check-task-status/<str:task_id>/', check_task_status, name='check_task_status'),
]

关键说明 #

• 异步任务避免了用户等待，提升了用户体验
• 轮询逻辑简单易实现，适合中小规模应用
• 翻译后的文件通过FileResponse返回，支持浏览器直接下载
• 模型中记录任务状态，便于后续排查问题

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