319 lines
15 KiB
Python
319 lines
15 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import time
|
|
import uuid
|
|
from typing import Any
|
|
|
|
from fastapi import HTTPException
|
|
|
|
from app.config import settings
|
|
from app.enums.private_portrait import (
|
|
ARK_PRIVATE_PORTRAIT_HOST,
|
|
ARK_PRIVATE_PORTRAIT_REGION,
|
|
ARK_PRIVATE_PORTRAIT_SERVICE_NAME,
|
|
ARK_PRIVATE_PORTRAIT_VERSION,
|
|
ArkPrivatePortraitAction,
|
|
PrivatePortraitEventSource,
|
|
)
|
|
from app.services.operation_log_service import log_ai_model_event
|
|
from app.services.private_portrait.rate_limiter import acquire_private_portrait_action_token
|
|
|
|
DOMAIN = "private_portrait"
|
|
|
|
|
|
class ArkPrivateAssetClientError(RuntimeError):
|
|
pass
|
|
|
|
|
|
class ArkPrivateAssetRemoteError(ArkPrivateAssetClientError):
|
|
def __init__(self, *, action: str, code: str, message: str, request_id: str | None = None, raw: dict[str, Any] | None = None):
|
|
self.action = action
|
|
self.code = code
|
|
self.message = message
|
|
self.request_id = request_id
|
|
self.raw = raw or {}
|
|
super().__init__(f"{action} 调用失败：{code} {message}" + (f" RequestId={request_id}" if request_id else ""))
|
|
|
|
|
|
def _remote_error_http_status(code: str) -> int:
|
|
normalized = (code or "").lower()
|
|
if "notfound" in normalized or normalized.startswith("not_found"):
|
|
return 404
|
|
if "invalid" in normalized or "missing" in normalized or "unsupported" in normalized or "limit" in normalized:
|
|
return 400
|
|
if "unauthorized" in normalized or "forbidden" in normalized or "permission" in normalized or "denied" in normalized:
|
|
return 403
|
|
if "throttl" in normalized or "rate" in normalized:
|
|
return 429
|
|
return 502
|
|
|
|
|
|
class ArkPrivateAssetClient:
|
|
"""火山 Ark 私域可信素材 Asset API Client。只做 AK/SK 鉴权调用与响应标准化。"""
|
|
|
|
def __init__(self, *, for_celery: bool = False):
|
|
self.ak = settings.VOLC_ACCESS_KEY_ID
|
|
self.sk = settings.VOLC_SECRET_ACCESS_KEY
|
|
self.for_celery = for_celery
|
|
if not self.ak or not self.sk:
|
|
raise ArkPrivateAssetClientError("火山 AK/SK 未配置：VOLC_SMS_ACCESS_KEY_ID / VOLC_SMS_SECRET_ACCESS_KEY")
|
|
|
|
async def create_visual_validate_session(self, *, project_name: str, callback_url: str) -> dict[str, Any]:
|
|
return await self._call(ArkPrivatePortraitAction.CREATE_VISUAL_VALIDATE_SESSION, {"CallbackURL": callback_url, "ProjectName": project_name})
|
|
|
|
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:
|
|
payload["Name"] = name
|
|
return await self._call(ArkPrivatePortraitAction.CREATE_ASSET, payload)
|
|
|
|
async def get_asset(self, *, project_name: str, asset_id: str) -> dict[str, Any]:
|
|
return await self._call(ArkPrivatePortraitAction.GET_ASSET, {"Id": asset_id, "ProjectName": project_name})
|
|
|
|
async def list_assets(self, *, project_name: str, filter_payload: dict[str, Any] | None = None, page_number: int = 1, page_size: int = 20) -> dict[str, Any]:
|
|
payload = {"Filter": filter_payload or {}, "PageNumber": page_number, "PageSize": page_size, "ProjectName": project_name}
|
|
return await self._call(ArkPrivatePortraitAction.LIST_ASSETS, payload)
|
|
|
|
async def list_asset_groups(self, *, project_name: str, filter_payload: dict[str, Any] | None = None, page_number: int = 1, page_size: int = 20) -> dict[str, Any]:
|
|
payload = {"Filter": filter_payload or {}, "PageNumber": page_number, "PageSize": page_size, "ProjectName": project_name}
|
|
return await self._call(ArkPrivatePortraitAction.LIST_ASSET_GROUPS, payload)
|
|
|
|
async def get_asset_group(self, *, project_name: str, group_id: str) -> dict[str, Any]:
|
|
return await self._call(ArkPrivatePortraitAction.GET_ASSET_GROUP, {"Id": group_id, "ProjectName": project_name})
|
|
|
|
async def update_asset_group(self, *, project_name: str, group_id: str, name: str | None = None, title: str | None = None, description: str | None = None) -> dict[str, Any]:
|
|
payload: dict[str, Any] = {"Id": group_id, "ProjectName": project_name}
|
|
if name is not None:
|
|
payload["Name"] = name
|
|
if title is not None:
|
|
payload["Title"] = title
|
|
if description is not None:
|
|
payload["Description"] = description
|
|
return await self._call(ArkPrivatePortraitAction.UPDATE_ASSET_GROUP, payload)
|
|
|
|
async def update_asset(self, *, project_name: str, asset_id: str, name: str | None = None) -> dict[str, Any]:
|
|
payload: dict[str, Any] = {"Id": asset_id, "ProjectName": project_name}
|
|
if name is not None:
|
|
payload["Name"] = name
|
|
return await self._call(ArkPrivatePortraitAction.UPDATE_ASSET, payload)
|
|
|
|
async def delete_asset(self, *, project_name: str, asset_id: str) -> dict[str, Any]:
|
|
return await self._call(ArkPrivatePortraitAction.DELETE_ASSET, {"Id": asset_id, "ProjectName": project_name})
|
|
|
|
async def delete_asset_group(self, *, project_name: str, group_id: str) -> dict[str, Any]:
|
|
return await self._call(ArkPrivatePortraitAction.DELETE_ASSET_GROUP, {"Id": group_id, "ProjectName": project_name})
|
|
|
|
async def _call(self, action: ArkPrivatePortraitAction, payload: dict[str, Any]) -> dict[str, Any]:
|
|
action_value = action.value
|
|
await acquire_private_portrait_action_token(action=action_value, wait_timeout_seconds=2.0, for_celery=self.for_celery)
|
|
call_id = uuid.uuid4().hex
|
|
source = PrivatePortraitEventSource.CELERY.value if self.for_celery else PrivatePortraitEventSource.SERVICE.value
|
|
started = time.perf_counter()
|
|
log_ai_model_event(
|
|
event_type="REQUEST",
|
|
event_phase="REQUEST",
|
|
event_status="started",
|
|
module=DOMAIN,
|
|
step_code=action_value,
|
|
call_id=call_id,
|
|
source=source,
|
|
remote_action=action_value,
|
|
provider="volcengine_ark",
|
|
request=payload,
|
|
)
|
|
try:
|
|
result = await asyncio.to_thread(self._call_sync, action, payload)
|
|
log_ai_model_event(
|
|
event_type="RESPONSE",
|
|
event_phase="RESPONSE",
|
|
event_status="success",
|
|
module=DOMAIN,
|
|
step_code=action_value,
|
|
call_id=call_id,
|
|
source=source,
|
|
remote_action=action_value,
|
|
remote_request_id=result.get("RequestId") or result.get("request_id"),
|
|
provider="volcengine_ark",
|
|
latency_ms=int((time.perf_counter() - started) * 1000),
|
|
response=result,
|
|
)
|
|
return result
|
|
except ArkPrivateAssetRemoteError as exc:
|
|
log_ai_model_event(
|
|
event_type="RESPONSE",
|
|
event_phase="RESPONSE",
|
|
event_status="failed",
|
|
module=DOMAIN,
|
|
step_code=action_value,
|
|
call_id=call_id,
|
|
source=source,
|
|
remote_action=action_value,
|
|
remote_request_id=exc.request_id,
|
|
provider="volcengine_ark",
|
|
latency_ms=int((time.perf_counter() - started) * 1000),
|
|
response=exc.raw,
|
|
detail={"remote_code": exc.code, "remote_message": exc.message},
|
|
error=exc.message,
|
|
)
|
|
log_ai_model_event(
|
|
event_type="ERROR",
|
|
event_phase="ERROR",
|
|
event_status="failed",
|
|
module=DOMAIN,
|
|
step_code=action_value,
|
|
call_id=call_id,
|
|
source=source,
|
|
remote_action=action_value,
|
|
remote_request_id=exc.request_id,
|
|
provider="volcengine_ark",
|
|
latency_ms=int((time.perf_counter() - started) * 1000),
|
|
detail={"remote_code": exc.code, "remote_message": exc.message},
|
|
error=str(exc),
|
|
)
|
|
if self.for_celery:
|
|
raise
|
|
raise HTTPException(status_code=_remote_error_http_status(exc.code), detail={"message": exc.message, "code": exc.code, "request_id": exc.request_id}) from exc
|
|
except HTTPException:
|
|
raise
|
|
except Exception as exc:
|
|
log_ai_model_event(
|
|
event_type="ERROR",
|
|
event_phase="ERROR",
|
|
event_status="failed",
|
|
module=DOMAIN,
|
|
step_code=action_value,
|
|
call_id=call_id,
|
|
source=source,
|
|
remote_action=action_value,
|
|
provider="volcengine_ark",
|
|
latency_ms=int((time.perf_counter() - started) * 1000),
|
|
error=str(exc),
|
|
)
|
|
if self.for_celery:
|
|
raise
|
|
raise HTTPException(status_code=502, detail=f"火山私域素材接口调用失败：{self._exception_message(exc)}") from exc
|
|
|
|
def _call_sync(self, action: ArkPrivatePortraitAction, payload: dict[str, Any]) -> dict[str, Any]:
|
|
try:
|
|
from volcengine.ApiInfo import ApiInfo
|
|
from volcengine.Credentials import Credentials
|
|
from volcengine.ServiceInfo import ServiceInfo
|
|
from volcengine.base.Service import Service
|
|
except Exception as exc:
|
|
raise ArkPrivateAssetClientError("缺少火山 volcengine Python SDK。请确认线上环境已安装 volcengine。") from exc
|
|
|
|
credentials = Credentials(self.ak, self.sk, ARK_PRIVATE_PORTRAIT_SERVICE_NAME, ARK_PRIVATE_PORTRAIT_REGION)
|
|
service_info = ServiceInfo(
|
|
ARK_PRIVATE_PORTRAIT_HOST,
|
|
{"Content-Type": "application/json", "Accept": "application/json"},
|
|
credentials,
|
|
10,
|
|
60,
|
|
"https",
|
|
)
|
|
# volcengine SDK 的 ApiInfo.query 必须是 dict，不能传 "Action=xxx&Version=xxx" 字符串。
|
|
# volcengine SDK 签名时要求 body 是 bytes。
|
|
# ServiceInfo 默认 scheme='http'，这里必须显式传 https，避免请求走 http://host:80。
|
|
api_info = {
|
|
action.value: ApiInfo(
|
|
"POST",
|
|
"/",
|
|
{"Action": action.value, "Version": ARK_PRIVATE_PORTRAIT_VERSION},
|
|
{},
|
|
{},
|
|
)
|
|
}
|
|
service = Service(service_info, api_info)
|
|
body = json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
|
|
try:
|
|
raw = service.json(action.value, {}, body)
|
|
except Exception as exc:
|
|
resp = self._extract_response_from_exception(exc)
|
|
if resp:
|
|
self._raise_remote_error_if_any(action.value, resp)
|
|
message = self._exception_message(exc)
|
|
if "ConnectTimeout" in message or "timed out" in message or "Connection" in message:
|
|
raise ArkPrivateAssetClientError(f"{action.value} 连接火山私域素材接口失败：{message}") from exc
|
|
raise ArkPrivateAssetClientError(f"{action.value} 请求火山私域素材接口异常：{message}") from exc
|
|
|
|
resp = self._normalize_response(raw)
|
|
self._raise_remote_error_if_any(action.value, resp)
|
|
metadata = resp.get("ResponseMetadata") if isinstance(resp, dict) else None
|
|
request_id = metadata.get("RequestId") if isinstance(metadata, dict) else None
|
|
if isinstance(resp, dict) and isinstance(resp.get("Result"), dict):
|
|
result = dict(resp["Result"])
|
|
if request_id:
|
|
result["RequestId"] = request_id
|
|
return result
|
|
if isinstance(resp, dict):
|
|
if request_id:
|
|
resp.setdefault("RequestId", request_id)
|
|
return resp
|
|
return {"raw": resp, "RequestId": request_id}
|
|
|
|
@classmethod
|
|
def _raise_remote_error_if_any(cls, action: str, resp: dict[str, Any]) -> None:
|
|
metadata = resp.get("ResponseMetadata") if isinstance(resp, dict) else None
|
|
error = metadata.get("Error") if isinstance(metadata, dict) else None
|
|
if not error:
|
|
return
|
|
request_id = metadata.get("RequestId") if isinstance(metadata, dict) else None
|
|
code = error.get("Code") or "ArkPrivateAssetError"
|
|
message = error.get("Message") or str(error)
|
|
raise ArkPrivateAssetRemoteError(action=action, code=code, message=message, request_id=request_id, raw=resp)
|
|
|
|
@classmethod
|
|
def _extract_response_from_exception(cls, exc: Exception) -> dict[str, Any] | None:
|
|
# volcengine SDK 在 HTTP 非 2xx 时会 raise Exception(resp.text.encode("utf-8"))，这里把 bytes JSON 还原，避免业务错误变 500。
|
|
if not exc.args:
|
|
return None
|
|
raw = exc.args[0]
|
|
if isinstance(raw, (bytes, bytearray)):
|
|
raw = raw.decode("utf-8", errors="ignore")
|
|
if not isinstance(raw, str):
|
|
return None
|
|
raw = raw.strip()
|
|
if not raw or not raw.startswith("{"):
|
|
return None
|
|
try:
|
|
obj = json.loads(raw)
|
|
except json.JSONDecodeError:
|
|
return None
|
|
return obj if isinstance(obj, dict) else None
|
|
|
|
@staticmethod
|
|
def _exception_message(exc: Exception) -> str:
|
|
if exc.args:
|
|
raw = exc.args[0]
|
|
if isinstance(raw, (bytes, bytearray)):
|
|
return raw.decode("utf-8", errors="ignore")
|
|
return str(raw)
|
|
return str(exc)
|
|
|
|
@staticmethod
|
|
def _normalize_response(raw: Any) -> dict[str, Any]:
|
|
if raw is None:
|
|
return {}
|
|
if isinstance(raw, dict):
|
|
return raw
|
|
if isinstance(raw, (bytes, bytearray)):
|
|
raw = raw.decode("utf-8", errors="ignore")
|
|
if isinstance(raw, str):
|
|
try:
|
|
obj = json.loads(raw)
|
|
return obj if isinstance(obj, dict) else {"raw": obj}
|
|
except json.JSONDecodeError:
|
|
return {"raw": raw}
|
|
return {"raw": raw}
|