Xiaoxia AI
44 lines
1.7 KiB
Python
44 lines
1.7 KiB
Python
from __future__ import annotations
|
|
|
|
from packages.domain import GeneratedVideo
|
|
from packages.ports.generated_video_repository import GeneratedVideoRepository
|
|
|
|
|
|
class ListGeneratedVideosUseCase:
|
|
def __init__(self, generated_video_repository: GeneratedVideoRepository):
|
|
self.generated_video_repository = generated_video_repository
|
|
|
|
def execute(self, project_id: str) -> list[GeneratedVideo]:
|
|
if not project_id.strip():
|
|
raise ValueError("project_id 不能为空")
|
|
return self.generated_video_repository.list_by_project(project_id.strip())
|
|
|
|
|
|
class GetGeneratedVideoUseCase:
|
|
def __init__(self, generated_video_repository: GeneratedVideoRepository):
|
|
self.generated_video_repository = generated_video_repository
|
|
|
|
def execute(self, video_id: str) -> GeneratedVideo | None:
|
|
return self.generated_video_repository.get(video_id)
|
|
|
|
|
|
class ListGeneratedVideosByTaskUseCase:
|
|
def __init__(self, generated_video_repository: GeneratedVideoRepository):
|
|
self.generated_video_repository = generated_video_repository
|
|
|
|
def execute(self, generation_task_id: str) -> list[GeneratedVideo]:
|
|
if not generation_task_id.strip():
|
|
raise ValueError("generation_task_id 不能为空")
|
|
return self.generated_video_repository.list_by_generation_task(generation_task_id.strip())
|
|
|
|
|
|
class GetGeneratedVideoDownloadUrlUseCase:
|
|
def __init__(self, generated_video_repository: GeneratedVideoRepository):
|
|
self.generated_video_repository = generated_video_repository
|
|
|
|
def execute(self, video_id: str) -> str | None:
|
|
item = self.generated_video_repository.get(video_id)
|
|
if item is None:
|
|
return None
|
|
return item.file_url
|