xiaoxia-saas/packages/application/generated_videos.py

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 ListGeneratedVideosPaginatedUseCase:
    def __init__(self, generated_video_repository: GeneratedVideoRepository):
        self.generated_video_repository = generated_video_repository

    def execute(
        self,
        *,
        user_id: str | None = None,
        project_id: str | None = None,
        status: str | None = None,
        review_status: str | None = None,
        page: int = 1,
        page_size: int = 20,
    ) -> tuple[list[GeneratedVideo], int]:
        if page < 1:
            page = 1
        if page_size < 1 or page_size > 100:
            page_size = 20
        return self.generated_video_repository.list_paginated(
            user_id=user_id,
            project_id=project_id,
            status=status,
            review_status=review_status,
            page=page,
            page_size=page_size,
        )


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


class UpdateVideoReviewStatusUseCase:
    def __init__(self, generated_video_repository: GeneratedVideoRepository):
        self.generated_video_repository = generated_video_repository

    def execute(self, video_id: str, review_status: str) -> GeneratedVideo | None:
        if not video_id.strip():
            raise ValueError("video_id 不能为空")
        if review_status not in ("pending_review", "approved", "rejected"):
            raise ValueError(f"无效的 review_status: {review_status}")
        return self.generated_video_repository.update_review_status(video_id.strip(), review_status)


class GetVideosByIdsUseCase:
    def __init__(self, generated_video_repository: GeneratedVideoRepository):
        self.generated_video_repository = generated_video_repository

    def execute(self, video_ids: list[str]) -> list[GeneratedVideo]:
        return self.generated_video_repository.get_by_ids(video_ids)