Jony
ba920f2b66
Replace the prior release tree with the current plugin, frontend, tests, and documentation. Document retired node IDs and the public Gitea update source.
75 lines
2.9 KiB
Python
75 lines
2.9 KiB
Python
"""
|
||
可灵主体（Element）API 客户端
|
||
封装对 {base}/kling/v1/general/* 的调用，统一注入 Authorization。
|
||
两类调用方：
|
||
- 后端代理路由（__init__.py）：面板的上传/创建/刷新/列表/删除。
|
||
- 节点提交（K3_video.py）：生视频前查列表拿 名称→element_id 映射。
|
||
所有方法返回后端的响应信封 {"success", "message", "data"} 解出的 data，
|
||
失败抛 RuntimeError（带 message），由调用方决定如何呈现。
|
||
"""
|
||
|
||
import aiohttp
|
||
|
||
from ..utils.config import get_api_key_or_raise, get_base_url_by_route
|
||
|
||
|
||
_ELEMENT_PREFIX = "/kling/v1/general"
|
||
|
||
|
||
def _headers(api_key: str = None) -> dict:
|
||
key = api_key or get_api_key_or_raise()
|
||
return {"Authorization": f"Bearer {key}"}
|
||
|
||
|
||
def _resolve_base(route: str = None, base_url: str = None) -> str:
|
||
"""Prefer an explicit URL, otherwise use the global network route."""
|
||
if base_url:
|
||
return base_url.rstrip("/")
|
||
return get_base_url_by_route().rstrip("/")
|
||
|
||
|
||
def _unwrap(payload: dict):
|
||
"""从响应信封取 data；success=false 时抛 RuntimeError。"""
|
||
if not isinstance(payload, dict):
|
||
raise RuntimeError(f"主体接口返回异常：{payload!r}")
|
||
if not payload.get("success", False):
|
||
raise RuntimeError(payload.get("message") or "主体接口调用失败")
|
||
return payload.get("data")
|
||
|
||
|
||
async def list_elements(session: aiohttp.ClientSession, *, route=None, base_url=None,
|
||
include_all=False, api_key=None):
|
||
"""GET /advanced-custom-elements：返回主体列表（默认仅 succeed）。"""
|
||
base = _resolve_base(route, base_url)
|
||
url = f"{base}{_ELEMENT_PREFIX}/advanced-custom-elements"
|
||
params = {"pageNum": "1", "pageSize": "100"}
|
||
async with session.get(url, headers=_headers(api_key), params=params) as resp:
|
||
data = await resp.json()
|
||
result = _unwrap(data)
|
||
# 新API返回: {"code": 0, "data": [...], "total": N}
|
||
if isinstance(result, dict) and "data" in result:
|
||
elements = result.get("data", [])
|
||
else:
|
||
elements = result if isinstance(result, list) else []
|
||
# 过滤：默认只返回 succeed 状态
|
||
if not include_all:
|
||
elements = [e for e in elements if e.get("status") == "succeed"]
|
||
return elements
|
||
|
||
|
||
async def fetch_name_to_id_map(session: aiohttp.ClientSession, *, route=None,
|
||
base_url=None, api_key=None) -> dict:
|
||
"""生视频用：返回 {主体名称: element_id}，仅含已成功的主体。
|
||
|
||
注意：新API返回的 element_id 是 int64 数字类型，不是字符串。
|
||
"""
|
||
elements = await list_elements(session, route=route, base_url=base_url, api_key=api_key)
|
||
mapping = {}
|
||
for e in elements:
|
||
name = (e.get("name") or "").strip()
|
||
eid = e.get("element_id")
|
||
# element_id 可能是数字或字符串，统一保持原始类型（生视频时需要数字）
|
||
if name and eid is not None:
|
||
mapping[name] = eid
|
||
return mapping
|