Daniil
145 lines
4.8 KiB
Python
145 lines
4.8 KiB
Python
from __future__ import annotations
|
|
|
|
import uuid
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Response, status
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from cpv3.infrastructure.auth import get_current_user
|
|
from cpv3.infrastructure.deps import get_storage
|
|
from cpv3.infrastructure.storage.base import StorageService
|
|
from cpv3.db.session import get_db
|
|
from cpv3.modules.transcription.schemas import (
|
|
Document,
|
|
GoogleSpeechParams,
|
|
TranscriptionCreate,
|
|
TranscriptionRead,
|
|
TranscriptionUpdate,
|
|
WhisperParams,
|
|
)
|
|
from cpv3.modules.transcription.service import (
|
|
transcribe_with_google_speech,
|
|
transcribe_with_whisper,
|
|
)
|
|
from cpv3.modules.transcription.repository import TranscriptionRepository
|
|
from cpv3.modules.users.models import User
|
|
|
|
router = APIRouter(prefix="/api/transcribe", tags=["Transcription"])
|
|
|
|
|
|
@router.get("/transcriptions/", response_model=list[TranscriptionRead])
|
|
async def list_all_transcriptions(
|
|
current_user: User = Depends(get_current_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
) -> list[TranscriptionRead]:
|
|
_ = current_user
|
|
repo = TranscriptionRepository(db)
|
|
items = await repo.list_all()
|
|
return [TranscriptionRead.model_validate(t) for t in items]
|
|
|
|
|
|
@router.post(
|
|
"/transcriptions/", response_model=TranscriptionRead, status_code=status.HTTP_201_CREATED
|
|
)
|
|
async def create_transcription_entry(
|
|
body: TranscriptionCreate,
|
|
current_user: User = Depends(get_current_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
) -> TranscriptionRead:
|
|
_ = current_user
|
|
repo = TranscriptionRepository(db)
|
|
transcription = await repo.create(body)
|
|
return TranscriptionRead.model_validate(transcription)
|
|
|
|
|
|
@router.get("/transcriptions/{transcription_id}/", response_model=TranscriptionRead)
|
|
async def retrieve_transcription_entry(
|
|
transcription_id: uuid.UUID,
|
|
current_user: User = Depends(get_current_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
) -> TranscriptionRead:
|
|
_ = current_user
|
|
repo = TranscriptionRepository(db)
|
|
transcription = await repo.get_by_id(transcription_id)
|
|
if transcription is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Not found")
|
|
|
|
return TranscriptionRead.model_validate(transcription)
|
|
|
|
|
|
@router.get("/transcriptions/by-artifact/{artifact_id}/", response_model=TranscriptionRead)
|
|
async def retrieve_transcription_by_artifact(
|
|
artifact_id: uuid.UUID,
|
|
current_user: User = Depends(get_current_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
) -> TranscriptionRead:
|
|
_ = current_user
|
|
repo = TranscriptionRepository(db)
|
|
transcription = await repo.get_by_artifact_id(artifact_id)
|
|
if transcription is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Not found")
|
|
|
|
return TranscriptionRead.model_validate(transcription)
|
|
|
|
|
|
@router.patch("/transcriptions/{transcription_id}/", response_model=TranscriptionRead)
|
|
async def patch_transcription_entry(
|
|
transcription_id: uuid.UUID,
|
|
body: TranscriptionUpdate,
|
|
current_user: User = Depends(get_current_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
) -> TranscriptionRead:
|
|
_ = current_user
|
|
repo = TranscriptionRepository(db)
|
|
transcription = await repo.get_by_id(transcription_id)
|
|
if transcription is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Not found")
|
|
|
|
transcription = await repo.update(transcription, body)
|
|
return TranscriptionRead.model_validate(transcription)
|
|
|
|
|
|
@router.delete("/transcriptions/{transcription_id}/", status_code=status.HTTP_204_NO_CONTENT)
|
|
async def delete_transcription_entry(
|
|
transcription_id: uuid.UUID,
|
|
current_user: User = Depends(get_current_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
) -> Response:
|
|
_ = current_user
|
|
repo = TranscriptionRepository(db)
|
|
transcription = await repo.get_by_id(transcription_id)
|
|
if transcription is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Not found")
|
|
|
|
await repo.deactivate(transcription)
|
|
return Response(status_code=status.HTTP_204_NO_CONTENT)
|
|
|
|
|
|
@router.post("/whisper/", response_model=Document)
|
|
async def whisper_transcribe(
|
|
body: WhisperParams,
|
|
current_user: User = Depends(get_current_user),
|
|
storage: StorageService = Depends(get_storage),
|
|
) -> Document:
|
|
_ = current_user
|
|
return await transcribe_with_whisper(
|
|
storage,
|
|
file_key=body.file_path,
|
|
model_name=body.model_name,
|
|
language=body.language,
|
|
)
|
|
|
|
|
|
@router.post("/google-speech/", response_model=Document)
|
|
async def google_speech_transcribe(
|
|
body: GoogleSpeechParams,
|
|
current_user: User = Depends(get_current_user),
|
|
storage: StorageService = Depends(get_storage),
|
|
) -> Document:
|
|
_ = current_user
|
|
return await transcribe_with_google_speech(
|
|
storage,
|
|
file_key=body.file_path,
|
|
language_codes=body.language_codes,
|
|
)
|