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

1

Browse Source
This commit is contained in:
root
2026-07-07 15:18:00 +08:00
parent e94225995c
commit 0fb2c5f165
52 changed files with 4017 additions and 1531 deletions
Download Patch File Download Diff File Expand all files Collapse all files
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