mz/video-gen
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

真人素材库V3

Browse Source
This commit is contained in:
GinHa
2026-07-07 13:20:50 +08:00
parent d80f5c3577
commit 6ae5e0d663
31 changed files with 1421 additions and 606 deletions
Download Patch File Download Diff File Expand all files Collapse all files
video-gen-api/app/services/private_portrait/admin/__init__.py
View File
video-gen-api/app/services/private_portrait/admin/asset_query_service.py
+21
View File
@@ -0,0 +1,21 @@
from __future__ import annotations
from sqlalchemy.ext.asyncio import AsyncSession
from app.services.private_portrait.asset_service import asset_to_out, list_assets
async def admin_list_assets(
db: AsyncSession,
*,
user_id: str | None = None,
project_id: str | None = None,
library_type: str | None = None,
asset_type: str | None = None,
keyword: str | None = None,
status: str | None = None,
page: int = 1,
page_size: int = 20,
):
assets, total, project_name_map = await list_assets(db, user_id=user_id, project_id=project_id, status=status, keyword=keyword, page=page, page_size=page_size, library_type=library_type, asset_type=asset_type)
return [asset_to_out(asset, project_name=project_name_map.get(asset.project_id), include_user=True) for asset in assets], total
video-gen-api/app/services/private_portrait/admin/project_query_service.py
+20
View File
@@ -0,0 +1,20 @@
from __future__ import annotations
from sqlalchemy.ext.asyncio import AsyncSession
from app.services.private_portrait.project_service import list_projects, project_to_out, refresh_project_counters
async def admin_list_projects(
db: AsyncSession,
*,
user_id: str | None = None,
library_type: str | None = None,
keyword: str | None = None,
status: str | None = None,
page: int = 1,
page_size: int = 20,
):
items, total = await list_projects(db, user_id=user_id, page=page, page_size=page_size, keyword=keyword, status=status, library_type=library_type)
await refresh_project_counters(db, [item.id for item in items])
return [project_to_out(item, include_user=True) for item in items], total
video-gen-api/app/services/private_portrait/admin/stats_query_service.py
+45
View File
@@ -0,0 +1,45 @@
from __future__ import annotations
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession
from app.enums.private_portrait import PrivatePortraitAssetStatus, PrivatePortraitAssetType, PrivatePortraitLibraryType
from app.models.private_portrait import PrivatePortraitAsset, PrivatePortraitProject
from app.schemas.private_portrait import PrivatePortraitAdminStatsOut
async def admin_get_private_portrait_stats(db: AsyncSession, *, user_id: str | None = None, library_type: str | None = None) -> PrivatePortraitAdminStatsOut:
project_filters = [PrivatePortraitProject.deleted_at.is_(None)]
asset_filters = [PrivatePortraitAsset.deleted_at.is_(None)]
if user_id:
project_filters.append(PrivatePortraitProject.user_id == user_id)
asset_filters.append(PrivatePortraitAsset.user_id == user_id)
if library_type:
project_filters.append(PrivatePortraitProject.library_type == library_type)
asset_filters.append(PrivatePortraitAsset.library_type == library_type)
total_projects = (await db.execute(select(func.count(PrivatePortraitProject.id)).where(*project_filters))).scalar_one()
rows = await db.execute(
select(PrivatePortraitAsset.library_type, PrivatePortraitAsset.asset_type, PrivatePortraitAsset.status, func.count(PrivatePortraitAsset.id))
.where(*asset_filters)
.group_by(PrivatePortraitAsset.library_type, PrivatePortraitAsset.asset_type, PrivatePortraitAsset.status)
)
stats = PrivatePortraitAdminStatsOut(total_projects=int(total_projects or 0))
for lib, asset_type, status, count in rows.all():
n = int(count or 0)
stats.total_assets += n
if asset_type == PrivatePortraitAssetType.IMAGE.value:
stats.image_assets += n
elif asset_type == PrivatePortraitAssetType.VIDEO.value:
stats.video_assets += n
if status == PrivatePortraitAssetStatus.ACTIVE.value:
stats.active_assets += n
elif status == PrivatePortraitAssetStatus.PROCESSING.value:
stats.processing_assets += n
elif status == PrivatePortraitAssetStatus.FAILED.value:
stats.failed_assets += n
if lib == PrivatePortraitLibraryType.REAL_PERSON.value:
stats.real_person_assets += n
elif lib == PrivatePortraitLibraryType.AIGC_VIRTUAL.value:
stats.virtual_assets += n
return stats
video-gen-api/app/services/private_portrait/ark_client.py
+7 -1
View File
@@ -51,7 +51,7 @@ def _remote_error_http_status(code: str) -> int:
class ArkPrivateAssetClient:
"""火山 Ark 私域真人人像素材 API Client。只做 AK/SK 鉴权调用与响应标准化。"""
"""火山 Ark 私域可信素材 Asset API Client。只做 AK/SK 鉴权调用与响应标准化。"""
def __init__(self, *, ak: str | None = None, sk: str | None = None, for_celery: bool = False):
self.ak = ak or settings.VOLC_SMS_ACCESS_KEY_ID
@@ -66,6 +66,12 @@ class ArkPrivateAssetClient:
async def get_visual_validate_result(self, *, project_name: str, byted_token: str) -> dict[str, Any]:
return await self._call(ArkPrivatePortraitAction.GET_VISUAL_VALIDATE_RESULT, {"BytedToken": byted_token, "ProjectName": project_name})
async def create_asset_group(self, *, project_name: str, name: str, description: str | None = None, group_type: str = "AIGC") -> dict[str, Any]:
payload: dict[str, Any] = {"Name": name, "GroupType": group_type, "ProjectName": project_name}
if description:
payload["Description"] = description
return await self._call(ArkPrivatePortraitAction.CREATE_ASSET_GROUP, payload)
async def create_asset(self, *, project_name: str, group_id: str, url: str, asset_type: str, name: str | None = None) -> dict[str, Any]:
payload: dict[str, Any] = {"GroupId": group_id, "URL": url, "AssetType": asset_type, "ProjectName": project_name}
if name:
video-gen-api/app/services/private_portrait/asset_service.py
+169 -248
View File
@@ -6,33 +6,42 @@ from typing import Any
from urllib.parse import urlencode
from fastapi import HTTPException
from sqlalchemy import and_, func, select, update
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession
from app.config import settings
from app.enums.private_portrait import (
PRIVATE_PORTRAIT_ASSET_POLL_INTERVAL_SECONDS,
PRIVATE_PORTRAIT_ASSET_POLL_MAX_COUNT,
PRIVATE_PORTRAIT_DEFAULT_IMAGE_LIMIT,
PRIVATE_PORTRAIT_GROUP_TYPE,
PRIVATE_PORTRAIT_ASSET_URI_PREFIX,
PRIVATE_PORTRAIT_ENABLED_ASSET_TYPES,
PRIVATE_PORTRAIT_REAL_PERSON_GROUP_TYPE,
PRIVATE_PORTRAIT_SUCCESS_RESULT_CODE,
PRIVATE_PORTRAIT_VALIDATE_TOKEN_EXPIRE_MINUTES,
PRIVATE_PORTRAIT_VIDEO_ASSET_POLL_INTERVAL_SECONDS,
PRIVATE_PORTRAIT_VIDEO_ASSET_POLL_MAX_COUNT,
PrivatePortraitAssetGroupStatus,
PrivatePortraitAssetStatus,
PrivatePortraitAssetType,
PrivatePortraitEventSource,
PrivatePortraitEventStatus,
PrivatePortraitEventType,
PrivatePortraitLibraryType,
PrivatePortraitProjectStatus,
PrivatePortraitRemoteDeleteStatus,
PrivatePortraitValidateSessionStatus,
)
from app.models.private_portrait import PrivatePortraitAsset, PrivatePortraitAssetGroup, PrivatePortraitProject, PrivatePortraitValidateSession
from app.models.user import User
from app.schemas.private_portrait import PrivatePortraitAssetCreate, PrivatePortraitAssetOut, PrivatePortraitConfigOut, PrivatePortraitSelectableAssetOut, PrivatePortraitValidateSessionOut
from app.schemas.private_portrait import PrivatePortraitAssetCreate, PrivatePortraitAssetOut, PrivatePortraitSelectableAssetOut, PrivatePortraitValidateSessionOut
from app.services.operation_log_service import log_operation_error, log_operation_event
from app.services.private_portrait.ark_client import ArkPrivateAssetClient
from app.services.private_portrait.project_service import get_user_project, refresh_project_counters
from app.services.private_portrait.quota_service import (
count_user_counting_assets,
ensure_private_portrait_asset_quota_available,
get_user_private_portrait_config,
set_user_private_portrait_limit,
)
from app.utils.id_gen import generate_id
DOMAIN = "private_portrait"
@@ -53,7 +62,6 @@ def _loads(data: str | None) -> Any:
return None
def _exception_message(exc: Exception) -> str:
if isinstance(exc, HTTPException):
detail = exc.detail
@@ -65,7 +73,7 @@ def _exception_message(exc: Exception) -> str:
def _public_url(url: str) -> str:
if url.startswith(("http://", "https://")):
if url.startswith(("http://", "https://", PRIVATE_PORTRAIT_ASSET_URI_PREFIX)):
return url
return f"{settings.BASE_URL.rstrip('/')}/{url.lstrip('/')}"
@@ -83,36 +91,31 @@ def _remote_group_name(user_id: str, project_name: str) -> str:
return f"{user_id}-{safe_name}"[:128]
async def get_user_private_portrait_config(db: AsyncSession, *, user_id: str) -> PrivatePortraitConfigOut:
user = (await db.execute(select(User).where(User.id == user_id).limit(1))).scalar_one_or_none()
if not user:
raise HTTPException(status_code=404, detail="用户不存在")
limit = int(getattr(user, "private_portrait_image_limit", PRIVATE_PORTRAIT_DEFAULT_IMAGE_LIMIT) or 0)
used = await count_user_counting_image_assets(db, user_id=user_id)
return PrivatePortraitConfigOut(enabled=limit > 0, image_limit=limit, used_image_count=used, remaining_image_count=max(0, limit - used) if limit > 0 else 0)
def _asset_display_url(asset: PrivatePortraitAsset) -> str | None:
return asset.preview_url or asset.remote_url or asset.source_url or None
async def set_user_private_portrait_limit(db: AsyncSession, *, user_id: str, limit: int) -> User:
user = (await db.execute(select(User).where(User.id == user_id).limit(1))).scalar_one_or_none()
if not user:
raise HTTPException(status_code=404, detail="用户不存在")
user.private_portrait_image_limit = max(0, int(limit))
await db.flush()
return user
def _provider_url(asset: PrivatePortraitAsset) -> str | None:
return f"{PRIVATE_PORTRAIT_ASSET_URI_PREFIX}{asset.remote_asset_id}" if asset.remote_asset_id else None
async def count_user_counting_image_assets(db: AsyncSession, *, user_id: str) -> int:
statuses = [PrivatePortraitAssetStatus.CREATING.value, PrivatePortraitAssetStatus.PROCESSING.value, PrivatePortraitAssetStatus.ACTIVE.value]
total = (await db.execute(select(func.count(PrivatePortraitAsset.id)).where(PrivatePortraitAsset.user_id == user_id, PrivatePortraitAsset.asset_type == PrivatePortraitAssetType.IMAGE.value, PrivatePortraitAsset.deleted_at.is_(None), PrivatePortraitAsset.status.in_(statuses)))).scalar_one()
return int(total or 0)
def _poll_interval_seconds(asset_type: str) -> int:
if asset_type == PrivatePortraitAssetType.VIDEO.value:
return PRIVATE_PORTRAIT_VIDEO_ASSET_POLL_INTERVAL_SECONDS
return PRIVATE_PORTRAIT_ASSET_POLL_INTERVAL_SECONDS
async def _lock_user_for_upload(db: AsyncSession, *, user_id: str) -> User:
# 锁 users 行，避免并发绕过用户总量限制。SQLite 会忽略 FOR UPDATE，不影响本地开发。
user = (await db.execute(select(User).where(User.id == user_id).with_for_update().limit(1))).scalar_one_or_none()
if not user:
raise HTTPException(status_code=404, detail="用户不存在")
return user
def _poll_max_count(asset_type: str) -> int:
if asset_type == PrivatePortraitAssetType.VIDEO.value:
return PRIVATE_PORTRAIT_VIDEO_ASSET_POLL_MAX_COUNT
return PRIVATE_PORTRAIT_ASSET_POLL_MAX_COUNT
def _assert_enabled_asset_type(asset_type: str) -> None:
if asset_type not in {item.value for item in PrivatePortraitAssetType}:
raise HTTPException(status_code=400, detail="asset_type 仅支持 Image/Video，Audio 暂未开放")
if asset_type not in PRIVATE_PORTRAIT_ENABLED_ASSET_TYPES:
raise HTTPException(status_code=400, detail="Audio 暂未开放，当前仅支持 Image/Video")
def validate_session_to_out(session: PrivatePortraitValidateSession, *, include_user: bool = False) -> PrivatePortraitValidateSessionOut:
@@ -137,12 +140,14 @@ def validate_session_to_out(session: PrivatePortraitValidateSession, *, include_
def asset_to_out(asset: PrivatePortraitAsset, *, project_name: str | None = None, include_user: bool = False) -> PrivatePortraitAssetOut:
display_url = _asset_display_url(asset)
return PrivatePortraitAssetOut(
id=asset.id,
user_id=asset.user_id if include_user else None,
project_id=asset.project_id,
project_name=project_name,
group_id=asset.group_id,
library_type=asset.library_type,
remote_group_id=asset.remote_group_id,
remote_asset_id=asset.remote_asset_id,
remote_project_name=asset.remote_project_name,
@@ -150,8 +155,14 @@ def asset_to_out(asset: PrivatePortraitAsset, *, project_name: str | None = None
name=asset.name,
source_url=asset.source_url,
preview_url=asset.preview_url,
display_url=display_url,
provider_url=_provider_url(asset),
remote_url=asset.remote_url,
remote_url_expired_at=asset.remote_url_expired_at,
video_duration=asset.video_duration,
video_cover_url=asset.video_cover_url,
file_size=asset.file_size,
mime_type=asset.mime_type,
status=asset.status,
moderation=_loads(asset.moderation_json),
last_poll_at=asset.last_poll_at,
@@ -166,15 +177,18 @@ def asset_to_out(asset: PrivatePortraitAsset, *, project_name: str | None = None
)
async def _get_existing_active_group(db: AsyncSession, *, project_id: str) -> PrivatePortraitAssetGroup | None:
async def _get_existing_active_group(db: AsyncSession, *, project_id: str, library_type: str | None = None) -> PrivatePortraitAssetGroup | None:
filters = [
PrivatePortraitAssetGroup.project_id == project_id,
PrivatePortraitAssetGroup.status == PrivatePortraitAssetGroupStatus.ACTIVE.value,
PrivatePortraitAssetGroup.deleted_at.is_(None),
]
if library_type:
filters.append(PrivatePortraitAssetGroup.library_type == library_type)
return (
await db.execute(
select(PrivatePortraitAssetGroup)
.where(
PrivatePortraitAssetGroup.project_id == project_id,
PrivatePortraitAssetGroup.status == PrivatePortraitAssetGroupStatus.ACTIVE.value,
PrivatePortraitAssetGroup.deleted_at.is_(None),
)
.where(*filters)
.order_by(PrivatePortraitAssetGroup.created_at.desc())
.limit(1)
)
@@ -182,7 +196,9 @@ async def _get_existing_active_group(db: AsyncSession, *, project_id: str) -> Pr
async def _ensure_project_can_validate(db: AsyncSession, *, project: PrivatePortraitProject) -> PrivatePortraitValidateSession | None:
active_group = await _get_existing_active_group(db, project_id=project.id)
if project.library_type != PrivatePortraitLibraryType.REAL_PERSON.value:
raise HTTPException(status_code=400, detail="虚拟人像项目不支持真人认证")
active_group = await _get_existing_active_group(db, project_id=project.id, library_type=project.library_type)
if active_group or project.status == PrivatePortraitProjectStatus.ACTIVE.value:
raise HTTPException(status_code=409, detail="该真人素材项目已完成认证，不能重复认证")
@@ -206,12 +222,7 @@ async def _ensure_project_can_validate(db: AsyncSession, *, project: PrivatePort
select(PrivatePortraitValidateSession)
.where(
PrivatePortraitValidateSession.project_id == project.id,
PrivatePortraitValidateSession.status.in_(
[
PrivatePortraitValidateSessionStatus.CREATED.value,
PrivatePortraitValidateSessionStatus.CALLBACK_SUCCESS.value,
]
),
PrivatePortraitValidateSession.status.in_([PrivatePortraitValidateSessionStatus.CREATED.value, PrivatePortraitValidateSessionStatus.CALLBACK_SUCCESS.value]),
PrivatePortraitValidateSession.expired_at.is_not(None),
PrivatePortraitValidateSession.expired_at > now,
)
@@ -223,7 +234,7 @@ async def _ensure_project_can_validate(db: AsyncSession, *, project: PrivatePort
async def create_validate_session(db: AsyncSession, *, user_id: str, project_id: str, callback_redirect_url: str | None = None) -> PrivatePortraitValidateSession:
project = await get_user_project(db, user_id=user_id, project_id=project_id)
project = await get_user_project(db, user_id=user_id, project_id=project_id, library_type=PrivatePortraitLibraryType.REAL_PERSON.value)
reusable_session = await _ensure_project_can_validate(db, project=project)
if reusable_session:
return reusable_session
@@ -246,10 +257,8 @@ async def create_validate_session(db: AsyncSession, *, user_id: str, project_id:
session.h5_link = resp.get("H5Link") or resp.get("h5Link")
session.raw_response_json = _json(resp)
await db.flush()
# created_at / updated_at 来自数据库默认值或 onupdate，flush 后可能处于 expired 状态。
# 在 async SQLAlchemy 下，响应转换时同步读取 expired 字段会触发 MissingGreenlet。
await db.refresh(session)
log_operation_event(domain=DOMAIN, event_type=PrivatePortraitEventType.VALIDATE_SESSION_CREATE.value, event_status=PrivatePortraitEventStatus.SUCCESS.value, source=PrivatePortraitEventSource.API.value, user_id=user_id, project_id=project.id, session_id=session.id, detail={"remote_project_name": project.remote_project_name})
log_operation_event(domain=DOMAIN, event_type=PrivatePortraitEventType.VALIDATE_SESSION_CREATE.value, event_status=PrivatePortraitEventStatus.SUCCESS.value, source=PrivatePortraitEventSource.API.value, user_id=user_id, project_id=project.id, session_id=session.id, detail={"remote_project_name": project.remote_project_name, "library_type": project.library_type})
return session
except Exception as exc:
session.status = PrivatePortraitValidateSessionStatus.FAILED.value
@@ -259,6 +268,7 @@ async def create_validate_session(db: AsyncSession, *, user_id: str, project_id:
log_operation_error(domain=DOMAIN, event_type=PrivatePortraitEventType.VALIDATE_SESSION_CREATE_FAILED.value, source=PrivatePortraitEventSource.API.value, user_id=user_id, project_id=project.id, session_id=session.id, exc=exc)
raise
async def get_validate_session(db: AsyncSession, *, user_id: str | None, session_id: str) -> PrivatePortraitValidateSession:
filters = [PrivatePortraitValidateSession.id == session_id]
if user_id is not None:
@@ -304,7 +314,7 @@ async def handle_validate_callback(db: AsyncSession, *, session_id: str, query_p
raise HTTPException(status_code=400, detail=session.error_message)
try:
existing_group = await _get_existing_active_group(db, project_id=session.project_id)
existing_group = await _get_existing_active_group(db, project_id=session.project_id, library_type=PrivatePortraitLibraryType.REAL_PERSON.value)
if existing_group:
session.remote_group_id = existing_group.remote_group_id
session.status = PrivatePortraitValidateSessionStatus.GROUP_ACTIVE.value
@@ -330,10 +340,11 @@ async def handle_validate_callback(db: AsyncSession, *, session_id: str, query_p
id=generate_id(),
user_id=session.user_id,
project_id=session.project_id,
library_type=PrivatePortraitLibraryType.REAL_PERSON.value,
remote_group_id=group_id,
remote_group_name=remote_group_name,
remote_project_name=session.remote_project_name,
group_type=PRIVATE_PORTRAIT_GROUP_TYPE,
group_type=PRIVATE_PORTRAIT_REAL_PERSON_GROUP_TYPE,
status=PrivatePortraitAssetGroupStatus.ACTIVE.value,
raw_response_json=_json(resp),
)
@@ -347,7 +358,7 @@ async def handle_validate_callback(db: AsyncSession, *, session_id: str, query_p
await refresh_project_counters(db, [session.project_id])
await db.flush()
await db.refresh(session)
log_operation_event(domain=DOMAIN, event_type=PrivatePortraitEventType.VALIDATE_GET_RESULT_SUCCESS.value, event_status=PrivatePortraitEventStatus.SUCCESS.value, source=PrivatePortraitEventSource.CALLBACK.value, user_id=session.user_id, project_id=session.project_id, session_id=session.id, group_id=group.id, detail={"remote_group_id": group_id, "remote_project_name": session.remote_project_name})
log_operation_event(domain=DOMAIN, event_type=PrivatePortraitEventType.VALIDATE_GET_RESULT_SUCCESS.value, event_status=PrivatePortraitEventStatus.SUCCESS.value, source=PrivatePortraitEventSource.CALLBACK.value, user_id=session.user_id, project_id=session.project_id, session_id=session.id, group_id=group.id, detail={"remote_group_id": group_id, "remote_project_name": session.remote_project_name, "library_type": PrivatePortraitLibraryType.REAL_PERSON.value})
return session
except Exception as exc:
session.status = PrivatePortraitValidateSessionStatus.FAILED.value
@@ -358,51 +369,67 @@ async def handle_validate_callback(db: AsyncSession, *, session_id: str, query_p
log_operation_error(domain=DOMAIN, event_type=PrivatePortraitEventType.VALIDATE_GET_RESULT_FAILED.value, source=PrivatePortraitEventSource.CALLBACK.value, user_id=session.user_id, project_id=session.project_id, session_id=session.id, exc=exc)
raise
async def get_project_active_group(db: AsyncSession, *, user_id: str, project_id: str) -> PrivatePortraitAssetGroup:
project = await get_user_project(db, user_id=user_id, project_id=project_id)
async def get_project_active_group(db: AsyncSession, *, user_id: str, project_id: str, library_type: str | None = None) -> PrivatePortraitAssetGroup:
project = await get_user_project(db, user_id=user_id, project_id=project_id, library_type=library_type)
if project.status != PrivatePortraitProjectStatus.ACTIVE.value:
raise HTTPException(status_code=400, detail="请先完成真人授权认证，再上传素材")
result = await db.execute(select(PrivatePortraitAssetGroup).where(PrivatePortraitAssetGroup.user_id == user_id, PrivatePortraitAssetGroup.project_id == project_id, PrivatePortraitAssetGroup.status == PrivatePortraitAssetGroupStatus.ACTIVE.value, PrivatePortraitAssetGroup.deleted_at.is_(None)).order_by(PrivatePortraitAssetGroup.created_at.desc()).limit(1))
detail = "请先完成真人授权认证，再上传素材" if project.library_type == PrivatePortraitLibraryType.REAL_PERSON.value else "虚拟人像素材组尚未创建成功，不能上传素材"
raise HTTPException(status_code=400, detail=detail)
result = await db.execute(
select(PrivatePortraitAssetGroup)
.where(
PrivatePortraitAssetGroup.user_id == user_id,
PrivatePortraitAssetGroup.project_id == project_id,
PrivatePortraitAssetGroup.library_type == project.library_type,
PrivatePortraitAssetGroup.status == PrivatePortraitAssetGroupStatus.ACTIVE.value,
PrivatePortraitAssetGroup.deleted_at.is_(None),
)
.order_by(PrivatePortraitAssetGroup.created_at.desc())
.limit(1)
)
group = result.scalar_one_or_none()
if not group:
raise HTTPException(status_code=400, detail="请先完成真人授权认证，再上传素材")
raise HTTPException(status_code=400, detail="项目没有可用的远程素材组")
return group
async def create_asset(db: AsyncSession, *, user_id: str, project_id: str, payload: PrivatePortraitAssetCreate) -> PrivatePortraitAsset:
project = await get_user_project(db, user_id=user_id, project_id=project_id)
async def create_asset(
db: AsyncSession,
*,
user_id: str,
project_id: str,
payload: PrivatePortraitAssetCreate,
library_type: str | None = None,
) -> PrivatePortraitAsset:
_assert_enabled_asset_type(payload.asset_type)
project = await get_user_project(db, user_id=user_id, project_id=project_id, library_type=library_type)
if project.status != PrivatePortraitProjectStatus.ACTIVE.value:
raise HTTPException(status_code=400, detail="项目正在真人认证或认证未通过，不能上传素材")
if payload.asset_type != PrivatePortraitAssetType.IMAGE.value:
raise HTTPException(status_code=400, detail="第一版真人素材库仅开放 Image 图片素材")
user = await _lock_user_for_upload(db, user_id=user_id)
limit = int(getattr(user, "private_portrait_image_limit", PRIVATE_PORTRAIT_DEFAULT_IMAGE_LIMIT) or 0)
if limit <= 0:
log_operation_event(domain=DOMAIN, event_type=PrivatePortraitEventType.ASSET_CREATE_REJECT_DISABLED.value, event_status=PrivatePortraitEventStatus.FAILED.value, source=PrivatePortraitEventSource.API.value, user_id=user_id, project_id=project_id, message="用户真人素材模块未启用")
raise HTTPException(status_code=403, detail="真人素材库未启用")
current_count = await count_user_counting_image_assets(db, user_id=user_id)
if current_count >= limit:
log_operation_event(domain=DOMAIN, event_type=PrivatePortraitEventType.ASSET_CREATE_REJECT_MAX_LIMIT.value, event_status=PrivatePortraitEventStatus.FAILED.value, source=PrivatePortraitEventSource.API.value, user_id=user_id, project_id=project_id, detail={"current_count": current_count, "limit": limit})
raise HTTPException(status_code=400, detail=f"你的真人素材库最多可上传 {limit} 张图片，请删除已有素材后再上传")
raise HTTPException(status_code=400, detail="项目未激活，不能上传素材")
group = await get_project_active_group(db, user_id=user_id, project_id=project.id)
limit, current_count = await ensure_private_portrait_asset_quota_available(db, user_id=user_id, project_id=project_id, library_type=project.library_type, asset_type=payload.asset_type)
group = await get_project_active_group(db, user_id=user_id, project_id=project.id, library_type=project.library_type)
public_url = _public_url(payload.url)
asset = PrivatePortraitAsset(
id=generate_id(),
user_id=user_id,
project_id=project.id,
group_id=group.id,
library_type=project.library_type,
remote_group_id=group.remote_group_id,
remote_project_name=project.remote_project_name,
asset_type=payload.asset_type,
name=payload.name,
source_url=public_url,
preview_url=payload.url,
video_duration=payload.video_duration,
video_cover_url=payload.video_cover_url,
file_size=payload.file_size,
mime_type=payload.mime_type,
status=PrivatePortraitAssetStatus.CREATING.value,
)
db.add(asset)
await db.flush()
log_operation_event(domain=DOMAIN, event_type=PrivatePortraitEventType.ASSET_CREATE_START.value, event_status=PrivatePortraitEventStatus.PENDING.value, source=PrivatePortraitEventSource.API.value, user_id=user_id, project_id=project.id, group_id=group.id, asset_id=asset.id, detail={"limit": limit, "current_count": current_count, "remote_project_name": project.remote_project_name})
log_operation_event(domain=DOMAIN, event_type=PrivatePortraitEventType.ASSET_CREATE_START.value, event_status=PrivatePortraitEventStatus.PENDING.value, source=PrivatePortraitEventSource.API.value, user_id=user_id, project_id=project.id, group_id=group.id, asset_id=asset.id, detail={"asset_limit": limit, "used_asset_count": current_count, "library_type": project.library_type, "asset_type": payload.asset_type, "remote_project_name": project.remote_project_name})
try:
remote_resp = await ArkPrivateAssetClient().create_asset(project_name=project.remote_project_name, group_id=group.remote_group_id, url=public_url, asset_type=payload.asset_type, name=payload.name)
remote_asset_id = remote_resp.get("Id") or remote_resp.get("AssetId") or remote_resp.get("assetId")
@@ -411,14 +438,12 @@ async def create_asset(db: AsyncSession, *, user_id: str, project_id: str, paylo
now = datetime.now(timezone.utc)
asset.remote_asset_id = remote_asset_id
asset.status = PrivatePortraitAssetStatus.PROCESSING.value
asset.next_poll_at = now + timedelta(seconds=PRIVATE_PORTRAIT_ASSET_POLL_INTERVAL_SECONDS)
asset.next_poll_at = now + timedelta(seconds=_poll_interval_seconds(asset.asset_type))
asset.raw_response_json = _json(remote_resp)
await refresh_project_counters(db, [project.id])
await db.flush()
# created_at / updated_at 来自数据库默认值或 onupdate，flush 后可能处于 expired 状态。
# 在 async SQLAlchemy 下，响应转换时同步读取 expired 字段会触发 MissingGreenlet。
await db.refresh(asset)
log_operation_event(domain=DOMAIN, event_type=PrivatePortraitEventType.ASSET_CREATE_SUCCESS.value, event_status=PrivatePortraitEventStatus.SUCCESS.value, source=PrivatePortraitEventSource.API.value, user_id=user_id, project_id=project.id, group_id=group.id, asset_id=asset.id, detail={"remote_asset_id": remote_asset_id, "remote_project_name": project.remote_project_name})
log_operation_event(domain=DOMAIN, event_type=PrivatePortraitEventType.ASSET_CREATE_SUCCESS.value, event_status=PrivatePortraitEventStatus.SUCCESS.value, source=PrivatePortraitEventSource.API.value, user_id=user_id, project_id=project.id, group_id=group.id, asset_id=asset.id, detail={"remote_asset_id": remote_asset_id, "remote_project_name": project.remote_project_name, "library_type": project.library_type, "asset_type": asset.asset_type})
return asset
except Exception as exc:
asset.status = PrivatePortraitAssetStatus.FAILED.value
@@ -434,11 +459,11 @@ async def sync_asset_status(db: AsyncSession, *, user_id: str | None, asset_id:
filters.append(PrivatePortraitAsset.user_id == user_id)
asset = (await db.execute(select(PrivatePortraitAsset).where(*filters).limit(1))).scalar_one_or_none()
if not asset:
raise HTTPException(status_code=404, detail="真人素材不存在")
raise HTTPException(status_code=404, detail="私域人像素材不存在")
if asset.deleted_at is not None:
raise HTTPException(status_code=400, detail="真人素材已删除")
raise HTTPException(status_code=400, detail="私域人像素材已删除")
if not asset.remote_asset_id:
raise HTTPException(status_code=400, detail="真人素材尚未创建远程 Asset")
raise HTTPException(status_code=400, detail="私域人像素材尚未创建远程 Asset")
source = PrivatePortraitEventSource.CELERY.value if user_id is None else PrivatePortraitEventSource.API.value
log_operation_event(
@@ -449,12 +474,7 @@ async def sync_asset_status(db: AsyncSession, *, user_id: str | None, asset_id:
user_id=asset.user_id,
project_id=asset.project_id,
asset_id=asset.id,
detail={
"status": asset.status,
"poll_count": int(asset.poll_count or 0),
"remote_asset_id": asset.remote_asset_id,
"remote_project_name": asset.remote_project_name,
},
detail={"status": asset.status, "poll_count": int(asset.poll_count or 0), "remote_asset_id": asset.remote_asset_id, "remote_project_name": asset.remote_project_name, "library_type": asset.library_type, "asset_type": asset.asset_type},
)
try:
remote_resp = await ArkPrivateAssetClient(for_celery=(user_id is None)).get_asset(project_name=asset.remote_project_name, asset_id=asset.remote_asset_id)
@@ -467,24 +487,15 @@ async def sync_asset_status(db: AsyncSession, *, user_id: str | None, asset_id:
asset.status = status
asset.remote_url = remote_resp.get("URL") or remote_resp.get("url") or asset.remote_url
asset.moderation_json = _json(remote_resp.get("Moderation") or remote_resp.get("moderation"))
max_count = _poll_max_count(asset.asset_type)
if asset.status == PrivatePortraitAssetStatus.PROCESSING.value and asset.poll_count >= PRIVATE_PORTRAIT_ASSET_POLL_MAX_COUNT:
if asset.status == PrivatePortraitAssetStatus.PROCESSING.value and asset.poll_count >= max_count:
asset.status = PrivatePortraitAssetStatus.FAILED.value
asset.error_message = "素材入库轮询超时"
asset.next_poll_at = None
log_operation_event(
domain=DOMAIN,
event_type=PrivatePortraitEventType.ASSET_POLL_TIMEOUT.value,
event_status=PrivatePortraitEventStatus.FAILED.value,
source=source,
user_id=asset.user_id,
project_id=asset.project_id,
asset_id=asset.id,
detail={"poll_count": asset.poll_count, "max_count": PRIVATE_PORTRAIT_ASSET_POLL_MAX_COUNT, "remote_asset_id": asset.remote_asset_id},
error=asset.error_message,
)
log_operation_event(domain=DOMAIN, event_type=PrivatePortraitEventType.ASSET_POLL_TIMEOUT.value, event_status=PrivatePortraitEventStatus.FAILED.value, source=source, user_id=asset.user_id, project_id=asset.project_id, asset_id=asset.id, detail={"poll_count": asset.poll_count, "max_count": max_count, "remote_asset_id": asset.remote_asset_id, "library_type": asset.library_type, "asset_type": asset.asset_type}, error=asset.error_message)
elif asset.status == PrivatePortraitAssetStatus.PROCESSING.value:
asset.next_poll_at = now + timedelta(seconds=PRIVATE_PORTRAIT_ASSET_POLL_INTERVAL_SECONDS)
asset.next_poll_at = now + timedelta(seconds=_poll_interval_seconds(asset.asset_type))
else:
asset.next_poll_at = None
@@ -493,23 +504,25 @@ async def sync_asset_status(db: AsyncSession, *, user_id: str | None, asset_id:
await refresh_project_counters(db, [asset.project_id])
await db.flush()
await db.refresh(asset)
log_operation_event(
domain=DOMAIN,
event_type=PrivatePortraitEventType.ASSET_SYNC_SUCCESS.value,
event_status=PrivatePortraitEventStatus.SUCCESS.value,
source=source,
user_id=asset.user_id,
project_id=asset.project_id,
asset_id=asset.id,
detail={"status": asset.status, "remote_asset_id": asset.remote_asset_id, "next_poll_at": asset.next_poll_at, "poll_count": asset.poll_count},
)
log_operation_event(domain=DOMAIN, event_type=PrivatePortraitEventType.ASSET_SYNC_SUCCESS.value, event_status=PrivatePortraitEventStatus.SUCCESS.value, source=source, user_id=asset.user_id, project_id=asset.project_id, asset_id=asset.id, detail={"status": asset.status, "remote_asset_id": asset.remote_asset_id, "next_poll_at": asset.next_poll_at, "poll_count": asset.poll_count, "library_type": asset.library_type, "asset_type": asset.asset_type})
return asset
except Exception as exc:
log_operation_error(domain=DOMAIN, event_type=PrivatePortraitEventType.ASSET_SYNC_FAILED.value, source=source, user_id=asset.user_id, project_id=asset.project_id, asset_id=asset.id, exc=exc)
raise
async def list_assets(db: AsyncSession, *, user_id: str | None, project_id: str | None = None, status: str | None = None, keyword: str | None = None, page: int = 1, page_size: int = 20) -> tuple[list[PrivatePortraitAsset], int, dict[str, str]]:
async def list_assets(
db: AsyncSession,
*,
user_id: str | None,
project_id: str | None = None,
status: str | None = None,
keyword: str | None = None,
page: int = 1,
page_size: int = 20,
library_type: str | None = None,
asset_type: str | None = None,
) -> tuple[list[PrivatePortraitAsset], int, dict[str, str]]:
page = max(1, page)
page_size = min(max(1, page_size), 100)
filters = [PrivatePortraitAsset.deleted_at.is_(None)]
@@ -517,6 +530,10 @@ async def list_assets(db: AsyncSession, *, user_id: str | None, project_id: str
filters.append(PrivatePortraitAsset.user_id == user_id)
if project_id:
filters.append(PrivatePortraitAsset.project_id == project_id)
if library_type:
filters.append(PrivatePortraitAsset.library_type == library_type)
if asset_type:
filters.append(PrivatePortraitAsset.asset_type == asset_type)
if status:
filters.append(PrivatePortraitAsset.status == status)
if keyword:
@@ -532,15 +549,35 @@ async def list_assets(db: AsyncSession, *, user_id: str | None, project_id: str
return assets, int(total or 0), project_name_map
async def list_selectable_assets(db: AsyncSession, *, user_id: str, project_id: str | None = None, keyword: str | None = None, page: int = 1, page_size: int = 20) -> tuple[list[PrivatePortraitSelectableAssetOut], int]:
assets, total, project_name_map = await list_assets(db, user_id=user_id, project_id=project_id, status=PrivatePortraitAssetStatus.ACTIVE.value, keyword=keyword, page=page, page_size=page_size)
return [PrivatePortraitSelectableAssetOut(id=asset.id, project_id=asset.project_id, project_name=project_name_map.get(asset.project_id, ""), name=asset.name, asset_type=asset.asset_type, preview_url=asset.preview_url or asset.remote_url, status=asset.status, created_at=asset.created_at) for asset in assets], total
async def list_selectable_assets(db: AsyncSession, *, user_id: str, project_id: str | None = None, keyword: str | None = None, page: int = 1, page_size: int = 20, library_type: str | None = None, asset_type: str | None = None) -> tuple[list[PrivatePortraitSelectableAssetOut], int]:
assets, total, project_name_map = await list_assets(db, user_id=user_id, project_id=project_id, status=PrivatePortraitAssetStatus.ACTIVE.value, keyword=keyword, page=page, page_size=page_size, library_type=library_type, asset_type=asset_type)
return [
PrivatePortraitSelectableAssetOut(
id=asset.id,
project_id=asset.project_id,
project_name=project_name_map.get(asset.project_id, ""),
library_type=asset.library_type,
name=asset.name,
asset_type=asset.asset_type,
preview_url=asset.preview_url or asset.remote_url,
display_url=_asset_display_url(asset),
provider_url=_provider_url(asset),
video_duration=asset.video_duration,
video_cover_url=asset.video_cover_url,
status=asset.status,
created_at=asset.created_at,
)
for asset in assets
], total
async def soft_delete_asset(db: AsyncSession, *, user_id: str, asset_id: str) -> PrivatePortraitAsset:
asset = (await db.execute(select(PrivatePortraitAsset).where(PrivatePortraitAsset.id == asset_id, PrivatePortraitAsset.user_id == user_id, PrivatePortraitAsset.deleted_at.is_(None)).limit(1))).scalar_one_or_none()
async def soft_delete_asset(db: AsyncSession, *, user_id: str, asset_id: str, library_type: str | None = None) -> PrivatePortraitAsset:
filters = [PrivatePortraitAsset.id == asset_id, PrivatePortraitAsset.user_id == user_id, PrivatePortraitAsset.deleted_at.is_(None)]
if library_type:
filters.append(PrivatePortraitAsset.library_type == library_type)
asset = (await db.execute(select(PrivatePortraitAsset).where(*filters).limit(1))).scalar_one_or_none()
if not asset:
raise HTTPException(status_code=404, detail="真人素材不存在")
raise HTTPException(status_code=404, detail="私域人像素材不存在")
now = datetime.now(timezone.utc)
asset.deleted_at = now
asset.status = PrivatePortraitAssetStatus.LOCAL_DELETED.value
@@ -548,64 +585,30 @@ async def soft_delete_asset(db: AsyncSession, *, user_id: str, asset_id: str) ->
await refresh_project_counters(db, [asset.project_id])
await db.flush()
await db.refresh(asset)
log_operation_event(domain=DOMAIN, event_type=PrivatePortraitEventType.ASSET_DELETE_LOCAL.value, event_status=PrivatePortraitEventStatus.SUCCESS.value, source=PrivatePortraitEventSource.API.value, user_id=user_id, project_id=asset.project_id, asset_id=asset.id, detail={"remote_asset_id": asset.remote_asset_id, "remote_project_name": asset.remote_project_name})
log_operation_event(domain=DOMAIN, event_type=PrivatePortraitEventType.ASSET_DELETE_LOCAL.value, event_status=PrivatePortraitEventStatus.SUCCESS.value, source=PrivatePortraitEventSource.API.value, user_id=user_id, project_id=asset.project_id, asset_id=asset.id, detail={"remote_asset_id": asset.remote_asset_id, "remote_project_name": asset.remote_project_name, "library_type": asset.library_type, "asset_type": asset.asset_type})
return asset
async def delete_asset_remote(db: AsyncSession, *, asset_id: str) -> None:
asset = (await db.execute(select(PrivatePortraitAsset).where(PrivatePortraitAsset.id == asset_id).limit(1))).scalar_one_or_none()
if not asset:
log_operation_event(
domain=DOMAIN,
event_type=PrivatePortraitEventType.ASSET_DELETE_REMOTE_START.value,
event_status=PrivatePortraitEventStatus.SKIPPED.value,
source=PrivatePortraitEventSource.CELERY.value,
asset_id=asset_id,
message="远程删除跳过：本地素材不存在",
)
log_operation_event(domain=DOMAIN, event_type=PrivatePortraitEventType.ASSET_DELETE_REMOTE_START.value, event_status=PrivatePortraitEventStatus.SKIPPED.value, source=PrivatePortraitEventSource.CELERY.value, asset_id=asset_id, message="远程删除跳过：本地素材不存在")
return
if not asset.remote_asset_id:
asset.remote_delete_status = PrivatePortraitRemoteDeleteStatus.SKIPPED.value
asset.remote_delete_error = None
await db.flush()
log_operation_event(
domain=DOMAIN,
event_type=PrivatePortraitEventType.ASSET_DELETE_REMOTE_SUCCESS.value,
event_status=PrivatePortraitEventStatus.SKIPPED.value,
source=PrivatePortraitEventSource.CELERY.value,
user_id=asset.user_id,
project_id=asset.project_id,
asset_id=asset.id,
message="远程删除跳过：素材没有 remote_asset_id",
)
log_operation_event(domain=DOMAIN, event_type=PrivatePortraitEventType.ASSET_DELETE_REMOTE_SUCCESS.value, event_status=PrivatePortraitEventStatus.SKIPPED.value, source=PrivatePortraitEventSource.CELERY.value, user_id=asset.user_id, project_id=asset.project_id, asset_id=asset.id, message="远程删除跳过：素材没有 remote_asset_id")
return
now = datetime.now(timezone.utc)
log_operation_event(
domain=DOMAIN,
event_type=PrivatePortraitEventType.ASSET_DELETE_REMOTE_START.value,
event_status=PrivatePortraitEventStatus.PENDING.value,
source=PrivatePortraitEventSource.CELERY.value,
user_id=asset.user_id,
project_id=asset.project_id,
asset_id=asset.id,
detail={"remote_asset_id": asset.remote_asset_id, "remote_project_name": asset.remote_project_name},
)
log_operation_event(domain=DOMAIN, event_type=PrivatePortraitEventType.ASSET_DELETE_REMOTE_START.value, event_status=PrivatePortraitEventStatus.PENDING.value, source=PrivatePortraitEventSource.CELERY.value, user_id=asset.user_id, project_id=asset.project_id, asset_id=asset.id, detail={"remote_asset_id": asset.remote_asset_id, "remote_project_name": asset.remote_project_name, "library_type": asset.library_type, "asset_type": asset.asset_type})
try:
await ArkPrivateAssetClient(for_celery=True).delete_asset(project_name=asset.remote_project_name, asset_id=asset.remote_asset_id)
asset.status = PrivatePortraitAssetStatus.REMOTE_DELETED.value
asset.remote_delete_status = PrivatePortraitRemoteDeleteStatus.SUCCESS.value
asset.remote_deleted_at = now
asset.remote_delete_error = None
log_operation_event(
domain=DOMAIN,
event_type=PrivatePortraitEventType.ASSET_DELETE_REMOTE_SUCCESS.value,
event_status=PrivatePortraitEventStatus.SUCCESS.value,
source=PrivatePortraitEventSource.CELERY.value,
user_id=asset.user_id,
project_id=asset.project_id,
asset_id=asset.id,
detail={"remote_asset_id": asset.remote_asset_id, "remote_project_name": asset.remote_project_name},
)
log_operation_event(domain=DOMAIN, event_type=PrivatePortraitEventType.ASSET_DELETE_REMOTE_SUCCESS.value, event_status=PrivatePortraitEventStatus.SUCCESS.value, source=PrivatePortraitEventSource.CELERY.value, user_id=asset.user_id, project_id=asset.project_id, asset_id=asset.id, detail={"remote_asset_id": asset.remote_asset_id, "remote_project_name": asset.remote_project_name, "library_type": asset.library_type})
except Exception as exc:
asset.status = PrivatePortraitAssetStatus.DELETE_FAILED.value
asset.remote_delete_status = PrivatePortraitRemoteDeleteStatus.FAILED.value
@@ -619,45 +622,18 @@ async def _delete_asset_group_remote(db: AsyncSession, *, group: PrivatePortrait
group.remote_delete_status = PrivatePortraitRemoteDeleteStatus.SKIPPED.value
group.remote_delete_error = None
await db.flush()
log_operation_event(
domain=DOMAIN,
event_type=PrivatePortraitEventType.PROJECT_DELETE_REMOTE_SUCCESS.value,
event_status=PrivatePortraitEventStatus.SKIPPED.value,
source=PrivatePortraitEventSource.CELERY.value,
user_id=group.user_id,
project_id=group.project_id,
group_id=group.id,
message="远程删除跳过：素材组没有 remote_group_id",
)
log_operation_event(domain=DOMAIN, event_type=PrivatePortraitEventType.PROJECT_DELETE_REMOTE_SUCCESS.value, event_status=PrivatePortraitEventStatus.SKIPPED.value, source=PrivatePortraitEventSource.CELERY.value, user_id=group.user_id, project_id=group.project_id, group_id=group.id, message="远程删除跳过：素材组没有 remote_group_id")
return
client = client or ArkPrivateAssetClient(for_celery=True)
now = datetime.now(timezone.utc)
log_operation_event(
domain=DOMAIN,
event_type=PrivatePortraitEventType.PROJECT_DELETE_REMOTE_START.value,
event_status=PrivatePortraitEventStatus.PENDING.value,
source=PrivatePortraitEventSource.CELERY.value,
user_id=group.user_id,
project_id=group.project_id,
group_id=group.id,
detail={"remote_group_id": group.remote_group_id, "remote_project_name": group.remote_project_name},
)
log_operation_event(domain=DOMAIN, event_type=PrivatePortraitEventType.PROJECT_DELETE_REMOTE_START.value, event_status=PrivatePortraitEventStatus.PENDING.value, source=PrivatePortraitEventSource.CELERY.value, user_id=group.user_id, project_id=group.project_id, group_id=group.id, detail={"remote_group_id": group.remote_group_id, "remote_project_name": group.remote_project_name, "library_type": group.library_type})
try:
await client.delete_asset_group(project_name=group.remote_project_name, group_id=group.remote_group_id)
group.status = PrivatePortraitAssetGroupStatus.REMOTE_DELETED.value
group.remote_delete_status = PrivatePortraitRemoteDeleteStatus.SUCCESS.value
group.remote_deleted_at = now
group.remote_delete_error = None
log_operation_event(
domain=DOMAIN,
event_type=PrivatePortraitEventType.PROJECT_DELETE_REMOTE_SUCCESS.value,
event_status=PrivatePortraitEventStatus.SUCCESS.value,
source=PrivatePortraitEventSource.CELERY.value,
user_id=group.user_id,
project_id=group.project_id,
group_id=group.id,
detail={"remote_group_id": group.remote_group_id, "remote_project_name": group.remote_project_name},
)
log_operation_event(domain=DOMAIN, event_type=PrivatePortraitEventType.PROJECT_DELETE_REMOTE_SUCCESS.value, event_status=PrivatePortraitEventStatus.SUCCESS.value, source=PrivatePortraitEventSource.CELERY.value, user_id=group.user_id, project_id=group.project_id, group_id=group.id, detail={"remote_group_id": group.remote_group_id, "remote_project_name": group.remote_project_name, "library_type": group.library_type})
except Exception as exc:
group.status = PrivatePortraitAssetGroupStatus.DELETE_FAILED.value
group.remote_delete_status = PrivatePortraitRemoteDeleteStatus.FAILED.value
@@ -667,14 +643,7 @@ async def _delete_asset_group_remote(db: AsyncSession, *, group: PrivatePortrait
async def delete_project_remote(db: AsyncSession, *, project_id: str) -> None:
log_operation_event(
domain=DOMAIN,
event_type=PrivatePortraitEventType.PROJECT_DELETE_REMOTE_START.value,
event_status=PrivatePortraitEventStatus.PENDING.value,
source=PrivatePortraitEventSource.CELERY.value,
project_id=project_id,
message="开始远程删除真人素材项目资源",
)
log_operation_event(domain=DOMAIN, event_type=PrivatePortraitEventType.PROJECT_DELETE_REMOTE_START.value, event_status=PrivatePortraitEventStatus.PENDING.value, source=PrivatePortraitEventSource.CELERY.value, project_id=project_id, message="开始远程删除私域人像素材项目资源")
rows = await db.execute(select(PrivatePortraitAsset).where(PrivatePortraitAsset.project_id == project_id))
for asset in rows.scalars().all():
await delete_asset_remote(db, asset_id=asset.id)
@@ -682,14 +651,7 @@ async def delete_project_remote(db: AsyncSession, *, project_id: str) -> None:
client = ArkPrivateAssetClient(for_celery=True)
for group in groups.scalars().all():
await _delete_asset_group_remote(db, group=group, client=client)
log_operation_event(
domain=DOMAIN,
event_type=PrivatePortraitEventType.PROJECT_DELETE_REMOTE_SUCCESS.value,
event_status=PrivatePortraitEventStatus.SUCCESS.value,
source=PrivatePortraitEventSource.CELERY.value,
project_id=project_id,
message="远程删除真人素材项目资源完成",
)
log_operation_event(domain=DOMAIN, event_type=PrivatePortraitEventType.PROJECT_DELETE_REMOTE_SUCCESS.value, event_status=PrivatePortraitEventStatus.SUCCESS.value, source=PrivatePortraitEventSource.CELERY.value, project_id=project_id, message="远程删除私域人像素材项目资源完成")
await db.flush()
@@ -707,13 +669,7 @@ async def poll_due_assets_once(db: AsyncSession, *, limit: int) -> int:
.limit(limit)
)
ids = [row[0] for row in rows.all()]
log_operation_event(
domain=DOMAIN,
event_type=PrivatePortraitEventType.SYNC_DUE_ASSETS_START.value,
event_status=PrivatePortraitEventStatus.PENDING.value,
source=PrivatePortraitEventSource.CELERY.value,
detail={"limit": limit, "matched_count": len(ids)},
)
log_operation_event(domain=DOMAIN, event_type=PrivatePortraitEventType.SYNC_DUE_ASSETS_START.value, event_status=PrivatePortraitEventStatus.PENDING.value, source=PrivatePortraitEventSource.CELERY.value, detail={"limit": limit, "matched_count": len(ids)})
success_count = 0
failed_count = 0
for asset_id in ids:
@@ -722,38 +678,15 @@ async def poll_due_assets_once(db: AsyncSession, *, limit: int) -> int:
success_count += 1
except Exception as exc:
failed_count += 1
log_operation_error(
domain=DOMAIN,
event_type=PrivatePortraitEventType.ASSET_POLL_FAILED.value,
source=PrivatePortraitEventSource.CELERY.value,
asset_id=asset_id,
exc=exc,
)
log_operation_event(
domain=DOMAIN,
event_type=PrivatePortraitEventType.SYNC_DUE_ASSETS_DONE.value,
event_status=PrivatePortraitEventStatus.SUCCESS.value if failed_count == 0 else PrivatePortraitEventStatus.WARNING.value,
source=PrivatePortraitEventSource.CELERY.value,
detail={"matched_count": len(ids), "success_count": success_count, "failed_count": failed_count},
)
log_operation_error(domain=DOMAIN, event_type=PrivatePortraitEventType.ASSET_POLL_FAILED.value, source=PrivatePortraitEventSource.CELERY.value, asset_id=asset_id, exc=exc)
log_operation_event(domain=DOMAIN, event_type=PrivatePortraitEventType.SYNC_DUE_ASSETS_DONE.value, event_status=PrivatePortraitEventStatus.SUCCESS.value if failed_count == 0 else PrivatePortraitEventStatus.WARNING.value, source=PrivatePortraitEventSource.CELERY.value, detail={"matched_count": len(ids), "success_count": success_count, "failed_count": failed_count})
return len(ids)
async def recover_remote_deletes_once(db: AsyncSession, *, limit: int) -> dict[str, int]:
log_operation_event(
domain=DOMAIN,
event_type=PrivatePortraitEventType.REMOTE_DELETE_RECOVERY_START.value,
event_status=PrivatePortraitEventStatus.PENDING.value,
source=PrivatePortraitEventSource.CELERY.value,
detail={"limit": limit},
)
log_operation_event(domain=DOMAIN, event_type=PrivatePortraitEventType.REMOTE_DELETE_RECOVERY_START.value, event_status=PrivatePortraitEventStatus.PENDING.value, source=PrivatePortraitEventSource.CELERY.value, detail={"limit": limit})
statuses = [PrivatePortraitRemoteDeleteStatus.PENDING.value, PrivatePortraitRemoteDeleteStatus.FAILED.value]
asset_rows = await db.execute(
select(PrivatePortraitAsset.id)
.where(PrivatePortraitAsset.remote_delete_status.in_(statuses))
.order_by(PrivatePortraitAsset.updated_at.asc())
.limit(limit)
)
asset_rows = await db.execute(select(PrivatePortraitAsset.id).where(PrivatePortraitAsset.remote_delete_status.in_(statuses)).order_by(PrivatePortraitAsset.updated_at.asc()).limit(limit))
asset_ids = [row[0] for row in asset_rows.all()]
for asset_id in asset_ids:
await delete_asset_remote(db, asset_id=asset_id)
@@ -761,12 +694,7 @@ async def recover_remote_deletes_once(db: AsyncSession, *, limit: int) -> dict[s
remaining = max(0, limit - len(asset_ids))
group_count = 0
if remaining > 0:
group_rows = await db.execute(
select(PrivatePortraitAssetGroup)
.where(PrivatePortraitAssetGroup.remote_delete_status.in_(statuses))
.order_by(PrivatePortraitAssetGroup.updated_at.asc())
.limit(remaining)
)
group_rows = await db.execute(select(PrivatePortraitAssetGroup).where(PrivatePortraitAssetGroup.remote_delete_status.in_(statuses)).order_by(PrivatePortraitAssetGroup.updated_at.asc()).limit(remaining))
client = ArkPrivateAssetClient(for_celery=True)
groups = list(group_rows.scalars().all())
group_count = len(groups)
@@ -774,12 +702,5 @@ async def recover_remote_deletes_once(db: AsyncSession, *, limit: int) -> dict[s
await _delete_asset_group_remote(db, group=group, client=client)
result = {"asset_count": len(asset_ids), "group_count": group_count, "total_count": len(asset_ids) + group_count}
log_operation_event(
domain=DOMAIN,
event_type=PrivatePortraitEventType.REMOTE_DELETE_RECOVERY_DONE.value,
event_status=PrivatePortraitEventStatus.SUCCESS.value,
source=PrivatePortraitEventSource.CELERY.value,
detail=result,
)
log_operation_event(domain=DOMAIN, event_type=PrivatePortraitEventType.REMOTE_DELETE_RECOVERY_DONE.value, event_status=PrivatePortraitEventStatus.SUCCESS.value, source=PrivatePortraitEventSource.CELERY.value, detail=result)
return result
video-gen-api/app/services/private_portrait/project_service.py
+124 -35
View File
@@ -11,9 +11,11 @@ from app.enums.private_portrait import (
PRIVATE_PORTRAIT_REMOTE_PROJECT_NAME,
PrivatePortraitAssetGroupStatus,
PrivatePortraitAssetStatus,
PrivatePortraitAssetType,
PrivatePortraitEventSource,
PrivatePortraitEventStatus,
PrivatePortraitEventType,
PrivatePortraitLibraryType,
PrivatePortraitProjectStatus,
PrivatePortraitRemoteDeleteStatus,
)
@@ -27,16 +29,24 @@ DOMAIN = "private_portrait"
def _safe_slug(value: str, *, max_length: int = 80) -> str:
value = (value or "").strip().lower()
# 先保留常见英文数字连字符；中文等字符统一转 _，仅用于本地项目 slug。
value = re.sub(r"[^a-z0-9_-]+", "_", value)
value = re.sub(r"_+", "_", value).strip("_-")
return (value[:max_length] or "project")
def _status_for_created_project(library_type: str) -> str:
if library_type == PrivatePortraitLibraryType.REAL_PERSON.value:
return PrivatePortraitProjectStatus.VALIDATING.value
if library_type == PrivatePortraitLibraryType.AIGC_VIRTUAL.value:
return PrivatePortraitProjectStatus.CREATING_REMOTE_GROUP.value
raise HTTPException(status_code=400, detail="library_type 不支持")
def project_to_out(project: PrivatePortraitProject, *, include_user: bool = False) -> PrivatePortraitProjectOut:
return PrivatePortraitProjectOut(
id=project.id,
user_id=project.user_id if include_user else None,
library_type=project.library_type,
name=project.name,
name_slug=project.name_slug,
remote_project_name=project.remote_project_name,
@@ -44,37 +54,57 @@ def project_to_out(project: PrivatePortraitProject, *, include_user: bool = Fals
status=project.status,
asset_group_count=project.asset_group_count or 0,
asset_count=project.asset_count or 0,
image_asset_count=getattr(project, "image_asset_count", 0) or 0,
video_asset_count=getattr(project, "video_asset_count", 0) or 0,
active_asset_count=project.active_asset_count or 0,
active_image_asset_count=getattr(project, "active_image_asset_count", 0) or 0,
active_video_asset_count=getattr(project, "active_video_asset_count", 0) or 0,
last_used_at=project.last_used_at,
created_at=project.created_at,
updated_at=project.updated_at,
)
async def get_user_project(db: AsyncSession, *, user_id: str, project_id: str) -> PrivatePortraitProject:
result = await db.execute(
select(PrivatePortraitProject).where(
PrivatePortraitProject.id == project_id,
PrivatePortraitProject.user_id == user_id,
PrivatePortraitProject.deleted_at.is_(None),
).limit(1)
)
async def get_user_project(
db: AsyncSession,
*,
user_id: str,
project_id: str,
library_type: str | None = None,
) -> PrivatePortraitProject:
filters = [
PrivatePortraitProject.id == project_id,
PrivatePortraitProject.user_id == user_id,
PrivatePortraitProject.deleted_at.is_(None),
]
if library_type:
filters.append(PrivatePortraitProject.library_type == library_type)
result = await db.execute(select(PrivatePortraitProject).where(*filters).limit(1))
project = result.scalar_one_or_none()
if not project:
raise HTTPException(status_code=404, detail="真人素材项目不存在")
raise HTTPException(status_code=404, detail="私域人像素材项目不存在")
return project
async def create_project(db: AsyncSession, *, user_id: str, payload: PrivatePortraitProjectCreate) -> PrivatePortraitProject:
async def create_project(
db: AsyncSession,
*,
user_id: str,
payload: PrivatePortraitProjectCreate,
library_type: str = PrivatePortraitLibraryType.REAL_PERSON.value,
status: str | None = None,
remote_project_name: str = PRIVATE_PORTRAIT_REMOTE_PROJECT_NAME,
) -> PrivatePortraitProject:
slug = _safe_slug(payload.name)
project = PrivatePortraitProject(
id=generate_id(),
user_id=user_id,
library_type=library_type,
name=payload.name.strip(),
name_slug=slug,
remote_project_name=PRIVATE_PORTRAIT_REMOTE_PROJECT_NAME,
remote_project_name=remote_project_name,
description=payload.description,
status=PrivatePortraitProjectStatus.VALIDATING.value,
status=status or _status_for_created_project(library_type),
)
db.add(project)
await db.flush()
@@ -85,35 +115,38 @@ async def create_project(db: AsyncSession, *, user_id: str, payload: PrivatePort
source=PrivatePortraitEventSource.API.value,
user_id=user_id,
project_id=project.id,
message="创建待认证真人素材项目",
detail={"name": project.name, "remote_project_name": project.remote_project_name, "status": project.status},
message="创建私域人像素材项目",
detail={"name": project.name, "library_type": library_type, "remote_project_name": project.remote_project_name, "status": project.status},
)
return project
async def update_project(db: AsyncSession, *, user_id: str, project_id: str, payload: PrivatePortraitProjectUpdate) -> PrivatePortraitProject:
project = await get_user_project(db, user_id=user_id, project_id=project_id)
async def update_project(
db: AsyncSession,
*,
user_id: str,
project_id: str,
payload: PrivatePortraitProjectUpdate,
library_type: str | None = None,
) -> PrivatePortraitProject:
project = await get_user_project(db, user_id=user_id, project_id=project_id, library_type=library_type)
before = {
"name": project.name,
"name_slug": project.name_slug,
"remote_project_name": project.remote_project_name,
"description": project.description,
"status": project.status,
"library_type": project.library_type,
}
if payload.name is not None:
new_name = payload.name.strip()
if new_name and new_name != project.name:
project.name = new_name
project.name_slug = _safe_slug(new_name)
project.remote_project_name = PRIVATE_PORTRAIT_REMOTE_PROJECT_NAME
if payload.description is not None:
project.description = payload.description
if payload.status is not None:
allowed_statuses = {
PrivatePortraitProjectStatus.VALIDATING.value,
PrivatePortraitProjectStatus.ACTIVE.value,
PrivatePortraitProjectStatus.VALIDATE_FAILED.value,
}
allowed_statuses = {item.value for item in PrivatePortraitProjectStatus if item != PrivatePortraitProjectStatus.DELETED}
if payload.status not in allowed_statuses:
raise HTTPException(status_code=400, detail="项目状态不支持")
project.status = payload.status
@@ -124,6 +157,7 @@ async def update_project(db: AsyncSession, *, user_id: str, project_id: str, pay
"remote_project_name": project.remote_project_name,
"description": project.description,
"status": project.status,
"library_type": project.library_type,
}
log_operation_event(
domain=DOMAIN,
@@ -132,7 +166,7 @@ async def update_project(db: AsyncSession, *, user_id: str, project_id: str, pay
source=PrivatePortraitEventSource.API.value,
user_id=user_id,
project_id=project.id,
message="更新真人素材项目",
message="更新私域人像素材项目",
detail={"before": before, "after": after},
)
return project
@@ -146,18 +180,27 @@ async def list_projects(
page_size: int = 20,
keyword: str | None = None,
status: str | None = None,
library_type: str | None = None,
) -> tuple[list[PrivatePortraitProject], int]:
page = max(1, page)
page_size = min(max(1, page_size), 100)
filters = [PrivatePortraitProject.deleted_at.is_(None)]
if user_id:
filters.append(PrivatePortraitProject.user_id == user_id)
if library_type:
filters.append(PrivatePortraitProject.library_type == library_type)
if keyword:
filters.append(PrivatePortraitProject.name.ilike(f"%{keyword.strip()}%"))
if status:
filters.append(PrivatePortraitProject.status == status)
total = (await db.execute(select(func.count(PrivatePortraitProject.id)).where(*filters))).scalar_one()
result = await db.execute(select(PrivatePortraitProject).where(*filters).order_by(PrivatePortraitProject.created_at.desc()).offset((page - 1) * page_size).limit(page_size))
result = await db.execute(
select(PrivatePortraitProject)
.where(*filters)
.order_by(PrivatePortraitProject.created_at.desc())
.offset((page - 1) * page_size)
.limit(page_size)
)
return list(result.scalars().all()), int(total or 0)
@@ -174,27 +217,73 @@ async def refresh_project_counters(db: AsyncSession, project_ids: list[str]) ->
select(
PrivatePortraitAsset.project_id,
func.count(PrivatePortraitAsset.id),
func.sum(case((PrivatePortraitAsset.asset_type == PrivatePortraitAssetType.IMAGE.value, 1), else_=0)),
func.sum(case((PrivatePortraitAsset.asset_type == PrivatePortraitAssetType.VIDEO.value, 1), else_=0)),
func.sum(case((PrivatePortraitAsset.status == PrivatePortraitAssetStatus.ACTIVE.value, 1), else_=0)),
func.sum(case(((PrivatePortraitAsset.status == PrivatePortraitAssetStatus.ACTIVE.value) & (PrivatePortraitAsset.asset_type == PrivatePortraitAssetType.IMAGE.value), 1), else_=0)),
func.sum(case(((PrivatePortraitAsset.status == PrivatePortraitAssetStatus.ACTIVE.value) & (PrivatePortraitAsset.asset_type == PrivatePortraitAssetType.VIDEO.value), 1), else_=0)),
)
.where(PrivatePortraitAsset.project_id.in_(project_ids), PrivatePortraitAsset.deleted_at.is_(None))
.group_by(PrivatePortraitAsset.project_id)
)
group_count_map = {pid: int(count or 0) for pid, count in group_rows.all()}
asset_count_map: dict[str, tuple[int, int]] = {}
for pid, total, active_total in asset_rows.all():
asset_count_map[pid] = (int(total or 0), int(active_total or 0))
asset_count_map: dict[str, tuple[int, int, int, int, int, int]] = {}
for pid, total, image_total, video_total, active_total, active_image_total, active_video_total in asset_rows.all():
asset_count_map[pid] = (
int(total or 0),
int(image_total or 0),
int(video_total or 0),
int(active_total or 0),
int(active_image_total or 0),
int(active_video_total or 0),
)
for pid in project_ids:
total, active_total = asset_count_map.get(pid, (0, 0))
await db.execute(update(PrivatePortraitProject).where(PrivatePortraitProject.id == pid).values(asset_group_count=group_count_map.get(pid, 0), asset_count=total, active_asset_count=active_total))
total, image_total, video_total, active_total, active_image_total, active_video_total = asset_count_map.get(pid, (0, 0, 0, 0, 0, 0))
await db.execute(
update(PrivatePortraitProject)
.where(PrivatePortraitProject.id == pid)
.values(
asset_group_count=group_count_map.get(pid, 0),
asset_count=total,
image_asset_count=image_total,
video_asset_count=video_total,
active_asset_count=active_total,
active_image_asset_count=active_image_total,
active_video_asset_count=active_video_total,
)
)
async def soft_delete_project(db: AsyncSession, *, user_id: str, project_id: str) -> PrivatePortraitProject:
project = await get_user_project(db, user_id=user_id, project_id=project_id)
async def soft_delete_project(
db: AsyncSession,
*,
user_id: str,
project_id: str,
library_type: str | None = None,
) -> PrivatePortraitProject:
project = await get_user_project(db, user_id=user_id, project_id=project_id, library_type=library_type)
now = datetime.now(timezone.utc)
project.deleted_at = now
project.status = PrivatePortraitProjectStatus.DELETED.value
await db.execute(update(PrivatePortraitAsset).where(PrivatePortraitAsset.project_id == project_id, PrivatePortraitAsset.deleted_at.is_(None)).values(deleted_at=now, status=PrivatePortraitAssetStatus.LOCAL_DELETED.value, remote_delete_status=PrivatePortraitRemoteDeleteStatus.PENDING.value))
await db.execute(update(PrivatePortraitAssetGroup).where(PrivatePortraitAssetGroup.project_id == project_id, PrivatePortraitAssetGroup.deleted_at.is_(None)).values(deleted_at=now, status=PrivatePortraitAssetGroupStatus.LOCAL_DELETED.value, remote_delete_status=PrivatePortraitRemoteDeleteStatus.PENDING.value))
await db.execute(
update(PrivatePortraitAsset)
.where(PrivatePortraitAsset.project_id == project_id, PrivatePortraitAsset.deleted_at.is_(None))
.values(deleted_at=now, status=PrivatePortraitAssetStatus.LOCAL_DELETED.value, remote_delete_status=PrivatePortraitRemoteDeleteStatus.PENDING.value)
)
await db.execute(
update(PrivatePortraitAssetGroup)
.where(PrivatePortraitAssetGroup.project_id == project_id, PrivatePortraitAssetGroup.deleted_at.is_(None))
.values(deleted_at=now, status=PrivatePortraitAssetGroupStatus.LOCAL_DELETED.value, remote_delete_status=PrivatePortraitRemoteDeleteStatus.PENDING.value)
)
await db.flush()
log_operation_event(domain=DOMAIN, event_type=PrivatePortraitEventType.PROJECT_DELETE.value, event_status=PrivatePortraitEventStatus.SUCCESS.value, source=PrivatePortraitEventSource.API.value, user_id=user_id, project_id=project.id, message="本地软删真人素材项目", detail={"remote_project_name": project.remote_project_name})
log_operation_event(
domain=DOMAIN,
event_type=PrivatePortraitEventType.PROJECT_DELETE.value,
event_status=PrivatePortraitEventStatus.SUCCESS.value,
source=PrivatePortraitEventSource.API.value,
user_id=user_id,
project_id=project.id,
message="本地软删私域人像素材项目",
detail={"library_type": project.library_type, "remote_project_name": project.remote_project_name},
)
return project
video-gen-api/app/services/private_portrait/quota_service.py
+141
View File
@@ -0,0 +1,141 @@
from __future__ import annotations
from fastapi import HTTPException
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession
from app.enums.private_portrait import (
PRIVATE_PORTRAIT_DEFAULT_ASSET_LIMIT,
PRIVATE_PORTRAIT_ENABLED_ASSET_TYPES,
PrivatePortraitAssetStatus,
PrivatePortraitEventSource,
PrivatePortraitEventStatus,
PrivatePortraitEventType,
PrivatePortraitLibraryType,
)
from app.models.private_portrait import PrivatePortraitAsset
from app.models.user import User
from app.schemas.private_portrait import PrivatePortraitConfigOut
from app.services.operation_log_service import log_operation_event
DOMAIN = "private_portrait"
_COUNTING_STATUSES = {
PrivatePortraitAssetStatus.CREATING.value,
PrivatePortraitAssetStatus.PROCESSING.value,
PrivatePortraitAssetStatus.ACTIVE.value,
}
_COUNTING_LIBRARY_TYPES = {
PrivatePortraitLibraryType.REAL_PERSON.value,
PrivatePortraitLibraryType.AIGC_VIRTUAL.value,
}
async def get_user_or_404(db: AsyncSession, *, user_id: str, for_update: bool = False) -> User:
stmt = select(User).where(User.id == user_id).limit(1)
if for_update:
stmt = stmt.with_for_update()
user = (await db.execute(stmt)).scalar_one_or_none()
if not user:
raise HTTPException(status_code=404, detail="用户不存在")
return user
def get_user_asset_limit_value(user: User) -> int:
return int(getattr(user, "private_portrait_asset_limit", PRIVATE_PORTRAIT_DEFAULT_ASSET_LIMIT) or 0)
async def count_user_counting_assets(db: AsyncSession, *, user_id: str) -> int:
total = (
await db.execute(
select(func.count(PrivatePortraitAsset.id)).where(
PrivatePortraitAsset.user_id == user_id,
PrivatePortraitAsset.library_type.in_(_COUNTING_LIBRARY_TYPES),
PrivatePortraitAsset.asset_type.in_(PRIVATE_PORTRAIT_ENABLED_ASSET_TYPES),
PrivatePortraitAsset.deleted_at.is_(None),
PrivatePortraitAsset.status.in_(_COUNTING_STATUSES),
)
)
).scalar_one()
return int(total or 0)
async def get_user_private_portrait_config(db: AsyncSession, *, user_id: str) -> PrivatePortraitConfigOut:
user = await get_user_or_404(db, user_id=user_id)
limit = get_user_asset_limit_value(user)
used = await count_user_counting_assets(db, user_id=user_id)
remaining = max(0, limit - used) if limit > 0 else 0
return PrivatePortraitConfigOut(
enabled=limit > 0,
asset_limit=limit,
used_asset_count=used,
remaining_asset_count=remaining,
image_limit=limit,
used_image_count=used,
remaining_image_count=remaining,
)
async def set_user_private_portrait_limit(db: AsyncSession, *, user_id: str, limit: int) -> User:
user = await get_user_or_404(db, user_id=user_id)
user.private_portrait_asset_limit = max(0, int(limit))
await db.flush()
return user
async def ensure_private_portrait_asset_quota_available(
db: AsyncSession,
*,
user_id: str,
project_id: str | None = None,
library_type: str | None = None,
asset_type: str | None = None,
) -> tuple[int, int]:
user = await get_user_or_404(db, user_id=user_id, for_update=True)
limit = get_user_asset_limit_value(user)
log_operation_event(
domain=DOMAIN,
event_type=PrivatePortraitEventType.QUOTA_CHECK_START.value,
event_status=PrivatePortraitEventStatus.PENDING.value,
source=PrivatePortraitEventSource.SERVICE.value,
user_id=user_id,
project_id=project_id,
detail={"asset_limit": limit, "library_type": library_type, "asset_type": asset_type},
)
if limit <= 0:
log_operation_event(
domain=DOMAIN,
event_type=PrivatePortraitEventType.QUOTA_CHECK_DENY.value,
event_status=PrivatePortraitEventStatus.FAILED.value,
source=PrivatePortraitEventSource.SERVICE.value,
user_id=user_id,
project_id=project_id,
detail={"asset_limit": limit, "library_type": library_type, "asset_type": asset_type, "reason": "disabled"},
message="用户私域人像素材库未启用",
)
raise HTTPException(status_code=403, detail="私域人像素材库未启用")
used = await count_user_counting_assets(db, user_id=user_id)
if used >= limit:
log_operation_event(
domain=DOMAIN,
event_type=PrivatePortraitEventType.QUOTA_CHECK_DENY.value,
event_status=PrivatePortraitEventStatus.FAILED.value,
source=PrivatePortraitEventSource.SERVICE.value,
user_id=user_id,
project_id=project_id,
detail={"asset_limit": limit, "used_asset_count": used, "library_type": library_type, "asset_type": asset_type, "reason": "max_limit"},
message="用户私域人像素材总量已达上限",
)
raise HTTPException(status_code=400, detail=f"你的私域人像素材库最多可上传 {limit} 个素材，请删除已有素材后再上传")
log_operation_event(
domain=DOMAIN,
event_type=PrivatePortraitEventType.QUOTA_CHECK_PASS.value,
event_status=PrivatePortraitEventStatus.SUCCESS.value,
source=PrivatePortraitEventSource.SERVICE.value,
user_id=user_id,
project_id=project_id,
detail={"asset_limit": limit, "used_asset_count": used, "remaining_asset_count": max(0, limit - used), "library_type": library_type, "asset_type": asset_type},
)
return limit, used
video-gen-api/app/services/private_portrait/real_person/__init__.py
+1
View File
@@ -0,0 +1 @@
from app.services.private_portrait.real_person.service import *
video-gen-api/app/services/private_portrait/real_person/service.py
+24
View File
@@ -0,0 +1,24 @@
from __future__ import annotations
from sqlalchemy.ext.asyncio import AsyncSession
from app.enums.private_portrait import PrivatePortraitLibraryType
from app.schemas.private_portrait import PrivatePortraitAssetCreate, PrivatePortraitProjectCreate, PrivatePortraitProjectUpdate
from app.services.private_portrait.asset_service import create_asset, create_validate_session
from app.services.private_portrait.project_service import create_project, update_project
async def create_real_person_project(db: AsyncSession, *, user_id: str, payload: PrivatePortraitProjectCreate):
return await create_project(db, user_id=user_id, payload=payload, library_type=PrivatePortraitLibraryType.REAL_PERSON.value)
async def update_real_person_project(db: AsyncSession, *, user_id: str, project_id: str, payload: PrivatePortraitProjectUpdate):
return await update_project(db, user_id=user_id, project_id=project_id, payload=payload, library_type=PrivatePortraitLibraryType.REAL_PERSON.value)
async def create_real_person_validate_session(db: AsyncSession, *, user_id: str, project_id: str, callback_redirect_url: str | None = None):
return await create_validate_session(db, user_id=user_id, project_id=project_id, callback_redirect_url=callback_redirect_url)
async def create_real_person_asset(db: AsyncSession, *, user_id: str, project_id: str, payload: PrivatePortraitAssetCreate):
return await create_asset(db, user_id=user_id, project_id=project_id, payload=payload, library_type=PrivatePortraitLibraryType.REAL_PERSON.value)
video-gen-api/app/services/private_portrait/reference_resolver.py
+10 -8
View File
@@ -10,6 +10,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
from app.enums.private_portrait import (
PRIVATE_PORTRAIT_ASSET_URI_PREFIX,
PrivatePortraitAssetStatus,
PRIVATE_PORTRAIT_ENABLED_ASSET_TYPES,
PrivatePortraitAssetType,
PrivatePortraitEventSource,
PrivatePortraitEventStatus,
@@ -25,7 +26,6 @@ DOMAIN = "private_portrait"
_ASSET_TYPE_TO_REFERENCE_TYPE = {
PrivatePortraitAssetType.IMAGE.value: "image",
PrivatePortraitAssetType.VIDEO.value: "video",
PrivatePortraitAssetType.AUDIO.value: "audio",
}
@@ -260,24 +260,26 @@ async def resolve_private_portrait_references(
asset_id = str(_ref_get(ref, "private_asset_id") or "")
if not asset_id:
raise HTTPException(status_code=400, detail="真人素材引用缺少 private_asset_id")
raise HTTPException(status_code=400, detail="私域人像素材引用缺少 private_asset_id")
asset = asset_map.get(asset_id)
if not asset:
raise HTTPException(status_code=400, detail="真人素材不存在")
raise HTTPException(status_code=400, detail="私域人像素材不存在")
if asset.user_id != user_id:
raise HTTPException(status_code=403, detail="真人素材不属于当前用户")
raise HTTPException(status_code=403, detail="私域人像素材不属于当前用户")
if asset.deleted_at is not None:
raise HTTPException(status_code=400, detail="真人素材已删除")
raise HTTPException(status_code=400, detail="私域人像素材已删除")
if asset.asset_type not in PRIVATE_PORTRAIT_ENABLED_ASSET_TYPES:
raise HTTPException(status_code=400, detail="Audio 暂未开放，当前仅支持 Image/Video 私域素材")
if asset.status != PrivatePortraitAssetStatus.ACTIVE.value:
raise HTTPException(status_code=400, detail=f"真人素材状态为 {asset.status}，Active 后才可用于生成")
raise HTTPException(status_code=400, detail=f"私域人像素材状态为 {asset.status}，Active 后才可用于生成")
if not asset.remote_asset_id:
raise HTTPException(status_code=400, detail="真人素材缺少远程 AssetId")
raise HTTPException(status_code=400, detail="私域人像素材缺少远程 AssetId")
expected_ref_type = _ASSET_TYPE_TO_REFERENCE_TYPE.get(asset.asset_type)
ref_type = _normalize_ref_type(_ref_get(ref, "type"))
if expected_ref_type and ref_type and ref_type != expected_ref_type:
raise HTTPException(status_code=400, detail=f"真人素材类型不匹配：引用为 {ref_type}，素材为 {expected_ref_type}")
raise HTTPException(status_code=400, detail=f"私域人像素材类型不匹配：引用为 {ref_type}，素材为 {expected_ref_type}")
provider_url = f"{PRIVATE_PORTRAIT_ASSET_URI_PREFIX}{asset.remote_asset_id}"
_ref_set(ref, "source", PrivatePortraitReferenceSource.PRIVATE_PORTRAIT_ASSET.value)
video-gen-api/app/services/private_portrait/virtual/__init__.py
+1
View File
@@ -0,0 +1 @@
from app.services.private_portrait.virtual.service import *
video-gen-api/app/services/private_portrait/virtual/service.py
+126
View File
@@ -0,0 +1,126 @@
from __future__ import annotations
import json
from fastapi import HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from app.enums.private_portrait import (
PRIVATE_PORTRAIT_REMOTE_PROJECT_NAME,
PRIVATE_PORTRAIT_VIRTUAL_GROUP_TYPE,
PrivatePortraitAssetGroupStatus,
PrivatePortraitEventSource,
PrivatePortraitEventStatus,
PrivatePortraitEventType,
PrivatePortraitLibraryType,
PrivatePortraitProjectStatus,
)
from app.models.private_portrait import PrivatePortraitAssetGroup
from app.schemas.private_portrait import PrivatePortraitAssetCreate, PrivatePortraitProjectUpdate, PrivatePortraitVirtualProjectCreate
from app.services.operation_log_service import log_operation_error, log_operation_event
from app.services.private_portrait.ark_client import ArkPrivateAssetClient
from app.services.private_portrait.asset_service import create_asset
from app.services.private_portrait.project_service import create_project, refresh_project_counters, update_project
from app.utils.id_gen import generate_id
DOMAIN = "private_portrait"
def _json(data) -> str | None:
if data is None:
return None
return json.dumps(data, ensure_ascii=False, default=str)
def _remote_group_name(user_id: str, project_name: str) -> str:
safe_name = "".join(ch if ch.isalnum() or ch in "-_" else "_" for ch in project_name.strip())[:80]
return f"virtual-{user_id}-{safe_name}"[:128]
async def create_virtual_project(db: AsyncSession, *, user_id: str, payload: PrivatePortraitVirtualProjectCreate):
project = await create_project(
db,
user_id=user_id,
payload=payload, # type: ignore[arg-type]
library_type=PrivatePortraitLibraryType.AIGC_VIRTUAL.value,
status=PrivatePortraitProjectStatus.CREATING_REMOTE_GROUP.value,
remote_project_name=PRIVATE_PORTRAIT_REMOTE_PROJECT_NAME,
)
remote_group_name = _remote_group_name(user_id, project.name)
log_operation_event(
domain=DOMAIN,
event_type=PrivatePortraitEventType.VIRTUAL_ASSET_GROUP_CREATE_REMOTE_START.value,
event_status=PrivatePortraitEventStatus.PENDING.value,
source=PrivatePortraitEventSource.API.value,
user_id=user_id,
project_id=project.id,
detail={"remote_group_name": remote_group_name, "remote_project_name": project.remote_project_name, "group_type": PRIVATE_PORTRAIT_VIRTUAL_GROUP_TYPE},
)
try:
resp = await ArkPrivateAssetClient().create_asset_group(
project_name=project.remote_project_name,
name=remote_group_name,
description=project.description,
group_type=PRIVATE_PORTRAIT_VIRTUAL_GROUP_TYPE,
)
remote_group_id = resp.get("Id") or resp.get("GroupId") or resp.get("groupId")
if not remote_group_id:
raise RuntimeError("CreateAssetGroup 未返回素材组 ID")
group = PrivatePortraitAssetGroup(
id=generate_id(),
user_id=user_id,
project_id=project.id,
library_type=PrivatePortraitLibraryType.AIGC_VIRTUAL.value,
remote_group_id=remote_group_id,
remote_group_name=remote_group_name,
remote_project_name=project.remote_project_name,
group_type=PRIVATE_PORTRAIT_VIRTUAL_GROUP_TYPE,
status=PrivatePortraitAssetGroupStatus.ACTIVE.value,
raw_response_json=_json(resp),
)
db.add(group)
project.status = PrivatePortraitProjectStatus.ACTIVE.value
await refresh_project_counters(db, [project.id])
await db.flush()
await db.refresh(project)
log_operation_event(
domain=DOMAIN,
event_type=PrivatePortraitEventType.VIRTUAL_ASSET_GROUP_CREATE_REMOTE_SUCCESS.value,
event_status=PrivatePortraitEventStatus.SUCCESS.value,
source=PrivatePortraitEventSource.API.value,
user_id=user_id,
project_id=project.id,
group_id=group.id,
detail={"remote_group_id": remote_group_id, "remote_group_name": remote_group_name, "remote_project_name": project.remote_project_name},
)
return project
except Exception as exc:
project.status = PrivatePortraitProjectStatus.CREATE_GROUP_FAILED.value
await db.flush()
log_operation_error(
domain=DOMAIN,
event_type=PrivatePortraitEventType.VIRTUAL_ASSET_GROUP_CREATE_REMOTE_FAILED.value,
source=PrivatePortraitEventSource.API.value,
user_id=user_id,
project_id=project.id,
exc=exc,
)
raise HTTPException(status_code=502, detail=f"创建火山虚拟人像素材组失败：{exc}") from exc
async def update_virtual_project(db: AsyncSession, *, user_id: str, project_id: str, payload: PrivatePortraitProjectUpdate):
project = await update_project(db, user_id=user_id, project_id=project_id, payload=payload, library_type=PrivatePortraitLibraryType.AIGC_VIRTUAL.value)
# 远程同步失败不影响本地更新，记录日志便于排查。
try:
# 只同步当前激活组。
from app.services.private_portrait.asset_service import get_project_active_group
group = await get_project_active_group(db, user_id=user_id, project_id=project_id, library_type=PrivatePortraitLibraryType.AIGC_VIRTUAL.value)
await ArkPrivateAssetClient().update_asset_group(project_name=project.remote_project_name, group_id=group.remote_group_id, name=group.remote_group_name, title=project.name, description=project.description)
except Exception as exc:
log_operation_error(domain=DOMAIN, event_type=PrivatePortraitEventType.ASSET_GROUP_UPDATE_REMOTE_FAILED.value, source=PrivatePortraitEventSource.API.value, user_id=user_id, project_id=project_id, exc=exc)
return project
async def create_virtual_asset(db: AsyncSession, *, user_id: str, project_id: str, payload: PrivatePortraitAssetCreate):
return await create_asset(db, user_id=user_id, project_id=project_id, payload=payload, library_type=PrivatePortraitLibraryType.AIGC_VIRTUAL.value)