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

import logging
from typing import Any, Optional

from app.api.routes._helpers import check_project_access, format_utc_datetime
from app.auth import AuthenticatedUser, get_current_user
from app.core.storage import get_storage_service
from app.dependencies import (
    get_asset_library_repository,
    get_asset_repository,
    get_project_repository,
    get_tag_repository,
)
from app.schemas.asset import (
    AssetResponse,
    BatchClassifyRequest,
    BatchDeleteRequest,
    BatchMarkRequest,
    BatchOperationResponse,
    BatchTagRequest,
    CreateAssetRequest,
    ListAssetsResponse,
    UpdateAssetRequest,
    UpdateAssetReviewRequest,
)
from app.schemas.tag import TagAssetsRequest
from fastapi import APIRouter, Depends, HTTPException, Query, Response

from packages.application import (
    CreateAssetCommand,
    CreateAssetUseCase,
)
from packages.domain import AssetStatus, ClassificationStatus

logger = logging.getLogger(__name__)

router = APIRouter()


def _to_asset_response(item, storage_service=None) -> AssetResponse:
    # 生成签名文件 URL（用于视频播放 / 文件下载）
    file_url = None
    if item.storage_key:
        try:
            svc = storage_service or get_storage_service()
            file_url = svc.get_download_url(item.storage_key)
        except Exception:
            logger.warning("生成签名URL失败: storage_key=%s", item.storage_key, exc_info=True)
            file_url = None

    # 缩略图：优先用已有 thumbnail_url，否则对视频素材复用文件签名 URL
    thumbnail_url = item.thumbnail_url
    if not thumbnail_url and item.mime_type and item.mime_type.startswith("video") and file_url:
        thumbnail_url = file_url

    return AssetResponse(
        id=item.id,
        project_id=item.project_id,
        library_id=item.library_id,
        name=item.name,
        storage_key=item.storage_key,
        mime_type=item.mime_type,
        metadata=item.metadata,
        file_size=item.file_size,
        file_url=file_url,
        thumbnail_url=thumbnail_url,
        duration=item.duration,
        width=item.width,
        height=item.height,
        fps=item.fps,
        codec=item.codec,
        status=item.status.value,
        classification_status=item.classification_status.value,
        quality_score=item.quality_score,
        created_at=format_utc_datetime(item.created_at),
        uploaded_by_user_id=item.uploaded_by_user_id,
        tag_ids=getattr(item, "tag_ids", []),
    )


@router.get("", response_model=ListAssetsResponse)
def list_assets(
    library_id: Optional[str] = Query(None),
    project_id: Optional[str] = Query(None),
    kind: Optional[str] = Query(None, pattern="^(video|voice|image)$"),
    keyword: Optional[str] = Query(None, description="按名称模糊匹配"),
    gender: Optional[str] = Query(None, description="按 metadata.gender 筛选"),
    style: Optional[str] = Query(None, description="按 metadata.style 筛选"),
    tag_ids: Optional[str] = Query(None, description="按标签 ID 筛选（逗号分隔，取交集）"),
    smart_view: Optional[str] = Query(
        None,
        description="智能视图筛选：recommended=推荐(质量分≥80)、cautious=慎用(60-79)、risky=高风险(<60或已驳回)、unused=未使用、used=已使用、pending_review=待复核",
        pattern="^(recommended|cautious|risky|unused|used|pending_review)$",
    ),
    classification: Optional[str] = Query(
        None,
        description="按内容分类筛选：scenic=风景、product=产品、person=人物、animal=动物、food=美食、tech=科技、sport=运动、music=音乐、other=其他",
    ),
    status: Optional[str] = Query(
        "default",
        description="按状态筛选，逗号分隔多值；默认返回除deleted外的所有状态；传deleted查看回收站；传all返回所有状态",
    ),
    page: Optional[int] = Query(None, ge=1, description="页码，从1开始；与 page_size 配对使用，优先于 skip/limit"),
    page_size: Optional[int] = Query(None, ge=1, le=500, description="每页数量；与 page 配对使用"),
    skip: int = Query(0, ge=0),
    limit: int = Query(100, ge=1, le=500),
    authenticated_user: AuthenticatedUser = Depends(get_current_user),
    asset_repository: Any = Depends(get_asset_repository),
    asset_library_repository: Any = Depends(get_asset_library_repository),
    project_repository: Any = Depends(get_project_repository),
) -> ListAssetsResponse:
    user_id = authenticated_user.user.id

    # ── 分页：page/page_size 优先于 skip/limit
    if page is not None and page_size is not None:
        skip = (page - 1) * page_size
        limit = page_size

    # ── 解析 status 过滤
    status_list: list[str] | None
    if status and status.lower() == "all":
        status_list = None  # None = 不过滤，返回所有状态
    elif status and status.lower() == "deleted":
        status_list = ["deleted"]  # 仅查回收站
    elif status and status.lower() == "default":
        status_list = ["ready", "uploading", "processing", "error"]  # 默认排除deleted
    elif status:
        status_list = [s.strip() for s in status.split(",") if s.strip()]
        if not status_list:
            status_list = ["ready", "uploading", "processing", "error"]
    else:
        status_list = ["ready", "uploading", "processing", "error"]

    # kind → file_type 映射（voice 对应 audio）
    kind_to_file_type = {"video": "video", "voice": "audio", "image": "image"}

    # 解析 tag_ids 参数（逗号分隔）
    filter_tag_ids: list[str] | None = None
    if tag_ids:
        filter_tag_ids = [t.strip() for t in tag_ids.split(",") if t.strip()]
        if not filter_tag_ids:
            filter_tag_ids = None

    # 需要内存过滤的标志（keyword/gender/style/tag_ids/smart_view/classification 无法在 DB 层过滤）
    needs_memory_filter = bool(keyword or gender or style or filter_tag_ids or smart_view or classification)

    def _apply_memory_filters(items):
        """应用 keyword / gender / style / tag_ids / smart_view / classification 内存过滤。"""
        result = items
        if keyword:
            kw = keyword.lower()
            result = [i for i in result if kw in (i.name or "").lower()]
        if gender:
            result = [i for i in result if (i.metadata or {}).get("gender") == gender]
        if style:
            result = [i for i in result if (i.metadata or {}).get("style") == style]
        if classification:
            result = [i for i in result if (i.metadata or {}).get("classification") == classification]
        if filter_tag_ids:
            tag_set = set(filter_tag_ids)
            result = [i for i in result if tag_set.issubset(set(getattr(i, "tag_ids", [])))]
        if smart_view:

            def __meta(a):
                return a.metadata or {}

            def __use_count(a):
                return int(__meta(a).get("generation_use_count") or 0)

            def __review_status(a):
                return __meta(a).get("review_status", "")

            if smart_view == "recommended":
                result = [i for i in result if i.quality_score is not None and i.quality_score >= 80]
            elif smart_view == "cautious":
                result = [i for i in result if i.quality_score is not None and 60 <= i.quality_score < 80]
            elif smart_view == "risky":
                result = [
                    i
                    for i in result
                    if (i.quality_score is not None and i.quality_score < 60) or __review_status(i) == "rejected"
                ]
            elif smart_view == "unused":
                result = [i for i in result if __use_count(i) == 0]
            elif smart_view == "used":
                result = [i for i in result if __use_count(i) > 0]
            elif smart_view == "pending_review":
                result = [i for i in result if __review_status(i) == "pending_review"]
        return result

    # ── 优化路径：无内存过滤时，使用 DB 级分页 ──
    if not needs_memory_filter:
        ft = kind_to_file_type.get(kind) if kind else None

        # 模式1：指定 library_id
        if library_id:
            library = asset_library_repository.get(library_id)
            if library is None:
                raise HTTPException(status_code=404, detail=f"AssetLibrary {library_id} not found")
            check_project_access(library.project_id, user_id, project_repository)
            if ft:
                items = asset_repository.find_by_library_and_file_type(
                    library_id, ft, skip=skip, limit=limit, status=status_list
                )
                total = asset_repository.count_by_library_and_file_type(library_id, ft, status=status_list)
            else:
                items = asset_repository.find_by_library(library_id, skip=skip, limit=limit, status=status_list)
                total = asset_repository.count_by_project(library.project_id, status=status_list)
            return ListAssetsResponse(
                items=[_to_asset_response(item) for item in items],
                total=total,
                skip=skip,
                limit=limit,
            )

        # 模式2：指定 project_id
        if project_id:
            check_project_access(project_id, user_id, project_repository)
            if ft:
                items = asset_repository.find_by_project_and_file_type(
                    project_id, ft, skip=skip, limit=limit, status=status_list
                )
                total = asset_repository.count_by_project_and_file_type(project_id, ft, status=status_list)
                paged = items
            else:
                items = asset_repository.find_by_project(project_id, skip=skip, limit=limit, status=status_list)
                total = asset_repository.count_by_project(project_id, status=status_list)
                paged = items
            return ListAssetsResponse(
                items=[_to_asset_response(item) for item in paged],
                total=total,
                skip=skip,
                limit=limit,
            )

        # 模式3：跨项目（无 library_id/project_id）
        try:
            projects = project_repository.find_accessible_projects(user_id)
        except Exception:
            logger.exception("查询用户可访问项目失败: user_id=%s", user_id)
            return ListAssetsResponse(items=[], total=0, skip=skip, limit=limit)

        project_ids = [p.id for p in projects]
        if not project_ids:
            return ListAssetsResponse(items=[], total=0, skip=skip, limit=limit)

        if ft:
            # 有 kind 过滤：逐项目查 file_type，凑够一页
            total = 0
            paged_items: list = []
            offset = skip
            remaining = limit
            for pid in project_ids:
                proj_total = asset_repository.count_by_project_and_file_type(pid, ft, status=status_list)
                total += proj_total
                if offset >= proj_total:
                    offset -= proj_total
                    continue
                proj_items = asset_repository.find_by_project_and_file_type(
                    pid, ft, skip=offset, limit=remaining, status=status_list
                )
                paged_items.extend(proj_items)
                remaining -= len(proj_items)
                offset = 0
                if remaining <= 0:
                    break
        else:
            total = asset_repository.count_by_project_ids(project_ids, status=status_list)
            # 跨项目分页：逐项目累积直到凑够一页
            paged_items: list = []
            offset = skip
            remaining = limit
            for pid in project_ids:
                proj_total = asset_repository.count_by_project(pid, status=status_list)
                if offset >= proj_total:
                    offset -= proj_total
                    continue
                proj_items = asset_repository.find_by_project(pid, skip=offset, limit=remaining, status=status_list)
                paged_items.extend(proj_items)
                remaining -= len(proj_items)
                offset = 0
                if remaining <= 0:
                    break

        return ListAssetsResponse(
            items=[_to_asset_response(item) for item in paged_items],
            total=total,
            skip=skip,
            limit=limit,
        )

    # ── 内存过滤路径：有 keyword/gender/style 时，加载全量后内存过滤 ──
    if library_id:
        library = asset_library_repository.get(library_id)
        if library is None:
            raise HTTPException(status_code=404, detail=f"AssetLibrary {library_id} not found")
        check_project_access(library.project_id, user_id, project_repository)
        if kind:
            all_items = asset_repository.find_by_library_and_file_type(
                library_id, kind_to_file_type[kind], status=status_list
            )
        else:
            all_items = asset_repository.find_by_library(library_id, status=status_list)
    elif project_id:
        check_project_access(project_id, user_id, project_repository)
        if kind:
            ft = kind_to_file_type.get(kind)
            if ft:
                all_items = asset_repository.find_by_project_and_file_type(project_id, ft, status=status_list)
            else:
                all_items = asset_repository.find_by_project(project_id, status=status_list)
        else:
            all_items = asset_repository.find_by_project(project_id, status=status_list)
    else:
        try:
            projects = project_repository.find_accessible_projects(user_id)
        except Exception:
            logger.exception("查询用户可访问项目失败: user_id=%s", user_id)
            return ListAssetsResponse(items=[], total=0, skip=skip, limit=limit)
        all_items = []
        for proj in projects:
            if kind and kind_to_file_type.get(kind):
                all_items.extend(
                    asset_repository.find_by_project_and_file_type(proj.id, kind_to_file_type[kind], status=status_list)
                )
            else:
                all_items.extend(asset_repository.find_by_project(proj.id, status=status_list))

    # 应用 kind 过滤（如果有）+ keyword/gender/style
    if kind:
        ft = kind_to_file_type.get(kind)
        if ft:
            all_items = [i for i in all_items if i.file_type == ft]
    filtered = _apply_memory_filters(all_items)
    total = len(filtered)
    paged = filtered[skip : skip + limit]
    return ListAssetsResponse(
        items=[_to_asset_response(item) for item in paged],
        total=total,
        skip=skip,
        limit=limit,
    )


def _apply_asset_review_status(item, review_status: str):
    item.metadata = {
        **item.metadata,
        "review_status": review_status,
    }
    return item


@router.patch("/{asset_id}/review", response_model=AssetResponse)
def update_asset_review_status(
    asset_id: str,
    request: UpdateAssetReviewRequest,
    authenticated_user: AuthenticatedUser = Depends(get_current_user),
    asset_repository: Any = Depends(get_asset_repository),
    project_repository: Any = Depends(get_project_repository),
) -> AssetResponse:
    item = asset_repository.get(asset_id)
    if item is None:
        raise HTTPException(status_code=404, detail=f"Asset {asset_id} not found")
    check_project_access(item.project_id, authenticated_user.user.id, project_repository)
    _apply_asset_review_status(item, request.review_status)
    updated = asset_repository.update(item)
    return _to_asset_response(updated)


@router.post("/batch-delete", response_model=BatchOperationResponse)
def batch_delete_assets(
    request: BatchDeleteRequest,
    authenticated_user: AuthenticatedUser = Depends(get_current_user),
    asset_repository: Any = Depends(get_asset_repository),
    project_repository: Any = Depends(get_project_repository),
) -> BatchOperationResponse:
    """批量删除素材（软删除，标记 status=deleted），需逐项校验项目权限。"""
    user_id = authenticated_user.user.id
    success_ids: list[str] = []
    failed_details: dict[str, str] = {}

    for asset_id in request.asset_ids:
        item = asset_repository.find_by_id(asset_id)
        if item is None:
            failed_details[asset_id] = "not_found"
            continue
        try:
            check_project_access(item.project_id, user_id, project_repository)
            success_ids.append(asset_id)
        except HTTPException:
            failed_details[asset_id] = "access_denied"

    if success_ids:
        asset_repository.batch_delete(success_ids)

    return BatchOperationResponse(
        success_count=len(success_ids),
        failed_ids=list(failed_details.keys()),
        failed_details=failed_details,
    )


@router.post("/batch-tag", response_model=BatchOperationResponse)
def batch_tag_assets(
    request: BatchTagRequest,
    authenticated_user: AuthenticatedUser = Depends(get_current_user),
    asset_repository: Any = Depends(get_asset_repository),
    project_repository: Any = Depends(get_project_repository),
    tag_repository: Any = Depends(get_tag_repository),
) -> BatchOperationResponse:
    """批量打标签（添加或替换模式），需逐项校验项目权限和标签权限。"""
    user_id = authenticated_user.user.id
    success_ids: list[str] = []
    failed_details: dict[str, str] = {}

    # 校验标签存在且属于当前用户
    for tag_id in request.tag_ids:
        tag = tag_repository.get(tag_id)
        if tag is None:
            return BatchOperationResponse(
                success_count=0,
                failed_ids=list(request.asset_ids),
                failed_details={aid: f"tag_not_found:{tag_id}" for aid in request.asset_ids},
            )
        if tag.user_id != user_id:
            return BatchOperationResponse(
                success_count=0,
                failed_ids=list(request.asset_ids),
                failed_details={aid: f"tag_access_denied:{tag_id}" for aid in request.asset_ids},
            )

    # 校验素材权限
    for asset_id in request.asset_ids:
        item = asset_repository.find_by_id(asset_id)
        if item is None:
            failed_details[asset_id] = "not_found"
            continue
        try:
            check_project_access(item.project_id, user_id, project_repository)
            success_ids.append(asset_id)
        except HTTPException:
            failed_details[asset_id] = "access_denied"

    if success_ids:
        if request.mode == "replace":
            asset_repository.batch_replace_tags(success_ids, request.tag_ids)
        else:
            asset_repository.batch_add_tags(success_ids, request.tag_ids)

    return BatchOperationResponse(
        success_count=len(success_ids),
        failed_ids=list(failed_details.keys()),
        failed_details=failed_details,
    )


@router.post("/batch-classify", response_model=BatchOperationResponse)
def batch_classify_assets(
    request: BatchClassifyRequest,
    authenticated_user: AuthenticatedUser = Depends(get_current_user),
    asset_repository: Any = Depends(get_asset_repository),
    project_repository: Any = Depends(get_project_repository),
) -> BatchOperationResponse:
    """批量修改素材内容分类（person/scenic/product等），存在metadata.category中。"""
    user_id = authenticated_user.user.id
    success_ids: list[str] = []
    failed_details: dict[str, str] = {}

    for asset_id in request.asset_ids:
        item = asset_repository.find_by_id(asset_id)
        if item is None:
            failed_details[asset_id] = "not_found"
            continue
        try:
            check_project_access(item.project_id, user_id, project_repository)
            success_ids.append(asset_id)
        except HTTPException:
            failed_details[asset_id] = "access_denied"

    if success_ids:
        asset_repository.batch_update_metadata(success_ids, {"category": request.category})

    return BatchOperationResponse(
        success_count=len(success_ids),
        failed_ids=list(failed_details.keys()),
        failed_details=failed_details,
    )


@router.post("/batch-mark", response_model=BatchOperationResponse)
def batch_mark_assets(
    request: BatchMarkRequest,
    authenticated_user: AuthenticatedUser = Depends(get_current_user),
    asset_repository: Any = Depends(get_asset_repository),
    project_repository: Any = Depends(get_project_repository),
) -> BatchOperationResponse:
    """批量设置智能视图标记（recommended/caution/high_risk），存在metadata.smart_view中。"""
    user_id = authenticated_user.user.id
    success_ids: list[str] = []
    failed_details: dict[str, str] = {}

    for asset_id in request.asset_ids:
        item = asset_repository.find_by_id(asset_id)
        if item is None:
            failed_details[asset_id] = "not_found"
            continue
        try:
            check_project_access(item.project_id, user_id, project_repository)
            success_ids.append(asset_id)
        except HTTPException:
            failed_details[asset_id] = "access_denied"

    if success_ids:
        asset_repository.batch_update_metadata(success_ids, {"smart_view": request.smart_view})

    return BatchOperationResponse(
        success_count=len(success_ids),
        failed_ids=list(failed_details.keys()),
        failed_details=failed_details,
    )


@router.get("/{asset_id}", response_model=AssetResponse)
def get_asset(
    asset_id: str,
    authenticated_user: AuthenticatedUser = Depends(get_current_user),
    asset_repository: Any = Depends(get_asset_repository),
    project_repository: Any = Depends(get_project_repository),
) -> AssetResponse:
    item = asset_repository.find_by_id(asset_id)
    if item is None:
        raise HTTPException(status_code=404, detail=f"Asset {asset_id} not found")
    check_project_access(item.project_id, authenticated_user.user.id, project_repository)
    return _to_asset_response(item)


@router.put("/{asset_id}", response_model=AssetResponse)
def update_asset(
    asset_id: str,
    request: UpdateAssetRequest,
    authenticated_user: AuthenticatedUser = Depends(get_current_user),
    asset_repository: Any = Depends(get_asset_repository),
    project_repository: Any = Depends(get_project_repository),
) -> AssetResponse:
    item = asset_repository.find_by_id(asset_id)
    if item is None:
        raise HTTPException(status_code=404, detail=f"Asset {asset_id} not found")
    check_project_access(item.project_id, authenticated_user.user.id, project_repository)

    # 合并可修改字段
    if request.name is not None:
        item.name = request.name
    if request.metadata is not None:
        item.metadata = {**item.metadata, **request.metadata}
    if request.tags is not None:
        item.metadata = {**item.metadata, "tags": request.tags}

    updated = asset_repository.update(item)
    return _to_asset_response(updated)


@router.delete("/{asset_id}", status_code=204, response_class=Response)
def delete_asset(
    asset_id: str,
    authenticated_user: AuthenticatedUser = Depends(get_current_user),
    asset_repository: Any = Depends(get_asset_repository),
    project_repository: Any = Depends(get_project_repository),
) -> None:
    item = asset_repository.find_by_id(asset_id)
    if item is None:
        raise HTTPException(status_code=404, detail=f"Asset {asset_id} not found")
    check_project_access(item.project_id, authenticated_user.user.id, project_repository)
    asset_repository.delete(asset_id)


@router.post("/{asset_id}/tags", response_model=AssetResponse)
def tag_asset(
    asset_id: str,
    request: TagAssetsRequest,
    authenticated_user: AuthenticatedUser = Depends(get_current_user),
    asset_repository: Any = Depends(get_asset_repository),
    project_repository: Any = Depends(get_project_repository),
    tag_repository: Any = Depends(get_tag_repository),
) -> AssetResponse:
    """给素材打标签。"""
    item = asset_repository.find_by_id(asset_id)
    if item is None:
        raise HTTPException(status_code=404, detail=f"Asset {asset_id} not found")
    check_project_access(item.project_id, authenticated_user.user.id, project_repository)
    for tag_id in request.tag_ids:
        tag = tag_repository.get(tag_id)
        if tag is None:
            raise HTTPException(status_code=404, detail=f"Tag {tag_id} not found")
        if tag.user_id != authenticated_user.user.id:
            raise HTTPException(status_code=403, detail=f"无权使用标签 {tag_id}")
        item.add_tag(tag_id)
    updated = asset_repository.update(item)
    return _to_asset_response(updated)


@router.delete("/{asset_id}/tags/{tag_id}", status_code=204, response_class=Response)
def untag_asset(
    asset_id: str,
    tag_id: str,
    authenticated_user: AuthenticatedUser = Depends(get_current_user),
    asset_repository: Any = Depends(get_asset_repository),
    project_repository: Any = Depends(get_project_repository),
) -> None:
    """取消素材的标签。"""
    item = asset_repository.find_by_id(asset_id)
    if item is None:
        raise HTTPException(status_code=404, detail=f"Asset {asset_id} not found")
    check_project_access(item.project_id, authenticated_user.user.id, project_repository)
    item.remove_tag(tag_id)
    asset_repository.update(item)


@router.post("", response_model=AssetResponse)
def create_asset(
    request: CreateAssetRequest,
    authenticated_user: AuthenticatedUser = Depends(get_current_user),
    asset_repository: Any = Depends(get_asset_repository),
    asset_library_repository: Any = Depends(get_asset_library_repository),
    project_repository: Any = Depends(get_project_repository),
) -> AssetResponse:
    project = project_repository.find_by_id(request.project_id)
    if project is None:
        raise HTTPException(status_code=404, detail=f"Project {request.project_id} not found")
    if not project.can_access(authenticated_user.user.id):
        raise HTTPException(status_code=403, detail="Access denied to project")

    library = asset_library_repository.get(request.library_id)
    if library is None or library.project_id != request.project_id:
        raise HTTPException(status_code=404, detail=f"AssetLibrary {request.library_id} not found")

    use_case = CreateAssetUseCase(asset_repository)
    item = use_case.execute(
        CreateAssetCommand(
            project_id=request.project_id,
            library_id=request.library_id,
            name=request.name,
            storage_key=request.storage_key,
            mime_type=request.mime_type,
            metadata=request.metadata,
            file_size=request.file_size,
            thumbnail_url=request.thumbnail_url,
            duration=request.duration,
            width=request.width,
            height=request.height,
            fps=request.fps,
            codec=request.codec,
            status=AssetStatus(request.status),
            classification_status=ClassificationStatus(request.classification_status),
            quality_score=request.quality_score,
            uploaded_by_user_id=authenticated_user.user.id,
        )
    )
    return _to_asset_response(item)