言映AI视频生成，我和我同伴的第一款AI产品

本项目是一个完整的AI视频生成平台，用户可以通过文字描述、图片或首尾帧生成视频内容。系统采用前后端分离架构，后端基于Django框架，前端使用Vue 3 + TypeScript，通过Celery处理耗时的视频生成任务。完整的用户认证和权限管理系统支持多种视频生成模式异步任务处理机制，提升用户体验前后端分离架构，便于维护和扩展完善的日志记录和错误处理机制。

惜月_treasure

195人浏览 · 2025-09-08 01:55:56

惜月_treasure · 2025-09-08 01:55:56 发布

在人工智能快速发展的今天，AI视频生成技术正逐渐成为热门应用领域。本文将详细介绍一个基于Django后端、Vue 3前端和Celery异步任务处理的完整AI视频生成平台，涵盖系统架构、核心功能和关键技术点。

一、项目概述

本项目是一个完整的AI视频生成平台，用户可以通过文字描述、图片或首尾帧生成视频内容。系统采用前后端分离架构，后端基于Django框架，前端使用Vue 3 + TypeScript，通过Celery处理耗时的视频生成任务。

二、技术架构

2.1 后端技术栈：

Django 5.2.5：Python Web框架
Django REST Framework：构建RESTful API
MySQL：关系型数据库
Redis：Celery消息队列
Celery：异步任务处理
JWT：用户认证授权

2.2 前端技术栈：

Vue 3：渐进式JavaScript框架
TypeScript：类型安全的JavaScript超集
Element Plus：Vue 3组件库
Vite：前端构建工具
Pinia：Vue状态管理

三、核心功能模块

3.1 用户管理系统

系统采用自定义的SysUser模型，包含用户基本信息、状态管理等功能：

class SysUser(models.Model):
    id = models.AutoField(primary_key=True)
    username = models.CharField(max_length=100, unique=True, verbose_name="用户名")
    password = models.CharField(max_length=100, verbose_name="密码")
    avatar = models.CharField(max_length=255, null=True, blank=True, verbose_name="用户头像")
    email = models.CharField(max_length=100, null=True, blank=True, verbose_name="用户邮箱")
    phonenumber = models.CharField(max_length=11, null=True, blank=True, verbose_name="手机号码")
    nickname = models.CharField(max_length=50, null=True, blank=True, verbose_name="用户昵称")
    gender = models.CharField(max_length=10, null=True, blank=True, choices=[('male', '男'), ('female', '女'), ('other', '其他')], verbose_name="性别")
    birthday = models.DateField(null=True, blank=True, verbose_name="生日")
    bio = models.TextField(max_length=200, null=True, blank=True, verbose_name="个人简介")
    login_date = models.DateField(null=True, verbose_name="最后登录时间")
    status = models.IntegerField(default=0, verbose_name="帐号状态（0正常 1停用）")
    create_time = models.DateField(auto_now_add=True, verbose_name="创建时间")
    update_time = models.DateField(auto_now=True, verbose_name="更新时间")
    remark = models.CharField(max_length=500, null=True, verbose_name="备注")

系统还实现了基于JWT的自定义认证机制：

class SysUserJWTAuthentication(JWTAuthentication):
    def get_user(self, validated_token):
        try:
            user_id = validated_token[api_settings.USER_ID_CLAIM]
        except KeyError:
            raise InvalidToken("Token contained no recognizable user identification")

        try:
            user = SysUser.objects.get(id=user_id)
        except SysUser.DoesNotExist:
            raise AuthenticationFailed('User not found', code='user_not_found')

        return user

3.2 权限角色系统

系统包含角色管理和用户角色关联功能

class Role(models.Model):
    id = models.AutoField(primary_key=True)
    name = models.CharField(max_length=100, unique=True, verbose_name="角色名称")
    key = models.CharField(max_length=100, unique=True, verbose_name="角色标识")
    status = models.IntegerField(default=0, verbose_name="状态（0正常 1停用）")
    remark = models.CharField(max_length=255, null=True, blank=True, verbose_name="备注")
    create_time = models.DateTimeField(auto_now_add=True, verbose_name="创建时间")
    update_time = models.DateTimeField(auto_now=True, verbose_name="更新时间")

class UserRole(models.Model):
    id = models.AutoField(primary_key=True)
    user = models.ForeignKey(SysUser, on_delete=models.CASCADE, related_name='user_roles')
    role = models.ForeignKey(Role, on_delete=models.CASCADE, related_name='role_users')
    create_time = models.DateTimeField(auto_now_add=True, verbose_name="创建时间")

3.3 AI视频生成功能

系统支持三种视频生成模式：

文生视频(Text-to-Video)
图生视频(Image-to-Video)
首尾帧生视频(Frames-to-Video)

核心任务处理：

使用Celery异步处理视频生成任务：

@shared_task
def generate_video(user_id, prompt, mode, image_url=None, confirmed=False, quality="quality", fps=30):
    try:
        if not confirmed:
            logger.info(f"用户 {user_id} 的任务，提示词 '{prompt}' 等待确认")
            return {'status': 'WAITING', 'prompt': prompt, 'image_url': image_url, 'mode': mode, 'quality': quality, 'fps': fps}

        api_key = os.getenv('ZHIPUAI_API_KEY')
        if not api_key:
            raise ValueError("未设置 ZHIPUAI_API_KEY 环境变量")
        client = ZhipuAiClient(api_key=api_key)
        response = client.videos.generations(
            model="cogvideox-3",
            prompt=prompt,
            image_url=image_url,
            quality=quality,
            with_audio=True,
            size="3840x2160",
            fps=fps,
        )
        
        # 轮询获取结果
        task_id = response.id
        max_attempts = 120  # 20 分钟
        for _ in range(max_attempts):
            status = client.videos.retrieve_videos_result(id=task_id)
            if status.task_status == 'SUCCESS':
                # 保存到数据库
                user = SysUser.objects.get(id=user_id)
                image_data = image_url if mode == 'image' else (','.join(image_url) if mode == 'frames' and image_url else '')
                history = VideoHistory.objects.create(
                    user=user,
                    task_id=task_id,
                    prompt=prompt,
                    mode=mode,
                    image_url=image_data,
                    video_url=video_url,
                    status='SUCCESS',
                    quality=quality,
                    fps=fps
                )
                return {'url': video_url}
            elif status.task_status == 'FAILURE':
                raise Exception(f"视频生成失败: {status.error_message or '未知错误'}")
            time.sleep(10)
        raise Exception("任务在20分钟后超时")
    except Exception as e:
        logger.error(f"视频生成错误: {str(e)}")
        raise

视频历史记录

class VideoHistory(models.Model):
    user = models.ForeignKey(SysUser, on_delete=models.CASCADE, related_name='video_histories', verbose_name="用户")
    task_id = models.CharField(max_length=100, unique=True, null=True, blank=True, verbose_name="任务ID")
    prompt = models.TextField(verbose_name="提示词")
    mode = models.CharField(max_length=20, choices=[('text', '文生视频'), ('image', '图生视频'), ('frames', '首尾帧生视频')], verbose_name="生成模式")
    image_url = models.TextField(null=True, blank=True, verbose_name="输入图片")
    video_url = models.URLField(verbose_name="视频URL")
    status = models.CharField(max_length=20, default='SUCCESS', choices=[('SUCCESS', '成功'), ('FAILURE', '失败'), ('PENDING', '处理中')], verbose_name="状态")
    quality = models.CharField(max_length=20, choices=[('speed', '快速'), ('quality', '高质量')], default='quality', null=True, blank=True, verbose_name="质量")
    fps = models.IntegerField(default=30, null=True, blank=True, verbose_name="帧率")
    created_at = models.DateTimeField(auto_now_add=True, verbose_name="创建时间")
    updated_at = models.DateTimeField(auto_now=True, verbose_name="更新时间")
    local_video_path = models.FileField(upload_to='videos/', null=True, blank=True, verbose_name="本地视频路径")

3.4 前端功能实现

前端使用Vue 3 + TypeScript构建，主要功能页面包括：

3.4.1 视频生成页面

支持三种生成模式切换，提供视频配置选项（帧率、质量等）：

// 表单数据
const formData = ref({
  content: '' // 提示词
});

// 视频配置数据
const videoConfig = ref({
  frameRate: 30, // 帧率
  quality: 'speed' // 质量
});

3.4.2 视频历史记录页面

展示用户生成的视频历史，支持在线播放和下载：

// 获取历史记录
const fetchHistory = async () => {
  try {
    const res = await axios.get('http://127.0.0.1:8000/api/videotest/video-history/');
    historyVideos.value = res.data.data || [];
  } catch (err) {
    error.value = '获取历史记录失败';
    console.error(err);
  }
};

四、关键技术点解析

4.1 异步任务处理

使用Celery处理耗时的视频生成任务，避免阻塞用户界面：

# 配置Celery
CELERY_BROKER_URL = 'redis://127.0.0.1:6379/0'
CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/0'
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'

4.2 跨域资源共享(CORS)

配置Django CORS中间件支持前后端分离架构：

# 允许所有来源的跨域请求
CORS_ORIGIN_ALLOW_ALL = True
# 允许跨域请求时携带用户凭证
CORS_ALLOW_CREDENTIALS = True
CORS_ALLOWED_ORIGINS = [
    'http://localhost:5173',  # 调整为你的前端地址
]

4.3 文件上传与代理

支持用户上传图片并实现视频文件代理下载：

@api_view(['GET'])
def proxy_video(request):
    video_url = request.GET.get('url')
    is_download = request.GET.get('download', 'false').lower() == 'true'
    
    # 实现视频文件代理和下载功能
    # ...