import base64 import tempfile from pathlib import Path from fastapi import APIRouter, HTTPException, UploadFile, Query from pydantic import BaseModel from .core import get_video_audio_service from plugins.audio.asr_drivers import get_asr_registry from app.plugins.run_log import get_run_log_service, RunStatus router = APIRouter() plugin = None def set_plugin_instance(plugin_instance): global plugin plugin = plugin_instance class VideoBase64Request(BaseModel): video_base64: str suffix: str = ".mp4" language: str | None = None provider: str | None = None @router.get("/status") async def get_status(): if plugin is None: return { "name": "video", "enabled": False, "message": "插件未加载", } return plugin.get_status() @router.post("/upload/transcribe") async def transcribe_video_upload( file: UploadFile, language: str | None = None, provider: str | None = None, wait: bool = Query(False, description="是否同步等待转录完成"), ): """上传视频文件，提取音频并转录 默认异步返回 run_id，设置 wait=true 同步等待结果。 """ if plugin is None or not plugin.enabled: raise HTTPException(status_code=400, detail="插件未启用") # 获取 ASR registry registry = get_asr_registry() provider_name = provider or registry.get_default_provider() driver = registry.get_driver(provider_name) if driver is None: raise HTTPException(status_code=400, detail=f"未知的 ASR provider: {provider_name}") # 检查 provider 可用性 availability = driver.check_availability() if availability.value != "available": raise HTTPException( status_code=400, detail=f"ASR provider {provider_name} 不可用: {availability.value}", ) # 创建 run run_service = get_run_log_service() run = run_service.create_run("video") # 写入上传事件 run_service.add_event( run_id=run.run_id, stage="upload", message=f"视频文件已上传: {file.filename}", detail=f"provider={provider_name}, language={language}", ) # 保存上传文件到临时目录 content = await file.read() suffix = "." + file.filename.rsplit(".", 1)[-1] if "." in file.filename else ".mp4" with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp_file: tmp_file.write(content) tmp_path = tmp_file.name try: # 写入音频提取开始事件 run_service.add_event( run_id=run.run_id, stage="extract_audio", message="开始从视频提取音频", ) # 提取音频 video_service = get_video_audio_service() extraction = video_service.extract_audio(tmp_path) if not extraction["success"]: # 写入提取失败事件 run_service.add_event( run_id=run.run_id, stage="extract_audio", message=f"音频提取失败: {extraction['error']}", level="error", ) run_service.finish_run( run_id=run.run_id, status=RunStatus.FAILED, error=extraction["error"], ) raise HTTPException(status_code=400, detail=extraction["error"]) # 写入音频提取完成事件 run_service.add_event( run_id=run.run_id, stage="extract_audio", message="音频提取完成", ) # 写入模型探测事件 run_service.add_event( run_id=run.run_id, stage="model_probe", message=f"模型探测: {driver.model_id}", detail=f"provider={provider_name}, status={availability.value}", ) # 写入转录开始事件 run_service.add_event( run_id=run.run_id, stage="transcribe", message="开始转录", ) # 执行转录 audio_path = extraction["audio_path"] result = driver.transcribe(audio_path, language) # 清理提取的音频文件 Path(audio_path).unlink(missing_ok=True) if result.success: # 写入转录完成事件 run_service.add_event( run_id=run.run_id, stage="transcribe", message=f"转录完成: {len(result.text)} 字符", detail=f"timings={result.timings}", ) # 完成 run run_service.finish_run( run_id=run.run_id, status=RunStatus.SUCCEEDED, result={ "text": result.text, "segments": result.segments, "provider": result.provider, "model": result.model, "timings": result.timings, }, ) else: # 写入转录失败事件 run_service.add_event( run_id=run.run_id, stage="transcribe", message=f"转录失败: {result.error}", level="error", ) # 完成 run（失败） run_service.finish_run( run_id=run.run_id, status=RunStatus.FAILED, error=result.error, ) # 如果同步等待，直接返回结果 if wait: return { "run_id": run.run_id, "status": "succeeded" if result.success else "failed", "provider": result.provider, "model": result.model, "text": result.text, "segments": result.segments, "timings": result.timings, "error": result.error, } # 异步返回 run_id return { "run_id": run.run_id, "status": "running", "provider": provider_name, "model": driver.model_id, } except HTTPException: raise except Exception as e: # 写入异常事件 run_service.add_event( run_id=run.run_id, stage="transcribe", message=f"转录异常: {str(e)}", level="error", ) run_service.finish_run( run_id=run.run_id, status=RunStatus.FAILED, error=str(e), ) raise HTTPException(status_code=500, detail=f"转录失败: {str(e)}") finally: # 清理临时文件 Path(tmp_path).unlink(missing_ok=True) @router.post("/video-base64/transcribe") async def transcribe_video_base64(request: VideoBase64Request): """Base64 视频转录（兼容旧接口）""" if plugin is None or not plugin.enabled: raise HTTPException(status_code=400, detail="插件未启用") # 获取 ASR registry registry = get_asr_registry() provider_name = request.provider or registry.get_default_provider() driver = registry.get_driver(provider_name) if driver is None: raise HTTPException(status_code=400, detail=f"未知的 ASR provider: {provider_name}") # 检查 provider 可用性 availability = driver.check_availability() if availability.value != "available": raise HTTPException( status_code=400, detail=f"ASR provider {provider_name} 不可用: {availability.value}", ) # 创建 run run_service = get_run_log_service() run = run_service.create_run("video") # 写入上传事件 run_service.add_event( run_id=run.run_id, stage="upload", message="Base64 视频已接收", detail=f"provider={provider_name}, language={request.language}", ) # 解码 Base64 并保存到临时文件 try: video_bytes = base64.b64decode(request.video_base64) except Exception as e: run_service.finish_run(run_id=run.run_id, status=RunStatus.FAILED, error=f"Base64 解码失败: {e}") raise HTTPException(status_code=400, detail=f"Base64 解码失败: {e}") suffix = request.suffix if request.suffix.startswith(".") else f".{request.suffix}" with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp_file: tmp_file.write(video_bytes) tmp_path = tmp_file.name try: # 写入音频提取开始事件 run_service.add_event( run_id=run.run_id, stage="extract_audio", message="开始从视频提取音频", ) # 提取音频 video_service = get_video_audio_service() extraction = video_service.extract_audio(tmp_path) if not extraction["success"]: # 写入提取失败事件 run_service.add_event( run_id=run.run_id, stage="extract_audio", message=f"音频提取失败: {extraction['error']}", level="error", ) run_service.finish_run( run_id=run.run_id, status=RunStatus.FAILED, error=extraction["error"], ) return { "run_id": run.run_id, "status": "failed", "error": extraction["error"], } # 写入音频提取完成事件 run_service.add_event( run_id=run.run_id, stage="extract_audio", message="音频提取完成", ) # 写入模型探测事件 run_service.add_event( run_id=run.run_id, stage="model_probe", message=f"模型探测: {driver.model_id}", detail=f"provider={provider_name}, status={availability.value}", ) # 写入转录开始事件 run_service.add_event( run_id=run.run_id, stage="transcribe", message="开始转录", ) # 执行转录 audio_path = extraction["audio_path"] result = driver.transcribe(audio_path, request.language) # 清理提取的音频文件 Path(audio_path).unlink(missing_ok=True) if result.success: # 写入转录完成事件 run_service.add_event( run_id=run.run_id, stage="transcribe", message=f"转录完成: {len(result.text)} 字符", detail=f"timings={result.timings}", ) # 完成 run run_service.finish_run( run_id=run.run_id, status=RunStatus.SUCCEEDED, result={ "text": result.text, "segments": result.segments, "provider": result.provider, "model": result.model, "timings": result.timings, }, ) else: # 写入转录失败事件 run_service.add_event( run_id=run.run_id, stage="transcribe", message=f"转录失败: {result.error}", level="error", ) # 完成 run（失败） run_service.finish_run( run_id=run.run_id, status=RunStatus.FAILED, error=result.error, ) return { "run_id": run.run_id, "status": "succeeded" if result.success else "failed", "provider": result.provider, "model": result.model, "text": result.text, "segments": result.segments, "timings": result.timings, "error": result.error, } except Exception as e: # 写入异常事件 run_service.add_event( run_id=run.run_id, stage="transcribe", message=f"转录异常: {str(e)}", level="error", ) run_service.finish_run( run_id=run.run_id, status=RunStatus.FAILED, error=str(e), ) raise HTTPException(status_code=500, detail=f"转录失败: {str(e)}") finally: # 清理临时文件 Path(tmp_path).unlink(missing_ok=True)