import os# 服务器配置HOST = "0.0.0.0"PORT = 8000# 视频存储配置BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))VIDEO_STORAGE_PATH = os.path.join(BASE_DIR, "video-storage/")TEMP_SHARD_PATH = os.path.join(VIDEO_STORAGE_PATH, "temp/")  # 临时分片目录HLS_STORAGE_PATH = os.path.join(VIDEO_STORAGE_PATH, "hls/")  # HLS转码后存储目录# 上传配置MAX_SHARD_SIZE = 2 * 1024 * 1024  # 分片大小（2MB）ALLOWED_VIDEO_TYPES = ["video/mp4", "video/avi", "video/mkv", "video/mov", "video/webm"]  # 允许上传的视频类型# 权限配置（示例：有效Token列表，生产环境需对接数据库）VALID_TOKENS = {"user_token_123", "admin_token_456"}# 创建目录（确保目录存在）for path in [VIDEO_STORAGE_PATH, TEMP_SHARD_PATH, HLS_STORAGE_PATH]:
    if not os.path.exists(path):
        os.makedirs(path, mode=0o755)

import osimport mimetypesfrom app.config import VIDEO_STORAGE_PATH, HLS_STORAGE_PATHdef get_video_mime_type(file_path: str) -> str:
    """获取文件MIME类型"""
    mime_type, _ = mimetypes.guess_type(file_path)
    if not mime_type:
        if file_path.endswith(".m3u8"):
            return "application/x-mpegURL"
        elif file_path.endswith(".ts"):
            return "video/MP2T"
        else:
            return "application/octet-stream"
    return mime_typedef get_video_abs_path(file_name: str = None, hls_path: str = None) -> str:
    """获取视频文件绝对路径（支持普通视频和HLS文件）"""
    if file_name:
        return os.path.join(VIDEO_STORAGE_PATH, file_name)
    elif hls_path:
        return os.path.join(HLS_STORAGE_PATH, hls_path)
    raise ValueError("必须提供file_name或hls_path")

from fastapi import Header, HTTPExceptionfrom app.config import VALID_TOKENSdef verify_token(token: str = Header(None)):
    """验证播放权限（通过Token）"""
    if not token or token not in VALID_TOKENS:
        raise HTTPException(status_code=403, detail="无播放权限，请提供有效Token")
    return token

import osimport hashlibimport asynciofrom fastapi import UploadFile, HTTPExceptionfrom app.config import TEMP_SHARD_PATH, VIDEO_STORAGE_PATH, HLS_STORAGE_PATH, MAX_SHARD_SIZEfrom app.services.transcode import generate_hlsasync def upload_video_shard(
    file: UploadFile,
    file_name: str,
    chunk_index: int,
    total_chunks: int) -> dict:
    """分片上传视频，所有分片上传完成后合并并转码"""
    # 验证文件类型
    if file.content_type not in ["video/mp4", "video/avi", "video/mkv", "video/mov", "video/webm"]:
        raise HTTPException(status_code=400, detail="不支持的视频格式")
    
    # 生成唯一文件标识（避免文件名重复）
    file_hash = hashlib.md5((file_name + str(os.path.getsize(file.file))).encode()).hexdigest()
    shard_dir = os.path.join(TEMP_SHARD_PATH, file_hash + "/")
    
    # 确保分片目录存在
    if not os.path.exists(shard_dir):
        os.makedirs(shard_dir, mode=0o755)
    
    # 保存当前分片（异步写入文件）
    shard_file_path = os.path.join(shard_dir, str(chunk_index))
    try:
        async with aiofiles.open(shard_file_path, "wb") as f:
            content = await file.read()
            await f.write(content)
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"分片保存失败：{str(e)}")
    
    # 检查是否所有分片上传完成
    if chunk_index == total_chunks - 1:
        # 合并分片
        final_file_ext = file_name.split(".")[-1]
        final_file_name = f"{file_hash}.{final_file_ext}"
        final_file_path = os.path.join(VIDEO_STORAGE_PATH, final_file_name)
        
        try:
            # 按索引顺序合并所有分片
            async with aiofiles.open(final_file_path, "wb") as final_f:
                for i in range(total_chunks):
                    current_shard_path = os.path.join(shard_dir, str(i))
                    if not os.path.exists(current_shard_path):
                        raise HTTPException(status_code=500, detail=f"分片{i}缺失，合并失败")
                    async with aiofiles.open(current_shard_path, "rb") as shard_f:
                        await final_f.write(await shard_f.read())
                    # 删除临时分片
                    os.remove(current_shard_path)
            # 删除临时分片目录
            os.rmdir(shard_dir)
        except Exception as e:
            raise HTTPException(status_code=500, detail=f"分片合并失败：{str(e)}")
        
        # 后台异步转码为HLS格式（不阻塞当前请求）
        asyncio.create_task(generate_hls(final_file_path, os.path.join(HLS_STORAGE_PATH, file_hash + "/")))
        
        # 返回播放地址
        play_url = f"http://{os.getenv('SERVER_HOST', 'localhost')}:{os.getenv('SERVER_PORT', 8000)}/api/video/play?file_name={final_file_name}"
        hls_play_url = f"http://{os.getenv('SERVER_HOST', 'localhost')}:{os.getenv('SERVER_PORT', 8000)}/api/video/play?hls_path={file_hash}/index.m3u8"
        return {
            "code": 200,
            "msg": "上传完成，转码中（HLS格式稍后可用）",
            "data": {
                "play_url": play_url,
                "hls_play_url": hls_play_url,
                "file_name": final_file_name            }
        }
    
    return {"code": 200, "msg": f"分片{chunk_index}上传成功", "data": {}}

import osimport subprocessfrom app.config import HLS_STORAGE_PATHasync def generate_hls(source_path: str, target_dir: str):
    """使用FFmpeg将视频转码为HLS格式（.m3u8 + ts分片）"""
    # 确保目标目录存在
    if not os.path.exists(target_dir):
        os.makedirs(target_dir, mode=0o755)
    
    # FFmpeg命令：10秒分片，支持自适应码率，无分片数量限制
    cmd = [
        "ffmpeg",
        "-i", source_path,
        "-profile:v", "baseline",  # 兼容移动端
        "-level", "3.0",
        "-hls_time", "10",  # 分片时长（10秒）
        "-hls_list_size", "0",  # 0表示无限制，显示所有分片
        "-hls_segment_filename", os.path.join(target_dir, "segment_%03d.ts"),  # ts分片命名格式
        "-f", "hls",
        os.path.join(target_dir, "index.m3u8")  # 输出m3u8索引文件
    ]
    
    try:
        # 执行命令（后台运行，捕获输出）
        subprocess.run(
            cmd,
            check=True,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True
        )
        print(f"HLS转码成功：{target_dir}index.m3u8")
    except subprocess.CalledProcessError as e:
        print(f"HLS转码失败：{e.stderr}")
        # 可选：删除失败的转码文件
        if os.path.exists(target_dir):
            for file in os.listdir(target_dir):
                os.remove(os.path.join(target_dir, file))
            os.rmdir(target_dir)

import osfrom fastapi import HTTPException, Response, statusfrom fastapi.responses import StreamingResponsefrom app.utils.file_utils import get_video_abs_path, get_video_mime_typeasync def stream_video(file_name: str = None, hls_path: str = None):
    """流式播放视频（支持Range请求，断点续传）"""
    # 获取视频绝对路径
    try:
        video_path = get_video_abs_path(file_name, hls_path)
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))
    
    # 检查文件是否存在
    if not os.path.exists(video_path) or not os.path.isfile(video_path):
        raise HTTPException(status_code=404, detail="视频文件不存在")
    
    file_size = os.path.getsize(video_path)
    mime_type = get_video_mime_type(video_path)
    
    # 解析Range请求头（格式：bytes=0-1023 或 bytes=1024-）
    range_header = os.getenv("HTTP_RANGE", "")
    start = 0
    end = file_size - 1
    
    if range_header.startswith("bytes="):
        range_str = range_header.split("=")[1]
        range_parts = range_str.split("-")
        start = int(range_parts[0]) if range_parts[0] else 0
        end = int(range_parts[1]) if len(range_parts) > 1 and range_parts[1] else file_size - 1
        # 修正范围（避免超出文件大小）
        start = min(start, end)
        end = min(end, file_size - 1)
    
    # 计算响应长度
    response_length = end - start + 1
    
    # 构建响应头
    headers = {
        "Content-Type": mime_type,
        "Content-Length": str(response_length),
        "Content-Range": f"bytes {start}-{end}/{file_size}",
        "Accept-Ranges": "bytes",
        "Cache-Control": "public, max-age=86400"  # 缓存1天
    }
    
    # 流式读取文件并返回
    async def file_stream():
        async with aiofiles.open(video_path, "rb") as f:
            await f.seek(start)
            remaining = response_length
            chunk_size = 4096  # 4KB缓冲区
            while remaining > 0:
                read_size = min(chunk_size, remaining)
                data = await f.read(read_size)
                if not data:
                    break
                yield data
                remaining -= read_size    
    return StreamingResponse(
        file_stream(),
        status_code=status.HTTP_206_PARTIAL_CONTENT,
        headers=headers    )

from fastapi import FastAPI, UploadFile, File, Form, Query, Dependsfrom fastapi.middleware.cors import CORSMiddlewarefrom app.config import HOST, PORTfrom app.services.upload import upload_video_shardfrom app.services.play import stream_videofrom app.utils.auth_utils import verify_token# 创建FastAPI应用app = FastAPI(title="Python 在线视频播放系统", version="1.0.0")# 允许跨域（开发环境，生产环境限制具体域名）app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],)# 健康检查接口@app.get("/health")async def health_check():
    return {"status": "healthy", "message": "视频服务正常运行"}# 视频分片上传接口@app.post("/api/video/upload/shard", summary="分片上传视频")async def upload_shard(
    file: UploadFile = File(...),
    file_name: str = Form(...),
    chunk_index: int = Form(...),
    total_chunks: int = Form(...)):
    return await upload_video_shard(file, file_name, chunk_index, total_chunks)# 视频播放接口（带权限验证）@app.get("/api/video/play", summary="流式播放视频（支持断点续传）")async def play_video(
    file_name: str = Query(None, description="普通视频文件名（如xxx.mp4）"),
    hls_path: str = Query(None, description="HLS文件路径（如xxx/index.m3u8）"),
    token: str = Depends(verify_token)):
    return await stream_video(file_name, hls_path)# 运行服务if __name__ == "__main__":
    import uvicorn
    uvicorn.run("app.main:app", host=HOST, port=PORT, reload=True)  # reload=True仅用于开发环境