小鱼开发
1a0679049e
Replace menu list (使用明细 + 设置) with recent transactions table: - Add back recentTx state and loading state - Fetch last 5 transactions in loadData - Display table with type/amount/description/time columns - Add '查看全部' link to usage-detail page - Remove unused icon components (FileTextIcon, SettingsIcon, ChevronRightIcon)
92 lines
2.9 KiB
Python
92 lines
2.9 KiB
Python
"""
|
|
火山引擎 MediaKit Adapter
|
|
==========================
|
|
|
|
实现 PlatformAdapter + SyncCapable。
|
|
直接接入 VolcengineMediakitProvider，提供标准 Protocol 接口。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import Any
|
|
|
|
from app.ai.adapters.base import AdapterResponse, PlatformAdapter, SyncCapable
|
|
from app.ai.adapters.constants import Method
|
|
from app.ai.providers.volcengine_mediakit_provider import VolcengineMediakitProvider
|
|
from app.core.exceptions import PlatformError, PlatformErrorType
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class VolcengineMediakitAdapter(PlatformAdapter, SyncCapable):
|
|
"""火山引擎 MediaKit 平台标准 Adapter"""
|
|
|
|
platform_id = "volcengine_mediakit"
|
|
|
|
def __init__(self, provider: VolcengineMediakitProvider):
|
|
self.provider = provider
|
|
|
|
# ── PlatformAdapter ──
|
|
|
|
async def health(self) -> AdapterResponse:
|
|
try:
|
|
# 用无效 URL 测试连通性（400 说明网络通且认证通过）
|
|
await self.provider.remove_background(
|
|
image_url="https://example.com/health-check.jpg",
|
|
scene="general",
|
|
)
|
|
return AdapterResponse(success=True)
|
|
except PlatformError as e:
|
|
if e.error_type in (
|
|
PlatformErrorType.AUTH_FAILED,
|
|
PlatformErrorType.BAD_REQUEST,
|
|
):
|
|
return AdapterResponse(success=True)
|
|
return AdapterResponse(
|
|
success=False,
|
|
error_message=str(e),
|
|
retryable=e.retryable,
|
|
)
|
|
except Exception as e:
|
|
return AdapterResponse(
|
|
success=False,
|
|
error_message=str(e),
|
|
retryable=False,
|
|
)
|
|
|
|
async def close(self) -> None:
|
|
await self.provider.close()
|
|
|
|
# ── SyncCapable ──
|
|
|
|
async def call(self, method: str, payload: dict[str, Any]) -> AdapterResponse:
|
|
try:
|
|
if method == Method.REMOVE_BACKGROUND:
|
|
result = await self.provider.remove_background(
|
|
image_url=payload["image_url"],
|
|
scene=payload.get("scene", "general"),
|
|
)
|
|
data = result.get("data", {})
|
|
return AdapterResponse(
|
|
success=True,
|
|
data={"image_url": data.get("image_url")},
|
|
)
|
|
|
|
else:
|
|
return AdapterResponse(
|
|
success=False,
|
|
error_message=f"不支持的方法: {method}",
|
|
retryable=False,
|
|
)
|
|
|
|
except PlatformError:
|
|
raise
|
|
except Exception as e:
|
|
raise PlatformError(
|
|
f"MediaKit {method} 调用失败: {e}",
|
|
platform="volcengine_mediakit",
|
|
retryable=False,
|
|
error_type=PlatformErrorType.UNKNOWN,
|
|
) from e
|