xiaoxia
58ff565c48
CI/CD Pipeline / Validate Code Quality And Tests (push) Successful in 42s
CI/CD Pipeline / Unit Tests (push) Successful in 1m18s
CI/CD Pipeline / Integration Tests (push) Has been cancelled
CI/CD Pipeline / Frontend Lint (push) Has been cancelled
CI/CD Pipeline / Build & Push Staging (Watchtower auto-deploy) (push) Has been cancelled
CI/CD Pipeline / Staging E2E Tests (push) Has been cancelled
CI/CD Pipeline / Staging API Integration Tests (push) Has been cancelled
CI/CD Pipeline / Build Production Runtime Images (push) Has been cancelled
CI/CD Pipeline / Deploy Production (push) Has been cancelled
CI/CD Pipeline / Production Browser E2E (push) Has been cancelled
feat: 素材批量操作接口（软删除/打标签/改分类/智能视图标记）
120 lines
3.1 KiB
Python
Executable File
120 lines
3.1 KiB
Python
Executable File
"""素材仓储接口定义。"""
|
|
|
|
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,
|
|
) -> list[Asset]:
|
|
pass
|
|
|
|
@abstractmethod
|
|
def find_by_library(
|
|
self,
|
|
library_id: str,
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
) -> list[Asset]:
|
|
pass
|
|
|
|
@abstractmethod
|
|
def find_by_library_and_file_type(
|
|
self,
|
|
library_id: str,
|
|
file_type: str,
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
) -> 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) -> int:
|
|
pass
|
|
|
|
@abstractmethod
|
|
def count_by_project_ids(self, project_ids: list[str]) -> 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
|