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

from pydantic import BaseModel, Field


class UploadAssetRequest(BaseModel):
    """素材上传请求（multipart form）"""

    project_id: str = Field(..., min_length=1, description="项目 ID")
    library_id: str = Field(..., min_length=1, description="素材库 ID")
    file_hash: str = Field(default="", max_length=64, description="文件 MD5 哈希，用于去重检测")


class DirectUploadPrepareRequest(BaseModel):
    project_id: str = Field(..., min_length=1)
    library_id: str = Field(..., min_length=1)
    filename: str = Field(..., min_length=1, max_length=255)
    content_type: str = Field(default="application/octet-stream", min_length=1, max_length=100)
    file_size: int = Field(..., gt=0)
    file_hash: str = Field(default="", max_length=64, description="文件 MD5 哈希，用于去重检测")
    client_upload_id: str = Field(default="", max_length=64, description="客户端幂等 token（同一次上传的重试保持一致）")


class DirectUploadPrepareResponse(BaseModel):
    upload_url: str
    method: str
    storage_key: str
    expires_at: str
    fields: dict[str, str]
    max_size_bytes: int
    duplicated: bool = False
    skip_transfer: bool = False
    asset_id: str = ""
    # duplicated=true 时填充已存在素材的公网 URL，前端可直接用而不必再调 complete
    url: str = Field(default="", description="duplicated=true 时已存在素材的公网 URL")


class DirectUploadCompleteRequest(BaseModel):
    project_id: str = Field(..., min_length=1)
    library_id: str = Field(..., min_length=1)
    storage_key: str = Field(..., min_length=1, max_length=255)
    file_hash: str = Field(default="", max_length=64, description="文件哈希，用于去重检测")
    client_upload_id: str = Field(default="", max_length=64, description="客户端幂等 token（同一次上传的重试保持一致）")
    file_size: int = Field(default=0, ge=0, description="文件大小（字节），用于无 hash 时的兜底去重")


class DirectUploadCompleteResponse(BaseModel):
    storage_key: str
    ingest_job_id: str
    duplicated: bool = Field(default=False, description="是否为重复素材/重复 complete（命中幂等去重）")
    asset_id: str = Field(default="", description="素材 asset_id（重复 complete 时返回已存在记录）")
    url: str = Field(default="", description="Public URL of uploaded file")


class UploadAssetResponse(BaseModel):
    storage_key: str
    ingest_job_id: str
    url: str = Field(..., description="Public URL of uploaded file")
    duplicated: bool = Field(default=False, description="是否为重复素材/重复提交（命中幂等去重）")
    asset_id: str = Field(default="", description="素材 asset_id（重复提交时返回已存在记录）")