xiaoxia-saas/apps/api/app/api/routes/asset_libraries.py

from typing import Any

from app.auth import AuthenticatedUser, get_current_user
from app.dependencies import (
    get_asset_library_repository,
    get_asset_repository,
    get_project_repository,
)
from app.schemas.asset_library import (
    AssetLibraryResponse,
    CreateAssetLibraryRequest,
    EnsureDefaultLibraryRequest,
    ListAssetLibrariesResponse,
)
from fastapi import APIRouter, Depends, HTTPException, Query, Response, status

from packages.application import (
    CreateAssetLibraryCommand,
    CreateAssetLibraryUseCase,
    GetProjectUseCase,
    ListAssetLibrariesUseCase,
)
from packages.domain import AssetLibraryKind

from ._helpers import check_project_access

router = APIRouter()


def _to_asset_library_response(item) -> AssetLibraryResponse:
    return AssetLibraryResponse(
        id=item.id,
        project_id=item.project_id,
        name=item.name,
        kind=item.kind.value,
        asset_count=item.asset_count,
        total_size=item.total_size,
    )


@router.get("", response_model=ListAssetLibrariesResponse)
def list_asset_libraries(
    project_id: str | None = Query(None),
    kind: str | None = Query(None, pattern="^(video|voice|image)$"),
    authenticated_user: AuthenticatedUser = Depends(get_current_user),
    asset_library_repository: Any = Depends(get_asset_library_repository),
    project_repository: Any = Depends(get_project_repository),
) -> ListAssetLibrariesResponse:
    user_id = authenticated_user.user.id
    use_case = ListAssetLibrariesUseCase(asset_library_repository)

    if project_id:
        # If project_id provided, check access and filter by project
        project = GetProjectUseCase(project_repository).execute(project_id)
        if project is None:
            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Project not found")
        if not project.can_access(user_id):
            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Access denied to project")
        items = use_case.execute(project_id)
    else:
        # If no project_id, list all libraries from accessible projects
        accessible_projects = project_repository.find_accessible_projects(user_id)
        all_items = []
        for proj in accessible_projects:
            all_items.extend(use_case.execute(proj.id))
        items = all_items

    # 按 kind 过滤（可选）
    if kind:
        kind_enum = AssetLibraryKind(kind)
        items = [item for item in items if item.kind == kind_enum]

    return ListAssetLibrariesResponse(items=[_to_asset_library_response(item) for item in items])


@router.post("", response_model=AssetLibraryResponse)
def create_asset_library(
    request: CreateAssetLibraryRequest,
    authenticated_user: AuthenticatedUser = Depends(get_current_user),
    asset_library_repository: Any = Depends(get_asset_library_repository),
    project_repository: Any = Depends(get_project_repository),
) -> AssetLibraryResponse:
    project = project_repository.find_by_id(request.project_id)
    if project is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Project not found")
    if not project.can_access(authenticated_user.user.id):
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Access denied to project")
    use_case = CreateAssetLibraryUseCase(asset_library_repository)
    item = use_case.execute(
        CreateAssetLibraryCommand(
            project_id=request.project_id,
            name=request.name,
            kind=AssetLibraryKind(request.kind),
        )
    )
    return _to_asset_library_response(item)


# 默认素材库名称映射
_DEFAULT_LIBRARY_NAMES = {
    "video": "视频素材库",
    "voice": "配音素材库",
    "image": "图片素材库",
}


@router.post("/ensure-default", response_model=AssetLibraryResponse)
def ensure_default_library(
    request: EnsureDefaultLibraryRequest,
    authenticated_user: AuthenticatedUser = Depends(get_current_user),
    asset_library_repository: Any = Depends(get_asset_library_repository),
    project_repository: Any = Depends(get_project_repository),
) -> AssetLibraryResponse:
    """确保项目下指定 kind 的默认素材库存在，已存在则直接返回，不存在则自动创建。"""
    project = project_repository.find_by_id(request.project_id)
    if project is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Project not found")
    if not project.can_access(authenticated_user.user.id):
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Access denied to project")

    kind = AssetLibraryKind(request.kind)

    # Issue #1775: 幂等获取/创建——依赖唯一约束 uq_asset_libraries_project_kind，
    # 并发创建冲突时回滚重查返回已有记录，不再依赖应用层"先查后插"，也不会 500。
    default_name = _DEFAULT_LIBRARY_NAMES.get(request.kind, f"{request.kind}素材库")
    library = asset_library_repository.get_or_create_default_library(request.project_id, kind, name=default_name)
    return _to_asset_library_response(library)


@router.delete("/{library_id}", status_code=status.HTTP_204_NO_CONTENT, response_class=Response)
def delete_asset_library(
    library_id: str,
    authenticated_user: AuthenticatedUser = Depends(get_current_user),
    asset_library_repository: Any = Depends(get_asset_library_repository),
    asset_repository: Any = Depends(get_asset_repository),
    project_repository: Any = Depends(get_project_repository),
) -> None:
    """删除素材库，同时删除库内所有素材。"""
    # 查找素材库
    library = asset_library_repository.find_by_id(library_id)
    if library is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="素材库不存在")

    # 权限校验：检查用户是否有项目访问权限
    check_project_access(library.project_id, authenticated_user.user.id, project_repository)

    # 删除库内所有素材（硬删除，素材库已删除，无需保留软删除状态）
    assets_in_library = asset_repository.find_by_library(library_id)
    for asset in assets_in_library:
        asset_repository.delete(asset.id)

    # 删除素材库本身
    asset_library_repository.delete(library_id)