import hashlib import hmac import logging import mimetypes import shutil import subprocess import uuid from pathlib import Path from typing import Any, BinaryIO, Dict, List, Optional import json from fastapi import ( BackgroundTasks, Depends, FastAPI, File, Form, Header, HTTPException, Path as ApiPath, Query, Request, UploadFile, ) from fastapi.responses import FileResponse, RedirectResponse, Response from fastapi.staticfiles import StaticFiles from starlette.concurrency import run_in_threadpool from starlette.requests import ClientDisconnect from .assessment import AssessmentService from .auth import AuthenticatedUser from .audio_metrics import AudioAnalysisError from .config import Settings from .models import ( AssessmentResult, AuthRequest, AuthResponse, CourseSummary, SentenceBoundaryAdjust, SentenceBoundary, SentenceBoundaryDocument, SentenceTextUpdate, UserDubShareSummary, UserPublic, UserResultSummary, VideoDetailResponse, VideoListResponse, VideoSummary, VideoUploadResponse, ) from .oss import VolcanoOSSUploader from .processing import VideoProcessor, split_sentence_text from .repository import VideoRepository from .store import BoundaryStore from .transcription import MossTranscriber, Transcriber logger = logging.getLogger(__name__) SHA256_PATH = ApiPath( min_length=64, max_length=64, pattern=r"^[A-Fa-f0-9]{64}$", description="SHA-256 hex digest of the exact uploaded video bytes", ) ALLOWED_VIDEO_EXTENSIONS = {".mp4", ".m4v", ".mov", ".mkv", ".webm"} ALLOWED_AUDIO_EXTENSIONS = {".m4a", ".mp4", ".wav"} class UploadTooLargeError(ValueError): pass def create_app( store: Optional[BoundaryStore] = None, *, settings: Optional[Settings] = None, repository: Optional[VideoRepository] = None, transcriber: Optional[Transcriber] = None, oss_uploader: Optional[VolcanoOSSUploader] = None, ) -> FastAPI: service_settings = settings or Settings.from_env() service_settings.ensure_directories() video_repository = repository or VideoRepository(service_settings.database_path) legacy_store = store or BoundaryStore(service_settings.legacy_boundaries_file) moss = transcriber or MossTranscriber( endpoint=service_settings.moss_transcribe_url, model=service_settings.moss_model, timeout_seconds=service_settings.moss_timeout_seconds, max_new_tokens=service_settings.moss_max_new_tokens, ) oss = oss_uploader or VolcanoOSSUploader(service_settings) processor = VideoProcessor(service_settings, video_repository, moss, oss) assessment_service = AssessmentService( video_repository, moss, service_settings.pass_score, ) application = FastAPI( title="Oral Trainer Video Service", version="2.0.0", description="Video ingestion, sentence transcription, streaming, and oral-reading assessment.", ) application.state.settings = service_settings application.state.boundary_store = legacy_store application.state.video_repository = video_repository application.state.transcriber = moss application.state.oss_uploader = oss application.state.video_processor = processor application.state.assessment_service = assessment_service static_dir = Path(__file__).resolve().parent / "static" if static_dir.is_dir(): application.mount("/static", StaticFiles(directory=static_dir), name="static") @application.exception_handler(ClientDisconnect) async def handle_client_disconnect(request: Request, exc: ClientDisconnect) -> Response: logger.warning( "Upload interrupted: the client or an intermediate proxy closed the " "connection before the request body completed." ) return Response(status_code=400, content="Upload connection was interrupted.") def require_admin(x_admin_key: Optional[str] = Header(default=None)) -> None: expected = service_settings.admin_api_key if expected and not hmac.compare_digest(x_admin_key or "", expected): raise HTTPException(status_code=401, detail="A valid X-Admin-Key header is required.") def require_client(x_client_key: Optional[str] = Header(default=None)) -> None: expected = service_settings.client_api_key if expected and not hmac.compare_digest(x_client_key or "", expected): raise HTTPException(status_code=401, detail="A valid X-Client-Key header is required.") def current_user( authorization: Optional[str] = Header(default=None), x_user_token: Optional[str] = Header(default=None), ) -> AuthenticatedUser: token = x_user_token if authorization and authorization.lower().startswith("bearer "): token = authorization[7:].strip() if not token: raise HTTPException(status_code=401, detail="Login required.") row = video_repository.resolve_session(token) if row is None: raise HTTPException(status_code=401, detail="Session expired. Please sign in again.") return AuthenticatedUser(id=row["id"], username=row["username"], nickname=row["nickname"]) def find_document(video_hash: str) -> Optional[SentenceBoundaryDocument]: return video_repository.get_document(video_hash) or legacy_store.get(video_hash) @application.get("/healthz") def healthz() -> Dict[str, Any]: videos = video_repository.list_videos() return { "status": "ok", "video_count": len(videos) + legacy_store.count(), "managed_video_count": len(videos), "moss_configured": moss.available, "volcano_oss_configured": oss.enabled, "scoring_version": "asr-fluency-v1", } @application.post("/api/v1/auth/register", response_model=AuthResponse, status_code=201) def register_user(payload: AuthRequest) -> AuthResponse: user = video_repository.create_user( username=payload.username, password=payload.password, nickname=payload.username, ) if user is None: raise HTTPException(status_code=409, detail="Username is already taken.") token = video_repository.create_session(user["id"]) return AuthResponse(token=token, user=UserPublic(**user)) @application.post("/api/v1/auth/login", response_model=AuthResponse) def login_user(payload: AuthRequest) -> AuthResponse: user = video_repository.authenticate_user(payload.username, payload.password) if user is None: raise HTTPException(status_code=401, detail="Incorrect username or password.") token = video_repository.create_session(user["id"]) return AuthResponse(token=token, user=UserPublic(**user)) @application.get("/api/v1/auth/me", response_model=UserPublic) def get_authenticated_user(user: AuthenticatedUser = Depends(current_user)) -> UserPublic: row = video_repository.get_user_by_id(user.id) if row is None: raise HTTPException(status_code=404, detail="User was not found.") return UserPublic(**row) @application.post("/api/v1/auth/logout", status_code=204) def logout_user( authorization: Optional[str] = Header(default=None), x_user_token: Optional[str] = Header(default=None), ) -> Response: token = x_user_token if authorization and authorization.lower().startswith("bearer "): token = authorization[7:].strip() if token: video_repository.delete_session(token) return Response(status_code=204) @application.get("/api/v1/courses", response_model=List[CourseSummary]) def list_courses(user: AuthenticatedUser = Depends(current_user)) -> List[CourseSummary]: return [ CourseSummary( video_hash=row["video_hash"], title=row["title"], duration_ms=row["duration_ms"], language=row["language"], sentence_count=row["sentence_count"], stream_url=_video_summary(row, service_settings).stream_url, enrolled=bool(row["enrolled"]), ) for row in video_repository.list_courses(user.id) ] @application.post("/api/v1/courses/{video_hash}/enroll", status_code=204) def enroll_course( video_hash: str = SHA256_PATH, user: AuthenticatedUser = Depends(current_user), ) -> Response: if not video_repository.enroll_course(user.id, video_hash): raise HTTPException(status_code=404, detail="Course is not available.") return Response(status_code=204) @application.delete("/api/v1/courses/{video_hash}/enroll", status_code=204) def unenroll_course( video_hash: str = SHA256_PATH, user: AuthenticatedUser = Depends(current_user), ) -> Response: if not video_repository.unenroll_course(user.id, video_hash): raise HTTPException(status_code=404, detail="Enrollment was not found.") return Response(status_code=204) @application.get("/api/v1/me/results", response_model=List[UserResultSummary]) def my_results(user: AuthenticatedUser = Depends(current_user)) -> List[UserResultSummary]: return [UserResultSummary(**row) for row in video_repository.list_user_results(user.id)] @application.get("/api/v1/me/dub-shares", response_model=List[UserDubShareSummary]) def my_dub_shares(user: AuthenticatedUser = Depends(current_user)) -> List[UserDubShareSummary]: return [UserDubShareSummary(**row) for row in video_repository.list_user_dub_shares(user.id)] @application.get("/admin", include_in_schema=False) def admin_page() -> FileResponse: page = static_dir / "admin.html" if not page.is_file(): raise HTTPException(status_code=404, detail="Admin UI is not installed.") return FileResponse(page) @application.get("/api/v1/videos", response_model=VideoListResponse) def list_videos() -> VideoListResponse: return VideoListResponse( videos=[_video_summary(row, service_settings) for row in video_repository.list_videos()] ) @application.get("/api/v1/videos/{video_hash}", response_model=VideoDetailResponse) def get_video(video_hash: str = SHA256_PATH) -> VideoDetailResponse: row = video_repository.get_video(video_hash) if row is None: raise HTTPException(status_code=404, detail="Video was not found.") return VideoDetailResponse( video=_video_summary(row, service_settings), boundaries=video_repository.get_document(video_hash), ) @application.get("/api/v1/videos/{video_hash}/content") def stream_video(video_hash: str = SHA256_PATH): row = video_repository.get_video(video_hash) if row is None: raise HTTPException(status_code=404, detail="Video was not found.") if row.get("remote_url"): return RedirectResponse(row["remote_url"], status_code=307) media_path = service_settings.videos_dir / row["stored_filename"] if not media_path.is_file(): raise HTTPException(status_code=404, detail="Stored video file is missing.") return FileResponse( media_path, media_type=row["content_type"], headers={"Accept-Ranges": "bytes", "Cache-Control": "public, max-age=3600"}, ) @application.get( "/api/v1/videos/{video_hash}/sentence-boundaries", response_model=SentenceBoundaryDocument, dependencies=[Depends(current_user)], ) @application.get( "/api/v1/sentence-boundaries/{video_hash}", response_model=SentenceBoundaryDocument, include_in_schema=False, dependencies=[Depends(current_user)], ) def get_sentence_boundaries(video_hash: str = SHA256_PATH) -> SentenceBoundaryDocument: document = find_document(video_hash) if document is None: raise HTTPException( status_code=404, detail={ "code": "SENTENCE_BOUNDARIES_NOT_FOUND", "message": "No sentence boundaries are registered for this video hash.", }, ) return document @application.post( "/api/v1/admin/videos", response_model=VideoUploadResponse, status_code=202, dependencies=[Depends(require_admin)], ) async def upload_video( background_tasks: BackgroundTasks, file: UploadFile = File(...), title: Optional[str] = Form(default=None), language: Optional[str] = Form(default=None), ) -> VideoUploadResponse: original_name = Path(file.filename or "video.mp4").name suffix = _validate_video_extension(original_name) temporary_path = service_settings.work_dir / f"upload-{uuid.uuid4().hex}.part" try: video_hash, size_bytes = await run_in_threadpool( _persist_upload, file.file, temporary_path, service_settings.max_upload_bytes, ) except UploadTooLargeError as exc: temporary_path.unlink(missing_ok=True) raise HTTPException(status_code=413, detail=str(exc)) from exc except ValueError as exc: temporary_path.unlink(missing_ok=True) raise HTTPException(status_code=400, detail=str(exc)) from exc finally: await file.close() content_type = file.content_type or mimetypes.guess_type(original_name)[0] or "video/mp4" response = _register_uploaded_video( settings=service_settings, repository=video_repository, temporary_path=temporary_path, video_hash=video_hash, size_bytes=size_bytes, original_name=original_name, suffix=suffix, content_type=content_type, title=title, language=language, ) background_tasks.add_task(_process_safely, processor, video_hash) return response @application.put( "/api/v1/admin/videos/raw", response_model=VideoUploadResponse, status_code=202, dependencies=[Depends(require_admin)], ) async def upload_video_stream( request: Request, background_tasks: BackgroundTasks, filename: str = Query(min_length=1, max_length=500), title: Optional[str] = Query(default=None, max_length=200), language: Optional[str] = Query(default=None, max_length=32), ) -> VideoUploadResponse: original_name = Path(filename).name suffix = _validate_video_extension(original_name) content_length = request.headers.get("content-length") if content_length: try: declared_size = int(content_length) except ValueError as exc: raise HTTPException(status_code=400, detail="Invalid Content-Length header.") from exc if declared_size > service_settings.max_upload_bytes: raise HTTPException(status_code=413, detail="Upload exceeds the configured limit.") temporary_path = service_settings.work_dir / f"upload-{uuid.uuid4().hex}.part" try: video_hash, size_bytes = await _persist_request_stream( request, temporary_path, service_settings.max_upload_bytes, ) except UploadTooLargeError as exc: raise HTTPException(status_code=413, detail=str(exc)) from exc except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc response = _register_uploaded_video( settings=service_settings, repository=video_repository, temporary_path=temporary_path, video_hash=video_hash, size_bytes=size_bytes, original_name=original_name, suffix=suffix, content_type=request.headers.get("content-type") or "application/octet-stream", title=title, language=language, ) background_tasks.add_task(_process_safely, processor, video_hash) return response @application.post( "/api/v1/admin/videos/{video_hash}/process", status_code=202, dependencies=[Depends(require_admin)], ) def reprocess_video( background_tasks: BackgroundTasks, video_hash: str = SHA256_PATH, ) -> Dict[str, str]: if video_repository.get_video(video_hash) is None: raise HTTPException(status_code=404, detail="Video was not found.") background_tasks.add_task(_process_safely, processor, video_hash) return {"video_hash": video_hash.lower(), "status": "processing_queued"} @application.put( "/api/v1/admin/videos/{video_hash}/sentences/{sentence_index}", response_model=SentenceBoundary, dependencies=[Depends(require_admin)], ) def update_sentence( payload: SentenceTextUpdate, video_hash: str = SHA256_PATH, sentence_index: int = ApiPath(ge=0), ) -> SentenceBoundary: sentence = video_repository.update_sentence_text( video_hash, sentence_index, payload.text, payload.language, ) if sentence is None: raise HTTPException(status_code=404, detail="Sentence was not found.") return sentence @application.put( "/api/v1/admin/videos/{video_hash}/sentences/{sentence_index}/split", response_model=List[SentenceBoundary], dependencies=[Depends(require_admin)], ) def split_sentence( payload: SentenceTextUpdate, video_hash: str = SHA256_PATH, sentence_index: int = ApiPath(ge=0), ) -> List[SentenceBoundary]: video = video_repository.get_video(video_hash) sentence = video_repository.get_sentence(video_hash, sentence_index) if video is None or sentence is None: raise HTTPException(status_code=404, detail="Sentence was not found.") media_path = service_settings.videos_dir / video["stored_filename"] replacements = split_sentence_text( transcriber=moss, media_path=media_path, sentence=sentence, text=payload.text, language=payload.language if payload.language is not None else sentence.language, ) try: result = video_repository.replace_sentence( video_hash, sentence_index, replacements, ) except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc if result is None: raise HTTPException(status_code=404, detail="Sentence was not found.") return result @application.put( "/api/v1/admin/videos/{video_hash}/sentences/{sentence_index}/boundary", response_model=List[SentenceBoundary], dependencies=[Depends(require_admin)], ) def adjust_sentence_boundary( payload: SentenceBoundaryAdjust, video_hash: str = SHA256_PATH, sentence_index: int = ApiPath(ge=0), ) -> List[SentenceBoundary]: try: result = video_repository.adjust_sentence_boundary( video_hash, sentence_index, payload.delta_ms, ) except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc if result is None: raise HTTPException(status_code=404, detail="Sentence or video was not found.") return result @application.delete( "/api/v1/admin/videos/{video_hash}", status_code=204, dependencies=[Depends(require_admin)], ) def delete_video(video_hash: str = SHA256_PATH) -> Response: row = video_repository.delete_video(video_hash) if row is None: raise HTTPException(status_code=404, detail="Video was not found.") (service_settings.videos_dir / row["stored_filename"]).unlink(missing_ok=True) return Response(status_code=204) @application.post( "/api/v1/videos/{video_hash}/sentences/{sentence_index}/assessments", response_model=AssessmentResult, ) async def assess_sentence( video_hash: str = SHA256_PATH, sentence_index: int = ApiPath(ge=0), audio: UploadFile = File(...), language: Optional[str] = Form(default=None), user: AuthenticatedUser = Depends(current_user), ) -> AssessmentResult: document = find_document(video_hash) if document is None: raise HTTPException(status_code=404, detail="Video or sentence boundaries were not found.") if not moss.available: raise HTTPException(status_code=503, detail="MOSS transcription is not configured.") suffix = Path(audio.filename or "attempt.wav").suffix.lower() or ".wav" temporary_name = f"attempt-{uuid.uuid4().hex}{suffix}" temporary_path = service_settings.attempts_dir / temporary_name completed = False try: await run_in_threadpool( _persist_upload, audio.file, temporary_path, service_settings.max_attempt_bytes, ) result = await run_in_threadpool( assessment_service.assess, document=document, sentence_index=sentence_index, audio_path=temporary_path, language=language, retained_audio_filename=temporary_name if service_settings.keep_attempt_audio else None, user_id=user.id, ) completed = True return result except UploadTooLargeError as exc: raise HTTPException(status_code=413, detail=str(exc)) from exc except LookupError as exc: raise HTTPException(status_code=404, detail=str(exc)) from exc except (ValueError, AudioAnalysisError) as exc: raise HTTPException(status_code=422, detail=str(exc)) from exc except RuntimeError as exc: raise HTTPException(status_code=502, detail=str(exc)) from exc finally: await audio.close() if not service_settings.keep_attempt_audio or not completed: temporary_path.unlink(missing_ok=True) @application.post("/api/v1/dub-shares", status_code=201) async def create_dub_share( request: Request, video_hash: str = Form(...), title: str = Form(default="我的口语配音"), segments: str = Form(...), scores: str = Form(default="[]"), files: List[UploadFile] = File(...), user: AuthenticatedUser = Depends(current_user), ) -> Dict[str, Any]: try: segment_items = json.loads(segments) except (TypeError, ValueError) as exc: raise HTTPException(status_code=400, detail="segments must be valid JSON.") from exc if not isinstance(segment_items, list) or not segment_items: raise HTTPException(status_code=400, detail="At least one dubbing segment is required.") if len(segment_items) != len(files): raise HTTPException(status_code=400, detail="Segment count does not match audio file count.") try: score_items = json.loads(scores) except (TypeError, ValueError) as exc: raise HTTPException(status_code=400, detail="scores must be valid JSON.") from exc if not isinstance(score_items, list): raise HTTPException(status_code=400, detail="scores must be a JSON array.") scores_by_index = { int(item["sentence_index"]): item for item in score_items if isinstance(item, dict) and "sentence_index" in item } document = find_document(video_hash.lower()) if document is None: raise HTTPException(status_code=404, detail="Video or sentence boundaries were not found.") prepared_segments: List[Dict[str, Any]] = [] saved_files: List[Path] = [] try: for item, upload in zip(segment_items, files): sentence_index = int(item["sentence_index"]) boundary = next((s for s in document.sentences if s.index == sentence_index), None) if boundary is None: raise HTTPException(status_code=404, detail=f"Sentence {sentence_index} was not found.") suffix = Path(upload.filename or "dub.m4a").suffix.lower() if suffix == ".mp4": suffix = ".m4a" if suffix not in ALLOWED_AUDIO_EXTENSIONS: raise HTTPException(status_code=415, detail=f"Unsupported audio extension: {suffix or '(none)'}") audio_name = f"{uuid.uuid4().hex}{suffix}" audio_path = service_settings.dub_shares_dir / audio_name await run_in_threadpool(_persist_upload, upload.file, audio_path, 50 * 1024 * 1024) await run_in_threadpool(_align_share_audio, audio_path, boundary.end_ms / 1000, moss) saved_files.append(audio_path) score = scores_by_index.get(sentence_index, {}) score_details = { key: float(score[key]) for key in ( "content_score", "fluency_score", "duration_score", "pause_score", "speech_rate_score", ) if score.get(key) is not None } prepared_segments.append({ "sentence_index": sentence_index, "start_ms": boundary.start_ms, "end_ms": boundary.end_ms, "text": boundary.text, "audio_filename": audio_name, "audio_size_bytes": audio_path.stat().st_size, "overall_score": float(score.get("overall_score", 0)), "recognized_text": str(score.get("recognized_text") or ""), "score_details": json.dumps(score_details, ensure_ascii=False), }) share = video_repository.create_dub_share( video_hash=video_hash.lower(), title=title, segments=prepared_segments, user_id=user.id, ) if share is None: raise HTTPException(status_code=404, detail="Video was not found.") return { **share, "share_url": f"/dub-shares/{share['share_id']}", } except UploadTooLargeError as exc: raise HTTPException(status_code=413, detail=str(exc)) from exc except HTTPException: raise finally: for upload in files: await upload.close() # Files intentionally remain until share deletion because the web page references them. del saved_files @application.get("/api/v1/dub-shares/{share_id}") def get_dub_share(share_id: str) -> Dict[str, Any]: share = video_repository.get_dub_share(share_id) if share is None: raise HTTPException(status_code=404, detail="Dubbing share was not found.") return { "share_id": share["share_id"], "video_hash": share["video_hash"], "title": share["title"], "created_at": share["created_at"], "video_hash": share["video_hash"], "segments": [ { "index": row["sentence_index"], "start_ms": row["start_ms"], "end_ms": row["end_ms"], "text": row["text"], "overall_score": row["overall_score"], "recognized_text": row["recognized_text"], "score_details": json.loads(row["score_details"] or "{}"), "audio_url": f"/api/v1/dub-shares/{share_id}/audio/{row['audio_filename']}", } for row in share["segments"] ], } @application.get("/api/v1/dub-shares/{share_id}/audio/{filename}") def get_dub_share_audio(share_id: str, filename: str) -> FileResponse: share = video_repository.get_dub_share(share_id) if share is None: raise HTTPException(status_code=404, detail="Dubbing share was not found.") safe_name = Path(filename).name row = next((item for item in share["segments"] if item["audio_filename"] == safe_name), None) if row is None: raise HTTPException(status_code=404, detail="Audio was not found.") media_path = service_settings.dub_shares_dir / safe_name if not media_path.is_file(): raise HTTPException(status_code=404, detail="Stored audio file is missing.") return FileResponse(media_path, media_type="audio/mp4", headers={"Cache-Control": "public, max-age=31536000"}) @application.get("/dub-shares/{share_id}", include_in_schema=False) def dub_share_page(share_id: str) -> FileResponse: if video_repository.get_dub_share(share_id) is None: raise HTTPException(status_code=404, detail="Dubbing share was not found.") page = static_dir / "dub-share.html" if not page.is_file(): raise HTTPException(status_code=404, detail="Dubbing share UI is not installed.") return FileResponse(page) return application def _persist_upload(source: BinaryIO, destination: Path, max_bytes: int) -> tuple[str, int]: digest = hashlib.sha256() size = 0 destination.parent.mkdir(parents=True, exist_ok=True) try: with destination.open("wb") as output: while True: chunk = source.read(4 * 1024 * 1024) if not chunk: break size += len(chunk) if size > max_bytes: raise UploadTooLargeError(f"Upload exceeds the {max_bytes}-byte limit.") digest.update(chunk) output.write(chunk) if size == 0: raise ValueError("Uploaded file is empty.") except Exception: destination.unlink(missing_ok=True) raise return digest.hexdigest(), size def _align_share_audio(audio_path: Path, target_seconds: float, transcriber: Transcriber) -> None: ffmpeg = shutil.which("ffmpeg") if ffmpeg is None: return aligned_path = audio_path.with_name(f"{audio_path.stem}-aligned{audio_path.suffix}") try: transcript = transcriber.transcribe(audio_path) speech_segments = [segment for segment in transcript.segments if segment.text.strip()] if not speech_segments: return start_seconds = max(0.0, min(segment.start_seconds for segment in speech_segments) - 0.05) end_seconds = min( _media_duration_seconds(audio_path), max(segment.end_seconds for segment in speech_segments) + 0.58, ) if end_seconds <= start_seconds + 0.1: return filters: List[str] = [ "silenceremove=start_periods=1:start_threshold=-45dB:start_silence=0.03" ] if target_seconds > 0.2: speech_seconds = max(0.1, end_seconds - start_seconds - 0.5) speed_factor = speech_seconds / target_seconds if 0.5 <= speed_factor <= 2.0: filters.append(f"atempo={speed_factor:.6f}") command = [ ffmpeg, "-hide_banner", "-loglevel", "error", "-y", "-ss", f"{start_seconds:.3f}", "-t", f"{end_seconds - start_seconds:.3f}", "-i", str(audio_path), "-af", ",".join(filters), "-c:a", "aac", str(aligned_path), ] subprocess.run(command, check=True) if aligned_path.stat().st_size > 0: aligned_path.replace(audio_path) except Exception: logger.warning("Could not align share audio %s; keeping original.", audio_path.name, exc_info=True) finally: aligned_path.unlink(missing_ok=True) def _media_duration_seconds(audio_path: Path) -> float: ffprobe = shutil.which("ffprobe") if ffprobe is None: return 0.0 try: completed = subprocess.run( [ffprobe, "-v", "error", "-show_entries", "format=duration", "-of", "csv=p=0", str(audio_path)], check=True, capture_output=True, text=True, ) return float(completed.stdout.strip()) except (OSError, ValueError, subprocess.SubprocessError): return 0.0 async def _persist_request_stream( request: Request, destination: Path, max_bytes: int, ) -> tuple[str, int]: digest = hashlib.sha256() size = 0 destination.parent.mkdir(parents=True, exist_ok=True) try: with destination.open("wb") as output: async for chunk in request.stream(): if not chunk: continue size += len(chunk) if size > max_bytes: raise UploadTooLargeError(f"Upload exceeds the {max_bytes}-byte limit.") digest.update(chunk) output.write(chunk) if size == 0: raise ValueError("Uploaded file is empty.") except Exception: destination.unlink(missing_ok=True) raise return digest.hexdigest(), size def _validate_video_extension(filename: str) -> str: suffix = Path(filename).suffix.lower() if suffix not in ALLOWED_VIDEO_EXTENSIONS: raise HTTPException( status_code=415, detail=f"Unsupported video extension: {suffix or '(none)'}", ) return suffix def _register_uploaded_video( *, settings: Settings, repository: VideoRepository, temporary_path: Path, video_hash: str, size_bytes: int, original_name: str, suffix: str, content_type: str, title: Optional[str], language: Optional[str], ) -> VideoUploadResponse: stored_filename = f"{video_hash}{suffix}" destination = settings.videos_dir / stored_filename if destination.exists(): temporary_path.unlink(missing_ok=True) else: temporary_path.replace(destination) repository.upsert_upload( video_hash=video_hash, title=(title or Path(original_name).stem).strip() or video_hash, filename=original_name, stored_filename=stored_filename, content_type=content_type, size_bytes=size_bytes, language=language.strip() if language else None, ) return VideoUploadResponse( video_hash=video_hash, status="uploaded", detail_url=f"/api/v1/videos/{video_hash}", ) def _process_safely(processor: VideoProcessor, video_hash: str) -> None: try: processor.process(video_hash) except Exception: logger.exception("Video processing failed for %s", video_hash) def _video_summary(row: Dict[str, Any], settings: Settings) -> VideoSummary: if row.get("remote_url"): stream_url = row["remote_url"] else: relative_stream_url = f"/api/v1/videos/{row['video_hash']}/content" stream_url = ( f"{settings.public_base_url}{relative_stream_url}" if settings.public_base_url else relative_stream_url ) return VideoSummary( video_hash=row["video_hash"], title=row["title"], filename=row["filename"], content_type=row["content_type"], size_bytes=row["size_bytes"], duration_ms=row["duration_ms"], language=row["language"], status=row["status"], error_message=row["error_message"], sentence_count=row["sentence_count"], stream_url=stream_url, created_at=row["created_at"], updated_at=row["updated_at"], ) app = create_app()