Python使用FastAPI制作一个视频流媒体平台

from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel
from passlib.context import CryptContext
import jwt
from datetime import datetime, timedelta

app = FastAPI()

# 密码加密上下文
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

# 用户数据模型
class User(BaseModel):
    username: str
    password: str

class Token(BaseModel):
    access_token: str
    token_type: str

# 模拟数据库
fake_users_db = {}

# 密码加密函数
def get_password_hash(password: str):
    return pwd_context.hash(password)

def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)

# 登录认证
@app.post("/login/", response_model=Token)
def login(user: User):
    db_user = fake_users_db.get(user.username)
    if not db_user or not verify_password(user.password, db_user['password']):
        raise HTTPException(status_code=401, detail="Invalid credentials")
    access_token = jwt.encode({"sub": user.username, "exp": datetime.utcnow() + timedelta(hours=1)}, "secret", algorithm="HS256")
    return {"access_token": access_token, "token_type": "bearer"}

from fastapi import UploadFile, File, HTTPException

@app.post("/upload_video/")
async def upload_video(file: UploadFile = File(...)):
    # 文件类型检查
    if not file.content_type.startswith('video'):
        raise HTTPException(status_code=400, detail="File is not a video")
    
    # 文件大小限制（例如限制为100MB）
    if file.size > 100 * 1024 * 1024:
        raise HTTPException(status_code=400, detail="File is too large")
    
    # 文件保存逻辑
    file_location = f"uploads/{file.filename}"
    with open(file_location, "wb") as f:
        f.write(file.file.read())
    
    return {"message": "Video uploaded successfully", "file_location": file_location}

import subprocess
import os

def transcode_video(input_path: str, output_path: str):
    command = [
        "ffmpeg", 
        "-i", input_path, 
        "-vcodec", "libx264", 
        "-acodec", "aac", 
        "-strict", "experimental", 
        output_path
    ]
    subprocess.run(command, check=True)

@app.post("/transcode_video/")
async def transcode_video(file_location: str):
    output_path = file_location.replace(".avi", ".mp4")  # 将AVI转换为MP4
    try:
        transcode_video(file_location, output_path)
        return {"message": "Video transcoded successfully", "output_path": output_path}
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Error during transcoding: {e}")

def save_video(file: UploadFile, file_location: str):
    with open(file_location, "wb") as f:
        f.write(file.file.read())

from fastapi import StreamingResponse
from pathlib import Path

@app.get("/video/{video_id}")
async def stream_video(video_id: int):
    video_path = Path(f"uploads/video{video_id}.mp4")
    if not video_path.exists():
        raise HTTPException(status_code=404, detail="Video not found")
    
    video_file = open(video_path, "rb")
    return StreamingResponse(video_file, media_type="video/mp4")

from pydub import AudioSegment

@app.post("/extract_audio/")
async def extract_audio(file_location: str):
    try:
        video_audio = AudioSegment.from_file(file_location)
        audio_path = file_location.replace(".mp4", ".mp3")
        video_audio.export(audio_path, format="mp3")
        return {"message": "Audio extracted successfully", "audio_path": audio_path}
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Error during audio extraction: {e}")