from __future__ import annotations from urllib.parse import urlencode, unquote from fastapi import APIRouter, Depends, File, HTTPException, Query, Request, UploadFile from fastapi.responses import RedirectResponse from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.dependencies import get_current_user, get_db from app.enums.private_portrait import ( PrivatePortraitEventSource, PrivatePortraitEventStatus, PrivatePortraitEventType, PrivatePortraitLibraryType, PrivatePortraitProjectStatus, PrivatePortraitRemoteDeleteStatus, ) from app.models.private_portrait import PrivatePortraitAsset, PrivatePortraitProject from app.models.user import User from app.enums.upload_resource import UploadResourceTypeEnum from app.schemas.private_portrait import ( PrivatePortraitAssetCreate, PrivatePortraitAssetListOut, PrivatePortraitAssetOut, PrivatePortraitConfigOut, PrivatePortraitDeleteOut, PrivatePortraitEnumMetaOut, PrivatePortraitProjectCreate, PrivatePortraitProjectCreateWithValidateOut, PrivatePortraitProjectListOut, PrivatePortraitProjectOut, PrivatePortraitProjectUpdate, PrivatePortraitSelectableAssetListOut, PrivatePortraitUploadOut, PrivatePortraitValidateSessionCreate, PrivatePortraitValidateSessionOut, build_private_portrait_enum_meta, ) from app.services.operation_log_service import log_operation_error, log_operation_event from app.services.private_portrait.asset_service import ( DOMAIN, asset_to_out, get_user_private_portrait_config, get_validate_session, handle_validate_callback, list_assets, list_selectable_assets, soft_delete_asset, sync_asset_status, validate_session_to_out, ) from app.services.private_portrait.project_service import ( get_user_project, list_projects, project_to_out, refresh_project_counters, soft_delete_project, ) from app.services.private_portrait.upload_service import upload_private_portrait_asset_file from app.services.upload_resource import cleanup_upload_resource_files_after_commit from app.services.private_portrait.real_person.service import ( create_real_person_asset, create_real_person_project, create_real_person_validate_session, update_real_person_project, ) router = APIRouter(tags=["私域真人素材库"]) _REAL_PERSON_API_DESCRIPTION = ( "私域真人素材库 API。真人和虚拟素材共用用户素材额度；user.private_portrait_asset_limit=0 表示关闭，" ">0 表示启用并限制总素材数量。真人项目创建后需要通过火山 CreateVisualValidateSession 进行人脸认证，" "每个真人项目只允许认证一次。认证回调 resultCode=10000 表示成功，成功后才允许上传可用于 AI 创作的真人素材。" ) def _log_task_dispatch_failed(*, task_name: str, user_id: str | None = None, project_id: str | None = None, asset_id: str | None = None, exc: BaseException) -> None: log_operation_error( domain=DOMAIN, event_type=PrivatePortraitEventType.TASK_DISPATCH_FAILED.value, source=PrivatePortraitEventSource.API.value, user_id=user_id, project_id=project_id, asset_id=asset_id, exc=exc, detail={"task_name": task_name}, ) def _log_task_dispatch_success(*, task_name: str, user_id: str | None = None, project_id: str | None = None, asset_id: str | None = None) -> None: log_operation_event( domain=DOMAIN, event_type=PrivatePortraitEventType.TASK_DISPATCH_SUCCESS.value, event_status=PrivatePortraitEventStatus.SUCCESS.value, source=PrivatePortraitEventSource.API.value, user_id=user_id, project_id=project_id, asset_id=asset_id, detail={"task_name": task_name}, ) @router.get( "/private-portrait/config", response_model=PrivatePortraitConfigOut, summary="获取当前用户私域真人素材额度配置", description=( _REAL_PERSON_API_DESCRIPTION + "返回 enabled、asset_limit、used_asset_count、remaining_asset_count 等字段。额度按用户维度限制，真人/虚拟、图片/视频共用；Audio 暂不开放。" ), ) async def get_my_private_portrait_config(current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db)): return await get_user_private_portrait_config(db, user_id=current_user.id) @router.get( "/private-portrait/meta/enums", response_model=PrivatePortraitEnumMetaOut, summary="获取私域真人素材库枚举说明", description=( "返回前端展示所需枚举：library_type=real_person/aigc_virtual；" "asset_type=Image/Video/Audio；project status、asset status、remote_delete_status 等。" "当前业务上传只开放 Image 和 Video，Audio 仅作为兼容枚举展示。" ), ) async def get_private_portrait_enum_meta(): return build_private_portrait_enum_meta() @router.post( "/private-portrait/uploads/image", response_model=PrivatePortraitUploadOut, summary="上传真人图片素材", description="上传真人图片素材并写入 UploadResource，module=private_portrait_real。创建素材时需回传 resource_id 到 upload_resource_id。", ) async def upload_private_portrait_image(file: UploadFile = File(...), current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db)): try: out = await upload_private_portrait_asset_file( db, file=file, current_user=current_user, library_type=PrivatePortraitLibraryType.REAL_PERSON.value, resource_type=UploadResourceTypeEnum.IMAGE.value, ) await db.commit() return out except HTTPException: await db.rollback() raise except Exception as exc: await db.rollback() raise HTTPException(status_code=500, detail=f"上传真人图片素材失败: {exc}") @router.post( "/private-portrait/uploads/video", response_model=PrivatePortraitUploadOut, summary="上传真人视频素材", description="上传真人视频素材并写入 UploadResource，module=private_portrait_real。创建素材时需回传 resource_id 到 upload_resource_id。", ) async def upload_private_portrait_video( file: UploadFile = File(...), duration_seconds: float | None = Query(None, description="客户端解析的视频秒数，服务端会写入 UploadResource 并在创建素材时回填"), current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): try: out = await upload_private_portrait_asset_file( db, file=file, current_user=current_user, library_type=PrivatePortraitLibraryType.REAL_PERSON.value, resource_type=UploadResourceTypeEnum.VIDEO.value, duration_seconds=duration_seconds, ) await db.commit() return out except HTTPException: await db.rollback() raise except Exception as exc: await db.rollback() raise HTTPException(status_code=500, detail=f"上传真人视频素材失败: {exc}") @router.post( "/private-portrait/projects", response_model=PrivatePortraitProjectCreateWithValidateOut, summary="创建真人认证素材项目并生成认证会话", description=( _REAL_PERSON_API_DESCRIPTION + "创建本地真人项目后立即调用火山 CreateVisualValidateSession，返回 H5Link/BytedToken 给 PC 前端展示二维码。" "手机扫码完成人脸认证后，PC 端通过 validate_session 查询状态；认证成功后才允许创建真人素材。" ), ) async def create_private_portrait_project(payload: PrivatePortraitProjectCreate, current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db)): project = await create_real_person_project(db, user_id=current_user.id, payload=payload) session = await create_real_person_validate_session(db, user_id=current_user.id, project_id=project.id, callback_redirect_url=payload.callback_redirect_url) out = PrivatePortraitProjectCreateWithValidateOut(project=project_to_out(project), validate_session=validate_session_to_out(session), poll_interval_ms=2000) await db.commit() return out @router.get( "/private-portrait/projects", response_model=PrivatePortraitProjectListOut, summary="查询当前用户真人认证素材项目列表", description="只返回 library_type=real_person 的项目。默认查询 active 项目，可通过 status 覆盖。", ) async def list_private_portrait_projects( page: int = Query(1, ge=1, description="页码，从 1 开始"), page_size: int = Query(20, ge=1, le=100, description="每页数量，最大 100"), keyword: str | None = Query(None, description="项目名称模糊搜索"), status: str | None = Query(None, description="项目状态筛选，不传默认 active"), current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): query_status = status or PrivatePortraitProjectStatus.ACTIVE.value items, total = await list_projects(db, user_id=current_user.id, page=page, page_size=page_size, keyword=keyword, status=query_status, library_type=PrivatePortraitLibraryType.REAL_PERSON.value) await refresh_project_counters(db, [item.id for item in items]) await db.commit() items, total = await list_projects(db, user_id=current_user.id, page=page, page_size=page_size, keyword=keyword, status=query_status, library_type=PrivatePortraitLibraryType.REAL_PERSON.value) return PrivatePortraitProjectListOut(items=[project_to_out(item) for item in items], total=total, page=page, page_size=page_size) @router.get( "/private-portrait/projects/{project_id}", response_model=PrivatePortraitProjectOut, summary="获取真人认证素材项目详情", description="获取当前用户真人项目详情，project_id 必须属于当前用户且 library_type=real_person。", ) async def get_private_portrait_project(project_id: str, current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db)): return project_to_out(await get_user_project(db, user_id=current_user.id, project_id=project_id, library_type=PrivatePortraitLibraryType.REAL_PERSON.value)) @router.put( "/private-portrait/projects/{project_id}", response_model=PrivatePortraitProjectOut, summary="更新真人认证素材项目", description="更新真人项目本地展示信息；不会重新发起真人认证。每个真人项目只允许认证一次。", ) async def update_private_portrait_project(project_id: str, payload: PrivatePortraitProjectUpdate, current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db)): project = await update_real_person_project(db, user_id=current_user.id, project_id=project_id, payload=payload) out = project_to_out(project) await db.commit() return out @router.delete( "/private-portrait/projects/{project_id}", response_model=PrivatePortraitDeleteOut, summary="删除真人认证素材项目", description="软删真人项目和本地素材记录。本地先 commit，commit 成功后投递 Celery 删除远端 AssetGroup/Asset；remote_delete_status=pending 表示远端删除处理中。", ) async def delete_private_portrait_project(project_id: str, current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db)): project = await soft_delete_project(db, user_id=current_user.id, project_id=project_id, library_type=PrivatePortraitLibraryType.REAL_PERSON.value) project_id_snapshot = project.id pending_upload_resource_ids = list(getattr(project, "_pending_upload_resource_ids", []) or []) await db.commit() try: from app.tasks.private_portrait_asset_tasks import delete_private_portrait_project_remote delete_private_portrait_project_remote.delay(project_id_snapshot) _log_task_dispatch_success(task_name="private_portrait.delete_project_remote", user_id=current_user.id, project_id=project_id_snapshot) except Exception as exc: _log_task_dispatch_failed(task_name="private_portrait.delete_project_remote", user_id=current_user.id, project_id=project_id_snapshot, exc=exc) if pending_upload_resource_ids: try: await cleanup_upload_resource_files_after_commit(db, resource_ids=pending_upload_resource_ids) await db.commit() except Exception as exc: await db.rollback() log_operation_error( domain=DOMAIN, event_type=PrivatePortraitEventType.ASSET_UPLOAD_RELEASE_FAILED.value, source=PrivatePortraitEventSource.API.value, user_id=current_user.id, project_id=project_id_snapshot, exc=exc, detail={"resource_ids": pending_upload_resource_ids}, ) return PrivatePortraitDeleteOut(success=True, remote_delete_status=PrivatePortraitRemoteDeleteStatus.PENDING.value) @router.post( "/private-portrait/projects/{project_id}/validate-sessions", response_model=PrivatePortraitValidateSessionOut, summary="重新创建真人认证会话", description="为真人项目创建新的认证会话。业务层会限制项目只能认证一次；已认证成功的项目不允许重复认证。", ) async def create_private_portrait_validate_session(project_id: str, payload: PrivatePortraitValidateSessionCreate, current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db)): session = await create_real_person_validate_session(db, user_id=current_user.id, project_id=project_id, callback_redirect_url=payload.callback_redirect_url) out = validate_session_to_out(session) await db.commit() return out @router.get( "/private-portrait/validate-sessions/{session_id}", response_model=PrivatePortraitValidateSessionOut, summary="查询真人认证会话状态", description="PC 端轮询该接口查看手机扫码认证结果。status/result_code/remote_group_id 可用于判断是否认证成功并提示用户回到 PC 查看项目。", ) async def get_private_portrait_validate_session(session_id: str, current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db)): return validate_session_to_out(await get_validate_session(db, user_id=current_user.id, session_id=session_id)) @router.get( "/private-portrait/validate-callback", summary="火山真人认证回调入口", description="火山真人认证 H5 回调入口。resultCode=10000 表示认证成功；成功后会创建或更新本地 AssetGroup 映射，并可 redirect 回前端提示页。", ) async def private_portrait_validate_callback(session_id: str, request: Request, redirect_url: str | None = None, db: AsyncSession = Depends(get_db)): params = dict(request.query_params) params.pop("session_id", None) params.pop("redirect_url", None) session = await handle_validate_callback(db, session_id=session_id, query_params=params) redirect_params = {"session_id": session.id, "status": session.status, "resultCode": session.result_code or ""} if session.remote_group_id: redirect_params["remote_group_id"] = session.remote_group_id response = {"session_id": session.id, "status": session.status, "resultCode": session.result_code, "remote_group_id": session.remote_group_id} await db.commit() if redirect_url: base_url = unquote(redirect_url) sep = "&" if "?" in base_url else "?" return RedirectResponse(url=f"{base_url}{sep}{urlencode(redirect_params)}") return response @router.post( "/private-portrait/projects/{project_id}/assets", response_model=PrivatePortraitAssetOut, summary="上传真人认证素材", description=( "在已认证成功的真人项目下创建素材。当前仅开放 asset_type=Image/Video，Audio 暂不开放。" "Video 必须携带 video_duration，建议前端限制 2~15 秒。CreateAsset 是异步接口，返回后会投递轮询任务，" "只有 status=Active 的素材才会出现在 selectable-assets 并可用于 AI 创作。" ), ) async def create_private_portrait_asset(project_id: str, payload: PrivatePortraitAssetCreate, current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db)): asset = await create_real_person_asset(db, user_id=current_user.id, project_id=project_id, payload=payload) asset_id_snapshot = asset.id project_id_snapshot = asset.project_id out = asset_to_out(asset) await db.commit() try: from app.tasks.private_portrait_asset_tasks import poll_private_portrait_asset_status poll_private_portrait_asset_status.delay(asset_id_snapshot) _log_task_dispatch_success(task_name="private_portrait.poll_asset_status", user_id=current_user.id, project_id=project_id_snapshot, asset_id=asset_id_snapshot) except Exception as exc: _log_task_dispatch_failed(task_name="private_portrait.poll_asset_status", user_id=current_user.id, project_id=project_id_snapshot, asset_id=asset_id_snapshot, exc=exc) return out @router.get( "/private-portrait/projects/{project_id}/assets", response_model=PrivatePortraitAssetListOut, summary="查询真人认证素材列表", description="查询指定真人项目下的素材。asset_type 可传 Image 或 Video；status 可筛选素材状态。", ) async def list_private_portrait_assets( project_id: str, page: int = Query(1, ge=1, description="页码，从 1 开始"), page_size: int = Query(20, ge=1, le=100, description="每页数量，最大 100"), status: str | None = Query(None, description="素材状态筛选，不传查全部"), keyword: str | None = Query(None, description="素材名称模糊搜索"), asset_type: str | None = Query(None, description="素材类型：Image=图片，Video=视频；Audio 暂不开放"), current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): assets, total, project_name_map = await list_assets(db, user_id=current_user.id, project_id=project_id, status=status, keyword=keyword, page=page, page_size=page_size, library_type=PrivatePortraitLibraryType.REAL_PERSON.value, asset_type=asset_type) return PrivatePortraitAssetListOut(items=[asset_to_out(asset, project_name=project_name_map.get(asset.project_id)) for asset in assets], total=total, page=page, page_size=page_size) @router.get( "/private-portrait/assets/{asset_id}", response_model=PrivatePortraitAssetOut, summary="获取真人认证素材详情", description="获取当前用户真人素材详情，asset_id 必须属于当前用户且 library_type=real_person。", ) async def get_private_portrait_asset(asset_id: str, current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db)): asset = (await db.execute(select(PrivatePortraitAsset).where(PrivatePortraitAsset.id == asset_id, PrivatePortraitAsset.user_id == current_user.id, PrivatePortraitAsset.library_type == PrivatePortraitLibraryType.REAL_PERSON.value).limit(1))).scalar_one_or_none() if not asset: raise HTTPException(status_code=404, detail="私域人像素材不存在") project = (await db.execute(select(PrivatePortraitProject).where(PrivatePortraitProject.id == asset.project_id).limit(1))).scalar_one_or_none() return asset_to_out(asset, project_name=project.name if project else None) @router.post( "/private-portrait/assets/{asset_id}/sync", response_model=PrivatePortraitAssetOut, summary="同步真人认证素材状态", description="主动向火山查询并同步真人素材状态。一般由轮询任务自动执行；前端排查或手动刷新时可调用。", ) async def sync_private_portrait_asset(asset_id: str, current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db)): asset = await sync_asset_status(db, user_id=current_user.id, asset_id=asset_id) if asset.library_type != PrivatePortraitLibraryType.REAL_PERSON.value: raise HTTPException(status_code=404, detail="私域人像素材不存在") out = asset_to_out(asset) await db.commit() return out @router.delete( "/private-portrait/assets/{asset_id}", response_model=PrivatePortraitDeleteOut, summary="删除真人认证素材", description="软删本地真人素材记录。本地先 commit，commit 成功后投递 Celery 删除远端 Asset；remote_delete_status=pending 表示远端删除处理中。", ) async def delete_private_portrait_asset(asset_id: str, current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db)): asset = await soft_delete_asset(db, user_id=current_user.id, asset_id=asset_id, library_type=PrivatePortraitLibraryType.REAL_PERSON.value) asset_id_snapshot = asset.id project_id_snapshot = asset.project_id pending_upload_resource_ids = list(getattr(asset, "_pending_upload_resource_ids", []) or []) await db.commit() try: from app.tasks.private_portrait_asset_tasks import delete_private_portrait_asset_remote delete_private_portrait_asset_remote.delay(asset_id_snapshot) _log_task_dispatch_success(task_name="private_portrait.delete_asset_remote", user_id=current_user.id, project_id=project_id_snapshot, asset_id=asset_id_snapshot) except Exception as exc: _log_task_dispatch_failed(task_name="private_portrait.delete_asset_remote", user_id=current_user.id, project_id=project_id_snapshot, asset_id=asset_id_snapshot, exc=exc) if pending_upload_resource_ids: try: await cleanup_upload_resource_files_after_commit(db, resource_ids=pending_upload_resource_ids) await db.commit() except Exception as exc: await db.rollback() log_operation_error( domain=DOMAIN, event_type=PrivatePortraitEventType.ASSET_UPLOAD_RELEASE_FAILED.value, source=PrivatePortraitEventSource.API.value, user_id=current_user.id, project_id=project_id_snapshot, asset_id=asset_id_snapshot, exc=exc, detail={"resource_ids": pending_upload_resource_ids}, ) return PrivatePortraitDeleteOut(success=True, remote_delete_status=PrivatePortraitRemoteDeleteStatus.PENDING.value) @router.get( "/private-portrait/selectable-assets", response_model=PrivatePortraitSelectableAssetListOut, summary="查询可用于 AI 创作的真人认证素材", description="只返回当前用户真人素材库中 status=Active 且未删除的 Image/Video 素材。该接口给 AI 创作参考内容选择器使用。", ) async def list_private_portrait_selectable_assets( page: int = Query(1, ge=1, description="页码，从 1 开始"), page_size: int = Query(20, ge=1, le=100, description="每页数量，最大 100"), project_id: str | None = Query(None, description="按真人项目 ID 筛选"), keyword: str | None = Query(None, description="素材名称模糊搜索"), asset_type: str | None = Query(None, description="素材类型：Image 或 Video，不传查全部"), current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): items, total = await list_selectable_assets(db, user_id=current_user.id, project_id=project_id, keyword=keyword, page=page, page_size=page_size, library_type=PrivatePortraitLibraryType.REAL_PERSON.value, asset_type=asset_type) return PrivatePortraitSelectableAssetListOut(items=items, total=total, page=page, page_size=page_size)