refactor(provider-ops): 重构认证配置为 schema-driven 模式

后端架构类通过 get_credentials_schema() 返回带 x-* 扩展字段的 JSON Schema,
前端根据 schema 动态渲染表单、构建请求、验证和格式化显示。
新增架构只需后端一个文件,前端零改动。

主要变更:
- 删除前端手写模板文件(anyrouter.ts, cubence.ts, nekocode.ts, new-api.ts, yescode.ts)
- 新增 schema-utils.ts 实现 schema 到表单的转换、请求构建、验证和格式化
- 新增 field-hooks.ts 支持 schema 声明式字段联动钩子
- 后端 base.py 提供 parse_verify_response 默认实现,减少子类重复代码
- 各架构余额查询逻辑复用 balance.py 中的通用函数
- 删除废弃的 one_api.py 架构,新增架构 hidden 属性
- API 返回 credentials_schema 供前端消费
This commit is contained in:
fawney19
2026-02-13 16:41:30 +08:00
parent 7e68882872
commit d6b39babf5
30 changed files with 1233 additions and 1835 deletions

View File

@@ -40,6 +40,7 @@ class ArchitectureInfo(BaseModel):
architecture_id: str
display_name: str
description: str
credentials_schema: dict[str, Any]
supported_auth_types: list[dict[str, str]]
supported_actions: list[dict[str, Any]]
default_connector: str | None

View File

@@ -2,14 +2,13 @@
Anyrouter 余额查询操作(含自动签到)
"""
import time
from typing import Any
import httpx
from src.core.logger import logger
from src.services.provider_ops.actions.balance import BalanceAction
from src.services.provider_ops.types import ActionResult, ActionStatus, BalanceInfo
from src.services.provider_ops.types import BalanceInfo
class AnyrouterBalanceAction(BalanceAction):
@@ -26,61 +25,7 @@ class AnyrouterBalanceAction(BalanceAction):
display_name = "查询余额(含自动签到)"
description = "查询账户余额,同时自动签到"
async def _do_query_balance(self, client: httpx.AsyncClient) -> ActionResult:
"""执行余额查询"""
endpoint = self.config.get("endpoint", "/api/user/self")
method = self.config.get("method", "GET")
start_time = time.time()
try:
response = await client.request(method, endpoint)
response_time_ms = int((time.time() - start_time) * 1000)
try:
data = response.json()
except Exception:
return self._make_error_result(
ActionStatus.PARSE_ERROR,
"响应不是有效的 JSON",
)
if response.status_code != 200:
return self._handle_http_error(response, data)
if data.get("success") is False:
message = data.get("message", "业务状态码表示失败")
return self._make_error_result(
ActionStatus.UNKNOWN_ERROR,
message,
raw_response=data,
)
balance = self._parse_balance(data)
return self._make_success_result(
data=balance,
response_time_ms=response_time_ms,
raw_response=data,
)
except httpx.TimeoutException:
return self._make_error_result(
ActionStatus.NETWORK_ERROR,
"请求超时",
retry_after_seconds=30,
)
except httpx.RequestError as e:
return self._make_error_result(
ActionStatus.NETWORK_ERROR,
f"网络错误: {str(e)}",
retry_after_seconds=30,
)
except Exception as e:
return self._make_error_result(
ActionStatus.UNKNOWN_ERROR,
f"未知错误: {str(e)}",
)
_cookie_auth = True
def _parse_balance(self, data: Any) -> BalanceInfo:
"""解析余额信息"""
@@ -100,23 +45,6 @@ class AnyrouterBalanceAction(BalanceAction):
currency=self.config.get("currency", "USD"),
)
def _handle_http_error(
self, response: httpx.Response, raw_data: dict[str, Any] | None = None
) -> ActionResult:
"""处理 HTTP 错误响应"""
status_code = response.status_code
if status_code == 401:
return self._make_error_result(
ActionStatus.AUTH_FAILED, "Cookie 已失效,请重新配置", raw_response=raw_data
)
elif status_code == 403:
return self._make_error_result(
ActionStatus.AUTH_FAILED, "Cookie 已失效或无权限", raw_response=raw_data
)
return super()._handle_http_error(response, raw_data)
async def _do_checkin(self, client: httpx.AsyncClient) -> dict[str, Any] | None:
"""
执行自动签到(始终执行)

View File

@@ -2,6 +2,7 @@
余额查询操作抽象基类
"""
import time
from abc import abstractmethod
from typing import Any
@@ -20,8 +21,11 @@ class BalanceAction(ProviderAction):
"""
余额查询操作抽象基类
子类必须实现 _do_query_balance() 方法来处理特定平台的余额查询逻辑。
子类可选实现 _do_checkin() 方法来在查询余额前执行签到
子类必须实现 _parse_balance() 方法来处理特定平台的余额解析逻辑。
子类可选重写 _do_query_balance() 或 _do_checkin() 进行自定义
如果子类使用 Cookie 认证,设置 _cookie_auth = True 可让 401/403 错误
显示 "Cookie 已失效" 而非 "认证失败"
"""
action_type = ProviderActionType.QUERY_BALANCE
@@ -29,6 +33,9 @@ class BalanceAction(ProviderAction):
description = "查询账户余额信息"
default_cache_ttl = 86400 # 24 小时
# 子类设为 True 即可在 401/403 时显示 "Cookie 已失效" 消息
_cookie_auth: bool = False
async def execute(self, client: httpx.AsyncClient) -> ActionResult:
"""
执行余额查询(模板方法)
@@ -67,10 +74,13 @@ class BalanceAction(ProviderAction):
return result
@abstractmethod
async def _do_query_balance(self, client: httpx.AsyncClient) -> ActionResult:
"""
执行余额查询(子类必须实现)
执行余额查询
默认实现处理通用的请求/响应/错误处理流程。
子类只需实现 _parse_balance() 即可。
如果查询逻辑不同(如并发调用多个接口),子类可重写此方法。
Args:
client: 已认证的 HTTP 客户端
@@ -78,8 +88,109 @@ class BalanceAction(ProviderAction):
Returns:
ActionResult其中 data 字段为 BalanceInfo
"""
endpoint = self.config.get("endpoint", "/api/user/self")
method = self.config.get("method", "GET")
start_time = time.time()
try:
response = await client.request(method, endpoint)
response_time_ms = int((time.time() - start_time) * 1000)
try:
data = response.json()
except Exception:
return self._make_error_result(
ActionStatus.PARSE_ERROR,
"响应不是有效的 JSON",
)
if response.status_code != 200:
return self._handle_http_error(response, data)
if data.get("success") is False:
message = data.get("message", "业务状态码表示失败")
return self._make_error_result(
ActionStatus.UNKNOWN_ERROR,
message,
raw_response=data,
)
balance = self._parse_balance(data)
return self._make_success_result(
data=balance,
response_time_ms=response_time_ms,
raw_response=data,
)
except httpx.TimeoutException:
return self._make_error_result(
ActionStatus.NETWORK_ERROR,
"请求超时",
retry_after_seconds=30,
)
except httpx.RequestError as e:
return self._make_error_result(
ActionStatus.NETWORK_ERROR,
f"网络错误: {str(e)}",
retry_after_seconds=30,
)
except Exception as e:
return self._make_error_result(
ActionStatus.UNKNOWN_ERROR,
f"未知错误: {str(e)}",
)
@abstractmethod
def _parse_balance(self, data: Any) -> BalanceInfo:
"""
解析余额数据(子类必须实现)
Args:
data: API 响应 JSON 数据
Returns:
BalanceInfo 对象
"""
pass
def _handle_http_error(
self, response: httpx.Response, raw_data: dict[str, Any] | None = None
) -> ActionResult:
"""
处理 HTTP 错误响应
Cookie 认证的子类设置 _cookie_auth = True 即可获得友好的错误提示,
无需再逐个重写此方法。
"""
status_code = response.status_code
if status_code == 401:
msg = "Cookie 已失效,请重新配置" if self._cookie_auth else "认证失败"
return self._make_error_result(ActionStatus.AUTH_FAILED, msg, raw_response=raw_data)
elif status_code == 403:
msg = "Cookie 已失效或无权限" if self._cookie_auth else "无权限访问"
return self._make_error_result(ActionStatus.AUTH_FAILED, msg, raw_response=raw_data)
elif status_code == 404:
return self._make_error_result(
ActionStatus.NOT_SUPPORTED, "功能未开放", raw_response=raw_data
)
elif status_code == 429:
retry_after = response.headers.get("Retry-After")
return self._make_error_result(
ActionStatus.RATE_LIMITED,
"请求频率限制",
retry_after_seconds=int(retry_after) if retry_after else 60,
raw_response=raw_data,
)
else:
return self._make_error_result(
ActionStatus.UNKNOWN_ERROR,
f"HTTP {status_code}: {response.reason_phrase}",
raw_response=raw_data,
)
async def _do_checkin(self, client: httpx.AsyncClient) -> dict[str, Any] | None:
"""
执行签到(子类可选实现)

View File

@@ -2,13 +2,10 @@
Cubence 余额查询操作
"""
import time
from typing import Any
import httpx
from src.services.provider_ops.actions.balance import BalanceAction
from src.services.provider_ops.types import ActionResult, ActionStatus, BalanceInfo
from src.services.provider_ops.types import BalanceInfo
class CubenceBalanceAction(BalanceAction):
@@ -24,84 +21,7 @@ class CubenceBalanceAction(BalanceAction):
display_name = "查询余额(含窗口限额)"
description = "查询账户余额和窗口限额信息"
async def _do_query_balance(self, client: httpx.AsyncClient) -> ActionResult:
"""执行 Cubence 余额查询(实现抽象方法)"""
endpoint = self.config.get("endpoint", "/api/v1/dashboard/overview")
method = self.config.get("method", "GET")
start_time = time.time()
try:
response = await client.request(method, endpoint)
response_time_ms = int((time.time() - start_time) * 1000)
# 尝试解析 JSON
try:
data = response.json()
except Exception:
return self._make_error_result(
ActionStatus.PARSE_ERROR,
"响应不是有效的 JSON",
)
# 检查 HTTP 状态
if response.status_code != 200:
return self._handle_http_error(response, data)
# 检查业务状态码
if data.get("success") is False:
message = data.get("message", "业务状态码表示失败")
return self._make_error_result(
ActionStatus.UNKNOWN_ERROR,
message,
raw_response=data,
)
# 解析余额信息
balance = self._parse_balance(data)
return self._make_success_result(
data=balance,
response_time_ms=response_time_ms,
raw_response=data,
)
except httpx.TimeoutException:
return self._make_error_result(
ActionStatus.NETWORK_ERROR,
"请求超时",
retry_after_seconds=30,
)
except httpx.RequestError as e:
return self._make_error_result(
ActionStatus.NETWORK_ERROR,
f"网络错误: {str(e)}",
retry_after_seconds=30,
)
except Exception as e:
return self._make_error_result(
ActionStatus.UNKNOWN_ERROR,
f"未知错误: {str(e)}",
)
def _handle_http_error(
self, response: httpx.Response, raw_data: dict[str, Any] | None = None
) -> ActionResult:
"""处理 HTTP 错误响应Cubence 专用)"""
status_code = response.status_code
# Cubence 使用 Cookie 认证,提供更友好的错误提示
if status_code == 401:
return self._make_error_result(
ActionStatus.AUTH_FAILED, "Cookie 已失效,请重新配置", raw_response=raw_data
)
elif status_code == 403:
return self._make_error_result(
ActionStatus.AUTH_FAILED, "Cookie 已失效或无权限", raw_response=raw_data
)
# 其他错误使用基类处理
return super()._handle_http_error(response, raw_data)
_cookie_auth = True
def _parse_balance(self, data: Any) -> BalanceInfo:
"""解析 Cubence 余额信息"""

View File

@@ -2,15 +2,12 @@
NekoCode 余额查询操作
"""
import time
from datetime import datetime
from typing import Any
import httpx
from src.core.logger import logger
from src.services.provider_ops.actions.balance import BalanceAction
from src.services.provider_ops.types import ActionResult, ActionStatus, BalanceInfo
from src.services.provider_ops.types import BalanceInfo
class NekoCodeBalanceAction(BalanceAction):
@@ -27,61 +24,7 @@ class NekoCodeBalanceAction(BalanceAction):
display_name = "查询余额"
description = "查询 NekoCode 账户余额和订阅信息"
async def _do_query_balance(self, client: httpx.AsyncClient) -> ActionResult:
"""执行余额查询"""
endpoint = self.config.get("endpoint", "/api/usage/summary")
method = self.config.get("method", "GET")
start_time = time.time()
try:
response = await client.request(method, endpoint)
response_time_ms = int((time.time() - start_time) * 1000)
try:
data = response.json()
except Exception:
return self._make_error_result(
ActionStatus.PARSE_ERROR,
"响应不是有效的 JSON",
)
if response.status_code != 200:
return self._handle_http_error(response, data)
if data.get("success") is False:
message = data.get("message", "业务状态码表示失败")
return self._make_error_result(
ActionStatus.UNKNOWN_ERROR,
message,
raw_response=data,
)
balance = self._parse_balance(data)
return self._make_success_result(
data=balance,
response_time_ms=response_time_ms,
raw_response=data,
)
except httpx.TimeoutException:
return self._make_error_result(
ActionStatus.NETWORK_ERROR,
"请求超时",
retry_after_seconds=30,
)
except httpx.RequestError as e:
return self._make_error_result(
ActionStatus.NETWORK_ERROR,
f"网络错误: {str(e)}",
retry_after_seconds=30,
)
except Exception as e:
return self._make_error_result(
ActionStatus.UNKNOWN_ERROR,
f"未知错误: {str(e)}",
)
_cookie_auth = True
def _parse_balance(self, data: Any) -> BalanceInfo:
"""解析余额信息"""
@@ -160,23 +103,6 @@ class NekoCodeBalanceAction(BalanceAction):
extra=extra,
)
def _handle_http_error(
self, response: httpx.Response, raw_data: dict[str, Any] | None = None
) -> ActionResult:
"""处理 HTTP 错误响应"""
status_code = response.status_code
if status_code == 401:
return self._make_error_result(
ActionStatus.AUTH_FAILED, "Cookie 已失效,请重新配置", raw_response=raw_data
)
elif status_code == 403:
return self._make_error_result(
ActionStatus.AUTH_FAILED, "Cookie 已失效或无权限", raw_response=raw_data
)
return super()._handle_http_error(response, raw_data)
@classmethod
def get_config_schema(cls) -> dict[str, Any]:
"""获取操作配置 schema"""

View File

@@ -2,14 +2,13 @@
New API 余额查询操作
"""
import time
from typing import Any
import httpx
from src.core.logger import logger
from src.services.provider_ops.actions.balance import BalanceAction
from src.services.provider_ops.types import ActionResult, ActionStatus, BalanceInfo
from src.services.provider_ops.types import BalanceInfo
class NewApiBalanceAction(BalanceAction):
@@ -25,66 +24,6 @@ class NewApiBalanceAction(BalanceAction):
display_name = "查询余额"
description = "查询 New API 账户余额信息"
async def _do_query_balance(self, client: httpx.AsyncClient) -> ActionResult:
"""执行余额查询(实现抽象方法)"""
endpoint = self.config.get("endpoint", "/api/user/self")
method = self.config.get("method", "GET")
start_time = time.time()
try:
response = await client.request(method, endpoint)
response_time_ms = int((time.time() - start_time) * 1000)
# 尝试解析 JSON
try:
data = response.json()
except Exception:
return self._make_error_result(
ActionStatus.PARSE_ERROR,
"响应不是有效的 JSON",
)
# 检查 HTTP 状态
if response.status_code != 200:
return self._handle_http_error(response, data)
# 检查业务状态码
if data.get("success") is False:
message = data.get("message", "业务状态码表示失败")
return self._make_error_result(
ActionStatus.UNKNOWN_ERROR,
message,
raw_response=data,
)
# 解析余额信息
balance = self._parse_balance(data)
return self._make_success_result(
data=balance,
response_time_ms=response_time_ms,
raw_response=data,
)
except httpx.TimeoutException:
return self._make_error_result(
ActionStatus.NETWORK_ERROR,
"请求超时",
retry_after_seconds=30,
)
except httpx.RequestError as e:
return self._make_error_result(
ActionStatus.NETWORK_ERROR,
f"网络错误: {str(e)}",
retry_after_seconds=30,
)
except Exception as e:
return self._make_error_result(
ActionStatus.UNKNOWN_ERROR,
f"未知错误: {str(e)}",
)
def _parse_balance(
self,
data: Any,

View File

@@ -168,6 +168,12 @@ class YesCodeBalanceAction(BalanceAction):
display_name = "查询余额(含每周限额)"
description = "查询账户余额和每周限额信息"
_cookie_auth = True
def _parse_balance(self, data: Any) -> BalanceInfo:
"""YesCode 不使用基类的 _do_query_balance此方法不会被调用"""
raise NotImplementedError("YesCode 使用自定义 _do_query_balance")
async def _do_query_balance(self, client: httpx.AsyncClient) -> ActionResult:
"""执行余额查询(实现抽象方法,复用 client 调用两个接口获取完整数据)"""
import time

View File

@@ -12,7 +12,6 @@ from src.services.provider_ops.architectures.cubence import CubenceArchitecture
from src.services.provider_ops.architectures.generic_api import GenericApiArchitecture
from src.services.provider_ops.architectures.nekocode import NekoCodeArchitecture
from src.services.provider_ops.architectures.new_api import NewApiArchitecture
from src.services.provider_ops.architectures.one_api import OneApiArchitecture
from src.services.provider_ops.architectures.yescode import YesCodeArchitecture
__all__ = [
@@ -24,6 +23,5 @@ __all__ = [
"GenericApiArchitecture",
"NekoCodeArchitecture",
"NewApiArchitecture",
"OneApiArchitecture",
"YesCodeArchitecture",
]

View File

@@ -18,9 +18,9 @@ from src.services.provider_ops.actions import (
from src.services.provider_ops.architectures.base import (
ProviderArchitecture,
ProviderConnector,
VerifyResult,
)
from src.services.provider_ops.types import ConnectorAuthType, ProviderActionType
from src.services.provider_ops.utils import extract_cookie_value
from src.utils.ssl_utils import get_ssl_context
# acw_sc__v2 算法常量
@@ -93,31 +93,6 @@ def _compute_acw_sc_v2(arg1: str) -> str:
return result
def _extract_session_from_cookie(cookie_string: str) -> str:
"""
从完整的 Cookie 字符串中提取 session 值
支持两种输入格式:
1. 完整 Cookie: "session=xxx; acw_tc=xxx; ..."
2. 仅 session 值: "MTc2ODc4..."
Args:
cookie_string: Cookie 字符串或 session 值
Returns:
session cookie 的值
"""
# 如果包含 "session=",说明是完整 Cookie 字符串
if "session=" in cookie_string:
# 解析 Cookie 字符串
for part in cookie_string.split(";"):
part = part.strip()
if part.startswith("session="):
return part[8:] # 去掉 "session=" 前缀
# 否则认为直接是 session 值
return cookie_string.strip()
def _parse_session_user_id(cookie_input: str) -> tuple[str | None, str | None]:
"""
从 session cookie 中解析用户 ID 和用户名
@@ -138,7 +113,7 @@ def _parse_session_user_id(cookie_input: str) -> tuple[str | None, str | None]:
"""
try:
# 先提取 session 值
session_cookie = _extract_session_from_cookie(cookie_input)
session_cookie = extract_cookie_value(cookie_input, "session")
# 1. URL-safe base64 解码外层
padding = 4 - len(session_cookie) % 4
if padding != 4:
@@ -292,7 +267,7 @@ class AnyrouterConnector(ProviderConnector):
return False
# 提取纯 session 值(支持完整 Cookie 字符串或仅 session 值)
self._session_cookie = _extract_session_from_cookie(session_cookie)
self._session_cookie = extract_cookie_value(session_cookie, "session")
# 解析 user_id
self._user_id, _ = _parse_session_user_id(session_cookie)
@@ -341,13 +316,36 @@ class AnyrouterConnector(ProviderConnector):
return {
"type": "object",
"properties": {
"base_url": {
"type": "string",
"title": "站点地址",
"description": "API 基础地址",
"x-default-value": "https://anyrouter.top",
},
"session_cookie": {
"type": "string",
"title": "Session Cookie",
"description": "从浏览器复制的 session Cookie 值",
"x-sensitive": True,
"x-input-type": "password",
},
},
"required": ["session_cookie"],
"x-field-groups": [
{"fields": ["base_url"]},
{"fields": ["session_cookie"]},
],
"x-auth-type": "cookie",
"x-default-base-url": "https://anyrouter.top",
"x-validation": [
{
"type": "required",
"fields": ["session_cookie"],
"message": "请填写 Session Cookie",
},
],
"x-quota-divisor": 500000,
"x-currency": "USD",
}
@@ -441,7 +439,7 @@ class AnyrouterArchitecture(ProviderArchitecture):
cookie_input = credentials.get("session_cookie")
if cookie_input:
# 提取 session 值(支持完整 Cookie 字符串或仅 session 值)
session_value = _extract_session_from_cookie(cookie_input)
session_value = extract_cookie_value(cookie_input, "session")
cookies.append(f"session={session_value}")
# 从 session 解析 user_id 并添加 New-Api-User header
@@ -454,33 +452,8 @@ class AnyrouterArchitecture(ProviderArchitecture):
return headers
def parse_verify_response(
self,
status_code: int,
data: dict[str, Any],
) -> VerifyResult:
"""解析 Anyrouter 验证响应"""
def _auth_fail_message(self, status_code: int) -> str:
"""Cookie 认证的错误消息"""
if status_code == 401:
return VerifyResult(success=False, message="Cookie 已失效,请重新配置")
if status_code == 403:
return VerifyResult(success=False, message="Cookie 已失效或无权限")
if status_code != 200:
return VerifyResult(success=False, message=f"验证失败HTTP {status_code}")
# Anyrouter 响应格式: {"success": true, "data": {...}}
if not data.get("success"):
message = data.get("message", "验证失败")
return VerifyResult(success=False, message=message)
user_data = data.get("data", {})
return VerifyResult(
success=True,
username=user_data.get("username"),
display_name=user_data.get("display_name") or user_data.get("username"),
email=user_data.get("email"),
quota=user_data.get("quota"),
used_quota=user_data.get("used_quota"),
request_count=user_data.get("request_count"),
extra=None,
)
return "Cookie 已失效,请重新配置"
return "Cookie 已失效或无权限"

View File

@@ -241,6 +241,9 @@ class ProviderArchitecture(ABC):
display_name: str = ""
description: str = ""
# 设为 True 时不在架构列表 API 中返回(内部使用的架构)
hidden: bool = False
# 支持的 Connector 类型列表(按优先级排序)
supported_connectors: list[type[ProviderConnector]] = []
@@ -259,7 +262,7 @@ class ProviderArchitecture(ABC):
"""
self.config = config or {}
# ==================== 认证验证相关方法(子类必须实现) ====================
# ==================== 认证验证相关方法 ====================
@abstractmethod
def get_credentials_schema(self) -> dict[str, Any]:
@@ -267,26 +270,9 @@ class ProviderArchitecture(ABC):
获取凭据字段定义JSON Schema 格式)
子类必须实现此方法定义需要的凭据字段。
这个 schema 可用于:
1. 前端表单生成(如果需要动态渲染)
2. 凭据验证
3. 文档生成
Returns:
JSON Schema 格式的字段定义
Example:
{
"type": "object",
"properties": {
"api_key": {
"type": "string",
"title": "API Key",
"description": "访问令牌",
},
},
"required": ["api_key"],
}
"""
pass
@@ -322,7 +308,6 @@ class ProviderArchitecture(ABC):
"""
pass
@abstractmethod
def parse_verify_response(
self,
status_code: int,
@@ -331,7 +316,8 @@ class ProviderArchitecture(ABC):
"""
解析认证验证响应
子类必须实现此方法解析响应
默认实现处理通用的 {"success": bool, "data": {...}} 格式
子类可重写 _auth_fail_message() 和 _build_verify_result() 进行自定义。
Args:
status_code: HTTP 状态码
@@ -340,7 +326,61 @@ class ProviderArchitecture(ABC):
Returns:
验证结果
"""
pass
if status_code == 401:
return VerifyResult(success=False, message=self._auth_fail_message(401))
if status_code == 403:
return VerifyResult(success=False, message=self._auth_fail_message(403))
if status_code != 200:
return VerifyResult(success=False, message=f"验证失败HTTP {status_code}")
# 解析通用响应格式
if data.get("success") is True and "data" in data:
user_data = data["data"]
elif data.get("success") is False:
message = data.get("message", "验证失败")
return VerifyResult(success=False, message=message)
else:
user_data = data
return self._build_verify_result(user_data, data)
def _auth_fail_message(self, status_code: int) -> str:
"""
获取认证失败消息
子类可重写以提供自定义消息(如 Cookie 认证场景)。
"""
if status_code == 401:
return "认证失败:无效的凭据"
return "认证失败:权限不足"
def _build_verify_result(
self, user_data: dict[str, Any], raw_data: dict[str, Any] | None = None
) -> VerifyResult:
"""
从用户数据构建验证结果
默认实现提取 username, display_name, email, quota, used_quota, request_count。
子类可重写以自定义字段提取。
"""
known_fields = (
"username",
"display_name",
"email",
"quota",
"used_quota",
"request_count",
)
return VerifyResult(
success=True,
username=user_data.get("username"),
display_name=user_data.get("display_name") or user_data.get("username"),
email=user_data.get("email"),
quota=user_data.get("quota"),
used_quota=user_data.get("used_quota"),
request_count=user_data.get("request_count"),
extra={k: v for k, v in user_data.items() if k not in known_fields},
)
# ==================== 可选的钩子方法 ====================

View File

@@ -16,31 +16,7 @@ from src.services.provider_ops.architectures.base import (
VerifyResult,
)
from src.services.provider_ops.types import ConnectorAuthType, ProviderActionType
def _extract_token_from_cookie(cookie_string: str) -> str:
"""
从完整的 Cookie 字符串中提取 token 值
支持两种输入格式:
1. 完整 Cookie: "token=xxx; other=yyy; ..."
2. 仅 token 值: "eyJhbGciOiJI..."
Args:
cookie_string: Cookie 字符串或 token 值
Returns:
token cookie 的值
"""
# 如果包含 "token=",说明是完整 Cookie 字符串
if "token=" in cookie_string:
# 解析 Cookie 字符串
for part in cookie_string.split(";"):
part = part.strip()
if part.startswith("token="):
return part[6:] # 去掉 "token=" 前缀
# 否则认为直接是 token 值
return cookie_string.strip()
from src.services.provider_ops.utils import extract_cookie_value
class CubenceConnector(ProviderConnector):
@@ -66,7 +42,7 @@ class CubenceConnector(ProviderConnector):
return False
# 提取纯 token 值
self._token_cookie = _extract_token_from_cookie(token_cookie)
self._token_cookie = extract_cookie_value(token_cookie, "token")
self._set_connected()
return True
@@ -93,13 +69,50 @@ class CubenceConnector(ProviderConnector):
return {
"type": "object",
"properties": {
"base_url": {
"type": "string",
"title": "站点地址",
"description": "API 基础地址",
"x-default-value": "https://cubence.com",
},
"token_cookie": {
"type": "string",
"title": "Token Cookie",
"description": "从浏览器复制的 token Cookie 值JWT 格式)",
"x-sensitive": True,
"x-input-type": "password",
},
},
"required": ["token_cookie"],
"x-field-groups": [
{"fields": ["base_url"]},
{"fields": ["token_cookie"]},
],
"x-auth-type": "cookie",
"x-default-base-url": "https://cubence.com",
"x-validation": [
{
"type": "required",
"fields": ["token_cookie"],
"message": "请填写 Token Cookie",
},
],
"x-quota-divisor": None,
"x-currency": "USD",
"x-balance-extra-format": [
{
"label": "5h",
"type": "window_limit",
"source": "five_hour_limit",
"unit_divisor": 1000000,
},
{
"label": "",
"type": "window_limit",
"source": "weekly_limit",
"unit_divisor": 1000000,
},
],
}
@@ -161,30 +174,21 @@ class CubenceArchitecture(ProviderArchitecture):
# 添加 token Cookie
cookie_input = credentials.get("token_cookie")
if cookie_input:
token_value = _extract_token_from_cookie(cookie_input)
token_value = extract_cookie_value(cookie_input, "token")
headers["Cookie"] = f"token={token_value}"
return headers
def parse_verify_response(
self,
status_code: int,
data: dict[str, Any],
) -> VerifyResult:
"""解析 Cubence 验证响应"""
def _auth_fail_message(self, status_code: int) -> str:
"""Cookie 认证的错误消息"""
if status_code == 401:
return VerifyResult(success=False, message="Cookie 已失效,请重新配置")
if status_code == 403:
return VerifyResult(success=False, message="Cookie 已失效或无权限")
if status_code != 200:
return VerifyResult(success=False, message=f"验证失败HTTP {status_code}")
return "Cookie 已失效,请重新配置"
return "Cookie 已失效或无权限"
# Cubence 响应格式: {"success": true, "data": {...}}
if not data.get("success"):
message = data.get("message", "验证失败")
return VerifyResult(success=False, message=message)
user_data = data.get("data", {})
def _build_verify_result(
self, user_data: dict[str, Any], raw_data: dict[str, Any] | None = None
) -> VerifyResult:
"""Cubence 自定义字段提取user/balance/subscription_limits"""
user_info = user_data.get("user", {})
balance_info = user_data.get("balance", {})
subscription_limits = user_data.get("subscription_limits", {})

View File

@@ -50,7 +50,6 @@ from src.services.provider_ops.actions import (
from src.services.provider_ops.architectures.base import (
ProviderArchitecture,
ProviderConnector,
VerifyResult,
)
from src.services.provider_ops.types import ConnectorAuthType, ProviderActionType
@@ -112,13 +111,35 @@ class GenericApiKeyConnector(ProviderConnector):
return {
"type": "object",
"properties": {
"base_url": {
"type": "string",
"title": "站点地址",
"description": "API 基础地址",
},
"api_key": {
"type": "string",
"title": "API Key",
"description": "提供商的 API Key",
"x-sensitive": True,
"x-input-type": "password",
},
},
"required": ["api_key"],
"x-field-groups": [
{"fields": ["base_url"]},
{"fields": ["api_key"]},
],
"x-auth-type": "api_key",
"x-auth-method": "bearer",
"x-validation": [
{
"type": "required",
"fields": ["api_key"],
"message": "请填写 API Key",
},
],
"x-quota-divisor": 500000,
"x-currency": "USD",
}
@@ -135,6 +156,7 @@ class GenericApiArchitecture(ProviderArchitecture):
architecture_id = "generic_api"
display_name = "通用 API"
description = "可配置的通用 API 架构,适用于各种中转站"
hidden = True
supported_connectors: list[type[ProviderConnector]] = [
GenericApiKeyConnector,
@@ -180,48 +202,3 @@ class GenericApiArchitecture(ProviderArchitecture):
headers[header_name] = api_key
return headers
def parse_verify_response(
self,
status_code: int,
data: dict[str, Any],
) -> VerifyResult:
"""解析通用 API 验证响应"""
if status_code == 401:
return VerifyResult(success=False, message="认证失败:无效的凭据")
if status_code == 403:
return VerifyResult(success=False, message="认证失败:权限不足")
if status_code != 200:
return VerifyResult(success=False, message=f"验证失败HTTP {status_code}")
# 尝试解析通用响应格式
if data.get("success") is True and "data" in data:
user_data = data["data"]
elif data.get("success") is False:
message = data.get("message", "验证失败")
return VerifyResult(success=False, message=message)
else:
user_data = data
return VerifyResult(
success=True,
username=user_data.get("username"),
display_name=user_data.get("display_name") or user_data.get("username"),
email=user_data.get("email"),
quota=user_data.get("quota"),
used_quota=user_data.get("used_quota"),
request_count=user_data.get("request_count"),
extra={
k: v
for k, v in user_data.items()
if k
not in (
"username",
"display_name",
"email",
"quota",
"used_quota",
"request_count",
)
},
)

View File

@@ -17,34 +17,10 @@ from src.services.provider_ops.architectures.base import (
VerifyResult,
)
from src.services.provider_ops.types import ConnectorAuthType, ProviderActionType
from src.services.provider_ops.utils import extract_cookie_value
from src.utils.ssl_utils import get_ssl_context
def _extract_session_from_cookie(cookie_string: str) -> str:
"""
从完整的 Cookie 字符串中提取 session 值
支持两种输入格式:
1. 完整 Cookie: "session=xxx; other=xxx; ..."
2. 仅 session 值: "MTc2OTYx..."
Args:
cookie_string: Cookie 字符串或 session 值
Returns:
session cookie 的值
"""
# 如果包含 "session=",说明是完整 Cookie 字符串
if "session=" in cookie_string:
# 解析 Cookie 字符串
for part in cookie_string.split(";"):
part = part.strip()
if part.startswith("session="):
return part[8:] # 去掉 "session=" 前缀
# 否则认为直接是 session 值
return cookie_string.strip()
class NekoCodeConnector(ProviderConnector):
"""
NekoCode 专用连接器
@@ -68,7 +44,7 @@ class NekoCodeConnector(ProviderConnector):
return False
# 提取纯 session 值(支持完整 Cookie 字符串或仅 session 值)
self._session_cookie = _extract_session_from_cookie(session_cookie)
self._session_cookie = extract_cookie_value(session_cookie, "session")
self._set_connected()
return True
@@ -95,13 +71,50 @@ class NekoCodeConnector(ProviderConnector):
return {
"type": "object",
"properties": {
"base_url": {
"type": "string",
"title": "站点地址",
"description": "API 基础地址",
"x-default-value": "https://nekocode.ai",
},
"session_cookie": {
"type": "string",
"title": "Session Cookie",
"description": "从浏览器复制的 session Cookie 值",
"x-sensitive": True,
"x-input-type": "password",
},
},
"required": ["session_cookie"],
"x-field-groups": [
{"fields": ["base_url"]},
{"fields": ["session_cookie"]},
],
"x-auth-type": "cookie",
"x-default-base-url": "https://nekocode.ai",
"x-validation": [
{
"type": "required",
"fields": ["session_cookie"],
"message": "请填写 Session Cookie",
},
],
"x-quota-divisor": None,
"x-currency": "USD",
"x-balance-extra-format": [
{
"label": "",
"type": "daily_quota",
"source_limit": "daily_quota_limit",
"source_remaining": "daily_remaining_quota",
"source_start_date": "effective_start_date",
},
{
"label": "",
"type": "monthly_expiry",
"source_end_date": "effective_end_date",
},
],
}
@@ -172,7 +185,7 @@ class NekoCodeArchitecture(ProviderArchitecture):
# 添加 Cookie
cookie_input = credentials.get("session_cookie")
if cookie_input:
session_value = _extract_session_from_cookie(cookie_input)
session_value = extract_cookie_value(cookie_input, "session")
headers["Cookie"] = f"session={session_value}"
# 构建 client 参数
@@ -218,31 +231,21 @@ class NekoCodeArchitecture(ProviderArchitecture):
cookie_input = credentials.get("session_cookie")
if cookie_input:
# 提取 session 值(支持完整 Cookie 字符串或仅 session 值)
session_value = _extract_session_from_cookie(cookie_input)
session_value = extract_cookie_value(cookie_input, "session")
headers["Cookie"] = f"session={session_value}"
return headers
def parse_verify_response(
self,
status_code: int,
data: dict[str, Any],
) -> VerifyResult:
"""解析 NekoCode 验证响应(/api/user/self + _usage_summary"""
def _auth_fail_message(self, status_code: int) -> str:
"""Cookie 认证的错误消息"""
if status_code == 401:
return VerifyResult(success=False, message="Cookie 已失效,请重新配置")
if status_code == 403:
return VerifyResult(success=False, message="Cookie 已失效或无权限")
if status_code != 200:
return VerifyResult(success=False, message=f"验证失败HTTP {status_code}")
# NekoCode 响应格式: {"success": true, "data": {...}}
if not data.get("success"):
message = data.get("message", "验证失败")
return VerifyResult(success=False, message=message)
user_data = data.get("data", {})
return "Cookie 已失效,请重新配置"
return "Cookie 已失效或无权限"
def _build_verify_result(
self, user_data: dict[str, Any], raw_data: dict[str, Any] | None = None
) -> VerifyResult:
"""NekoCode 自定义字段提取(合并 _usage_summary 天卡数据)"""
# 转换余额字符串为数字
balance = user_data.get("balance")
try:
@@ -252,11 +255,10 @@ class NekoCodeArchitecture(ProviderArchitecture):
# 从 prepare_verify_config 获取的 _usage_summary 数据(天卡信息)
extra: dict[str, Any] = {}
usage_summary = data.get("_usage_summary", {})
usage_summary = (raw_data or {}).get("_usage_summary", {})
subscription = usage_summary.get("subscription", {})
if subscription:
# 转换字符串为数字
daily_limit = subscription.get("daily_quota_limit")
daily_remaining = subscription.get("daily_remaining_quota")
try:

View File

@@ -15,7 +15,6 @@ from src.services.provider_ops.actions import (
from src.services.provider_ops.architectures.base import (
ProviderArchitecture,
ProviderConnector,
VerifyResult,
)
from src.services.provider_ops.types import ConnectorAuthType, ProviderActionType
@@ -90,10 +89,17 @@ class NewApiConnector(ProviderConnector):
return {
"type": "object",
"properties": {
"base_url": {
"type": "string",
"title": "站点地址",
"description": "API 基础地址",
},
"api_key": {
"type": "string",
"title": "访问令牌 (API Key)",
"description": "New API 的访问令牌,与 Cookie 二选一",
"x-sensitive": True,
"x-input-type": "password",
},
"user_id": {
"type": "string",
@@ -104,9 +110,47 @@ class NewApiConnector(ProviderConnector):
"type": "string",
"title": "Cookie",
"description": "用于 Cookie 认证,与访问令牌二选一",
"x-sensitive": True,
"x-input-type": "password",
},
},
"required": [],
"x-field-groups": [
{"fields": ["base_url"]},
{
"fields": ["cookie"],
"x-help": "从浏览器开发者工具复制完整 Cookie",
},
{
"layout": "inline",
"fields": ["api_key", "user_id"],
"x-flex": {"api_key": 3, "user_id": 1},
},
],
"x-auth-type": "api_key",
"x-auth-method": "bearer",
"x-validation": [
{
"type": "any_required",
"fields": ["api_key", "cookie"],
"message": "访问令牌和 Cookie 至少需要填写一个",
},
{
"type": "conditional_required",
"if": "api_key",
"unless": "cookie",
"then": ["user_id"],
"message": "使用访问令牌时,用户 ID 不能为空",
},
],
"x-quota-divisor": 500000,
"x-currency": "USD",
"x-field-hooks": {
"cookie": {
"action": "parse_new_api_user_id",
"target": "user_id",
},
},
}
@@ -180,48 +224,3 @@ class NewApiArchitecture(ProviderArchitecture):
headers["Cookie"] = cookie
return headers
def parse_verify_response(
self,
status_code: int,
data: dict[str, Any],
) -> VerifyResult:
"""解析 New API 验证响应"""
if status_code == 401:
return VerifyResult(success=False, message="认证失败:无效的凭据")
if status_code == 403:
return VerifyResult(success=False, message="认证失败:权限不足")
if status_code != 200:
return VerifyResult(success=False, message=f"验证失败HTTP {status_code}")
# New API 响应格式: {"success": true, "data": {...}}
if data.get("success") is True and "data" in data:
user_data = data["data"]
elif data.get("success") is False:
message = data.get("message", "验证失败")
return VerifyResult(success=False, message=message)
else:
user_data = data
return VerifyResult(
success=True,
username=user_data.get("username"),
display_name=user_data.get("display_name") or user_data.get("username"),
email=user_data.get("email"),
quota=user_data.get("quota"),
used_quota=user_data.get("used_quota"),
request_count=user_data.get("request_count"),
extra={
k: v
for k, v in user_data.items()
if k
not in (
"username",
"display_name",
"email",
"quota",
"used_quota",
"request_count",
)
},
)

View File

@@ -1,178 +0,0 @@
"""
One API 架构
针对 One API 风格的中转站优化的预设配置。
"""
from typing import Any
import httpx
from src.services.provider_ops.actions import NewApiBalanceAction, ProviderAction
from src.services.provider_ops.architectures.base import (
ProviderArchitecture,
ProviderConnector,
VerifyResult,
)
from src.services.provider_ops.types import ConnectorAuthType, ProviderActionType
class OneApiConnector(ProviderConnector):
"""
One API 专用连接器
特点:
- 使用 Bearer Token 认证
- 不需要额外的 Header
"""
auth_type = ConnectorAuthType.API_KEY
display_name = "One API Key"
def __init__(self, base_url: str, config: dict[str, Any] | None = None):
super().__init__(base_url, config)
self._api_key: str | None = None
async def connect(self, credentials: dict[str, Any]) -> bool:
"""建立连接"""
api_key = credentials.get("api_key")
if not api_key:
self._set_error("API Key 不能为空")
return False
self._api_key = api_key
self._set_connected()
return True
async def disconnect(self) -> None:
"""断开连接"""
self._api_key = None
self._set_disconnected()
async def is_authenticated(self) -> bool:
"""检查是否已认证"""
return self._api_key is not None
def _apply_auth(self, request: httpx.Request) -> httpx.Request:
"""为请求应用认证信息"""
if self._api_key:
request.headers["Authorization"] = f"Bearer {self._api_key}"
return request
@classmethod
def get_credentials_schema(cls) -> dict[str, Any]:
"""获取凭据配置 schema"""
return {
"type": "object",
"properties": {
"api_key": {
"type": "string",
"title": "访问令牌 (API Key)",
"description": "One API 的访问令牌",
},
},
"required": ["api_key"],
}
class OneApiArchitecture(ProviderArchitecture):
"""
One API 架构预设
针对 One API 风格的中转站优化的预设配置。
特点:
- 使用 Bearer Token 认证
- 验证端点: /api/user/self
- 不需要额外的 Header
"""
architecture_id = "one_api"
display_name = "One API"
description = "One API 风格中转站的预设配置"
supported_connectors: list[type[ProviderConnector]] = [
OneApiConnector,
]
supported_actions: list[type[ProviderAction]] = [
NewApiBalanceAction,
]
default_action_configs: dict[ProviderActionType, dict[str, Any]] = {
ProviderActionType.QUERY_BALANCE: {
"endpoint": "/api/user/self",
"method": "GET",
"response_mapping": {
"total_granted": "data.quota",
"total_used": "data.used_quota",
},
},
}
def get_credentials_schema(self) -> dict[str, Any]:
"""One API 只需要 api_key"""
return OneApiConnector.get_credentials_schema()
def get_verify_endpoint(self) -> str:
"""One API 验证端点"""
return "/api/user/self"
def build_verify_headers(
self,
config: dict[str, Any],
credentials: dict[str, Any],
) -> dict[str, str]:
"""构建 One API 的验证请求 Headers"""
headers: dict[str, str] = {}
api_key = credentials.get("api_key", "")
if api_key:
headers["Authorization"] = f"Bearer {api_key}"
return headers
def parse_verify_response(
self,
status_code: int,
data: dict[str, Any],
) -> VerifyResult:
"""解析 One API 验证响应"""
if status_code == 401:
return VerifyResult(success=False, message="认证失败:无效的凭据")
if status_code == 403:
return VerifyResult(success=False, message="认证失败:权限不足")
if status_code != 200:
return VerifyResult(success=False, message=f"验证失败HTTP {status_code}")
# One API 响应格式: {"success": true, "data": {...}}
if data.get("success") is True and "data" in data:
user_data = data["data"]
elif data.get("success") is False:
message = data.get("message", "验证失败")
return VerifyResult(success=False, message=message)
else:
user_data = data
return VerifyResult(
success=True,
username=user_data.get("username"),
display_name=user_data.get("display_name") or user_data.get("username"),
email=user_data.get("email"),
quota=user_data.get("quota"),
used_quota=user_data.get("used_quota"),
request_count=user_data.get("request_count"),
extra={
k: v
for k, v in user_data.items()
if k
not in (
"username",
"display_name",
"email",
"quota",
"used_quota",
"request_count",
)
},
)

View File

@@ -121,13 +121,52 @@ class YesCodeConnector(ProviderConnector):
return {
"type": "object",
"properties": {
"base_url": {
"type": "string",
"title": "站点地址",
"description": "API 基础地址",
"x-default-value": "https://co.yes.vg",
},
"auth_cookie": {
"type": "string",
"title": "Auth Cookie",
"description": "从浏览器复制的 Cookie包含 yescode_auth 和 yescode_csrf",
"x-sensitive": True,
"x-input-type": "password",
},
},
"required": ["auth_cookie"],
"x-field-groups": [
{"fields": ["base_url"]},
{"fields": ["auth_cookie"]},
],
"x-auth-type": "cookie",
"x-default-base-url": "https://co.yes.vg",
"x-validation": [
{
"type": "required",
"fields": ["auth_cookie"],
"message": "请填写 Auth Cookie",
},
],
"x-quota-divisor": None,
"x-currency": "USD",
"x-balance-extra-format": [
{
"label": "",
"type": "weekly_spent",
"source_limit": "daily_limit",
"source_spent": "daily_spent",
"source_resets_at": "daily_resets_at",
},
{
"label": "",
"type": "weekly_spent",
"source_limit": "weekly_limit",
"source_spent": "weekly_spent",
"source_resets_at": "weekly_resets_at",
},
],
}

View File

@@ -15,7 +15,6 @@ from src.services.provider_ops.architectures import (
GenericApiArchitecture,
NekoCodeArchitecture,
NewApiArchitecture,
OneApiArchitecture,
ProviderArchitecture,
YesCodeArchitecture,
)
@@ -58,7 +57,6 @@ class ArchitectureRegistry:
GenericApiArchitecture,
NekoCodeArchitecture,
NewApiArchitecture,
OneApiArchitecture,
YesCodeArchitecture,
]
@@ -133,8 +131,8 @@ class ArchitectureRegistry:
return list(self._architectures.keys())
def to_dict_list(self) -> list[dict]:
"""获取所有架构的字典表示(用于 API 响应)"""
return [arch.to_dict() for arch in self._architectures.values()]
"""获取所有架构的字典表示(用于 API 响应,隐藏 hidden 架构"""
return [arch.to_dict() for arch in self._architectures.values() if not arch.hidden]
# 全局注册表实例

View File

@@ -0,0 +1,26 @@
"""
Provider Ops 通用工具函数
"""
def extract_cookie_value(cookie_string: str, key: str) -> str:
"""
从 Cookie 字符串中提取指定 key 的值
支持两种输入格式:
1. 完整 Cookie: "key=xxx; other=yyy; ..."
2. 仅值: "MTc2ODc4..."
Args:
cookie_string: Cookie 字符串或直接的值
key: 要提取的 Cookie key
Returns:
对应的值
"""
if f"{key}=" in cookie_string:
for part in cookie_string.split(";"):
part = part.strip()
if part.startswith(f"{key}="):
return part[len(key) + 1 :]
return cookie_string.strip()