xiaoxia-saas/apps/api/app/schemas/asset.py

from pydantic import BaseModel, Field


class CreateAssetRequest(BaseModel):
    project_id: str | None = Field(default=None, description="可选，不传时从 library.project_id 自动推导")
    library_id: str = Field(..., min_length=1)
    name: str = Field(..., min_length=1, max_length=100)
    storage_key: str = Field(..., min_length=1, max_length=255)
    mime_type: str = Field(..., min_length=1, max_length=100)
    metadata: dict[str, object] = Field(default_factory=dict)
    file_size: int = Field(default=0, ge=0)
    thumbnail_url: str | None = None
    duration: float | None = Field(default=None, ge=0)
    width: int | None = Field(default=None, ge=0)
    height: int | None = Field(default=None, ge=0)
    fps: float | None = Field(default=None, ge=0)
    codec: str | None = None
    status: str = Field(default="uploading")
    classification_status: str = Field(default="pending")
    quality_score: float | None = Field(default=None, ge=0, le=100)
    uploaded_by_user_id: str = Field(default="", max_length=100)


class UpdateAssetReviewRequest(BaseModel):
    review_status: str = Field(..., pattern="^(pending_review|approved|rejected)$")


class UpdateAssetRequest(BaseModel):
    name: str | None = Field(default=None, min_length=1, max_length=100)
    metadata: dict[str, object] | None = None
    tags: list[str] | None = None


class AssetResponse(BaseModel):
    id: str
    project_id: str
    library_id: str
    name: str
    storage_key: str
    mime_type: str
    metadata: dict[str, object]
    file_size: int
    file_url: str | None = None
    thumbnail_url: str | None = None
    duration: float | None = None
    width: int | None = None
    height: int | None = None
    fps: float | None = None
    codec: str | None = None
    status: str
    classification_status: str
    quality_score: float | None = None
    created_at: str
    uploaded_by_user_id: str
    tag_ids: list[str] = Field(default_factory=list)


MAX_BATCH_SIZE = 200


class BatchDeleteRequest(BaseModel):
    """批量删除请求（软删除）。"""

    asset_ids: list[str] = Field(..., min_length=1, max_length=MAX_BATCH_SIZE, description="要删除的素材 ID 列表")


class BatchOperationResponse(BaseModel):
    """批量操作通用响应。"""

    success_count: int = Field(..., ge=0, description="成功数量")
    failed_ids: list[str] = Field(default_factory=list, description="失败的 ID 列表")
    failed_details: dict[str, str] = Field(default_factory=dict, description="失败详情 {asset_id: reason}")


class BatchTagRequest(BaseModel):
    """批量打标签请求。"""

    asset_ids: list[str] = Field(..., min_length=1, max_length=MAX_BATCH_SIZE, description="素材 ID 列表")
    tag_ids: list[str] = Field(..., min_length=1, max_length=50, description="标签 ID 列表")
    mode: str = Field(default="add", pattern="^(add|replace)$", description="add=添加合并，replace=全量替换")


class BatchClassifyRequest(BaseModel):
    """批量修改分类请求。"""

    asset_ids: list[str] = Field(..., min_length=1, max_length=MAX_BATCH_SIZE, description="素材 ID 列表")
    category: str = Field(..., min_length=1, max_length=50, description="内容分类，如 person/scenic/product")


class BatchMarkRequest(BaseModel):
    """批量设置智能视图标记请求。"""

    asset_ids: list[str] = Field(..., min_length=1, max_length=MAX_BATCH_SIZE, description="素材 ID 列表")
    smart_view: str = Field(
        ..., pattern="^(recommended|caution|high_risk)$", description="智能视图标记：recommended/caution/high_risk"
    )


class ListAssetsResponse(BaseModel):
    items: list[AssetResponse]
    total: int = Field(default=0, ge=0)
    skip: int = Field(default=0, ge=0)
    limit: int = Field(default=100, ge=1)


class SmartMatchRequest(BaseModel):
    """智能选素材请求。"""

    library_id: str = Field(..., min_length=1, description="素材库 ID")
    limit: int | None = Field(default=None, ge=1, le=200, description="最大返回数量，不传则返回全部匹配素材")
    kind: str | None = Field(
        default=None,
        pattern="^(video|image|audio)$",
        description="按文件类型过滤，不传则返回所有类型",
    )


class SmartMatchItem(BaseModel):
    """智能选素材结果条目。"""

    asset: AssetResponse
    score: float = Field(..., ge=0, le=100, description="综合得分 0-100")
    breakdown: dict[str, float] = Field(default_factory=dict, description="各维度得分明细")


class SmartMatchResponse(BaseModel):
    """智能选素材响应。"""

    items: list[SmartMatchItem]
    total_candidates: int = Field(default=0, ge=0, description="参与评分的候选素材总数")