"""Script (口播文案库) CRUD routes — Issue #1795.""" from __future__ import annotations from typing import Optional from app.auth import AuthenticatedUser, get_current_user from app.dependencies import get_db_session from app.schemas.script import ( CreateScriptRequest, ScriptListResponse, ScriptResponse, ScriptSegment, UpdateScriptRequest, ) from app.services.script_service import ScriptNotFoundError, ScriptService from fastapi import APIRouter, Depends, HTTPException, Query, Response, status from sqlalchemy.orm import Session router = APIRouter() def _get_service(session: Session = Depends(get_db_session)) -> ScriptService: return ScriptService(session) def _to_response(script) -> ScriptResponse: segments = script.segments or [] return ScriptResponse( id=script.id, user_id=script.user_id, title=script.title, content=script.content, segments=[ ScriptSegment(text=s.get("text", ""), duration=s.get("duration")) if isinstance(s, dict) else s for s in segments ], tags=script.tags or [], created_at=script.created_at, updated_at=script.updated_at, ) @router.get("", response_model=ScriptListResponse) def list_scripts( skip: int = Query(0, ge=0), limit: int = Query(50, ge=1, le=200), tag: Optional[str] = Query(None, description="按标签筛选"), authenticated_user: AuthenticatedUser = Depends(get_current_user), svc: ScriptService = Depends(_get_service), ) -> ScriptListResponse: user_id = authenticated_user.user.id items, total = svc.list_scripts(user_id, skip=skip, limit=limit, tag=tag) return ScriptListResponse( items=[_to_response(i) for i in items], total=total, ) @router.post("", response_model=ScriptResponse, status_code=status.HTTP_201_CREATED) def create_script( request: CreateScriptRequest, authenticated_user: AuthenticatedUser = Depends(get_current_user), svc: ScriptService = Depends(_get_service), ) -> ScriptResponse: user_id = authenticated_user.user.id script = svc.create_script( user_id=user_id, title=request.title, content=request.content, segments=[s.model_dump() for s in request.segments], tags=request.tags, ) return _to_response(script) @router.get("/{script_id}", response_model=ScriptResponse) def get_script( script_id: str, authenticated_user: AuthenticatedUser = Depends(get_current_user), svc: ScriptService = Depends(_get_service), ) -> ScriptResponse: user_id = authenticated_user.user.id try: script = svc.get_script(script_id, user_id) except ScriptNotFoundError as exc: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Script not found") from exc return _to_response(script) @router.put("/{script_id}", response_model=ScriptResponse) def update_script( script_id: str, request: UpdateScriptRequest, authenticated_user: AuthenticatedUser = Depends(get_current_user), svc: ScriptService = Depends(_get_service), ) -> ScriptResponse: user_id = authenticated_user.user.id try: script = svc.update_script( script_id=script_id, user_id=user_id, title=request.title, content=request.content, segments=[s.model_dump() for s in request.segments] if request.segments is not None else None, tags=request.tags, ) except ScriptNotFoundError as exc: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Script not found") from exc return _to_response(script) @router.delete("/{script_id}", status_code=status.HTTP_204_NO_CONTENT, response_model=None, response_class=Response) def delete_script( script_id: str, authenticated_user: AuthenticatedUser = Depends(get_current_user), svc: ScriptService = Depends(_get_service), ) -> Response: user_id = authenticated_user.user.id deleted = svc.delete_script(script_id, user_id) if not deleted: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Script not found") return