基于Django Rest Framework实现视频流RestAPI的方案咨询

以下是实现接收视频ID和时间戳、返回对应视频分片或Base64数据的完整方案：

一、前置准备 #

• 安装依赖：确保系统已安装ffmpeg（用于视频分片处理），Python环境安装django、djangorestframework
• 初始化DRF项目：如果还没有项目，先通过django-admin startproject video_stream创建，再添加rest_framework到INSTALLED_APPS

二、数据库模型设计 #

首先创建存储视频元数据的模型，用于记录视频路径、总时长等关键信息：

# models.py
from django.db import models

class Video(models.Model):
    title = models.CharField(max_length=255, verbose_name="视频标题")
    file_path = models.CharField(max_length=500, verbose_name="视频文件路径")  # 支持本地路径或云存储路径
    duration = models.FloatField(verbose_name="视频总时长(秒)")
    created_at = models.DateTimeField(auto_now_add=True, verbose_name="创建时间")

    def __str__(self):
        return self.title

注意：视频文件建议存储在云存储（如OSS、S3）或Django的MEDIA_ROOT目录下，避免直接存储在代码目录。

三、请求参数序列化验证 #

用DRF序列化器做请求参数的合法性校验，避免非法输入：

# serializers.py
from rest_framework import serializers
from .models import Video

class StreamRequestSerializer(serializers.Serializer):
    video_id = serializers.IntegerField(help_text="视频ID")
    timestamp = serializers.FloatField(min_value=0, help_text="当前播放时间戳(秒)")
    return_type = serializers.ChoiceField(choices=['file', 'base64'], default='file', help_text="返回类型：file-视频文件，base64-Base64编码字符串")

    def validate_video_id(self, value):
        try:
            Video.objects.get(pk=value)
        except Video.DoesNotExist:
            raise serializers.ValidationError("指定视频不存在")
        return value

四、核心视图实现 #

编写处理视频流请求的视图，包含视频分片切割、数据返回逻辑：

# views.py
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework import status
from .models import Video
from .serializers import StreamRequestSerializer
import subprocess
import os
import base64

class StreamVideoView(APIView):
    def post(self, request):
        # 校验请求参数
        serializer = StreamRequestSerializer(data=request.data)
        if not serializer.is_valid():
            return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
        
        video_id = serializer.validated_data['video_id']
        timestamp = serializer.validated_data['timestamp']
        return_type = serializer.validated_data['return_type']
        
        video = Video.objects.get(pk=video_id)
        
        # 处理时间戳超出视频时长的情况
        if timestamp >= video.duration:
            return Response({"msg": "已播放至视频末尾"}, status=status.HTTP_204_NO_CONTENT)
        
        # 定义分片时长（可根据需求调整，比如10秒/段）
        segment_length = 10
        start_time = timestamp
        end_time = min(start_time + segment_length, video.duration)
        
        # 临时存储分片文件的路径
        temp_segment_path = f"/tmp/video_segment_{video_id}_{start_time}.mp4"
        
        # 调用ffmpeg切割视频
        ffmpeg_cmd = [
            "ffmpeg",
            "-ss", str(start_time),  # 起始时间
            "-to", str(end_time),    # 结束时间
            "-i", video.file_path,   # 源视频路径
            "-c:v", "copy",          # 视频编码直接复制（快速切割）
            "-c:a", "copy",          # 音频编码直接复制
            "-y",                    # 覆盖已有临时文件
            temp_segment_path
        ]
        
        try:
            # 执行ffmpeg命令，捕获输出
            subprocess.run(ffmpeg_cmd, check=True, capture_output=True)
        except subprocess.CalledProcessError as e:
            return Response(
                {"error": f"视频分片失败: {e.stderr.decode('utf-8')}"},
                status=status.HTTP_500_INTERNAL_SERVER_ERROR
            )
        except FileNotFoundError:
            return Response(
                {"error": "系统未安装ffmpeg，请先安装"},
                status=status.HTTP_500_INTERNAL_SERVER_ERROR
            )
        
        # 根据返回类型处理数据
        if return_type == "base64":
            with open(temp_segment_path, "rb") as f:
                segment_data = f.read()
                base64_str = base64.b64encode(segment_data).decode("utf-8")
            # 删除临时文件
            os.remove(temp_segment_path)
            return Response({
                "start_time": start_time,
                "end_time": end_time,
                "remaining_duration": video.duration - end_time,
                "video_base64": base64_str
            })
        else:
            # 返回视频文件流
            response = Response(open(temp_segment_path, "rb"), content_type="video/mp4")
            response["Content-Disposition"] = f"inline; filename=segment_{video_id}_{start_time}.mp4"
            # 可通过定时任务清理临时文件，避免占用磁盘空间
            return response

五、路由配置 #

在项目的urls.py中添加API路由：

# urls.py
from django.urls import path
from .views import StreamVideoView

urlpatterns = [
    path("api/video/stream/", StreamVideoView.as_view(), name="video-stream"),
]

六、关键优化与注意事项 #

• 临时文件清理：可通过Celery Beat或系统定时任务（如Linux的cron）定期清理/tmp目录下的过期分片文件
• 性能优化：对热门视频的分片做缓存，避免重复切割；高并发场景下可将视频切割逻辑异步化（用Celery）
• 权限控制：可添加DRF的权限类（如IsAuthenticated），限制仅授权用户访问API
• 视频时长获取：上传视频时可通过ffprobe自动获取时长并保存到Video模型，示例代码：

  def get_video_duration(file_path):
      cmd = [
          "ffprobe", "-v", "error",
          "-show_entries", "format=duration",
          "-of", "default=noprint_wrappers=1:nokey=1",
          file_path
      ]
      result = subprocess.run(cmd, capture_output=True, text=True)
      return float(result.stdout.strip())
  

• Base64注意事项：Base64编码会使数据体积增加约33%，大分片场景建议优先返回文件流而非Base64

内容的提问来源于stack exchange，提问作者ES Ref HYD-AMP