refactor: 重构供应商操作基类为抽象类并支持 New API Cookie 认证

主要更改：
- 重构 BalanceAction/CheckinAction 为抽象基类，子类实现具体查询逻辑
- 重构 ProviderArchitecture 认证相关方法为抽象方法
- New API 支持 Cookie 认证（与 api_key+user_id 二选一）
- 添加 NewApiBalanceAction，支持查询余额前自动签到
- 修复 session cookie gob 编码解析逻辑
- 前端认证表单支持 inline 布局、字段联动和 onFieldChange 回调

src/services/provider_ops/actions/__init__.py

@@ -7,12 +7,14 @@ from src.services.provider_ops.actions.balance import BalanceAction
 from src.services.provider_ops.actions.base import ProviderAction
 from src.services.provider_ops.actions.checkin import CheckinAction
 from src.services.provider_ops.actions.cubence_balance import CubenceBalanceAction
+from src.services.provider_ops.actions.new_api_balance import NewApiBalanceAction
 from src.services.provider_ops.actions.yescode_balance import YesCodeBalanceAction
 
 __all__ = [
     "ProviderAction",
     "BalanceAction",
     "CheckinAction",
+    "NewApiBalanceAction",
     "AnyrouterBalanceAction",
     "CubenceBalanceAction",
     "YesCodeBalanceAction",

src/services/provider_ops/actions/anyrouter_balance.py

@@ -2,35 +2,110 @@
 Anyrouter 余额查询操作（含自动签到）
 """
 
-from typing import Any, Dict, Optional, Tuple
+import time
+from typing import Any, Dict, Optional
 
 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
+from src.services.provider_ops.types import ActionResult, ActionStatus, BalanceInfo
 
 
 class AnyrouterBalanceAction(BalanceAction):
     """
-    Anyrouter 专用余额查询
+    Anyrouter 余额查询
 
     特点：
-    - 查询余额前自动触发签到
-    - 签到结果附加到余额信息的 extra 字段
+    - 查询余额前始终自动签到
+    - 签到端点为 /api/user/sign_in
     - Cookie 失效时返回友好的错误提示
+    - quota 单位是 1/500000 美元（与 New API 相同）
     """
 
     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)}",
+            )
+
+    def _parse_balance(self, data: Any) -> BalanceInfo:
+        """解析余额信息"""
+        user_data = data.get("data", {}) if isinstance(data, dict) else {}
+        quota_divisor = self.config.get("quota_divisor", 500000)
+
+        # 注意：Anyrouter 中 quota 是剩余额度（total_available），不是总额度
+        raw_quota = self._to_float(user_data.get("quota"))
+        raw_used = self._to_float(user_data.get("used_quota"))
+
+        total_available = raw_quota / quota_divisor if raw_quota is not None else None
+        total_used = raw_used / quota_divisor if raw_used is not None else None
+
+        return self._create_balance_info(
+            total_available=total_available,
+            total_used=total_used,
+            currency=self.config.get("currency", "USD"),
+        )
+
     def _handle_http_error(
         self, response: httpx.Response, raw_data: Optional[Dict[str, Any]] = None
     ) -> ActionResult:
-        """处理 HTTP 错误响应（Anyrouter 专用）"""
+        """处理 HTTP 错误响应"""
         status_code = response.status_code
 
-        # Anyrouter 使用 Cookie 认证，提供更友好的错误提示
         if status_code == 401:
             return self._make_error_result(
                 ActionStatus.AUTH_FAILED, "Cookie 已失效，请重新配置", raw_response=raw_data
@@ -40,68 +115,42 @@ class AnyrouterBalanceAction(BalanceAction):
                 ActionStatus.AUTH_FAILED, "Cookie 已失效或无权限", raw_response=raw_data
             )
 
-        # 其他错误使用基类处理
         return super()._handle_http_error(response, raw_data)
 
-    async def execute(self, client) -> ActionResult:
-        """执行余额查询（含自动签到）"""
-        # 先尝试签到
-        checkin_success, checkin_message = await self._auto_checkin(client)
-
-        # 执行余额查询
-        result = await super().execute(client)
-
-        # 将签到结果附加到 extra 字段
-        if result.data and hasattr(result.data, "extra"):
-            if result.data.extra is None:
-                result.data.extra = {}
-            result.data.extra["checkin_success"] = checkin_success
-            result.data.extra["checkin_message"] = checkin_message
-
-        return result
-
-    async def _auto_checkin(self, client) -> Tuple[Optional[bool], str]:
+    async def _do_checkin(self, client: httpx.AsyncClient) -> Optional[Dict[str, Any]]:
         """
-        自动签到
+        执行自动签到（始终执行）
 
         Returns:
-            (success, message) 元组:
-            - success: True=签到成功, False=签到失败, None=已签到/跳过
-            - message: 签到消息
+            签到结果字典，包含 success 和 message 字段
         """
         checkin_endpoint = self.config.get("checkin_endpoint", "/api/user/sign_in")
+        site = client.base_url.host or str(client.base_url)
 
         try:
             response = await client.post(checkin_endpoint)
 
-            if response.status_code == 200:
-                try:
-                    data = response.json()
-                    success = data.get("success", False)
-                    message = data.get("message", "")
+            try:
+                data = response.json()
+                success = data.get("success", False)
+                message = data.get("message", "")
 
-                    if success:
-                        logger.debug(f"Anyrouter 自动签到成功: {message}")
-                        return True, message or "签到成功"
+                if success:
+                    logger.debug(f"[{site}] 签到成功: {message}")
+                    return {"success": True, "message": message or "签到成功"}
+                else:
+                    already_indicators = ["已签到", "已签", "今日已", "already"]
+                    is_already = any(ind in message.lower() for ind in already_indicators)
+                    if is_already:
+                        logger.debug(f"[{site}] 今日已签到: {message}")
+                        return {"success": None, "message": message or "今日已签到"}
                     else:
-                        # 检查是否是"已签到"
-                        is_already = (
-                            any(ind in message for ind in ["已签到", "已签", "今日已"])
-                            or "already" in message.lower()
-                        )
-                        if is_already:
-                            logger.debug(f"Anyrouter 今日已签到: {message}")
-                            return None, message or "今日已签到"
-                        else:
-                            logger.debug(f"Anyrouter 签到失败: {message}")
-                            return False, message or "签到失败"
-                except Exception as e:
-                    logger.debug(f"Anyrouter 签到响应解析失败: {e}")
-                    return False, "响应解析失败"
-            else:
-                logger.debug(f"Anyrouter 签到请求失败: HTTP {response.status_code}")
-                return False, f"HTTP {response.status_code}"
+                        logger.debug(f"[{site}] 签到失败: {message}")
+                        return {"success": False, "message": message or "签到失败"}
+            except Exception as e:
+                logger.debug(f"[{site}] 签到响应解析失败: {e}")
+                return {"success": False, "message": "响应解析失败"}
 
         except Exception as e:
-            logger.debug(f"Anyrouter 自动签到异常: {e}")
-            return False, str(e)
+            logger.debug(f"[{site}] 签到异常: {e}")
+            return {"success": False, "message": str(e)}

src/services/provider_ops/actions/balance.py

@@ -1,8 +1,8 @@
 """
-余额查询操作
+余额查询操作抽象基类
 """
 
-import time
+from abc import abstractmethod
 from typing import Any, Dict, Optional
 
 import httpx
@@ -10,7 +10,6 @@ import httpx
 from src.services.provider_ops.actions.base import ProviderAction
 from src.services.provider_ops.types import (
     ActionResult,
-    ActionStatus,
     BalanceInfo,
     ProviderActionType,
 )
@@ -18,9 +17,10 @@ from src.services.provider_ops.types import (
 
 class BalanceAction(ProviderAction):
     """
-    余额查询操作
+    余额查询操作抽象基类
 
-    支持可配置的 endpoint 和响应字段映射。
+    子类必须实现 _do_query_balance() 方法来处理特定平台的余额查询逻辑。
+    子类可选实现 _do_checkin() 方法来在查询余额前执行签到。
     """
 
     action_type = ProviderActionType.QUERY_BALANCE
@@ -29,108 +29,88 @@ class BalanceAction(ProviderAction):
     default_cache_ttl = 86400  # 24 小时
 
     async def execute(self, client: httpx.AsyncClient) -> ActionResult:
-        """执行余额查询"""
-        endpoint = self.config.get("endpoint", "/api/user/balance")
-        method = self.config.get("method", "GET")
-        mapping = self.config.get("response_mapping", {})
+        """
+        执行余额查询（模板方法）
 
-        start_time = time.time()
+        1. 先尝试签到（如果子类实现了 _do_checkin）
+        2. 执行余额查询
 
-        try:
-            response = await client.request(method, endpoint)
-            response_time_ms = int((time.time() - start_time) * 1000)
+        Args:
+            client: 已认证的 HTTP 客户端
 
-            # 尝试解析 JSON
-            try:
-                data = response.json()
-            except Exception:
-                return self._make_error_result(
-                    ActionStatus.PARSE_ERROR,
-                    "响应不是有效的 JSON",
-                )
+        Returns:
+            ActionResult，其中 data 字段为 BalanceInfo
+        """
+        from src.core.logger import logger
 
-            # 检查 HTTP 状态
-            if response.status_code != 200:
-                return self._handle_http_error(response, data)
+        # 先尝试签到
+        checkin_result = await self._do_checkin(client)
 
-            # 检查业务状态码（如果配置了）
-            success_field = self.config.get("success_field")
-            if success_field:
-                is_success = self._extract_field(data, success_field)
-                if is_success is False or is_success == 0:
-                    message = self._extract_field(data, self.config.get("message_field", "message"))
-                    return self._make_error_result(
-                        ActionStatus.UNKNOWN_ERROR,
-                        message or "业务状态码表示失败",
-                        raw_response=data,
-                    )
+        # 执行余额查询
+        result = await self._do_query_balance(client)
 
-            # 解析余额信息
-            balance = self._parse_balance(data, mapping)
+        # 将签到结果附加到 extra 字段
+        if checkin_result and result.data and hasattr(result.data, "extra"):
+            if result.data.extra is None:
+                result.data.extra = {}
+            result.data.extra["checkin_success"] = checkin_result.get("success")
+            result.data.extra["checkin_message"] = checkin_result.get("message", "")
+            logger.debug(f"签到结果已附加到 extra: {checkin_result}")
 
-            return self._make_success_result(
-                data=balance,
-                response_time_ms=response_time_ms,
-                raw_response=data,
-            )
+        return result
 
-        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
+    async def _do_query_balance(self, client: httpx.AsyncClient) -> ActionResult:
+        """
+        执行余额查询（子类必须实现）
 
-    def _parse_balance(self, data: Any, mapping: Dict[str, str]) -> BalanceInfo:
-        """解析余额信息"""
-        # 默认映射（常见字段名）
-        default_mappings = {
-            "total_granted": ["data.total_quota", "data.quota", "total_quota", "quota"],
-            "total_used": ["data.used_quota", "data.used", "used_quota", "used"],
-            "total_available": [
-                "data.balance",
-                "data.remaining",
-                "data.available",
-                "balance",
-                "remaining",
-            ],
-        }
+        Args:
+            client: 已认证的 HTTP 客户端
 
-        # 获取 quota 除数（用于将原始值转换为美元，如 New API 的 1/500000）
-        quota_divisor = self.config.get("quota_divisor", 1)
+        Returns:
+            ActionResult，其中 data 字段为 BalanceInfo
+        """
+        pass
 
-        def get_value(field: str, default_paths: list) -> Optional[float]:
-            # 优先使用用户配置的映射
-            if field in mapping:
-                value = self._extract_field(data, mapping[field])
-                if value is not None:
-                    raw = self._to_float(value)
-                    return raw / quota_divisor if raw is not None else None
+    async def _do_checkin(self, client: httpx.AsyncClient) -> Optional[Dict[str, Any]]:
+        """
+        执行签到（子类可选实现）
 
-            # 尝试默认映射
-            for path in default_paths:
-                value = self._extract_field(data, path)
-                if value is not None:
-                    raw = self._to_float(value)
-                    return raw / quota_divisor if raw is not None else None
+        默认实现返回 None（不签到）。
+        子类可重写此方法实现平台特定的签到逻辑。
 
-            return None
+        Args:
+            client: 已认证的 HTTP 客户端
 
-        total_granted = get_value("total_granted", default_mappings["total_granted"])
-        total_used = get_value("total_used", default_mappings["total_used"])
-        total_available = get_value("total_available", default_mappings["total_available"])
+        Returns:
+            签到结果字典 {"success": bool, "message": str}，或 None 表示不签到
+        """
+        return None
 
+    def _create_balance_info(
+        self,
+        total_granted: Optional[float] = None,
+        total_used: Optional[float] = None,
+        total_available: Optional[float] = None,
+        currency: str = "USD",
+        extra: Optional[Dict[str, Any]] = None,
+    ) -> BalanceInfo:
+        """
+        创建余额信息对象
+
+        辅助方法，用于创建统一格式的 BalanceInfo。
+        如果只有部分数据，会尝试计算缺失的值。
+
+        Args:
+            total_granted: 总额度
+            total_used: 已用额度
+            total_available: 可用余额
+            currency: 货币单位
+            extra: 额外信息
+
+        Returns:
+            BalanceInfo 对象
+        """
         # 如果只有部分数据，尝试计算
         if total_available is None and total_granted is not None and total_used is not None:
             total_available = total_granted - total_used
@@ -139,20 +119,12 @@ class BalanceAction(ProviderAction):
         if total_granted is None and total_used is not None and total_available is not None:
             total_granted = total_used + total_available
 
-        # 提取额外字段
-        extra = {}
-        for key, path in mapping.items():
-            if key not in ["total_granted", "total_used", "total_available", "expires_at"]:
-                value = self._extract_field(data, path)
-                if value is not None:
-                    extra[key] = value
-
         return BalanceInfo(
             total_granted=total_granted,
             total_used=total_used,
             total_available=total_available,
-            currency=self.config.get("currency", "USD"),
-            extra=extra,
+            currency=currency,
+            extra=extra if extra is not None else {},
         )
 
     def _to_float(self, value: Any) -> Optional[float]:
@@ -166,66 +138,15 @@ class BalanceAction(ProviderAction):
 
     @classmethod
     def get_config_schema(cls) -> Dict[str, Any]:
-        """获取操作配置 schema"""
+        """获取操作配置 schema（子类可重写）"""
         return {
             "type": "object",
             "properties": {
-                "endpoint": {
-                    "type": "string",
-                    "title": "API 路径",
-                    "description": "余额查询 API 路径",
-                    "default": "/api/user/balance",
-                },
-                "method": {
-                    "type": "string",
-                    "title": "请求方法",
-                    "enum": ["GET", "POST"],
-                    "default": "GET",
-                },
-                "quota_divisor": {
-                    "type": "number",
-                    "title": "额度除数",
-                    "description": "将原始额度值转换为美元的除数（如 New API 为 500000）",
-                    "default": 1,
-                },
-                "success_field": {
-                    "type": "string",
-                    "title": "成功状态字段",
-                    "description": "响应中表示成功的字段路径（如 success, code）",
-                },
-                "message_field": {
-                    "type": "string",
-                    "title": "消息字段",
-                    "description": "响应中的消息字段路径",
-                    "default": "message",
-                },
-                "response_mapping": {
-                    "type": "object",
-                    "title": "响应字段映射",
-                    "description": "响应字段到余额字段的映射",
-                    "properties": {
-                        "total_granted": {
-                            "type": "string",
-                            "title": "总额度字段",
-                            "description": "响应中总额度的字段路径",
-                        },
-                        "total_used": {
-                            "type": "string",
-                            "title": "已用额度字段",
-                            "description": "响应中已用额度的字段路径",
-                        },
-                        "total_available": {
-                            "type": "string",
-                            "title": "可用余额字段",
-                            "description": "响应中可用余额的字段路径",
-                        },
-                    },
-                },
                 "currency": {
                     "type": "string",
                     "title": "货币单位",
                     "default": "USD",
                 },
             },
-            "required": ["endpoint"],
+            "required": [],
         }

src/services/provider_ops/actions/base.py

@@ -132,6 +132,11 @@ class ProviderAction(ABC):
             return self._make_error_result(
                 ActionStatus.AUTH_FAILED, "无权限访问", raw_response=raw_data
             )
+        elif status_code == 404:
+            # 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(

src/services/provider_ops/actions/checkin.py

@@ -1,16 +1,15 @@
 """
-签到操作
+签到操作抽象基类
 """
 
-import time
-from typing import Any, Dict
+from abc import abstractmethod
+from typing import Any, Dict, Optional
 
 import httpx
 
 from src.services.provider_ops.actions.base import ProviderAction
 from src.services.provider_ops.types import (
     ActionResult,
-    ActionStatus,
     CheckinInfo,
     ProviderActionType,
 )
@@ -18,9 +17,9 @@ from src.services.provider_ops.types import (
 
 class CheckinAction(ProviderAction):
     """
-    签到操作
+    签到操作抽象基类
 
-    支持可配置的 endpoint 和响应字段映射。
+    子类必须实现 execute() 方法来处理特定平台的签到逻辑。
     """
 
     action_type = ProviderActionType.CHECKIN
@@ -28,209 +27,54 @@ class CheckinAction(ProviderAction):
     description = "每日签到领取额度"
     default_cache_ttl = 3600  # 签到结果缓存 1 小时
 
+    @abstractmethod
     async def execute(self, client: httpx.AsyncClient) -> ActionResult:
-        """执行签到"""
-        endpoint = self.config.get("endpoint", "/api/user/checkin")
-        method = self.config.get("method", "POST")
-
-        start_time = time.time()
-
-        try:
-            # 构建请求
-            request_body = self.config.get("request_body", {})
-
-            if method == "POST":
-                response = await client.post(endpoint, json=request_body or None)
-            else:
-                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)
-
-            # 解析签到结果
-            checkin_info, status, message = self._parse_checkin_result(data)
-
-            if status == ActionStatus.SUCCESS:
-                return self._make_success_result(
-                    data=checkin_info,
-                    message=message,
-                    response_time_ms=response_time_ms,
-                    raw_response=data,
-                )
-            else:
-                return self._make_error_result(
-                    status,
-                    message,
-                    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_checkin_result(
-        self, data: Any
-    ) -> tuple[CheckinInfo, ActionStatus, str | None]:
         """
-        解析签到结果
+        执行签到
+
+        子类必须实现此方法。
+
+        Args:
+            client: 已认证的 HTTP 客户端
 
         Returns:
-            (CheckinInfo, 状态, 消息)
+            ActionResult，其中 data 字段为 CheckinInfo
         """
-        mapping = self.config.get("response_mapping", {})
+        pass
 
-        # 检查成功状态
-        success_field = self.config.get("success_field", "success")
-        is_success = self._extract_field(data, success_field)
+    def _create_checkin_info(
+        self,
+        reward: Optional[float] = None,
+        streak_days: Optional[int] = None,
+        message: Optional[str] = None,
+        extra: Optional[Dict[str, Any]] = None,
+    ) -> CheckinInfo:
+        """
+        创建签到信息对象
 
-        # 获取消息
-        message_field = self.config.get("message_field", "message")
-        message = self._extract_field(data, message_field)
-        if message is not None:
-            message = str(message)
+        辅助方法，用于创建统一格式的 CheckinInfo。
 
-        # 检查是否已签到
-        already_checked_indicators = self.config.get(
-            "already_checked_indicators", ["already", "已签到", "今日已签", "重复签到"]
-        )
-        if message:
-            for indicator in already_checked_indicators:
-                if indicator.lower() in message.lower():
-                    return (
-                        CheckinInfo(message=message),
-                        ActionStatus.ALREADY_DONE,
-                        message,
-                    )
+        Args:
+            reward: 签到奖励额度
+            streak_days: 连续签到天数
+            message: 签到消息
+            extra: 额外信息
 
-        # 判断是否成功
-        if is_success is False or is_success == 0:
-            return (
-                CheckinInfo(message=message),
-                ActionStatus.UNKNOWN_ERROR,
-                message or "签到失败",
-            )
-
-        # 解析签到信息
-        reward = None
-        reward_field = mapping.get("reward") or self.config.get("reward_field")
-        if reward_field:
-            reward_value = self._extract_field(data, reward_field)
-            if reward_value is not None:
-                try:
-                    reward = float(reward_value)
-                except (TypeError, ValueError):
-                    pass
-
-        streak_days = None
-        streak_field = mapping.get("streak_days") or self.config.get("streak_field")
-        if streak_field:
-            streak_value = self._extract_field(data, streak_field)
-            if streak_value is not None:
-                try:
-                    streak_days = int(streak_value)
-                except (TypeError, ValueError):
-                    pass
-
-        # 提取额外字段
-        extra = {}
-        for key, path in mapping.items():
-            if key not in ["reward", "streak_days", "message"]:
-                value = self._extract_field(data, path)
-                if value is not None:
-                    extra[key] = value
-
-        checkin_info = CheckinInfo(
+        Returns:
+            CheckinInfo 对象
+        """
+        return CheckinInfo(
             reward=reward,
             streak_days=streak_days,
             message=message,
-            extra=extra,
+            extra=extra if extra is not None else {},
         )
 
-        return (checkin_info, ActionStatus.SUCCESS, message or "签到成功")
-
     @classmethod
     def get_config_schema(cls) -> Dict[str, Any]:
-        """获取操作配置 schema"""
+        """获取操作配置 schema（子类可重写）"""
         return {
             "type": "object",
-            "properties": {
-                "endpoint": {
-                    "type": "string",
-                    "title": "API 路径",
-                    "description": "签到 API 路径",
-                    "default": "/api/user/checkin",
-                },
-                "method": {
-                    "type": "string",
-                    "title": "请求方法",
-                    "enum": ["GET", "POST"],
-                    "default": "POST",
-                },
-                "request_body": {
-                    "type": "object",
-                    "title": "请求体",
-                    "description": "签到请求的 JSON 体（可选）",
-                },
-                "success_field": {
-                    "type": "string",
-                    "title": "成功状态字段",
-                    "description": "响应中表示成功的字段路径",
-                    "default": "success",
-                },
-                "message_field": {
-                    "type": "string",
-                    "title": "消息字段",
-                    "description": "响应中的消息字段路径",
-                    "default": "message",
-                },
-                "reward_field": {
-                    "type": "string",
-                    "title": "奖励字段",
-                    "description": "响应中奖励额度的字段路径",
-                },
-                "streak_field": {
-                    "type": "string",
-                    "title": "连续签到天数字段",
-                    "description": "响应中连续签到天数的字段路径",
-                },
-                "already_checked_indicators": {
-                    "type": "array",
-                    "title": "已签到标识",
-                    "description": "消息中表示已签到的关键词",
-                    "items": {"type": "string"},
-                    "default": ["already", "已签到", "今日已签", "重复签到"],
-                },
-                "response_mapping": {
-                    "type": "object",
-                    "title": "响应字段映射",
-                    "description": "响应字段到签到信息的映射",
-                },
-            },
-            "required": ["endpoint"],
+            "properties": {},
+            "required": [],
         }

src/services/provider_ops/actions/cubence_balance.py

@@ -2,6 +2,7 @@
 Cubence 余额查询操作
 """
 
+import time
 from typing import Any, Dict, Optional
 
 import httpx
@@ -23,6 +24,66 @@ 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: Optional[Dict[str, Any]] = None
     ) -> ActionResult:
@@ -42,8 +103,8 @@ class CubenceBalanceAction(BalanceAction):
         # 其他错误使用基类处理
         return super()._handle_http_error(response, raw_data)
 
-    def _parse_balance(self, data: Any, mapping: Dict[str, str]) -> BalanceInfo:
-        """解析 Cubence 余额信息（覆盖基类方法）"""
+    def _parse_balance(self, data: Any) -> BalanceInfo:
+        """解析 Cubence 余额信息"""
         # Cubence 响应格式：data.balance 和 data.subscription_limits
         response_data = data.get("data", {}) if isinstance(data, dict) else {}
         balance_data = response_data.get("balance", {})
@@ -86,9 +147,7 @@ class CubenceBalanceAction(BalanceAction):
         if charity_balance is not None:
             extra["charity_balance"] = charity_balance
 
-        return BalanceInfo(
-            total_granted=None,  # Cubence 不提供总额度
-            total_used=None,
+        return self._create_balance_info(
             total_available=total_available,
             currency=self.config.get("currency", "USD"),
             extra=extra if extra else None,

src/services/provider_ops/actions/new_api_balance.py

@@ -0,0 +1,217 @@
+"""
+New API 余额查询操作
+"""
+
+import time
+from typing import Any, Dict, Optional
+
+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
+
+
+class NewApiBalanceAction(BalanceAction):
+    """
+    New API 风格的余额查询
+
+    特点：
+    - 使用 /api/user/self 端点
+    - quota 单位是 1/500000 美元
+    - 支持查询前自动签到（通过基类的模板方法）
+    """
+
+    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,
+    ) -> BalanceInfo:
+        """解析 New API 余额信息"""
+        # New API 响应格式: {"success": true, "data": {...}}
+        user_data = data.get("data", {}) if isinstance(data, dict) else {}
+
+        # 获取 quota 除数（默认 500000，New API 的标准）
+        quota_divisor = self.config.get("quota_divisor", 500000)
+
+        # 提取原始值
+        # 注意：New API 中 quota 是剩余额度（total_available），不是总额度
+        raw_quota = self._to_float(user_data.get("quota"))
+        raw_used = self._to_float(user_data.get("used_quota"))
+
+        # 转换为美元
+        total_available = raw_quota / quota_divisor if raw_quota is not None else None
+        total_used = raw_used / quota_divisor if raw_used is not None else None
+
+        return self._create_balance_info(
+            total_available=total_available,
+            total_used=total_used,
+            currency=self.config.get("currency", "USD"),
+        )
+
+    async def _do_checkin(self, client: httpx.AsyncClient) -> Optional[Dict[str, Any]]:
+        """
+        执行签到（静默，不抛出异常）
+
+        New API 签到需要 Cookie 认证。没有 Cookie 时跳过签到。
+        失败时仅记录日志，不影响余额查询。
+
+        Returns:
+            签到结果字典，包含 success 和 message 字段；
+            如果功能未开放或认证失败，返回 None
+        """
+        site = client.base_url.host or str(client.base_url)
+        checkin_endpoint = self.config.get("checkin_endpoint", "/api/user/checkin")
+
+        # 检查 client 是否配置了 Cookie（通过检查默认 headers）
+        # httpx.AsyncClient 的 headers 是 httpx.Headers 类型
+        has_cookie = "cookie" in {k.lower() for k in client.headers.keys()}
+        if not has_cookie:
+            logger.debug(f"[{site}] 未配置 Cookie，跳过签到")
+            return None
+
+        try:
+            response = await client.post(checkin_endpoint)
+
+            # 404 表示签到功能未开放
+            if response.status_code == 404:
+                logger.debug(f"[{site}] 签到功能未开放")
+                return None
+
+            # 401/403 表示签到需要额外认证（如 Cookie），当前配置不支持
+            if response.status_code in (401, 403):
+                logger.debug(f"[{site}] 签到需要额外认证")
+                return None
+
+            try:
+                data = response.json()
+                message = data.get("message", "")
+                success = data.get("success", False)
+
+                if success:
+                    logger.debug(f"[{site}] 签到成功: {message}")
+                    return {"success": True, "message": message or "签到成功"}
+                else:
+                    # 检查是否是"已签到"的情况
+                    already_indicators = ["already", "已签到", "今日已签", "重复签到"]
+                    is_already = any(ind in message.lower() for ind in already_indicators)
+                    if is_already:
+                        logger.debug(f"[{site}] 今日已签到: {message}")
+                        return {"success": None, "message": message or "今日已签到"}
+
+                    # 检查是否是认证失败（未登录、无权限等）- 静默跳过
+                    auth_fail_indicators = [
+                        "未登录", "请登录", "login", "unauthorized", "无权限", "权限不足",
+                        "turnstile", "captcha", "验证码",  # 需要人机验证
+                    ]
+                    is_auth_fail = any(ind in message.lower() for ind in auth_fail_indicators)
+                    if is_auth_fail:
+                        logger.debug(f"[{site}] 签到需要额外认证，跳过")
+                        return None
+
+                    # 其他失败情况
+                    logger.debug(f"[{site}] 签到失败: {message}")
+                    return {"success": False, "message": message or "签到失败"}
+            except Exception as e:
+                logger.debug(f"[{site}] 签到响应解析失败: {e}")
+                return {"success": False, "message": "响应解析失败"}
+
+        except Exception as e:
+            # 签到失败不影响余额查询
+            logger.debug(f"[{site}] 签到请求失败（不影响余额查询）: {e}")
+            return None
+
+    @classmethod
+    def get_config_schema(cls) -> Dict[str, Any]:
+        """获取操作配置 schema"""
+        return {
+            "type": "object",
+            "properties": {
+                "endpoint": {
+                    "type": "string",
+                    "title": "API 路径",
+                    "description": "余额查询 API 路径",
+                    "default": "/api/user/self",
+                },
+                "method": {
+                    "type": "string",
+                    "title": "请求方法",
+                    "enum": ["GET", "POST"],
+                    "default": "GET",
+                },
+                "quota_divisor": {
+                    "type": "number",
+                    "title": "额度除数",
+                    "description": "将原始额度值转换为美元的除数",
+                    "default": 500000,
+                },
+                "currency": {
+                    "type": "string",
+                    "title": "货币单位",
+                    "default": "USD",
+                },
+            },
+            "required": [],
+        }

src/services/provider_ops/actions/yescode_balance.py

@@ -168,8 +168,8 @@ class YesCodeBalanceAction(BalanceAction):
     display_name = "查询余额（含每周限额）"
     description = "查询账户余额和每周限额信息"
 
-    async def execute(self, client: httpx.AsyncClient) -> ActionResult:
-        """执行余额查询（复用 client 调用两个接口获取完整数据）"""
+    async def _do_query_balance(self, client: httpx.AsyncClient) -> ActionResult:
+        """执行余额查询（实现抽象方法，复用 client 调用两个接口获取完整数据）"""
         import time
 
         start_time = time.time()
@@ -198,7 +198,7 @@ class YesCodeBalanceAction(BalanceAction):
                 total_used=None,
                 total_available=total_available,
                 currency=self.config.get("currency", "USD"),
-                extra=extra if extra else None,
+                extra=extra if extra else {},
             )
 
             return self._make_success_result(

src/services/provider_ops/architectures/anyrouter.py

@@ -12,7 +12,10 @@ import httpx
 
 from src.core.logger import logger
 from src.utils.ssl_utils import get_ssl_context
-from src.services.provider_ops.actions import AnyrouterBalanceAction, ProviderAction
+from src.services.provider_ops.actions import (
+    AnyrouterBalanceAction,
+    ProviderAction,
+)
 from src.services.provider_ops.architectures.base import (
     ProviderArchitecture,
     ProviderConnector,
@@ -123,8 +126,8 @@ def _parse_session_user_id(cookie_input: str) -> Tuple[Optional[str], Optional[s
     base64(timestamp|gob_base64|signature)
 
     gob 数据中包含:
-    - id: 内部数字 ID
-    - username: 用户名 (如 linuxdo_129083)
+    - id: 内部数字 ID (gob 编码的整数)
+    - username: 用户名
     - role, status, group 等
 
     Args:
@@ -157,24 +160,53 @@ def _parse_session_user_id(cookie_input: str) -> Tuple[Optional[str], Optional[s
 
         gob_data = base64.urlsafe_b64decode(gob_b64)
 
-        # 4. 从 gob 数据中提取用户名
-        gob_text = gob_data.decode("utf-8", errors="ignore")
+        # 4. 从 gob 数据中解析 id 字段
+        # 查找 "\x02id\x03int" 模式，后面跟着 gob 编码的整数
+        id_pattern = b"\x02id\x03int"
+        id_idx = gob_data.find(id_pattern)
+        user_id = None
+        if id_idx != -1:
+            # 跳过 "\x02id\x03int" (7字节) 和类型标记 (2字节)
+            value_start = id_idx + 7 + 2
+            if value_start < len(gob_data):
+                # 读取第一个字节，检查是否是 00（正数标记）
+                first_byte = gob_data[value_start]
+                if first_byte == 0:
+                    # 下一个字节是长度标记
+                    marker = gob_data[value_start + 1]
+                    if marker >= 0x80:
+                        # 负的表示长度: 256 - marker = 字节数
+                        length = 256 - marker
+                        if value_start + 2 + length <= len(gob_data):
+                            # 读取 length 字节，大端序转整数
+                            val = int.from_bytes(
+                                gob_data[value_start + 2 : value_start + 2 + length],
+                                "big",
+                            )
+                            # gob zigzag 解码：正整数用 2*n 编码
+                            user_id = str(val >> 1)
 
-        # 查找 linuxdo_xxx 模式 (LinuxDo OAuth)
-        linuxdo_match = re.search(r"linuxdo_(\d+)", gob_text)
-        if linuxdo_match:
-            user_id = linuxdo_match.group(1)
-            username = linuxdo_match.group(0)
-            return user_id, username
+        # 5. 从 gob 数据中提取用户名
+        username = None
 
-        # 查找其他 OAuth 格式 (github_xxx, google_xxx 等)
-        oauth_match = re.search(r"(github|google|discord|twitter)_(\d+)", gob_text, re.IGNORECASE)
-        if oauth_match:
-            user_id = oauth_match.group(2)
-            username = oauth_match.group(0)
-            return user_id, username
+        # 查找 username 字段后的值
+        # 格式: \x08username\x06string\x0c\x10\x00\x0elinuxdo_129083
+        # 其中 \x0e (14) 是用户名的长度
+        username_pattern = b"\x08username\x06string"
+        username_idx = gob_data.find(username_pattern)
+        if username_idx != -1:
+            # 跳过模式本身 (17字节) 和 \x0c\x10\x00 (3字节)
+            # 第 4 个字节是长度
+            length_pos = username_idx + len(username_pattern) + 3
+            if length_pos < len(gob_data):
+                length_byte = gob_data[length_pos]
+                value_start = length_pos + 1
+                if length_byte < 128 and value_start + length_byte <= len(gob_data):
+                    username = gob_data[value_start : value_start + length_byte].decode(
+                        "utf-8", errors="ignore"
+                    )
 
-        return None, None
+        return user_id, username
     except Exception as e:
         logger.debug(f"解析 Anyrouter session cookie 失败: {e}")
         return None, None
@@ -340,9 +372,7 @@ class AnyrouterArchitecture(ProviderArchitecture):
         AnyrouterConnector,
     ]
 
-    supported_actions: List[Type[ProviderAction]] = [
-        AnyrouterBalanceAction,
-    ]
+    supported_actions: List[Type[ProviderAction]] = [AnyrouterBalanceAction]
 
     default_action_configs: Dict[ProviderActionType, Dict[str, Any]] = {
         ProviderActionType.QUERY_BALANCE: {
@@ -350,11 +380,6 @@ class AnyrouterArchitecture(ProviderArchitecture):
             "method": "GET",
             "quota_divisor": 500000,  # 与 New API 相同
             "checkin_endpoint": "/api/user/sign_in",  # 自动签到端点
-            "response_mapping": {
-                "total_granted": "data.quota",
-                "total_used": "data.used_quota",
-                "total_available": "data.quota",
-            },
         },
     }
 

src/services/provider_ops/architectures/base.py

@@ -213,7 +213,7 @@ class VerifyResult:
 
 class ProviderArchitecture(ABC):
     """
-    提供商架构基类
+    提供商架构抽象基类
 
     架构 = Connector（鉴权方式） + Actions（支持的操作）
 
@@ -226,7 +226,8 @@ class ProviderArchitecture(ABC):
     2. 继承 ProviderArchitecture 和 ProviderConnector
     3. 定义类属性：architecture_id, display_name, description
     4. 实现连接器子类和架构类
-    5. 重写认证相关方法：
+    5. 实现认证相关的抽象方法：
+       - get_credentials_schema(): 返回凭据字段定义
        - get_verify_endpoint(): 返回验证端点
        - build_verify_headers(): 构建验证请求 headers
        - parse_verify_response(): 解析验证响应
@@ -256,13 +257,14 @@ class ProviderArchitecture(ABC):
         """
         self.config = config or {}
 
-    # ==================== 认证验证相关方法 ====================
+    # ==================== 认证验证相关方法（子类必须实现） ====================
 
+    @abstractmethod
     def get_credentials_schema(self) -> Dict[str, Any]:
         """
         获取凭据字段定义（JSON Schema 格式）
 
-        子类应重写此方法定义需要的凭据字段。
+        子类必须实现此方法定义需要的凭据字段。
         这个 schema 可用于：
         1. 前端表单生成（如果需要动态渲染）
         2. 凭据验证
@@ -280,37 +282,65 @@ class ProviderArchitecture(ABC):
                         "title": "API Key",
                         "description": "访问令牌",
                     },
-                    "user_id": {
-                        "type": "string",
-                        "title": "用户 ID",
-                        "description": "New API 用户 ID",
-                    },
                 },
-                "required": ["api_key", "user_id"],
+                "required": ["api_key"],
             }
         """
-        return {
-            "type": "object",
-            "properties": {
-                "api_key": {
-                    "type": "string",
-                    "title": "API Key",
-                    "description": "访问令牌",
-                },
-            },
-            "required": ["api_key"],
-        }
+        pass
 
+    @abstractmethod
     def get_verify_endpoint(self) -> str:
         """
         获取认证验证端点
 
-        子类可重写以自定义验证端点。
+        子类必须实现此方法返回验证端点。
 
         Returns:
-            验证端点路径（如 /api/user/self）
+            验证端点路径（如 /api/user/self, /api/v1/auth/profile）
         """
-        return "/api/user/self"
+        pass
+
+    @abstractmethod
+    def build_verify_headers(
+        self,
+        config: Dict[str, Any],
+        credentials: Dict[str, Any],
+    ) -> Dict[str, str]:
+        """
+        构建认证验证请求的 Headers
+
+        子类必须实现此方法构建认证 Headers。
+
+        Args:
+            config: 连接器配置（可能包含 prepare_verify_config 返回的额外配置）
+            credentials: 凭据信息
+
+        Returns:
+            Headers 字典
+        """
+        pass
+
+    @abstractmethod
+    def parse_verify_response(
+        self,
+        status_code: int,
+        data: Dict[str, Any],
+    ) -> VerifyResult:
+        """
+        解析认证验证响应
+
+        子类必须实现此方法解析响应。
+
+        Args:
+            status_code: HTTP 状态码
+            data: 响应 JSON 数据
+
+        Returns:
+            验证结果
+        """
+        pass
+
+    # ==================== 可选的钩子方法 ====================
 
     async def prepare_verify_config(
         self,
@@ -319,7 +349,7 @@ class ProviderArchitecture(ABC):
         credentials: Dict[str, Any],
     ) -> Dict[str, Any]:
         """
-        验证前的异步预处理
+        验证前的异步预处理（可选）
 
         子类可重写以执行异步操作（如获取动态 Cookie）。
         返回的配置会传递给 build_verify_headers。
@@ -334,95 +364,6 @@ class ProviderArchitecture(ABC):
         """
         return {}
 
-    def build_verify_headers(
-        self,
-        config: Dict[str, Any],
-        credentials: Dict[str, Any],
-    ) -> Dict[str, str]:
-        """
-        构建认证验证请求的 Headers
-
-        子类可重写以添加特定的 Headers。
-
-        Args:
-            config: 连接器配置
-            credentials: 凭据信息
-
-        Returns:
-            Headers 字典
-        """
-        headers: Dict[str, str] = {}
-
-        # 处理 API Key 认证
-        api_key = credentials.get("api_key", "")
-        if api_key:
-            auth_method = config.get("auth_method", "bearer")
-            if auth_method == "bearer":
-                headers["Authorization"] = f"Bearer {api_key}"
-            elif auth_method == "header":
-                header_name = config.get("header_name", "X-API-Key")
-                headers[header_name] = api_key
-
-        return headers
-
-    def parse_verify_response(
-        self,
-        status_code: int,
-        data: Dict[str, Any],
-    ) -> VerifyResult:
-        """
-        解析认证验证响应
-
-        子类可重写以处理特定的响应格式。
-
-        Args:
-            status_code: HTTP 状态码
-            data: 响应 JSON 数据
-
-        Returns:
-            验证结果
-        """
-        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}")
-
-        # 尝试解析通用响应格式
-        # 格式1: {"success": true, "data": {...}}
-        # 格式2: 直接返回用户数据 {...}
-        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",
-                )
-            },
-        )
-
     # ==================== 连接器和操作相关方法 ====================
 
     def get_connector(

src/services/provider_ops/architectures/generic_api.py

@@ -43,8 +43,15 @@ from typing import Any, Dict, List, Optional, Type
 
 import httpx
 
-from src.services.provider_ops.actions import BalanceAction, CheckinAction, ProviderAction
-from src.services.provider_ops.architectures.base import ProviderArchitecture, ProviderConnector
+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
 
 
@@ -133,10 +140,7 @@ class GenericApiArchitecture(ProviderArchitecture):
         GenericApiKeyConnector,
     ]
 
-    supported_actions: List[Type[ProviderAction]] = [
-        BalanceAction,
-        CheckinAction,
-    ]
+    supported_actions: List[Type[ProviderAction]] = [NewApiBalanceAction]
 
     # 默认操作配置（可被用户配置覆盖）
     default_action_configs: Dict[ProviderActionType, Dict[str, Any]] = {
@@ -153,3 +157,71 @@ class GenericApiArchitecture(ProviderArchitecture):
     def get_credentials_schema(self) -> Dict[str, Any]:
         """通用架构只需要 api_key"""
         return GenericApiKeyConnector.get_credentials_schema()
+
+    def get_verify_endpoint(self) -> str:
+        """通用架构验证端点"""
+        return "/api/user/self"
+
+    def build_verify_headers(
+        self,
+        config: Dict[str, Any],
+        credentials: Dict[str, Any],
+    ) -> Dict[str, str]:
+        """构建通用 API 的验证请求 Headers"""
+        headers: Dict[str, str] = {}
+
+        api_key = credentials.get("api_key", "")
+        if api_key:
+            auth_method = config.get("auth_method", "bearer")
+            if auth_method == "bearer":
+                headers["Authorization"] = f"Bearer {api_key}"
+            elif auth_method == "header":
+                header_name = config.get("header_name", "X-API-Key")
+                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",
+                )
+            },
+        )

src/services/provider_ops/architectures/new_api.py

@@ -8,8 +8,15 @@ from typing import Any, Dict, List, Optional, Type
 
 import httpx
 
-from src.services.provider_ops.actions import BalanceAction, CheckinAction, ProviderAction
-from src.services.provider_ops.architectures.base import ProviderArchitecture, ProviderConnector
+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
 
 
@@ -29,21 +36,27 @@ class NewApiConnector(ProviderConnector):
         super().__init__(base_url, config)
         self._api_key: Optional[str] = None
         self._user_id: Optional[str] = None
+        self._cookie: Optional[str] = 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 不能为空")
+        cookie = credentials.get("cookie")
+        user_id = credentials.get("user_id")
+
+        # api_key 和 cookie 至少需要一个
+        if not api_key and not cookie:
+            self._set_error("访问令牌和 Cookie 至少需要填写一个")
             return False
 
-        user_id = credentials.get("user_id")
-        if not user_id:
-            self._set_error("用户 ID 不能为空")
+        # 使用 api_key 时必须提供 user_id，使用 cookie 时 user_id 可选
+        if api_key and not cookie and not user_id:
+            self._set_error("使用访问令牌时，用户 ID 不能为空")
             return False
 
         self._api_key = api_key
-        self._user_id = str(user_id)
+        self._user_id = str(user_id) if user_id else None
+        self._cookie = cookie
         self._set_connected()
         return True
 
@@ -51,10 +64,14 @@ class NewApiConnector(ProviderConnector):
         """断开连接"""
         self._api_key = None
         self._user_id = None
+        self._cookie = None
         self._set_disconnected()
 
     async def is_authenticated(self) -> bool:
         """检查是否已认证"""
+        # 有 cookie 就行，或者有 api_key + user_id
+        if self._cookie:
+            return True
         return self._api_key is not None and self._user_id is not None
 
     def _apply_auth(self, request: httpx.Request) -> httpx.Request:
@@ -63,6 +80,8 @@ class NewApiConnector(ProviderConnector):
             request.headers["Authorization"] = f"Bearer {self._api_key}"
         if self._user_id:
             request.headers["New-Api-User"] = self._user_id
+        if self._cookie:
+            request.headers["Cookie"] = self._cookie
         return request
 
     @classmethod
@@ -74,15 +93,20 @@ class NewApiConnector(ProviderConnector):
                 "api_key": {
                     "type": "string",
                     "title": "访问令牌 (API Key)",
-                    "description": "New API 的访问令牌",
+                    "description": "New API 的访问令牌，与 Cookie 二选一",
                 },
                 "user_id": {
                     "type": "string",
                     "title": "用户 ID",
-                    "description": "New API 用户 ID，用于 New-Api-User Header",
+                    "description": "使用访问令牌时必填，使用 Cookie 时可选",
+                },
+                "cookie": {
+                    "type": "string",
+                    "title": "Cookie",
+                    "description": "用于 Cookie 认证，与访问令牌二选一",
                 },
             },
-            "required": ["api_key", "user_id"],
+            "required": [],
         }
 
 
@@ -108,8 +132,7 @@ class NewApiArchitecture(ProviderArchitecture):
     ]
 
     supported_actions: List[Type[ProviderAction]] = [
-        BalanceAction,
-        CheckinAction,
+        NewApiBalanceAction,
     ]
 
     default_action_configs: Dict[ProviderActionType, Dict[str, Any]] = {
@@ -117,17 +140,7 @@ class NewApiArchitecture(ProviderArchitecture):
             "endpoint": "/api/user/self",
             "method": "GET",
             "quota_divisor": 500000,  # New API 的 quota 单位是 1/500000 美元
-            "response_mapping": {
-                "total_granted": "data.quota",
-                "total_used": "data.used_quota",
-                "total_available": "data.quota",  # New API 通常只返回剩余额度
-            },
-        },
-        ProviderActionType.CHECKIN: {
-            "endpoint": "/api/user/checkin",
-            "method": "POST",
-            "success_field": "success",
-            "message_field": "message",
+            "checkin_endpoint": "/api/user/checkin",  # 签到端点
         },
     }
 
@@ -135,6 +148,10 @@ class NewApiArchitecture(ProviderArchitecture):
         """New API 需要 api_key 和 user_id"""
         return NewApiConnector.get_credentials_schema()
 
+    def get_verify_endpoint(self) -> str:
+        """New API 验证端点"""
+        return "/api/user/self"
+
     def build_verify_headers(
         self,
         config: Dict[str, Any],
@@ -145,11 +162,66 @@ class NewApiArchitecture(ProviderArchitecture):
 
         New API 特有：需要 New-Api-User Header 传递用户 ID
         """
-        headers = super().build_verify_headers(config, credentials)
+        headers: Dict[str, str] = {}
+
+        # Bearer Token 认证
+        api_key = credentials.get("api_key", "")
+        if api_key:
+            headers["Authorization"] = f"Bearer {api_key}"
 
         # New API 特有的 header
         user_id = credentials.get("user_id", "")
         if user_id:
             headers["New-Api-User"] = str(user_id)
 
+        # 可选的 Cookie
+        cookie = credentials.get("cookie", "")
+        if cookie:
+            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",
+                )
+            },
+        )

src/services/provider_ops/architectures/one_api.py

@@ -8,8 +8,12 @@ from typing import Any, Dict, List, Optional, Type
 
 import httpx
 
-from src.services.provider_ops.actions import BalanceAction, ProviderAction
-from src.services.provider_ops.architectures.base import ProviderArchitecture, ProviderConnector
+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
 
 
@@ -92,7 +96,7 @@ class OneApiArchitecture(ProviderArchitecture):
     ]
 
     supported_actions: List[Type[ProviderAction]] = [
-        BalanceAction,
+        NewApiBalanceAction,
     ]
 
     default_action_configs: Dict[ProviderActionType, Dict[str, Any]] = {
@@ -109,3 +113,66 @@ class OneApiArchitecture(ProviderArchitecture):
     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",
+                )
+            },
+        )

src/services/provider_ops/registry.py

@@ -28,6 +28,7 @@ class ArchitectureRegistry:
 
     _instance: Optional["ArchitectureRegistry"] = None
     _lock = threading.Lock()
+    _initialized: bool = False
 
     def __new__(cls) -> "ArchitectureRegistry":
         if cls._instance is None:
@@ -49,7 +50,7 @@ class ArchitectureRegistry:
 
     def _register_builtin_architectures(self) -> None:
         """注册内置架构"""
-        builtin = [
+        builtin: List[Type[ProviderArchitecture]] = [
             AnyrouterArchitecture,
             CubenceArchitecture,
             GenericApiArchitecture,

src/services/provider_ops/service.py

@@ -5,7 +5,6 @@ Provider 操作服务
 """
 
 import asyncio
-import json
 from dataclasses import asdict
 from datetime import datetime, timezone
 from typing import Any, Dict, List, Optional
@@ -45,7 +44,7 @@ class ProviderOpsService:
     """
 
     # 凭据中需要加密的字段
-    SENSITIVE_FIELDS = {"api_key", "password", "session_token", "session_cookie", "token_cookie", "auth_cookie", "cookie_string", "cookies"}
+    SENSITIVE_FIELDS = {"api_key", "password", "session_token", "session_cookie", "token_cookie", "auth_cookie", "cookie_string", "cookie"}
 
     def __init__(self, db: Session):
         self.db = db
@@ -411,6 +410,7 @@ class ProviderOpsService:
         }
 
         await CacheService.set(cache_key, cache_data, BALANCE_CACHE_TTL)
+        logger.debug(f"余额缓存已写入: provider_id={provider_id}, extra={data.get('extra') if data else None}")
 
     async def _cache_balance_from_verify(
         self,
@@ -539,9 +539,6 @@ class ProviderOpsService:
                 else:
                     logger.warning(f"跳过空值字段 {key}")
                     encrypted[key] = value
-            elif key == "cookies" and isinstance(value, dict):
-                # cookies 整体加密
-                encrypted[key] = self.crypto.encrypt(json.dumps(value))
             else:
                 encrypted[key] = value
         return encrypted
@@ -556,12 +553,6 @@ class ProviderOpsService:
                 except Exception as e:
                     logger.warning(f"解密字段 {key} 失败: {e}")
                     decrypted[key] = value  # 解密失败则保持原值
-            elif key == "cookies" and isinstance(value, str):
-                try:
-                    decrypted[key] = json.loads(self.crypto.decrypt(value))
-                except Exception as e:
-                    logger.warning(f"解密 cookies 失败: {e}")
-                    decrypted[key] = value
             else:
                 decrypted[key] = value
         return decrypted
@@ -616,7 +607,7 @@ class ProviderOpsService:
         if saved_config:
             saved_credentials = self._decrypt_credentials(saved_config.connector_credentials)
             sensitive_fields = [
-                "api_key", "password", "session_token", "cookie_string", "cookies",
+                "api_key", "password", "session_token", "cookie_string", "cookie",
                 "token_cookie", "auth_cookie", "session_cookie",  # Cookie 认证字段
             ]