前端vue3 setup，后端fastapi

介绍

在 Web 开发中，视频播放是一个常见的需求。HLS（HTTP Live Streaming）和 FLV（Flash Video）是两种常见的视频流媒体传输协议。以下是它们的详细介绍和实现方法。

FLV（Flash Video）

简介

FLV 是 Adobe 开发的视频格式，曾经广泛用于 Flash 播放器。
虽然 Flash 已被淘汰，但 FLV 格式仍然可以通过 HTML5 和 JavaScript 播放。

优点

文件体积较小，适合低带宽环境。
支持流式传输，延迟较低。

缺点

兼容性较差，现代浏览器不再支持 Flash。
需要额外的 JavaScript 库（如 flv.js）来播放 FLV 视频。

后端

依赖

FastAPI:
- 用于构建 API 的 Web 框架。
- 安装命令：
```
pip install fastapi
```
Uvicorn:
- 用于运行 FastAPI 应用的 ASGI 服务器。
- 安装命令：
```
pip install uvicorn
```

代码

python">from fastapi import HTTPException, FastAPI, Query, Header
from fastapi.responses import JSONResponse, FileResponse
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import StreamingResponse
from typing import Generator, Optional
import json
import os
import uvicornapp = FastAPI()# 添加CORS中间件
app.add_middleware(CORSMiddleware,allow_origins=["*"],  # 允许所有来源allow_credentials=True,allow_methods=["*"],  # 允许所有方法allow_headers=["*"],  # 允许所有头部
)class BaseResponse:def __init__(self, code, msg):self.code = codeself.msg = msgclass MyFile:def __init__(self, filename, knowledge_base_name):self.filename = filenameself.filepath = os.path.join(knowledge_base_name, filename)def iter_file(file_path: str, start: int = 0, end: Optional[int] = None) -> Generator[bytes, None, None]:with open(file_path, "rb") as file:file.seek(start)while chunk := file.read(8192):yield chunkif end and file.tell() >= end:break@app.get("/knowledge_base/download_video")
def stream_video(filename: str = Query(..., description="FLV文件名称", examples=["test.flv"]),range: Optional[str] = Header(None)
):"""流式传输FLV视频，支持范围请求"""file_path = os.path.join(os.getcwd(), "files", filename)if not os.path.exists(file_path):raise HTTPException(status_code=404, detail="文件未找到")file_size = os.path.getsize(file_path)start = 0end = file_size - 1if range:# Parse the range headermatch = range.replace("bytes=", "").split("-")start = int(match[0])if len(match) > 1 and match[1]:end = int(match[1])content_length = end - start + 1headers = {"Content-Range": f"bytes {start}-{end}/{file_size}","Accept-Ranges": "bytes","Content-Length": str(content_length),"Content-Type": "video/x-flv"}return StreamingResponse(iter_file(file_path, start=start, end=end + 1),status_code=206,headers=headers,media_type="video/x-flv")def convert_json_to_vtt(json_file, vtt_file):# Read JSON data from a filewith open(json_file, 'r', encoding='utf-8') as f:json_data = json.load(f)vtt_data = "WEBVTT\n\n"for i, item in enumerate(json_data):start_time = float(item['start'])end_time = start_time + float(item['duration'])start_time_min, start_time_sec = divmod(start_time, 60)end_time_min, end_time_sec = divmod(end_time, 60)vtt_data += f"{i+1}\n"vtt_data += f"{int(start_time_min):02}:{start_time_sec:06.3f} --> {int(end_time_min):02}:{end_time_sec:06.3f}\n"vtt_data += f"{item['text']}\n\n"# Write VTT data to a filewith open(vtt_file, 'w', encoding='utf-8') as f:f.write(vtt_data)@app.get("/knowledge_base/download_vtt")
def download_vtt(file_name: str = Query(..., description="文件名称"),preview: bool = Query(False, description="是：浏览器内预览；否：下载")
):"""下载字幕文件"""vtt_dir = "./files/vtt_dir"json_dir = "./files/vtt_dir"name =  os.path.splitext(os.path.basename(file_name))[0]vtt_name = name + ".vtt"vtt_file_path = vtt_dir + "/" +vtt_nameif not os.path.exists(vtt_file_path):json_file_path = json_dir + "/" + name + ".json"convert_json_to_vtt(json_file_path,vtt_file_path)if preview:content_disposition_type = "inline"else:content_disposition_type = Nonetry:kb_file = MyFile(filename=vtt_name, knowledge_base_name=vtt_dir)if os.path.exists(kb_file.filepath):response = FileResponse(path=kb_file.filepath,filename=kb_file.filename,media_type="text/vtt",content_disposition_type=content_disposition_type,)# 添加CORS头部response.headers["Access-Control-Allow-Origin"] = "*"return responseexcept Exception as e:msg = f"{vtt_name} 读取文件失败，错误信息是：{e}"raise HTTPException(status_code=500, detail=msg)raise HTTPException(status_code=404, detail=f"{kb_file.filename} 文件未找到")if __name__ == "__main__":uvicorn.run(app, host="127.0.0.1", port=7861)

前端要访问接口地址

访问视频地址（flv形式非能直接访问下载）：http://127.0.0.1:7861/knowledge_base/download_video?filename=robot3.mp4

字幕文件格式

.json文件格式（使用json文件，代码convert_json_to_vtt函数会将其转化为vtt文件）

python">[{"text": "wwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwww","start": 0.0,"duration": 3.0},{"text": "eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee","start": 81.56,"duration": 3.0},{"text": "rrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrr","start": 119.2,"duration": 3.68}
]

.vtt文件格式

python">WEBVTT1
00:00.000 --> 00:03.000
wwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwwww2
01:21.560 --> 01:24.560
eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee3
01:59.200 --> 02:02.880
rrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrr4
03:11.120 --> 03:14.959
肖恩，嘿，怎么样，

字幕下载地址：http://127.0.0.1:7861/knowledge_base/download_vtt?preview=false&file_name=robot3.mp4&lang=zh

前端

下载依赖

你可以使用以下命令通过 npm 或 yarn 来安装这个依赖：

使用 npm 安装

python">npm install flv.js

使用 yarn 安装

python">yarn add flv.js

安装完成后，你就可以在项目中正常使用 flv.js 库了。这个库用于在支持 HTML5 <video> 标签的浏览器中播放 FLV 格式的视频。

代码

base.js

python">const ip = "127.0.0.1"//公共的配置文件
const base = {ip: ip,host: "http://" + ip + ":7861",download_vtt: "/knowledge_base/download_vtt"
}
export default base

FlvVideoPlayer.vue

python"><template><div class="mainContainer" style="width: 100%; overflow: hidden"><videoid="videoElement"class="centeredVideo"controlsautoplaystyle="width: 100%; height: calc(100vh - 6.1px)">Your browser is too old which doesn't support HTML5 video.</video></div>
</template><script>
import { ref, onMounted, onBeforeUnmount } from "vue";
import flvjs from "flv.js";
import base from "@/api/base.js";export default {setup() {const player = ref(null);const flvPlayer = ref(null);const seekPoint = ref("");const filename = ref("");let path = "http://127.0.0.1:7861/knowledge_base/download_video?filename=robot3.mp4"// 创建一个URL对象const urlObj = new URL(path);// 使用URLSearchParams获取filename参数const params = new URLSearchParams(urlObj.search);// 读取filename参数的值const fileValue = params.get("filename");filename.value = fileValue;onMounted(async () => {player.value = document.getElementById("videoElement");if (flvjs.isSupported()) {flvPlayer.value = flvjs.createPlayer({type: "auto",url: path,stashInitialSize: 32 * 1024,stashInitialTime: 0.2,seekType: "range",lazyLoad: false,lazyLoadMaxDuration: 0.2,networkStreamIO: true,nativeMP4H264Playback: true,},{enableWorker: false,enableStashBuffer: false,reuseRedirectedURL: true,autoCleanupSourceBuffer: true,});flvPlayer.value.attachMediaElement(player.value);flvPlayer.value.load();// flvStart();}player.value.addEventListener(flvjs.Events.ERROR,(errorType, errorDetail, errorInfo) => {console.log("errorType:", errorType);console.log("errorDetail:", errorDetail);console.log("errorInfo:", errorInfo);});// 动态加载字幕文件await loadSubtitles();});const loadSubtitles = async () => {try {const response = await fetch(base.host +base.download_vtt +"?preview=false&file_name=" +filename.value +"&lang=zh");if (!response.ok) {throw new Error("Network response was not ok");}const blob = await response.blob();const url = URL.createObjectURL(blob);const response2 = await fetch(base.host +base.download_vtt +"?preview=false&file_name=" +filename.value +"&lang=en");if (!response.ok) {throw new Error("Network response was not ok");}const blob2 = await response2.blob();const url2 = URL.createObjectURL(blob2);const track = document.createElement("track");const track2 = document.createElement("track");track2.src = url2;track2.kind = "subtitles";track2.srclang = "en";track2.label = "English";track2.default = true;track.src = url;track.kind = "subtitles";track.srclang = "zh";track.label = "Chinese";track.default = true;player.value.appendChild(track);player.value.appendChild(track2);} catch (error) {console.error("Failed to load subtitles:", error);}};const flvStart = () => {player.value.play();};const flvPause = () => {player.value.pause();};const flvDestroy = () => {player.value.pause();flvPlayer.value.unload();flvPlayer.value.detachMediaElement();flvPlayer.value.destroy();flvPlayer.value = null;};const flvSeekTo = () => {console.debug("seekPoint", seekPoint.value);const buffsize = player.value.buffered.length;if (buffsize) {const end = player.value.buffered.end(0);const diff = end - player.value.currentTime;}player.value.currentTime = parseFloat(seekPoint.value);};onBeforeUnmount(() => {if (flvPlayer.value) {flvDestroy();}});return {flvStart,flvPause,flvDestroy,flvSeekTo,seekPoint,};},
};
</script><style>
.mainContainer {display: block;width: 1024px;margin-left: auto;margin-right: auto;
}.urlInput {display: block;width: 100%;margin-left: auto;margin-right: auto;margin-top: 8px;margin-bottom: 8px;
}.centeredVideo {display: block;margin-left: auto;margin-right: auto;margin-bottom: auto;
}.controls {display: block;width: 100%;text-align: left;margin-left: auto;margin-right: auto;
}
</style>

注册路由后访问即可

展示结果

HLS（HTTP Live Streaming）

简介

HLS 是由 Apple 开发的流媒体传输协议，广泛用于 iOS 和 macOS 平台，但也支持其他平台。
它将视频文件分割成多个小片段（通常是 .ts 文件），并通过 HTTP 协议传输。
支持自适应码率（Adaptive Bitrate Streaming，ABR），根据网络状况动态切换视频质量。

优点

兼容性好，支持大多数现代浏览器和设备。
支持自适应码率，适合不同网络环境。
使用 HTTP 协议，易于部署和缓存。

缺点

延迟较高（通常 10-30 秒），不适合实时性要求高的场景。
需要将视频文件分割成多个小片段，增加了处理复杂度。

后端

视频转换函数

MP4->M3U8 播放列表和 .ts 分段文件->将下载 URL 添加到 .m3u8 文件中
例如

python">#EXTM3U
#EXT-X-VERSION:3
#EXT-X-TARGETDURATION:12
#EXT-X-MEDIA-SEQUENCE:0
#EXTINF:12.500000,
http://127.0.0.1:8000/knowledge_base/video_ts?preview=false&tsdir=001&file_name=segment_000.ts
#EXTINF:12.500000,
http://127.0.0.1:8000/knowledge_base/video_ts?preview=false&tsdir=001&file_name=segment_001.ts
#EXTINF:5.850000,
http://127.0.0.1:8000/knowledge_base/video_ts?preview=false&tsdir=001&file_name=segment_002.ts
#EXT-X-ENDLIST

函数

python">import subprocess
import os
import jsondef convert_to_hls(input_file, output_folder, output_m3u8, download_url):"""将视频文件转换为 HLS 格式（包括 M3U8 播放列表和 .ts 分段文件）。Args:input_file (str): 输入视频文件的路径。output_folder (str): .ts 分段文件的输出文件夹路径。output_m3u8 (str): 输出的 .m3u8 播放列表文件路径。download_url (str): 下载 .ts 文件的基本 URL。"""# 如果输出文件夹不存在，则创建它os.makedirs(output_folder, exist_ok=True)# 构建 FFmpeg 命令ffmpeg_command = (f"E:/web_work/python_work_space/Langchain-Chatchat/ffmpeg-6.1.1-essentials_build/bin/ffmpeg -i {input_file} -c:v h264 -c:a aac "f"-hls_time 10 -hls_list_size 0 "f"-hls_segment_filename {output_folder}/segment_%03d.ts "f"{output_m3u8}")# 执行 FFmpeg 命令subprocess.call(ffmpeg_command, shell=True)re_w_m3u8(output_m3u8, download_url)def re_w_m3u8(output_m3u8_path, download_url):# 将下载 URL 添加到 .m3u8 文件中with open(output_m3u8_path, 'r') as m3u8_file:m3u8_content = m3u8_file.read()m3u8_content = m3u8_content.replace('segment_', f'{download_url}segment_')with open(output_m3u8_path, 'w') as m3u8_file:m3u8_file.write(m3u8_content)# Example usage:
# input_video = r'E:\web_work\python_work_space\Langchain-Chatchat\files\new_robot3.mp4'
# tsdir = "001"
# output_ts_folder = "output_ts_segments/" + tsdir
# file_name = "output.m3u8"
# output_m3u8_file = "output_ts_segments/" + file_name
# download_base_url = "http://127.0.0.1:8000/knowledge_base/video_ts?preview=false" + "&tsdir=" + tsdir + "&file_name="  # Replace with your actual base URL
# convert_to_hls(input_video, output_ts_folder, output_m3u8_file, download_base_url)def create_srt_file(sub_content, output_file='1.srt'):"""创建一个 SRT 字幕文件。Args:sub_content (list): 包含字幕内容的列表，每个元素是一个字典，包含以下字段：- 'from'：字幕开始的时间（以秒为单位）。- 'to'：字幕结束的时间（以秒为单位）。- 'content'：字幕的文本内容。output_file (str): 输出的 SRT 文件路径，默认为 '1.srt'。"""def s2hms(x):m, s = divmod(x, 60)h, m = divmod(m, 60)hms = "%02d:%02d:%s" % (h, m, str('%.3f' % s).zfill(6))hms = hms.replace('.', ',')  # 把小数点改为逗号return hmswith open(output_file, 'w', encoding="utf-8") as f:write_content = []for n, i in enumerate(sub_content):write_content.append(str(n + 1) + '\n' + s2hms(i['from']) + ' --> ' + s2hms(i['to']) + '\n' + i['content'] + '\n\n')f.writelines(write_content)# 使用示例
# sub_content = [{"from": 0, "to": 3.39, "content": "在本视频中"},
#                {"from": 4, "to": 6, "content": "在本视频中2"},
#                {"from": 7, "to": 8, "content": "在本视频中3"}]
# create_srt_file(sub_content, output_file='my_subtitles.srt')def add_subtitles_to_video(video_file, subtitle_file, output_file):"""为视频添加字幕。Args:video_file (str): 输入视频文件的路径。subtitle_file (str): 字幕文件的路径。output_file (str): 输出的带有字幕的视频文件路径。"""# 构建FFmpeg命令cmd = ['E:/web_work/python_work_space/Langchain-Chatchat/ffmpeg-6.1.1-essentials_build/bin/ffmpeg','-i', video_file,  # 输入视频文件'-vf', 'subtitles={}'.format(subtitle_file),  # 添加字幕output_file]print(cmd)subprocess.run(cmd, check=True)  # 添加 check=True 以在出错时抛出异常# input_video = r'E:\web_work\python_work_space\Langchain-Chatchat\files\robot3.mp4'
# subtitle_file = r'my_subtitles.srt'
# output_file = r'E:\web_work\python_work_space\Langchain-Chatchat\files\new_robot3.mp4'
#
# # 使用示例
# add_subtitles_to_video(input_video, subtitle_file, output_file)

代码

python">from fastapi import HTTPException, FastAPI, Query, Header
from fastapi.responses import JSONResponse, FileResponse
from fastapi.middleware.cors import CORSMiddleware
import os
import uvicornapp = FastAPI()app.add_middleware(CORSMiddleware,allow_origins=["*"],  # 允许所有来源allow_credentials=True,allow_methods=["*"],  # 允许所有方法allow_headers=["*"],  # 允许所有头部
)class BaseResponse:def __init__(self, code, msg):self.code = codeself.msg = msgclass MyFile:def __init__(self, filename, knowledge_base_name):self.filename = filenameself.filepath = os.path.join(knowledge_base_name, filename)app.get("/knowledge_base/video_m3u8")
def download_doc(file_name: str = Query(..., description="文件名称", examples=["test.txt"]),preview: bool = Query(False, description="是：浏览器内预览；否：下载"),
):"""下载文档"""print(file_name)if preview:content_disposition_type = "inline"else:content_disposition_type = Nonetry:kb_file = MyFile(filename=file_name, knowledge_base_name=os.getcwd() + "\\output_ts_segments")if os.path.exists(kb_file.filepath):return FileResponse(path=kb_file.filepath,filename=kb_file.filename,media_type="multipart/form-data",content_disposition_type=content_disposition_type,)except Exception as e:msg = f"{file_name} 读取文件失败，错误信息是：{e}"return BaseResponse(code=500, msg=msg)return BaseResponse(code=500, msg=f"{kb_file.filename} 读取文件失败")@app.get("/knowledge_base/video_ts")
def download_doc(file_name: str = Query(..., description="文件名称", examples=["test.txt"]),tsdir: str = Query(..., description="文件夹", examples=["test"]),preview: bool = Query(False, description="是：浏览器内预览；否：下载"),
):"""下载文档"""print(file_name)if preview:content_disposition_type = "inline"else:content_disposition_type = Nonetry:kb_file = MyFile(filename=file_name, knowledge_base_name=os.getcwd() + "\\output_ts_segments\\" + tsdir)if os.path.exists(kb_file.filepath):return FileResponse(path=kb_file.filepath,filename=kb_file.filename,media_type="multipart/form-data",content_disposition_type=content_disposition_type,)except Exception as e:msg = f"{file_name} 读取文件失败，错误信息是：{e}"return BaseResponse(code=500, msg=msg)return BaseResponse(code=500, msg=f"{kb_file.filename} 读取文件失败")if __name__ == "__main__":uvicorn.run(app, host="127.0.0.1", port=7861)

前端要访问接口地址

http://127.0.0.1:7862/knowledge_base/video_m3u8?file_name=output.m3u8

前端

代码

子组件

HlsVideoPlayer.vue

python"><template><video:muted="isMuted"ref="videoEl"controlsstyle="width: 100%; height: 100%"></video>
</template>  <script setup>
import Hls from "hls.js";
import { onMounted, ref } from "vue";const videoEl = ref(null);
const isMuted = ref(false); // 默认情况下不是静音
const props = defineProps({url: {type: String,required: true,},adDuration: {type: Number,default: 0,},
});onMounted(() => {if (Hls.isSupported()) {const hls = new Hls();hls.loadSource(props.url);hls.attachMedia(videoEl.value);hls.on(Hls.Events.MANIFEST_PARSED, () => {// 这里应该添加代码来播放视频，但当前是空的});} else if (videoEl.value) {// 如果HLS.js不支持，则尝试使用原生HTML5视频播放videoEl.value.src = props.url;// 注意：m3u8文件通常不是HTML5 <video>直接支持的，所以这里可能不会工作videoEl.value.addEventListener("loadedmetadata", () => {// 这里也是空的，但通常你会在这里调用videoEl.value.play()来播放视频});}videoEl.value.currentTime = props.adDuration;
});
</script>

父组件

python"><template><div style="width: 100%; overflow: hidden"><div v-if="extension === 'm3u8'"><VideoPlayer:adDuration="adDuration":url="url"style="width: 100%; height: calc(100vh - 6.1px)"/></div></div>
</template><script setup>
import { onMounted, ref } from "vue";
import { useRouter } from "vue-router";
import VideoPlayer from "@/views/fileshow/HlsVideoPlayer.vue";const router = useRouter();
let querys = router.currentRoute.value.query;
let path = "http://127.0.0.1:7862/knowledge_base/video_m3u8?file_name=output.m3u8"
let extension = ref("");
let url = ref("");
let adDuration = 0;onMounted(() => {if (path.endsWith(".m3u8")) {extension.value = "m3u8";if ("videotime" in querys) {adDuration = Number(querys["videotime"]);}url.value = path;} })
</script><style scoped>
</style>

注册路由后访问即可

展示结果

HLS 和 FLV 的对比

特性	HLS	FLV
兼容性	高（支持大多数现代浏览器和设备）	低（需要 `flv.js` 支持）
延迟	较高（10-30 秒）	较低（2-5 秒）
自适应码率	支持	不支持
文件格式	`.m3u8` + `.ts`	`.flv`
部署复杂度	较高（需要分割视频文件）	较低（直接传输 FLV 文件）
适用场景	点播、直播（高延迟）	点播、低延迟直播

选择建议

HLS:
- 适合需要高兼容性和自适应码率的场景（如点播、高延迟直播）。
- 推荐用于 iOS、macOS 和现代浏览器。
FLV:
- 适合低延迟场景（如实时直播）。
- 需要确保用户浏览器支持 flv.js。

总结

HLS 适合高兼容性和自适应码率的场景。
FLV 适合低延迟场景，但需要额外的 JavaScript 库支持。
根据项目需求选择合适的视频传输协议，并结合 FastAPI 实现后端服务。