xiaoxia-saas/packages/ports/asset_repository.py

"""素材仓储接口定义。"""

from abc import ABC, abstractmethod

from packages.domain import Asset


class AssetRepository(ABC):
    @abstractmethod
    def create(self, asset: Asset) -> Asset:
        pass

    @abstractmethod
    def find_by_id(self, asset_id: str) -> Asset | None:
        pass

    @abstractmethod
    def find_by_project(
        self,
        project_id: str,
        skip: int = 0,
        limit: int = 100,
        status: list[str] | None = None,
    ) -> list[Asset]:
        pass

    @abstractmethod
    def find_by_library(
        self,
        library_id: str,
        skip: int = 0,
        limit: int = 100,
        status: list[str] | None = None,
    ) -> list[Asset]:
        pass

    @abstractmethod
    def find_by_library_and_file_type(
        self,
        library_id: str,
        file_type: str,
        skip: int = 0,
        limit: int = 100,
        status: list[str] | None = None,
    ) -> list[Asset]:
        pass

    @abstractmethod
    def update(self, asset: Asset) -> Asset:
        pass

    @abstractmethod
    def delete(self, asset_id: str) -> bool:
        pass

    @abstractmethod
    def batch_delete(self, asset_ids: list[str]) -> int:
        """批量删除素材（软删除，标记 status=deleted），返回实际影响数量。"""
        pass

    @abstractmethod
    def batch_update_metadata(self, asset_ids: list[str], metadata_patch: dict[str, object]) -> int:
        """批量更新素材 metadata（合并 patch），返回实际影响数量。"""
        pass

    @abstractmethod
    def batch_add_tags(self, asset_ids: list[str], tag_ids: list[str]) -> int:
        """批量给素材添加标签（合并去重），返回实际影响数量。"""
        pass

    @abstractmethod
    def batch_replace_tags(self, asset_ids: list[str], tag_ids: list[str]) -> int:
        """批量替换素材标签（全量覆盖），返回实际影响数量。"""
        pass

    @abstractmethod
    def count_by_project(self, project_id: str, status: list[str] | None = None) -> int:
        pass

    @abstractmethod
    def count_by_project_ids(self, project_ids: list[str], status: list[str] | None = None) -> int:
        pass

    @abstractmethod
    def sum_storage_by_project_ids(self, project_ids: list[str]) -> int:
        pass

    @abstractmethod
    def search_candidates(
        self,
        project_id: str,
        *,
        file_type: str | None = None,
        min_quality_score: float | None = None,
        min_duration: float | None = None,
        max_duration: float | None = None,
        classification_category: str | None = None,
        tags: list[str] | None = None,
        status: str | None = None,
        limit: int = 50,
    ) -> list[Asset]:
        """按筛选条件搜索候选素材，按质量分降序排列。"""
        pass

    @abstractmethod
    def find_by_tag_ids(
        self,
        tag_ids: list[str],
        skip: int = 0,
        limit: int = 100,
    ) -> list[Asset]:
        """查找包含所有指定标签的素材。"""
        pass

    @abstractmethod
    def find_by_library_and_file_hash(
        self,
        library_id: str,
        file_hash: str,
    ) -> Asset | None:
        """按素材库 + 文件哈希查找已有素材（去重检测）。"""
        pass