Xiaoxia AI
98 lines
3.7 KiB
Python
98 lines
3.7 KiB
Python
from fastapi import APIRouter, Depends, HTTPException
|
|
|
|
from app.core.celery_app import celery_app
|
|
from app.dependencies import get_generation_task_repository, get_generated_video_repository
|
|
from app.schemas.generation_task import CreateGenerationTaskRequest, GenerationTaskResponse
|
|
from app.schemas.generated_video import GeneratedVideoResponse, ListGeneratedVideosResponse
|
|
from packages.adapters.sqlalchemy_impl import SQLAlchemyGeneratedVideoRepository, SQLAlchemyGenerationTaskRepository
|
|
from packages.application import (
|
|
CreateGenerationTaskCommand,
|
|
CreateGenerationTaskUseCase,
|
|
GetGenerationTaskUseCase,
|
|
ListGeneratedVideosByTaskUseCase,
|
|
)
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.post("/tasks", response_model=GenerationTaskResponse)
|
|
def create_generation_task(
|
|
request: CreateGenerationTaskRequest,
|
|
generation_task_repository: SQLAlchemyGenerationTaskRepository = Depends(get_generation_task_repository),
|
|
) -> GenerationTaskResponse:
|
|
use_case = CreateGenerationTaskUseCase(generation_task_repository)
|
|
task = use_case.execute(
|
|
CreateGenerationTaskCommand(
|
|
workspace_id=request.workspace_id,
|
|
project_id=request.project_id,
|
|
asset_library_id=request.asset_library_id,
|
|
strategy_id=request.strategy_id,
|
|
voice_library_id=request.voice_library_id,
|
|
created_by_user_id=request.created_by_user_id,
|
|
)
|
|
)
|
|
celery_app.send_task("worker.generate_video", args=[task.id])
|
|
return GenerationTaskResponse(
|
|
id=task.id,
|
|
workspace_id=task.workspace_id,
|
|
project_id=task.project_id,
|
|
asset_library_id=task.asset_library_id,
|
|
strategy_id=task.strategy_id,
|
|
voice_library_id=task.voice_library_id,
|
|
status=task.status.value,
|
|
progress=task.progress,
|
|
result_count=task.result_count,
|
|
error_message=task.error_message,
|
|
)
|
|
|
|
|
|
@router.get("/tasks/{task_id}", response_model=GenerationTaskResponse)
|
|
def get_generation_task(
|
|
task_id: str,
|
|
generation_task_repository: SQLAlchemyGenerationTaskRepository = Depends(get_generation_task_repository),
|
|
) -> GenerationTaskResponse:
|
|
use_case = GetGenerationTaskUseCase(generation_task_repository)
|
|
task = use_case.execute(task_id)
|
|
if task is None:
|
|
raise HTTPException(status_code=404, detail=f"GenerationTask {task_id} not found")
|
|
return GenerationTaskResponse(
|
|
id=task.id,
|
|
workspace_id=task.workspace_id,
|
|
project_id=task.project_id,
|
|
asset_library_id=task.asset_library_id,
|
|
strategy_id=task.strategy_id,
|
|
voice_library_id=task.voice_library_id,
|
|
status=task.status.value,
|
|
progress=task.progress,
|
|
result_count=task.result_count,
|
|
error_message=task.error_message,
|
|
)
|
|
|
|
|
|
@router.get("/tasks/{task_id}/results", response_model=ListGeneratedVideosResponse)
|
|
def list_generation_results(
|
|
task_id: str,
|
|
generated_video_repository: SQLAlchemyGeneratedVideoRepository = Depends(get_generated_video_repository),
|
|
) -> ListGeneratedVideosResponse:
|
|
use_case = ListGeneratedVideosByTaskUseCase(generated_video_repository)
|
|
items = use_case.execute(task_id)
|
|
return ListGeneratedVideosResponse(
|
|
items=[
|
|
GeneratedVideoResponse(
|
|
id=item.id,
|
|
workspace_id=item.workspace_id,
|
|
project_id=item.project_id,
|
|
generation_task_id=item.generation_task_id,
|
|
name=item.name,
|
|
file_url=item.file_url,
|
|
file_size=item.file_size,
|
|
duration=item.duration,
|
|
thumbnail_url=item.thumbnail_url,
|
|
width=item.width,
|
|
height=item.height,
|
|
fps=item.fps,
|
|
)
|
|
for item in items
|
|
]
|
|
)
|