Daniil
123 lines
4.3 KiB
Python
123 lines
4.3 KiB
Python
from __future__ import annotations
|
|
|
|
import uuid
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from cpv3.db.session import get_db
|
|
from cpv3.infrastructure.auth import get_current_user
|
|
from cpv3.modules.captions.schemas import (
|
|
CaptionPresetCreate,
|
|
CaptionPresetRead,
|
|
CaptionPresetUpdate,
|
|
CaptionsRequest,
|
|
CaptionsResponse,
|
|
)
|
|
from cpv3.modules.captions.service import CaptionPresetService, generate_captions
|
|
from cpv3.modules.users.models import User
|
|
|
|
router = APIRouter(prefix="/api/captions", tags=["Captions"])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Legacy direct render endpoint
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@router.post("/get_video/", response_model=CaptionsResponse)
|
|
async def get_video(
|
|
body: CaptionsRequest, current_user: User = Depends(get_current_user)
|
|
) -> CaptionsResponse:
|
|
_ = current_user
|
|
result = await generate_captions(
|
|
folder=body.folder,
|
|
video_s3_path=body.video_s3_path,
|
|
transcription=body.transcription,
|
|
)
|
|
return CaptionsResponse(result=result)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Preset CRUD
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@router.get("/presets/", response_model=list[CaptionPresetRead])
|
|
async def list_presets(
|
|
current_user: User = Depends(get_current_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
) -> list[CaptionPresetRead]:
|
|
"""List system presets + user's own presets."""
|
|
service = CaptionPresetService(db)
|
|
presets = await service.list_presets(user_id=current_user.id)
|
|
return [CaptionPresetRead.model_validate(p, from_attributes=True) for p in presets]
|
|
|
|
|
|
@router.post(
|
|
"/presets/",
|
|
response_model=CaptionPresetRead,
|
|
status_code=status.HTTP_201_CREATED,
|
|
)
|
|
async def create_preset(
|
|
body: CaptionPresetCreate,
|
|
current_user: User = Depends(get_current_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
) -> CaptionPresetRead:
|
|
"""Create a user preset."""
|
|
service = CaptionPresetService(db)
|
|
preset = await service.create_preset(user_id=current_user.id, data=body)
|
|
return CaptionPresetRead.model_validate(preset, from_attributes=True)
|
|
|
|
|
|
@router.get("/presets/{preset_id}/", response_model=CaptionPresetRead)
|
|
async def get_preset(
|
|
preset_id: uuid.UUID,
|
|
current_user: User = Depends(get_current_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
) -> CaptionPresetRead:
|
|
"""Get a single preset."""
|
|
_ = current_user
|
|
service = CaptionPresetService(db)
|
|
try:
|
|
preset = await service.get_preset(preset_id=preset_id)
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc))
|
|
return CaptionPresetRead.model_validate(preset, from_attributes=True)
|
|
|
|
|
|
@router.patch("/presets/{preset_id}/", response_model=CaptionPresetRead)
|
|
async def update_preset(
|
|
preset_id: uuid.UUID,
|
|
body: CaptionPresetUpdate,
|
|
current_user: User = Depends(get_current_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
) -> CaptionPresetRead:
|
|
"""Update a user preset (cannot edit system or others' presets)."""
|
|
service = CaptionPresetService(db)
|
|
try:
|
|
preset = await service.update_preset(
|
|
preset_id=preset_id, user_id=current_user.id, data=body
|
|
)
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc))
|
|
except PermissionError as exc:
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=str(exc))
|
|
return CaptionPresetRead.model_validate(preset, from_attributes=True)
|
|
|
|
|
|
@router.delete("/presets/{preset_id}/", status_code=status.HTTP_204_NO_CONTENT)
|
|
async def delete_preset(
|
|
preset_id: uuid.UUID,
|
|
current_user: User = Depends(get_current_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
) -> None:
|
|
"""Delete a user preset (cannot delete system or others' presets)."""
|
|
service = CaptionPresetService(db)
|
|
try:
|
|
await service.delete_preset(preset_id=preset_id, user_id=current_user.id)
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc))
|
|
except PermissionError as exc:
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=str(exc))
|