feat: 添加 Provider Ops 扩展操作系统，支持余额监控

主要更改：
- 新增 Provider Ops 服务框架，支持通过架构配置执行余额查询等扩展操作
- 后端：添加 provider_ops 服务层和 API 路由
- 前端：添加 ProviderAuthDialog 组件配置认证信息
- 前端：添加 providerOps API 和认证模板系统

UI/组件优化：
- Input 组件：新增 masked 属性，使用 CSS 遮蔽敏感信息，避免触发密码管理器
- Pagination 组件：移除首页/末页/上下页按钮，改为页码跳转输入框
- KeyFormDialog：使用 masked 属性简化 API Key 输入逻辑
- ProviderManagement：重新设计表格布局，显示余额监控数据

src/api/admin/__init__.py

@@ -9,6 +9,7 @@ from .management_tokens import router as management_tokens_router
 from .modules import router as modules_router
 from .models import router as models_router
 from .monitoring import router as monitoring_router
+from .provider_ops import router as provider_ops_router
 from .provider_query import router as provider_query_router
 from .provider_strategy import router as provider_strategy_router
 from .providers import router as providers_router
@@ -32,6 +33,7 @@ router.include_router(security_router)
 router.include_router(provider_query_router)
 router.include_router(management_tokens_router)
 router.include_router(modules_router)
+router.include_router(provider_ops_router)
 
 # 注意：ldap_router 已迁移到模块系统，由 ModuleRegistry 动态注册
 # 当 LDAP_AVAILABLE=true 时才会注册路由

src/api/admin/provider_ops/__init__.py

@@ -0,0 +1,5 @@
+"""Provider 操作 API 模块"""
+
+from .routes import router
+
+__all__ = ["router"]

src/api/admin/provider_ops/routes.py

@@ -0,0 +1,510 @@
+"""
+Provider 操作 API 路由
+
+提供 Provider 操作相关的 API 端点：
+- 架构列表
+- 连接管理
+- 操作执行（余额查询、签到等）
+- 配置管理
+"""
+
+from dataclasses import asdict, is_dataclass
+from typing import Any, Dict, List, Optional
+
+from fastapi import APIRouter, Depends, HTTPException
+from pydantic import BaseModel, Field
+from sqlalchemy.orm import Session
+
+from src.database import get_db
+from src.models.database import Provider, User
+from src.services.provider_ops import (
+    ActionStatus,
+    ConnectorAuthType,
+    ConnectorStatus,
+    ProviderActionType,
+    ProviderOpsConfig,
+    ProviderOpsService,
+    get_registry,
+)
+from src.utils.auth_utils import require_admin
+
+router = APIRouter(prefix="/api/admin/provider-ops", tags=["Provider Operations"])
+
+
+# ==================== Request/Response Models ====================
+
+
+class ArchitectureInfo(BaseModel):
+    """架构信息"""
+
+    architecture_id: str
+    display_name: str
+    description: str
+    supported_auth_types: List[Dict[str, str]]
+    supported_actions: List[Dict[str, Any]]
+    default_connector: Optional[str]
+
+
+class ConnectorConfigRequest(BaseModel):
+    """连接器配置请求"""
+
+    auth_type: str = Field(..., description="认证类型")
+    config: Dict[str, Any] = Field(default_factory=dict, description="连接器配置")
+    credentials: Dict[str, Any] = Field(default_factory=dict, description="凭据信息")
+
+
+class ActionConfigRequest(BaseModel):
+    """操作配置请求"""
+
+    enabled: bool = Field(True, description="是否启用")
+    config: Dict[str, Any] = Field(default_factory=dict, description="操作配置")
+
+
+class SaveConfigRequest(BaseModel):
+    """保存配置请求"""
+
+    architecture_id: str = Field("generic_api", description="架构 ID")
+    base_url: Optional[str] = Field(None, description="API 基础地址")
+    connector: ConnectorConfigRequest
+    actions: Dict[str, ActionConfigRequest] = Field(default_factory=dict)
+    schedule: Dict[str, str] = Field(default_factory=dict, description="定时任务配置")
+
+
+class ConnectRequest(BaseModel):
+    """连接请求"""
+
+    credentials: Optional[Dict[str, Any]] = Field(None, description="凭据（可选，使用已保存的）")
+
+
+class ExecuteActionRequest(BaseModel):
+    """执行操作请求"""
+
+    config: Optional[Dict[str, Any]] = Field(None, description="操作配置（覆盖默认）")
+
+
+class ConnectionStatusResponse(BaseModel):
+    """连接状态响应"""
+
+    status: str
+    auth_type: str
+    connected_at: Optional[str]
+    expires_at: Optional[str]
+    last_error: Optional[str]
+
+
+class ActionResultResponse(BaseModel):
+    """操作结果响应"""
+
+    status: str
+    action_type: str
+    data: Optional[Any]
+    message: Optional[str]
+    executed_at: str
+    response_time_ms: Optional[int]
+    cache_ttl_seconds: int
+
+
+class ProviderOpsStatusResponse(BaseModel):
+    """Provider 操作状态响应"""
+
+    provider_id: str
+    is_configured: bool
+    architecture_id: Optional[str]
+    connection_status: ConnectionStatusResponse
+    enabled_actions: List[str]
+
+
+class ProviderOpsConfigResponse(BaseModel):
+    """Provider 操作配置响应（脱敏）"""
+
+    provider_id: str
+    is_configured: bool
+    architecture_id: Optional[str] = None
+    base_url: Optional[str] = None
+    connector: Optional[Dict[str, Any]] = None  # 脱敏后的连接器配置
+
+
+class VerifyAuthResponse(BaseModel):
+    """验证认证响应"""
+
+    success: bool
+    message: Optional[str] = None
+    data: Optional[Dict[str, Any]] = None
+
+
+# ==================== Helper Functions ====================
+
+
+def _serialize_data(data: Any) -> Any:
+    """序列化 dataclass 为字典，用于 JSON 响应"""
+    if data is None:
+        return None
+    if is_dataclass(data) and not isinstance(data, type):
+        return asdict(data)
+    return data
+
+
+# ==================== Routes ====================
+
+
+@router.get("/architectures", response_model=List[ArchitectureInfo])
+async def list_architectures(_: User = Depends(require_admin)):
+    """获取所有可用的架构"""
+    registry = get_registry()
+    return registry.to_dict_list()
+
+
+@router.get("/architectures/{architecture_id}", response_model=ArchitectureInfo)
+async def get_architecture(architecture_id: str, _: User = Depends(require_admin)):
+    """获取指定架构的详情"""
+    registry = get_registry()
+    arch = registry.get(architecture_id)
+    if not arch:
+        raise HTTPException(status_code=404, detail=f"架构 {architecture_id} 不存在")
+    return arch.to_dict()
+
+
+@router.get("/providers/{provider_id}/status", response_model=ProviderOpsStatusResponse)
+async def get_provider_ops_status(
+    provider_id: str,
+    db: Session = Depends(get_db),
+    _: User = Depends(require_admin),
+):
+    """获取 Provider 的操作状态"""
+    service = ProviderOpsService(db)
+
+    config = service.get_config(provider_id)
+    conn_state = service.get_connection_status(provider_id)
+
+    enabled_actions = []
+    if config:
+        for action_type, action_config in config.actions.items():
+            if action_config.get("enabled", True):
+                enabled_actions.append(action_type)
+
+    return ProviderOpsStatusResponse(
+        provider_id=provider_id,
+        is_configured=config is not None,
+        architecture_id=config.architecture_id if config else None,
+        connection_status=ConnectionStatusResponse(
+            status=conn_state.status.value,
+            auth_type=conn_state.auth_type.value,
+            connected_at=conn_state.connected_at.isoformat() if conn_state.connected_at else None,
+            expires_at=conn_state.expires_at.isoformat() if conn_state.expires_at else None,
+            last_error=conn_state.last_error,
+        ),
+        enabled_actions=enabled_actions,
+    )
+
+
+@router.get("/providers/{provider_id}/config", response_model=ProviderOpsConfigResponse)
+async def get_provider_ops_config(
+    provider_id: str,
+    db: Session = Depends(get_db),
+    _: User = Depends(require_admin),
+):
+    """
+    获取 Provider 的操作配置（脱敏）
+
+    返回已保存的配置，但敏感字段（如 api_key）会被脱敏处理。
+    """
+    service = ProviderOpsService(db)
+    config = service.get_config(provider_id)
+
+    if not config:
+        return ProviderOpsConfigResponse(
+            provider_id=provider_id,
+            is_configured=False,
+        )
+
+    # 获取 base_url
+    provider = db.query(Provider).filter(Provider.id == provider_id).first()
+    base_url = None
+    if provider:
+        provider_config = provider.config or {}
+        # base_url 可能存储在 provider_ops 配置中，也可能从 provider 获取
+        if provider.endpoints:
+            for endpoint in provider.endpoints:
+                if endpoint.base_url:
+                    base_url = endpoint.base_url
+                    break
+        if not base_url:
+            base_url = provider_config.get("base_url") or provider.website
+
+    # 获取脱敏后的凭据
+    masked_credentials = service.get_masked_credentials(config.connector_credentials)
+
+    return ProviderOpsConfigResponse(
+        provider_id=provider_id,
+        is_configured=True,
+        architecture_id=config.architecture_id,
+        base_url=base_url,
+        connector={
+            "auth_type": config.connector_auth_type.value,
+            "config": config.connector_config,
+            "credentials": masked_credentials,
+        },
+    )
+
+
+@router.put("/providers/{provider_id}/config")
+async def save_provider_ops_config(
+    provider_id: str,
+    request: SaveConfigRequest,
+    db: Session = Depends(get_db),
+    _: User = Depends(require_admin),
+):
+    """保存 Provider 的操作配置"""
+    service = ProviderOpsService(db)
+
+    # 合并凭据：如果请求中的敏感字段为空，使用已保存的凭据
+    credentials = service.merge_credentials_with_saved(
+        provider_id, dict(request.connector.credentials)
+    )
+
+    # 构建配置对象
+    config = ProviderOpsConfig(
+        architecture_id=request.architecture_id,
+        connector_auth_type=ConnectorAuthType(request.connector.auth_type),
+        connector_config=request.connector.config,
+        connector_credentials=credentials,
+        actions={
+            action_type: {"enabled": action_config.enabled, "config": action_config.config}
+            for action_type, action_config in request.actions.items()
+        },
+        schedule=request.schedule,
+    )
+
+    success = service.save_config(provider_id, config)
+    if not success:
+        raise HTTPException(status_code=404, detail="Provider 不存在")
+
+    return {"success": True, "message": "配置保存成功"}
+
+
+@router.post("/providers/{provider_id}/verify", response_model=VerifyAuthResponse)
+async def verify_provider_auth(
+    provider_id: str,
+    request: SaveConfigRequest,
+    db: Session = Depends(get_db),
+    _: User = Depends(require_admin),
+):
+    """
+    验证 Provider 认证配置
+
+    在保存前测试认证是否有效。
+    如果凭据中的敏感字段为空，会使用已保存的凭据。
+    """
+    service = ProviderOpsService(db)
+
+    # 获取 base_url
+    base_url = request.base_url
+    if not base_url:
+        # 尝试从 Provider 获取
+        provider = db.query(Provider).filter(Provider.id == provider_id).first()
+        if provider:
+            # 从 endpoints 或 config 获取
+            if provider.endpoints:
+                for endpoint in provider.endpoints:
+                    if endpoint.base_url:
+                        base_url = endpoint.base_url
+                        break
+            if not base_url and provider.config:
+                base_url = provider.config.get("base_url")
+
+    if not base_url:
+        return VerifyAuthResponse(
+            success=False,
+            message="请提供 API 地址",
+        )
+
+    # 合并凭据：如果请求中的敏感字段为空，使用已保存的凭据
+    credentials = service.merge_credentials_with_saved(
+        provider_id, dict(request.connector.credentials)
+    )
+
+    result = await service.verify_auth(
+        base_url=base_url,
+        architecture_id=request.architecture_id,
+        auth_type=ConnectorAuthType(request.connector.auth_type),
+        config=request.connector.config,
+        credentials=credentials,
+    )
+
+    return VerifyAuthResponse(
+        success=result.get("success", False),
+        message=result.get("message"),
+        data=result.get("data"),
+    )
+
+
+@router.delete("/providers/{provider_id}/config")
+async def delete_provider_ops_config(
+    provider_id: str,
+    db: Session = Depends(get_db),
+    _: User = Depends(require_admin),
+):
+    """删除 Provider 的操作配置"""
+    service = ProviderOpsService(db)
+    success = service.delete_config(provider_id)
+
+    if not success:
+        raise HTTPException(status_code=404, detail="Provider 不存在")
+
+    return {"success": True, "message": "配置已删除"}
+
+
+@router.post("/providers/{provider_id}/connect")
+async def connect_provider(
+    provider_id: str,
+    request: ConnectRequest,
+    db: Session = Depends(get_db),
+    _: User = Depends(require_admin),
+):
+    """建立与 Provider 的连接"""
+    service = ProviderOpsService(db)
+
+    success, message = await service.connect(provider_id, request.credentials)
+
+    if not success:
+        raise HTTPException(status_code=400, detail=message)
+
+    return {"success": True, "message": message}
+
+
+@router.post("/providers/{provider_id}/disconnect")
+async def disconnect_provider(
+    provider_id: str,
+    db: Session = Depends(get_db),
+    _: User = Depends(require_admin),
+):
+    """断开与 Provider 的连接"""
+    service = ProviderOpsService(db)
+    await service.disconnect(provider_id)
+
+    return {"success": True, "message": "已断开连接"}
+
+
+@router.post(
+    "/providers/{provider_id}/actions/{action_type}",
+    response_model=ActionResultResponse,
+)
+async def execute_action(
+    provider_id: str,
+    action_type: str,
+    request: ExecuteActionRequest,
+    db: Session = Depends(get_db),
+    _: User = Depends(require_admin),
+):
+    """执行指定操作"""
+    service = ProviderOpsService(db)
+
+    try:
+        action_type_enum = ProviderActionType(action_type)
+    except ValueError:
+        raise HTTPException(status_code=400, detail=f"无效的操作类型: {action_type}")
+
+    result = await service.execute_action(provider_id, action_type_enum, request.config)
+
+    return ActionResultResponse(
+        status=result.status.value,
+        action_type=result.action_type.value,
+        data=_serialize_data(result.data),
+        message=result.message,
+        executed_at=result.executed_at.isoformat(),
+        response_time_ms=result.response_time_ms,
+        cache_ttl_seconds=result.cache_ttl_seconds,
+    )
+
+
+@router.get("/providers/{provider_id}/balance", response_model=ActionResultResponse)
+async def get_balance(
+    provider_id: str,
+    refresh: bool = True,
+    db: Session = Depends(get_db),
+    _: User = Depends(require_admin),
+):
+    """
+    获取余额（优先返回缓存，后台异步刷新）
+
+    - refresh=True（默认）：返回缓存并触发后台刷新
+    - refresh=False：仅返回缓存，不触发刷新
+    """
+    service = ProviderOpsService(db)
+    result = await service.query_balance_with_cache(provider_id, trigger_refresh=refresh)
+
+    return ActionResultResponse(
+        status=result.status.value,
+        action_type=result.action_type.value,
+        data=_serialize_data(result.data),
+        message=result.message,
+        executed_at=result.executed_at.isoformat(),
+        response_time_ms=result.response_time_ms,
+        cache_ttl_seconds=result.cache_ttl_seconds,
+    )
+
+
+@router.post("/providers/{provider_id}/balance", response_model=ActionResultResponse)
+async def refresh_balance(
+    provider_id: str,
+    db: Session = Depends(get_db),
+    _: User = Depends(require_admin),
+):
+    """立即刷新余额（同步等待结果）"""
+    service = ProviderOpsService(db)
+    result = await service.query_balance(provider_id)
+
+    return ActionResultResponse(
+        status=result.status.value,
+        action_type=result.action_type.value,
+        data=_serialize_data(result.data),
+        message=result.message,
+        executed_at=result.executed_at.isoformat(),
+        response_time_ms=result.response_time_ms,
+        cache_ttl_seconds=result.cache_ttl_seconds,
+    )
+
+
+@router.post("/providers/{provider_id}/checkin", response_model=ActionResultResponse)
+async def checkin(
+    provider_id: str,
+    db: Session = Depends(get_db),
+    _: User = Depends(require_admin),
+):
+    """签到（快捷方法）"""
+    service = ProviderOpsService(db)
+    result = await service.checkin(provider_id)
+
+    return ActionResultResponse(
+        status=result.status.value,
+        action_type=result.action_type.value,
+        data=_serialize_data(result.data),
+        message=result.message,
+        executed_at=result.executed_at.isoformat(),
+        response_time_ms=result.response_time_ms,
+        cache_ttl_seconds=result.cache_ttl_seconds,
+    )
+
+
+@router.post("/batch/balance")
+async def batch_query_balance(
+    provider_ids: Optional[List[str]] = None,
+    db: Session = Depends(get_db),
+    _: User = Depends(require_admin),
+):
+    """批量查询余额"""
+    service = ProviderOpsService(db)
+    results = await service.batch_query_balance(provider_ids)
+
+    return {
+        provider_id: ActionResultResponse(
+            status=result.status.value,
+            action_type=result.action_type.value,
+            data=_serialize_data(result.data),
+            message=result.message,
+            executed_at=result.executed_at.isoformat(),
+            response_time_ms=result.response_time_ms,
+            cache_ttl_seconds=result.cache_ttl_seconds,
+        )
+        for provider_id, result in results.items()
+    }

src/api/admin/providers/summary.py

@@ -289,6 +289,9 @@ def _build_provider_summary(db: Session, provider: Provider) -> ProviderWithEndp
         for e in endpoints
     ]
 
+    # 检查是否配置了 Provider Ops（余额监控等）
+    ops_configured = bool((provider.config or {}).get("provider_ops"))
+
     return ProviderWithEndpointsSummary(
         id=provider.id,
         name=provider.name,
@@ -314,6 +317,7 @@ def _build_provider_summary(db: Session, provider: Provider) -> ProviderWithEndp
         unhealthy_endpoints=unhealthy_endpoints,
         api_formats=api_formats,
         endpoint_health_details=endpoint_health_details,
+        ops_configured=ops_configured,
         created_at=provider.created_at,
         updated_at=provider.updated_at,
     )

src/models/endpoint_models.py

@@ -639,6 +639,9 @@ class ProviderWithEndpointsSummary(BaseModel):
         default=0, description="不健康的端点数量（health_score < 0.5）"
     )
 
+    # Provider Ops 配置状态
+    ops_configured: bool = Field(default=False, description="是否配置了扩展操作（余额监控等）")
+
     # 时间戳
     created_at: datetime
     updated_at: datetime

src/services/provider_ops/__init__.py

@@ -0,0 +1,39 @@
+"""
+Provider 操作模块
+
+提供对提供商的扩展操作支持：
+- 多种鉴权方式（API Key、登录、Cookie）
+- 可扩展的操作类型（余额查询、签到等）
+"""
+
+from src.services.provider_ops.registry import ArchitectureRegistry, get_registry
+from src.services.provider_ops.service import ProviderOpsService
+from src.services.provider_ops.types import (
+    ActionResult,
+    ActionStatus,
+    BalanceInfo,
+    CheckinInfo,
+    ConnectorAuthType,
+    ConnectorState,
+    ConnectorStatus,
+    ProviderActionType,
+    ProviderOpsConfig,
+)
+
+__all__ = [
+    # 服务
+    "ProviderOpsService",
+    # 注册表
+    "ArchitectureRegistry",
+    "get_registry",
+    # 类型
+    "ActionResult",
+    "ActionStatus",
+    "BalanceInfo",
+    "CheckinInfo",
+    "ConnectorAuthType",
+    "ConnectorState",
+    "ConnectorStatus",
+    "ProviderActionType",
+    "ProviderOpsConfig",
+]

src/services/provider_ops/actions/__init__.py

@@ -0,0 +1,13 @@
+"""
+Provider 操作模块
+"""
+
+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
+
+__all__ = [
+    "ProviderAction",
+    "BalanceAction",
+    "CheckinAction",
+]

src/services/provider_ops/actions/balance.py

@@ -0,0 +1,231 @@
+"""
+余额查询操作
+"""
+
+import time
+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,
+    BalanceInfo,
+    ProviderActionType,
+)
+
+
+class BalanceAction(ProviderAction):
+    """
+    余额查询操作
+
+    支持可配置的 endpoint 和响应字段映射。
+    """
+
+    action_type = ProviderActionType.QUERY_BALANCE
+    display_name = "查询余额"
+    description = "查询账户余额信息"
+    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()
+
+        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)
+
+            # 检查业务状态码（如果配置了）
+            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,
+                    )
+
+            # 解析余额信息
+            balance = self._parse_balance(data, mapping)
+
+            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, 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",
+            ],
+        }
+
+        # 获取 quota 除数（用于将原始值转换为美元，如 New API 的 1/500000）
+        quota_divisor = self.config.get("quota_divisor", 1)
+
+        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
+
+            # 尝试默认映射
+            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
+
+            return None
+
+        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"])
+
+        # 如果只有部分数据，尝试计算
+        if total_available is None and total_granted is not None and total_used is not None:
+            total_available = total_granted - total_used
+        if total_used is None and total_granted is not None and total_available is not None:
+            total_used = total_granted - total_available
+        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,
+        )
+
+    def _to_float(self, value: Any) -> Optional[float]:
+        """转换为浮点数"""
+        if value is None:
+            return None
+        try:
+            return float(value)
+        except (TypeError, ValueError):
+            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/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"],
+        }

src/services/provider_ops/actions/base.py

@@ -0,0 +1,157 @@
+"""
+Provider 操作抽象基类
+"""
+
+from abc import ABC, abstractmethod
+from datetime import datetime, timezone
+from typing import Any, Dict, Optional
+
+import httpx
+
+from src.services.provider_ops.types import (
+    ActionResult,
+    ActionStatus,
+    ProviderActionType,
+)
+
+
+class ProviderAction(ABC):
+    """
+    提供商操作基类
+
+    定义具体的操作逻辑（如查询余额、签到等）。
+    """
+
+    # 子类需要定义的类属性
+    action_type: ProviderActionType = ProviderActionType.CUSTOM
+    display_name: str = "Base Action"
+    description: str = ""
+
+    # 默认缓存时间（秒）
+    default_cache_ttl: int = 300
+
+    def __init__(self, config: Optional[Dict[str, Any]] = None):
+        """
+        初始化操作
+
+        Args:
+            config: 操作配置
+        """
+        self.config = config or {}
+
+    @abstractmethod
+    async def execute(self, client: httpx.AsyncClient) -> ActionResult:
+        """
+        执行操作
+
+        Args:
+            client: 已认证的 HTTP 客户端
+
+        Returns:
+            操作结果
+        """
+        pass
+
+    def _extract_field(self, data: Any, path: Optional[str]) -> Any:
+        """
+        从响应数据中提取字段
+
+        支持点号分隔的路径，如 "data.user.balance"
+
+        Args:
+            data: 响应数据
+            path: 字段路径
+
+        Returns:
+            提取的值，如果路径无效则返回 None
+        """
+        if not path:
+            return None
+
+        current = data
+        for key in path.split("."):
+            if isinstance(current, dict):
+                current = current.get(key)
+            elif isinstance(current, list) and key.isdigit():
+                index = int(key)
+                current = current[index] if 0 <= index < len(current) else None
+            else:
+                return None
+
+            if current is None:
+                return None
+
+        return current
+
+    def _make_success_result(
+        self,
+        data: Any = None,
+        message: Optional[str] = None,
+        response_time_ms: Optional[int] = None,
+        raw_response: Optional[Dict[str, Any]] = None,
+    ) -> ActionResult:
+        """创建成功结果"""
+        return ActionResult(
+            status=ActionStatus.SUCCESS,
+            action_type=self.action_type,
+            data=data,
+            message=message,
+            response_time_ms=response_time_ms,
+            raw_response=raw_response,
+            cache_ttl_seconds=self.default_cache_ttl,
+        )
+
+    def _make_error_result(
+        self,
+        status: ActionStatus,
+        message: Optional[str] = None,
+        retry_after_seconds: Optional[int] = None,
+        raw_response: Optional[Dict[str, Any]] = None,
+    ) -> ActionResult:
+        """创建错误结果"""
+        return ActionResult(
+            status=status,
+            action_type=self.action_type,
+            message=message,
+            retry_after_seconds=retry_after_seconds,
+            raw_response=raw_response,
+            cache_ttl_seconds=0,  # 错误不缓存
+        )
+
+    def _handle_http_error(
+        self, response: httpx.Response, raw_data: Optional[Dict[str, Any]] = None
+    ) -> ActionResult:
+        """处理 HTTP 错误响应"""
+        status_code = response.status_code
+
+        if status_code == 401:
+            return self._make_error_result(
+                ActionStatus.AUTH_FAILED, "认证失败", raw_response=raw_data
+            )
+        elif status_code == 403:
+            return self._make_error_result(
+                ActionStatus.AUTH_FAILED, "无权限访问", 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,
+            )
+
+    @classmethod
+    def get_config_schema(cls) -> Dict[str, Any]:
+        """
+        获取操作配置 JSON Schema（用于前端表单生成）
+
+        子类应重写此方法
+        """
+        return {"type": "object", "properties": {}, "required": []}

src/services/provider_ops/actions/checkin.py

@@ -0,0 +1,236 @@
+"""
+签到操作
+"""
+
+import time
+from typing import Any, Dict
+
+import httpx
+
+from src.services.provider_ops.actions.base import ProviderAction
+from src.services.provider_ops.types import (
+    ActionResult,
+    ActionStatus,
+    CheckinInfo,
+    ProviderActionType,
+)
+
+
+class CheckinAction(ProviderAction):
+    """
+    签到操作
+
+    支持可配置的 endpoint 和响应字段映射。
+    """
+
+    action_type = ProviderActionType.CHECKIN
+    display_name = "签到"
+    description = "每日签到领取额度"
+    default_cache_ttl = 3600  # 签到结果缓存 1 小时
+
+    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]:
+        """
+        解析签到结果
+
+        Returns:
+            (CheckinInfo, 状态, 消息)
+        """
+        mapping = self.config.get("response_mapping", {})
+
+        # 检查成功状态
+        success_field = self.config.get("success_field", "success")
+        is_success = self._extract_field(data, success_field)
+
+        # 获取消息
+        message_field = self.config.get("message_field", "message")
+        message = self._extract_field(data, message_field)
+        if message is not None:
+            message = str(message)
+
+        # 检查是否已签到
+        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,
+                    )
+
+        # 判断是否成功
+        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(
+            reward=reward,
+            streak_days=streak_days,
+            message=message,
+            extra=extra,
+        )
+
+        return (checkin_info, ActionStatus.SUCCESS, message or "签到成功")
+
+    @classmethod
+    def get_config_schema(cls) -> Dict[str, Any]:
+        """获取操作配置 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"],
+        }

src/services/provider_ops/architectures/__init__.py

@@ -0,0 +1,21 @@
+"""
+Provider 架构模块
+"""
+
+from src.services.provider_ops.architectures.base import (
+    ProviderArchitecture,
+    ProviderConnector,
+    VerifyResult,
+)
+from src.services.provider_ops.architectures.generic_api import GenericApiArchitecture
+from src.services.provider_ops.architectures.new_api import NewApiArchitecture
+from src.services.provider_ops.architectures.one_api import OneApiArchitecture
+
+__all__ = [
+    "ProviderArchitecture",
+    "ProviderConnector",
+    "VerifyResult",
+    "GenericApiArchitecture",
+    "NewApiArchitecture",
+    "OneApiArchitecture",
+]

src/services/provider_ops/architectures/base.py

@@ -0,0 +1,529 @@
+"""
+Provider 架构抽象基类
+"""
+
+from abc import ABC, abstractmethod
+from contextlib import asynccontextmanager
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from typing import Any, AsyncIterator, Dict, List, Optional, Type
+
+import httpx
+
+from src.services.provider_ops.actions.base import ProviderAction
+from src.services.provider_ops.types import (
+    ConnectorAuthType,
+    ConnectorState,
+    ConnectorStatus,
+    ProviderActionType,
+)
+
+
+# ==================== 连接器基类 ====================
+
+
+class ProviderConnector(ABC):
+    """
+    提供商连接器基类
+
+    负责建立与提供商的认证连接，管理凭据状态。
+    每个架构应在自己的文件中实现对应的连接器子类。
+    """
+
+    # 子类需要定义的类属性
+    auth_type: ConnectorAuthType = ConnectorAuthType.NONE
+    display_name: str = "Base Connector"
+
+    def __init__(self, base_url: str, config: Optional[Dict[str, Any]] = None):
+        """
+        初始化连接器
+
+        Args:
+            base_url: 提供商 API 基础 URL
+            config: 连接器配置
+        """
+        self.base_url = base_url.rstrip("/")
+        self.config = config or {}
+        self._status = ConnectorStatus.DISCONNECTED
+        self._connected_at: Optional[datetime] = None
+        self._expires_at: Optional[datetime] = None
+        self._last_error: Optional[str] = None
+
+        # 代理配置
+        self._proxy: Optional[str] = self.config.get("proxy")
+
+        # HTTP 客户端配置
+        self._timeout = self.config.get("timeout", 30)
+        self._headers: Dict[str, str] = {}
+
+    @abstractmethod
+    async def connect(self, credentials: Dict[str, Any]) -> bool:
+        """
+        建立认证连接
+
+        Args:
+            credentials: 凭据信息（如用户名密码、API Key 等）
+
+        Returns:
+            是否连接成功
+        """
+        pass
+
+    @abstractmethod
+    async def disconnect(self) -> None:
+        """断开连接，清理状态"""
+        pass
+
+    @abstractmethod
+    async def is_authenticated(self) -> bool:
+        """检查当前是否已认证"""
+        pass
+
+    @abstractmethod
+    def _apply_auth(self, request: httpx.Request) -> httpx.Request:
+        """
+        为请求应用认证信息
+
+        Args:
+            request: 原始请求
+
+        Returns:
+            添加认证信息后的请求
+        """
+        pass
+
+    async def refresh_auth(self, credentials: Dict[str, Any]) -> bool:
+        """
+        刷新认证（如 Token 过期）
+
+        默认实现：重新连接
+
+        Args:
+            credentials: 凭据信息
+
+        Returns:
+            是否刷新成功
+        """
+        return await self.connect(credentials)
+
+    @asynccontextmanager
+    async def get_client(self) -> AsyncIterator[httpx.AsyncClient]:
+        """
+        获取已认证的 HTTP 客户端
+
+        使用 context manager 确保资源正确释放
+
+        Yields:
+            已配置认证信息的 AsyncClient
+        """
+        transport = None
+        if self._proxy:
+            transport = httpx.AsyncHTTPTransport(proxy=self._proxy)
+
+        async with httpx.AsyncClient(
+            base_url=self.base_url,
+            timeout=self._timeout,
+            transport=transport,
+            event_hooks={"request": [self._auth_hook]},
+        ) as client:
+            yield client
+
+    async def _auth_hook(self, request: httpx.Request) -> None:
+        """请求钩子：应用认证信息"""
+        self._apply_auth(request)
+
+    def get_state(self) -> ConnectorState:
+        """获取连接器当前状态"""
+        return ConnectorState(
+            status=self._status,
+            auth_type=self.auth_type,
+            connected_at=self._connected_at,
+            expires_at=self._expires_at,
+            last_error=self._last_error,
+        )
+
+    def _set_connected(self, expires_at: Optional[datetime] = None) -> None:
+        """设置为已连接状态"""
+        self._status = ConnectorStatus.CONNECTED
+        self._connected_at = datetime.now(timezone.utc)
+        self._expires_at = expires_at
+        self._last_error = None
+
+    def _set_error(self, error: str) -> None:
+        """设置错误状态"""
+        self._status = ConnectorStatus.ERROR
+        self._last_error = error
+
+    def _set_disconnected(self) -> None:
+        """设置为断开状态"""
+        self._status = ConnectorStatus.DISCONNECTED
+        self._connected_at = None
+        self._expires_at = None
+
+    @classmethod
+    def get_credentials_schema(cls) -> Dict[str, Any]:
+        """
+        获取凭据配置 JSON Schema（用于前端表单生成）
+
+        子类应重写此方法
+        """
+        return {"type": "object", "properties": {}, "required": []}
+
+
+# ==================== 验证结果 ====================
+
+
+@dataclass
+class VerifyResult:
+    """认证验证结果"""
+
+    success: bool
+    message: Optional[str] = None
+    username: Optional[str] = None
+    display_name: Optional[str] = None
+    email: Optional[str] = None
+    quota: Optional[float] = None
+    used_quota: Optional[float] = None
+    request_count: Optional[int] = None
+    extra: Optional[Dict[str, Any]] = None
+
+    def to_dict(self) -> Dict[str, Any]:
+        """转换为字典"""
+        if not self.success:
+            return {"success": False, "message": self.message}
+
+        return {
+            "success": True,
+            "data": {
+                "username": self.username,
+                "display_name": self.display_name or self.username,
+                "email": self.email,
+                "quota": self.quota,
+                "used_quota": self.used_quota,
+                "request_count": self.request_count,
+                "extra": self.extra or {},
+            },
+        }
+
+
+# ==================== 架构基类 ====================
+
+
+class ProviderArchitecture(ABC):
+    """
+    提供商架构基类
+
+    架构 = Connector（鉴权方式） + Actions（支持的操作）
+
+    一个架构可以被多个 Provider 复用。
+    例如：generic_api 架构可用于各种中转站。
+
+    ## 添加新认证模板的步骤
+
+    1. 在 architectures/ 目录创建新文件
+    2. 继承 ProviderArchitecture 和 ProviderConnector
+    3. 定义类属性：architecture_id, display_name, description
+    4. 实现连接器子类和架构类
+    5. 重写认证相关方法：
+       - get_verify_endpoint(): 返回验证端点
+       - build_verify_headers(): 构建验证请求 headers
+       - parse_verify_response(): 解析验证响应
+    6. 在 registry.py 的 _register_builtin_architectures() 中注册
+    """
+
+    # 子类需要定义的类属性
+    architecture_id: str = ""
+    display_name: str = ""
+    description: str = ""
+
+    # 支持的 Connector 类型列表（按优先级排序）
+    supported_connectors: List[Type[ProviderConnector]] = []
+
+    # 支持的 Action 类型列表
+    supported_actions: List[Type[ProviderAction]] = []
+
+    # 默认操作配置
+    default_action_configs: Dict[ProviderActionType, Dict[str, Any]] = {}
+
+    def __init__(self, config: Optional[Dict[str, Any]] = None):
+        """
+        初始化架构
+
+        Args:
+            config: 架构配置
+        """
+        self.config = config or {}
+
+    # ==================== 认证验证相关方法 ====================
+
+    def get_credentials_schema(self) -> Dict[str, Any]:
+        """
+        获取凭据字段定义（JSON Schema 格式）
+
+        子类应重写此方法定义需要的凭据字段。
+        这个 schema 可用于：
+        1. 前端表单生成（如果需要动态渲染）
+        2. 凭据验证
+        3. 文档生成
+
+        Returns:
+            JSON Schema 格式的字段定义
+
+        Example:
+            {
+                "type": "object",
+                "properties": {
+                    "api_key": {
+                        "type": "string",
+                        "title": "API Key",
+                        "description": "访问令牌",
+                    },
+                    "user_id": {
+                        "type": "string",
+                        "title": "用户 ID",
+                        "description": "New API 用户 ID",
+                    },
+                },
+                "required": ["api_key", "user_id"],
+            }
+        """
+        return {
+            "type": "object",
+            "properties": {
+                "api_key": {
+                    "type": "string",
+                    "title": "API Key",
+                    "description": "访问令牌",
+                },
+            },
+            "required": ["api_key"],
+        }
+
+    def get_verify_endpoint(self) -> str:
+        """
+        获取认证验证端点
+
+        子类可重写以自定义验证端点。
+
+        Returns:
+            验证端点路径（如 /api/user/self）
+        """
+        return "/api/user/self"
+
+    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(
+        self,
+        base_url: str,
+        auth_type: Optional[ConnectorAuthType] = None,
+        config: Optional[Dict[str, Any]] = None,
+    ) -> ProviderConnector:
+        """
+        获取连接器实例
+
+        Args:
+            base_url: 提供商 API 基础 URL
+            auth_type: 指定的认证类型，None 则使用默认
+            config: 连接器配置
+
+        Returns:
+            连接器实例
+
+        Raises:
+            ValueError: 不支持的认证类型
+        """
+        if not self.supported_connectors:
+            raise ValueError(f"架构 {self.architecture_id} 未配置支持的连接器")
+
+        # 查找匹配的连接器
+        connector_cls: Optional[Type[ProviderConnector]] = None
+
+        if auth_type:
+            for cls in self.supported_connectors:
+                if cls.auth_type == auth_type:
+                    connector_cls = cls
+                    break
+
+            if not connector_cls:
+                supported = [c.auth_type.value for c in self.supported_connectors]
+                raise ValueError(
+                    f"架构 {self.architecture_id} 不支持 {auth_type.value} 认证，"
+                    f"支持的类型: {supported}"
+                )
+        else:
+            # 使用第一个（默认）连接器
+            connector_cls = self.supported_connectors[0]
+
+        return connector_cls(base_url, config)
+
+    def get_action(
+        self,
+        action_type: ProviderActionType,
+        config: Optional[Dict[str, Any]] = None,
+    ) -> ProviderAction:
+        """
+        获取操作实例
+
+        Args:
+            action_type: 操作类型
+            config: 操作配置（会与默认配置合并）
+
+        Returns:
+            操作实例
+
+        Raises:
+            ValueError: 不支持的操作类型
+        """
+        action_cls: Optional[Type[ProviderAction]] = None
+
+        for cls in self.supported_actions:
+            if cls.action_type == action_type:
+                action_cls = cls
+                break
+
+        if not action_cls:
+            supported = [a.action_type.value for a in self.supported_actions]
+            raise ValueError(
+                f"架构 {self.architecture_id} 不支持 {action_type.value} 操作，"
+                f"支持的操作: {supported}"
+            )
+
+        # 合并默认配置和用户配置
+        merged_config = dict(self.default_action_configs.get(action_type, {}))
+        if config:
+            merged_config.update(config)
+
+        return action_cls(merged_config)
+
+    def supports_action(self, action_type: ProviderActionType) -> bool:
+        """检查是否支持指定操作"""
+        return any(a.action_type == action_type for a in self.supported_actions)
+
+    def supports_auth_type(self, auth_type: ConnectorAuthType) -> bool:
+        """检查是否支持指定认证类型"""
+        return any(c.auth_type == auth_type for c in self.supported_connectors)
+
+    def get_supported_auth_types(self) -> List[ConnectorAuthType]:
+        """获取支持的认证类型列表"""
+        return [c.auth_type for c in self.supported_connectors]
+
+    def get_supported_action_types(self) -> List[ProviderActionType]:
+        """获取支持的操作类型列表"""
+        return [a.action_type for a in self.supported_actions]
+
+    def to_dict(self) -> Dict[str, Any]:
+        """转换为字典（用于 API 响应）"""
+        return {
+            "architecture_id": self.architecture_id,
+            "display_name": self.display_name,
+            "description": self.description,
+            "credentials_schema": self.get_credentials_schema(),
+            "verify_endpoint": self.get_verify_endpoint(),
+            "supported_auth_types": [
+                {"type": c.auth_type.value, "display_name": c.display_name}
+                for c in self.supported_connectors
+            ],
+            "supported_actions": [
+                {
+                    "type": a.action_type.value,
+                    "display_name": a.display_name,
+                    "description": a.description,
+                    "config_schema": a.get_config_schema(),
+                }
+                for a in self.supported_actions
+            ],
+            "default_connector": (
+                self.supported_connectors[0].auth_type.value
+                if self.supported_connectors
+                else None
+            ),
+        }

src/services/provider_ops/architectures/generic_api.py

@@ -0,0 +1,155 @@
+"""
+通用 API 架构
+
+支持各种中转站的可配置架构。
+
+## 添加新认证模板示例
+
+如需添加新的中转站模板（如 MyApi），参考以下步骤：
+
+1. 在 architectures/ 目录创建新文件，如 my_api.py：
+
+    from src.services.provider_ops.architectures.base import ProviderArchitecture
+    from src.services.provider_ops.connectors.base import ProviderConnector
+
+    class MyApiConnector(ProviderConnector):
+        # 实现自己的连接器
+        pass
+
+    class MyApiArchitecture(ProviderArchitecture):
+        architecture_id = "my_api"
+        display_name = "My API"
+        description = "My API 风格中转站"
+
+        supported_connectors = [MyApiConnector]
+        supported_actions = [BalanceAction]
+
+        # 如果需要特殊的认证 headers，重写此方法
+        def build_verify_headers(self, config, credentials):
+            headers = super().build_verify_headers(config, credentials)
+            if "custom_field" in credentials:
+                headers["X-Custom-Header"] = credentials["custom_field"]
+            return headers
+
+2. 在 registry.py 的 _register_builtin_architectures() 中注册：
+
+    from .my_api import MyApiArchitecture
+    builtin = [..., MyApiArchitecture]
+
+3. 在前端 auth-templates/ 添加对应的模板定义
+"""
+
+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.types import ConnectorAuthType, ProviderActionType
+
+
+class GenericApiKeyConnector(ProviderConnector):
+    """
+    通用 API Key 连接器
+
+    支持多种 API Key 传递方式：
+    - Bearer Token (Authorization: Bearer xxx)
+    - Custom Header (X-API-Key: xxx)
+    """
+
+    auth_type = ConnectorAuthType.API_KEY
+    display_name = "API Key"
+
+    def __init__(self, base_url: str, config: Optional[Dict[str, Any]] = None):
+        super().__init__(base_url, config)
+        self._api_key: Optional[str] = None
+        # 支持配置认证方式
+        self._auth_method = self.config.get("auth_method", "bearer")
+        self._header_name = self.config.get("header_name", "Authorization")
+
+    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 not self._api_key:
+            return request
+
+        if self._auth_method == "bearer":
+            request.headers["Authorization"] = f"Bearer {self._api_key}"
+        elif self._auth_method == "header":
+            request.headers[self._header_name] = 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": "提供商的 API Key",
+                },
+            },
+            "required": ["api_key"],
+        }
+
+
+class GenericApiArchitecture(ProviderArchitecture):
+    """
+    通用 API 架构
+
+    适用于各种中转站，支持所有认证方式和操作类型。
+    用户可以完全自定义 endpoint 和响应映射。
+
+    这是"自定义"模板对应的后端架构。
+    """
+
+    architecture_id = "generic_api"
+    display_name = "通用 API"
+    description = "可配置的通用 API 架构，适用于各种中转站"
+
+    supported_connectors: List[Type[ProviderConnector]] = [
+        GenericApiKeyConnector,
+    ]
+
+    supported_actions: List[Type[ProviderAction]] = [
+        BalanceAction,
+        CheckinAction,
+    ]
+
+    # 默认操作配置（可被用户配置覆盖）
+    default_action_configs: Dict[ProviderActionType, Dict[str, Any]] = {
+        ProviderActionType.QUERY_BALANCE: {
+            "endpoint": "/api/user/balance",
+            "method": "GET",
+        },
+        ProviderActionType.CHECKIN: {
+            "endpoint": "/api/user/checkin",
+            "method": "POST",
+        },
+    }
+
+    def get_credentials_schema(self) -> Dict[str, Any]:
+        """通用架构只需要 api_key"""
+        return GenericApiKeyConnector.get_credentials_schema()

src/services/provider_ops/architectures/new_api.py

@@ -0,0 +1,155 @@
+"""
+New API 架构
+
+针对 New API 风格的中转站优化的预设配置。
+"""
+
+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.types import ConnectorAuthType, ProviderActionType
+
+
+class NewApiConnector(ProviderConnector):
+    """
+    New API 专用连接器
+
+    特点：
+    - 使用 Bearer Token 认证
+    - 需要 New-Api-User Header 传递用户 ID
+    """
+
+    auth_type = ConnectorAuthType.API_KEY
+    display_name = "New API Key"
+
+    def __init__(self, base_url: str, config: Optional[Dict[str, Any]] = None):
+        super().__init__(base_url, config)
+        self._api_key: Optional[str] = None
+        self._user_id: 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 不能为空")
+            return False
+
+        user_id = credentials.get("user_id")
+        if not user_id:
+            self._set_error("用户 ID 不能为空")
+            return False
+
+        self._api_key = api_key
+        self._user_id = str(user_id)
+        self._set_connected()
+        return True
+
+    async def disconnect(self) -> None:
+        """断开连接"""
+        self._api_key = None
+        self._user_id = None
+        self._set_disconnected()
+
+    async def is_authenticated(self) -> bool:
+        """检查是否已认证"""
+        return self._api_key is not None and self._user_id is not None
+
+    def _apply_auth(self, request: httpx.Request) -> httpx.Request:
+        """为请求应用认证信息"""
+        if self._api_key:
+            request.headers["Authorization"] = f"Bearer {self._api_key}"
+        if self._user_id:
+            request.headers["New-Api-User"] = self._user_id
+        return request
+
+    @classmethod
+    def get_credentials_schema(cls) -> Dict[str, Any]:
+        """获取凭据配置 schema"""
+        return {
+            "type": "object",
+            "properties": {
+                "api_key": {
+                    "type": "string",
+                    "title": "访问令牌 (API Key)",
+                    "description": "New API 的访问令牌",
+                },
+                "user_id": {
+                    "type": "string",
+                    "title": "用户 ID",
+                    "description": "New API 用户 ID，用于 New-Api-User Header",
+                },
+            },
+            "required": ["api_key", "user_id"],
+        }
+
+
+class NewApiArchitecture(ProviderArchitecture):
+    """
+    New API 架构预设
+
+    针对 New API 风格的中转站优化的预设配置。
+
+    特点：
+    - 使用 Bearer Token 认证
+    - 需要 New-Api-User Header 传递用户 ID
+    - 验证端点: /api/user/self
+    - quota 单位通常是 1/500000 美元
+    """
+
+    architecture_id = "new_api"
+    display_name = "New API"
+    description = "New API 风格中转站的预设配置"
+
+    supported_connectors: List[Type[ProviderConnector]] = [
+        NewApiConnector,
+    ]
+
+    supported_actions: List[Type[ProviderAction]] = [
+        BalanceAction,
+        CheckinAction,
+    ]
+
+    default_action_configs: Dict[ProviderActionType, Dict[str, Any]] = {
+        ProviderActionType.QUERY_BALANCE: {
+            "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",
+        },
+    }
+
+    def get_credentials_schema(self) -> Dict[str, Any]:
+        """New API 需要 api_key 和 user_id"""
+        return NewApiConnector.get_credentials_schema()
+
+    def build_verify_headers(
+        self,
+        config: Dict[str, Any],
+        credentials: Dict[str, Any],
+    ) -> Dict[str, str]:
+        """
+        构建 New API 的验证请求 Headers
+
+        New API 特有：需要 New-Api-User Header 传递用户 ID
+        """
+        headers = super().build_verify_headers(config, credentials)
+
+        # New API 特有的 header
+        user_id = credentials.get("user_id", "")
+        if user_id:
+            headers["New-Api-User"] = str(user_id)
+
+        return headers

src/services/provider_ops/architectures/one_api.py

@@ -0,0 +1,111 @@
+"""
+One API 架构
+
+针对 One API 风格的中转站优化的预设配置。
+"""
+
+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.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: Optional[Dict[str, Any]] = None):
+        super().__init__(base_url, config)
+        self._api_key: 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 不能为空")
+            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]] = [
+        BalanceAction,
+    ]
+
+    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()

src/services/provider_ops/registry.py

@@ -0,0 +1,136 @@
+"""
+架构注册表
+
+管理所有可用的 Provider 架构。
+"""
+
+import threading
+from typing import Dict, List, Optional, Type
+
+from src.core.logger import logger
+from src.services.provider_ops.architectures import (
+    GenericApiArchitecture,
+    NewApiArchitecture,
+    OneApiArchitecture,
+    ProviderArchitecture,
+)
+
+
+class ArchitectureRegistry:
+    """
+    架构注册表
+
+    单例模式，管理所有可用的 Provider 架构。
+    """
+
+    _instance: Optional["ArchitectureRegistry"] = None
+    _lock = threading.Lock()
+
+    def __new__(cls) -> "ArchitectureRegistry":
+        if cls._instance is None:
+            with cls._lock:
+                if cls._instance is None:
+                    cls._instance = super().__new__(cls)
+                    cls._instance._initialized = False
+        return cls._instance
+
+    def __init__(self) -> None:
+        if self._initialized:
+            return
+
+        self._architectures: Dict[str, ProviderArchitecture] = {}
+        self._initialized = True
+
+        # 注册内置架构
+        self._register_builtin_architectures()
+
+    def _register_builtin_architectures(self) -> None:
+        """注册内置架构"""
+        builtin = [
+            GenericApiArchitecture,
+            NewApiArchitecture,
+            OneApiArchitecture,
+        ]
+
+        for arch_cls in builtin:
+            self.register(arch_cls())
+
+    def register(self, architecture: ProviderArchitecture) -> None:
+        """
+        注册架构
+
+        Args:
+            architecture: 架构实例
+        """
+        if architecture.architecture_id in self._architectures:
+            logger.warning(f"架构 {architecture.architecture_id} 已存在，将被覆盖")
+
+        self._architectures[architecture.architecture_id] = architecture
+        logger.debug(f"注册架构: {architecture.architecture_id}")
+
+    def unregister(self, architecture_id: str) -> bool:
+        """
+        注销架构
+
+        Args:
+            architecture_id: 架构 ID
+
+        Returns:
+            是否成功注销
+        """
+        if architecture_id in self._architectures:
+            del self._architectures[architecture_id]
+            return True
+        return False
+
+    def get(self, architecture_id: str) -> Optional[ProviderArchitecture]:
+        """
+        获取架构
+
+        Args:
+            architecture_id: 架构 ID
+
+        Returns:
+            架构实例，不存在则返回 None
+        """
+        return self._architectures.get(architecture_id)
+
+    def get_or_default(self, architecture_id: Optional[str] = None) -> ProviderArchitecture:
+        """
+        获取架构，如果不存在则返回默认架构
+
+        Args:
+            architecture_id: 架构 ID
+
+        Returns:
+            架构实例
+        """
+        if architecture_id and architecture_id in self._architectures:
+            return self._architectures[architecture_id]
+
+        # 返回默认架构（generic_api）
+        return self._architectures.get("generic_api", GenericApiArchitecture())
+
+    def list_all(self) -> List[ProviderArchitecture]:
+        """获取所有已注册的架构"""
+        return list(self._architectures.values())
+
+    def list_ids(self) -> List[str]:
+        """获取所有已注册的架构 ID"""
+        return list(self._architectures.keys())
+
+    def to_dict_list(self) -> List[Dict]:
+        """获取所有架构的字典表示（用于 API 响应）"""
+        return [arch.to_dict() for arch in self._architectures.values()]
+
+
+# 全局注册表实例
+_registry: Optional[ArchitectureRegistry] = None
+
+
+def get_registry() -> ArchitectureRegistry:
+    """获取全局注册表实例"""
+    global _registry
+    if _registry is None:
+        _registry = ArchitectureRegistry()
+    return _registry

src/services/provider_ops/service.py

@@ -0,0 +1,689 @@
+"""
+Provider 操作服务
+
+提供操作执行、凭据管理、缓存等业务逻辑。
+"""
+
+import asyncio
+import json
+from dataclasses import asdict
+from datetime import datetime, timezone
+from typing import Any, Dict, List, Optional
+
+from sqlalchemy.orm import Session
+
+from src.core.cache_service import CacheService
+from src.core.crypto import CryptoService
+from src.core.logger import logger
+from src.database import create_session
+from src.models.database import Provider
+from src.services.provider_ops.architectures import ProviderArchitecture, ProviderConnector
+from src.services.provider_ops.registry import get_registry
+from src.services.provider_ops.types import (
+    ActionResult,
+    ActionStatus,
+    BalanceInfo,
+    ConnectorAuthType,
+    ConnectorState,
+    ConnectorStatus,
+    ProviderActionType,
+    ProviderOpsConfig,
+)
+
+# 余额缓存 TTL（24 小时）
+BALANCE_CACHE_TTL = 86400
+
+
+class ProviderOpsService:
+    """
+    Provider 操作服务
+
+    提供：
+    - 凭据管理（加密存储、读取）
+    - 连接管理（建立、断开、状态检查）
+    - 操作执行（余额查询、签到等）
+    """
+
+    # 凭据中需要加密的字段
+    SENSITIVE_FIELDS = {"api_key", "password", "session_token", "cookie_string", "cookies"}
+
+    def __init__(self, db: Session):
+        self.db = db
+        self.crypto = CryptoService()
+
+        # 连接器缓存 {provider_id: ProviderConnector}
+        self._connectors: Dict[str, ProviderConnector] = {}
+
+    # ==================== 配置管理 ====================
+
+    def get_config(self, provider_id: str) -> Optional[ProviderOpsConfig]:
+        """
+        获取 Provider 的操作配置
+
+        Args:
+            provider_id: Provider ID
+
+        Returns:
+            配置对象，未配置则返回 None
+        """
+        provider = self._get_provider(provider_id)
+        if not provider:
+            return None
+
+        config_data = (provider.config or {}).get("provider_ops")
+        if not config_data:
+            return None
+
+        return ProviderOpsConfig.from_dict(config_data)
+
+    def save_config(
+        self,
+        provider_id: str,
+        config: ProviderOpsConfig,
+    ) -> bool:
+        """
+        保存 Provider 的操作配置
+
+        Args:
+            provider_id: Provider ID
+            config: 配置对象
+
+        Returns:
+            是否保存成功
+        """
+        provider = self._get_provider(provider_id)
+        if not provider:
+            return False
+
+        # 加密敏感凭据
+        encrypted_credentials = self._encrypt_credentials(config.connector_credentials)
+        logger.debug(
+            f"加密凭据: provider_id={provider_id}, "
+            f"input_keys={list(config.connector_credentials.keys())}, "
+            f"output_keys={list(encrypted_credentials.keys())}, "
+            f"has_api_key={bool(config.connector_credentials.get('api_key'))}"
+        )
+
+        # 构建配置
+        config_dict = config.to_dict()
+        config_dict["connector"]["credentials"] = encrypted_credentials
+
+        # 更新 Provider 配置
+        provider_config = dict(provider.config or {})
+        provider_config["provider_ops"] = config_dict
+        provider.config = provider_config
+
+        self.db.commit()
+
+        # 清除连接器缓存
+        if provider_id in self._connectors:
+            del self._connectors[provider_id]
+
+        logger.info(f"保存 Provider 操作配置: provider_id={provider_id}")
+        return True
+
+    def delete_config(self, provider_id: str) -> bool:
+        """
+        删除 Provider 的操作配置
+
+        Args:
+            provider_id: Provider ID
+
+        Returns:
+            是否删除成功
+        """
+        provider = self._get_provider(provider_id)
+        if not provider:
+            return False
+
+        provider_config = dict(provider.config or {})
+        if "provider_ops" in provider_config:
+            del provider_config["provider_ops"]
+            provider.config = provider_config
+            self.db.commit()
+
+        # 清除连接器缓存
+        if provider_id in self._connectors:
+            del self._connectors[provider_id]
+
+        return True
+
+    # ==================== 连接管理 ====================
+
+    async def connect(
+        self,
+        provider_id: str,
+        credentials: Optional[Dict[str, Any]] = None,
+    ) -> tuple[bool, str]:
+        """
+        建立与 Provider 的连接
+
+        Args:
+            provider_id: Provider ID
+            credentials: 凭据（如果为 None 则使用已保存的凭据）
+
+        Returns:
+            (是否成功, 消息)
+        """
+        provider = self._get_provider(provider_id)
+        if not provider:
+            return False, "Provider 不存在"
+
+        config = self.get_config(provider_id)
+        if not config:
+            return False, "未配置操作设置"
+
+        # 获取架构
+        registry = get_registry()
+        architecture = registry.get_or_default(config.architecture_id)
+
+        # 获取 base_url
+        base_url = self._get_provider_base_url(provider)
+        if not base_url:
+            return False, "Provider 未配置 base_url"
+
+        # 创建连接器
+        try:
+            connector = architecture.get_connector(
+                base_url=base_url,
+                auth_type=config.connector_auth_type,
+                config=config.connector_config,
+            )
+        except ValueError as e:
+            return False, str(e)
+
+        # 使用提供的凭据或已保存的凭据
+        if credentials:
+            actual_credentials = credentials
+        else:
+            actual_credentials = self._decrypt_credentials(config.connector_credentials)
+            logger.debug(
+                f"解密凭据: provider_id={provider_id}, "
+                f"encrypted_keys={list(config.connector_credentials.keys())}, "
+                f"decrypted_keys={list(actual_credentials.keys())}, "
+                f"has_api_key={bool(actual_credentials.get('api_key'))}"
+            )
+
+        if not actual_credentials:
+            return False, "未提供凭据"
+
+        # 建立连接
+        logger.info(
+            f"尝试连接: provider_id={provider_id}, "
+            f"credentials_keys={list(actual_credentials.keys())}"
+        )
+        success = await connector.connect(actual_credentials)
+        if success:
+            self._connectors[provider_id] = connector
+            return True, "连接成功"
+        else:
+            state = connector.get_state()
+            return False, state.last_error or "连接失败"
+
+    async def disconnect(self, provider_id: str) -> bool:
+        """
+        断开与 Provider 的连接
+
+        Args:
+            provider_id: Provider ID
+
+        Returns:
+            是否成功
+        """
+        connector = self._connectors.get(provider_id)
+        if connector:
+            await connector.disconnect()
+            del self._connectors[provider_id]
+        return True
+
+    def get_connection_status(self, provider_id: str) -> ConnectorState:
+        """
+        获取连接状态
+
+        Args:
+            provider_id: Provider ID
+
+        Returns:
+            连接器状态
+        """
+        connector = self._connectors.get(provider_id)
+        if connector:
+            return connector.get_state()
+
+        # 未连接
+        config = self.get_config(provider_id)
+        return ConnectorState(
+            status=ConnectorStatus.DISCONNECTED,
+            auth_type=config.connector_auth_type if config else ConnectorAuthType.NONE,
+        )
+
+    # ==================== 操作执行 ====================
+
+    async def execute_action(
+        self,
+        provider_id: str,
+        action_type: ProviderActionType,
+        action_config: Optional[Dict[str, Any]] = None,
+    ) -> ActionResult:
+        """
+        执行操作
+
+        Args:
+            provider_id: Provider ID
+            action_type: 操作类型
+            action_config: 操作配置（覆盖默认配置）
+
+        Returns:
+            操作结果
+        """
+        # 检查连接状态
+        connector = self._connectors.get(provider_id)
+        if not connector:
+            # 尝试自动连接
+            success, message = await self.connect(provider_id)
+            if not success:
+                return ActionResult(
+                    status=ActionStatus.AUTH_FAILED,
+                    action_type=action_type,
+                    message=f"连接失败: {message}",
+                )
+            connector = self._connectors.get(provider_id)
+
+        if not connector or not await connector.is_authenticated():
+            return ActionResult(
+                status=ActionStatus.AUTH_EXPIRED,
+                action_type=action_type,
+                message="认证已过期，请重新连接",
+            )
+
+        # 获取配置
+        config = self.get_config(provider_id)
+        if not config:
+            return ActionResult(
+                status=ActionStatus.NOT_CONFIGURED,
+                action_type=action_type,
+                message="未配置操作设置",
+            )
+
+        # 获取架构
+        registry = get_registry()
+        architecture = registry.get_or_default(config.architecture_id)
+
+        # 检查是否支持该操作
+        if not architecture.supports_action(action_type):
+            return ActionResult(
+                status=ActionStatus.NOT_SUPPORTED,
+                action_type=action_type,
+                message=f"架构 {architecture.architecture_id} 不支持 {action_type.value} 操作",
+            )
+
+        # 合并操作配置
+        saved_action_config = config.actions.get(action_type.value, {}).get("config", {})
+        merged_config = {**saved_action_config, **(action_config or {})}
+
+        # 创建操作实例
+        action = architecture.get_action(action_type, merged_config)
+
+        # 执行操作
+        async with connector.get_client() as client:
+            result = await action.execute(client)
+
+        return result
+
+    async def query_balance(
+        self,
+        provider_id: str,
+        config: Optional[Dict[str, Any]] = None,
+    ) -> ActionResult:
+        """
+        查询余额（快捷方法）
+
+        Args:
+            provider_id: Provider ID
+            config: 操作配置
+
+        Returns:
+            操作结果
+        """
+        result = await self.execute_action(
+            provider_id, ProviderActionType.QUERY_BALANCE, config
+        )
+
+        # 成功时更新缓存
+        if result.status == ActionStatus.SUCCESS and result.data:
+            await self._cache_balance(provider_id, result)
+
+        return result
+
+    async def query_balance_with_cache(
+        self,
+        provider_id: str,
+        trigger_refresh: bool = True,
+    ) -> ActionResult:
+        """
+        查询余额（优先返回缓存，可触发异步刷新）
+
+        Args:
+            provider_id: Provider ID
+            trigger_refresh: 是否触发后台异步刷新
+
+        Returns:
+            操作结果（可能是缓存的）
+        """
+        # 尝试从缓存获取
+        cached = await self._get_cached_balance(provider_id)
+
+        if cached:
+            # 有缓存，可选触发后台刷新
+            if trigger_refresh:
+                # 后台任务内部已处理异常并记录日志，无需额外回调
+                asyncio.create_task(self._refresh_balance_async(provider_id))
+            return cached
+
+        # 没有缓存，同步查询一次（首次访问）
+        logger.info(f"余额缓存未命中，同步查询: provider_id={provider_id}")
+        return await self.query_balance(provider_id)
+
+    async def _refresh_balance_async(self, provider_id: str) -> None:
+        """后台异步刷新余额（使用独立的数据库 session）"""
+        try:
+            # 后台任务需要创建独立的 session，因为原请求的 session 可能已关闭
+            with create_session() as db:
+                service = ProviderOpsService(db)
+                await service.query_balance(provider_id)
+        except Exception as e:
+            logger.warning(f"异步刷新余额失败: provider_id={provider_id}, error={e}")
+
+    async def _cache_balance(self, provider_id: str, result: ActionResult) -> None:
+        """缓存余额结果"""
+        cache_key = f"provider_ops:balance:{provider_id}"
+
+        # 序列化 BalanceInfo
+        data = result.data
+        if isinstance(data, BalanceInfo):
+            data = asdict(data)
+
+        cache_data = {
+            "status": result.status.value,
+            "data": data,
+            "executed_at": result.executed_at.isoformat(),
+            "response_time_ms": result.response_time_ms,
+        }
+
+        await CacheService.set(cache_key, cache_data, BALANCE_CACHE_TTL)
+
+    async def _get_cached_balance(self, provider_id: str) -> Optional[ActionResult]:
+        """获取缓存的余额"""
+        cache_key = f"provider_ops:balance:{provider_id}"
+        cached = await CacheService.get(cache_key)
+
+        if not cached:
+            return None
+
+        # 反序列化
+        try:
+            data = cached.get("data")
+            if data and isinstance(data, dict):
+                # 转回 BalanceInfo
+                data = BalanceInfo(
+                    total_granted=data.get("total_granted"),
+                    total_used=data.get("total_used"),
+                    total_available=data.get("total_available"),
+                    currency=data.get("currency", "USD"),
+                    extra=data.get("extra", {}),
+                )
+
+            executed_at_str = cached.get("executed_at")
+            executed_at = (
+                datetime.fromisoformat(executed_at_str)
+                if executed_at_str
+                else datetime.now(timezone.utc)
+            )
+
+            return ActionResult(
+                status=ActionStatus(cached.get("status", "success")),
+                action_type=ProviderActionType.QUERY_BALANCE,
+                data=data,
+                executed_at=executed_at,
+                response_time_ms=cached.get("response_time_ms"),
+                cache_ttl_seconds=BALANCE_CACHE_TTL,
+            )
+        except Exception as e:
+            logger.warning(f"解析缓存余额失败: provider_id={provider_id}, error={e}")
+            return None
+
+    async def checkin(
+        self,
+        provider_id: str,
+        config: Optional[Dict[str, Any]] = None,
+    ) -> ActionResult:
+        """
+        签到（快捷方法）
+
+        Args:
+            provider_id: Provider ID
+            config: 操作配置
+
+        Returns:
+            操作结果
+        """
+        return await self.execute_action(provider_id, ProviderActionType.CHECKIN, config)
+
+    # ==================== 辅助方法 ====================
+
+    def _get_provider(self, provider_id: str) -> Optional[Provider]:
+        """获取 Provider"""
+        return self.db.query(Provider).filter(Provider.id == provider_id).first()
+
+    def _get_provider_base_url(self, provider: Provider) -> Optional[str]:
+        """从 Provider 获取 base_url"""
+        # 优先从第一个 endpoint 获取
+        if provider.endpoints:
+            for endpoint in provider.endpoints:
+                if endpoint.base_url:
+                    return endpoint.base_url
+
+        # 从 config 获取
+        config = provider.config or {}
+        if "base_url" in config:
+            return config["base_url"]
+
+        # 从 website 获取
+        if provider.website:
+            return provider.website
+
+        return None
+
+    def _encrypt_credentials(self, credentials: Dict[str, Any]) -> Dict[str, Any]:
+        """加密凭据中的敏感字段"""
+        encrypted = {}
+        for key, value in credentials.items():
+            if key in self.SENSITIVE_FIELDS and isinstance(value, str):
+                if value:  # 只加密非空值
+                    encrypted[key] = self.crypto.encrypt(value)
+                    logger.debug(f"加密字段 {key}: 原始长度={len(value)}, 加密后长度={len(encrypted[key])}")
+                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
+
+    def _decrypt_credentials(self, credentials: Dict[str, Any]) -> Dict[str, Any]:
+        """解密凭据中的敏感字段"""
+        decrypted = {}
+        for key, value in credentials.items():
+            if key in self.SENSITIVE_FIELDS and isinstance(value, str):
+                try:
+                    decrypted[key] = self.crypto.decrypt(value)
+                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
+
+    def get_masked_credentials(self, credentials: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        获取脱敏后的凭据
+
+        解密凭据并对敏感字段进行脱敏处理（显示部分字符）。
+
+        Args:
+            credentials: 加密的凭据
+
+        Returns:
+            脱敏后的凭据
+        """
+        decrypted = self._decrypt_credentials(credentials)
+
+        for field in self.SENSITIVE_FIELDS:
+            if field in decrypted and decrypted[field]:
+                value = str(decrypted[field])
+                # 显示前4位和后4位，中间固定4个 *（如 sk-x****a12k）
+                if len(value) > 12:
+                    decrypted[field] = value[:4] + "****" + value[-4:]
+                elif len(value) > 8:
+                    decrypted[field] = value[:2] + "****" + value[-2:]
+                else:
+                    decrypted[field] = "*" * len(value)
+
+        return decrypted
+
+    def merge_credentials_with_saved(
+        self,
+        provider_id: str,
+        credentials: Dict[str, Any],
+    ) -> Dict[str, Any]:
+        """
+        合并凭据：如果请求中的敏感字段为空，使用已保存的凭据
+
+        用于验证和保存配置时，当用户未重新输入敏感字段时保留原有值。
+
+        Args:
+            provider_id: Provider ID
+            credentials: 请求中的凭据
+
+        Returns:
+            合并后的凭据
+        """
+        merged = dict(credentials)
+        saved_config = self.get_config(provider_id)
+
+        if saved_config:
+            saved_credentials = self._decrypt_credentials(saved_config.connector_credentials)
+            sensitive_fields = ["api_key", "password", "session_token", "cookie_string", "cookies"]
+
+            for field in sensitive_fields:
+                # 如果请求中该字段为空或只包含星号（脱敏值），使用已保存的值
+                req_value = merged.get(field, "")
+                if not req_value or (isinstance(req_value, str) and set(req_value) <= {"*"}):
+                    if field in saved_credentials:
+                        merged[field] = saved_credentials[field]
+                        logger.debug(f"合并凭据 - 使用已保存的 {field}")
+
+        return merged
+
+    # ==================== 批量操作 ====================
+
+    async def batch_query_balance(
+        self, provider_ids: Optional[List[str]] = None
+    ) -> Dict[str, ActionResult]:
+        """
+        批量查询余额
+
+        Args:
+            provider_ids: Provider ID 列表，None 表示查询所有已配置的
+
+        Returns:
+            {provider_id: result}
+        """
+        if provider_ids is None:
+            # 查询所有已配置的 Provider
+            providers = self.db.query(Provider).filter(Provider.is_active.is_(True)).all()
+            provider_ids = [
+                p.id
+                for p in providers
+                if p.config and p.config.get("provider_ops")
+            ]
+
+        results = {}
+        for provider_id in provider_ids:
+            results[provider_id] = await self.query_balance(provider_id)
+
+        return results
+
+    # ==================== 认证验证 ====================
+
+    async def verify_auth(
+        self,
+        base_url: str,
+        architecture_id: str,
+        auth_type: ConnectorAuthType,
+        config: Dict[str, Any],
+        credentials: Dict[str, Any],
+    ) -> Dict[str, Any]:
+        """
+        验证认证配置
+
+        在保存前测试认证是否有效。
+        认证逻辑委托给对应的 Architecture 实现。
+
+        Args:
+            base_url: API 基础地址
+            architecture_id: 架构 ID
+            auth_type: 认证类型
+            config: 连接器配置
+            credentials: 凭据
+
+        Returns:
+            验证结果
+        """
+        import httpx
+
+        # 移除 base_url 末尾的斜杠
+        base_url = base_url.rstrip("/")
+
+        # 获取架构实例
+        registry = get_registry()
+        architecture = registry.get_or_default(architecture_id)
+
+        # 使用架构的方法构建请求
+        verify_endpoint = f"{base_url}{architecture.get_verify_endpoint()}"
+        headers = architecture.build_verify_headers(config, credentials)
+
+        logger.debug(
+            f"验证认证: architecture={architecture_id}, "
+            f"endpoint={verify_endpoint}, headers={list(headers.keys())}"
+        )
+
+        try:
+            async with httpx.AsyncClient(timeout=30.0) as client:
+                response = await client.get(verify_endpoint, headers=headers)
+
+                # 尝试解析 JSON
+                try:
+                    data = response.json()
+                except Exception:
+                    data = {}
+
+                # 使用架构的方法解析响应
+                result = architecture.parse_verify_response(response.status_code, data)
+                return result.to_dict()
+
+        except httpx.TimeoutException:
+            return {"success": False, "message": "连接超时"}
+        except httpx.ConnectError as e:
+            return {"success": False, "message": f"连接失败: {str(e)}"}
+        except Exception as e:
+            logger.error(f"验证认证失败: {e}")
+            return {"success": False, "message": f"验证失败: {str(e)}"}

src/services/provider_ops/types.py

@@ -0,0 +1,157 @@
+"""
+Provider 操作模块类型定义
+"""
+
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+from enum import Enum
+from typing import Any, Dict, List, Optional
+
+
+class ConnectorAuthType(str, Enum):
+    """连接器认证类型"""
+
+    API_KEY = "api_key"  # API Key 直接认证
+    SESSION_LOGIN = "session_login"  # 用户名密码登录获取 Session
+    OAUTH = "oauth"  # OAuth 流程
+    COOKIE = "cookie"  # 直接使用 Cookie
+    NONE = "none"  # 无需认证
+
+
+class ProviderActionType(str, Enum):
+    """提供商操作类型"""
+
+    QUERY_BALANCE = "query_balance"  # 查询余额
+    CHECKIN = "checkin"  # 签到
+    CLAIM_QUOTA = "claim_quota"  # 领取额度
+    REFRESH_TOKEN = "refresh_token"  # 刷新 Token
+    GET_USAGE = "get_usage"  # 获取使用记录
+    GET_MODELS = "get_models"  # 获取可用模型列表
+    CUSTOM = "custom"  # 自定义操作
+
+
+class ActionStatus(str, Enum):
+    """操作执行状态"""
+
+    SUCCESS = "success"  # 成功
+    AUTH_FAILED = "auth_failed"  # 认证失败
+    AUTH_EXPIRED = "auth_expired"  # 认证过期
+    RATE_LIMITED = "rate_limited"  # 频率限制
+    NETWORK_ERROR = "network_error"  # 网络错误
+    PARSE_ERROR = "parse_error"  # 响应解析错误
+    NOT_CONFIGURED = "not_configured"  # 未配置
+    NOT_SUPPORTED = "not_supported"  # 不支持
+    ALREADY_DONE = "already_done"  # 已完成（如今日已签到）
+    UNKNOWN_ERROR = "unknown_error"  # 未知错误
+
+
+class ConnectorStatus(str, Enum):
+    """连接器状态"""
+
+    DISCONNECTED = "disconnected"  # 未连接
+    CONNECTING = "connecting"  # 连接中
+    CONNECTED = "connected"  # 已连接
+    EXPIRED = "expired"  # 已过期
+    ERROR = "error"  # 错误
+
+
+@dataclass
+class BalanceInfo:
+    """余额信息"""
+
+    total_granted: Optional[float] = None  # 总授予额度
+    total_used: Optional[float] = None  # 已使用额度
+    total_available: Optional[float] = None  # 可用余额
+    expires_at: Optional[datetime] = None  # 过期时间
+    currency: str = "USD"  # 货币单位
+    extra: Dict[str, Any] = field(default_factory=dict)  # 额外信息
+
+
+@dataclass
+class CheckinInfo:
+    """签到信息"""
+
+    reward: Optional[float] = None  # 奖励额度
+    streak_days: Optional[int] = None  # 连续签到天数
+    next_reward: Optional[float] = None  # 下次奖励
+    message: Optional[str] = None  # 签到消息
+    extra: Dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass
+class ActionResult:
+    """操作执行结果"""
+
+    status: ActionStatus
+    action_type: ProviderActionType
+    data: Optional[Any] = None  # 操作返回的数据（如 BalanceInfo, CheckinInfo）
+    message: Optional[str] = None  # 消息
+    executed_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
+    response_time_ms: Optional[int] = None  # 响应时间（毫秒）
+    raw_response: Optional[Dict[str, Any]] = None  # 原始响应（调试用）
+    cache_ttl_seconds: int = 300  # 建议缓存时间
+    retry_after_seconds: Optional[int] = None  # 失败后重试间隔
+
+    @property
+    def is_success(self) -> bool:
+        return self.status == ActionStatus.SUCCESS
+
+
+@dataclass
+class ConnectorState:
+    """连接器状态信息"""
+
+    status: ConnectorStatus
+    auth_type: ConnectorAuthType
+    connected_at: Optional[datetime] = None
+    expires_at: Optional[datetime] = None
+    last_error: Optional[str] = None
+    extra: Dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass
+class ProviderOpsConfig:
+    """Provider 操作配置（存储在 Provider.config['provider_ops'] 中）"""
+
+    architecture_id: str = "generic_api"
+
+    # 连接器配置
+    connector_auth_type: ConnectorAuthType = ConnectorAuthType.API_KEY
+    connector_config: Dict[str, Any] = field(default_factory=dict)
+    connector_credentials: Dict[str, Any] = field(default_factory=dict)  # 加密存储
+
+    # 操作配置
+    actions: Dict[str, Dict[str, Any]] = field(default_factory=dict)
+
+    # 定时任务配置
+    schedule: Dict[str, str] = field(default_factory=dict)  # {action_type: cron_expression}
+
+    @classmethod
+    def from_dict(cls, data: Optional[Dict[str, Any]]) -> "ProviderOpsConfig":
+        """从字典创建配置"""
+        if not data:
+            return cls()
+
+        return cls(
+            architecture_id=data.get("architecture_id", "generic_api"),
+            connector_auth_type=ConnectorAuthType(
+                data.get("connector", {}).get("auth_type", "api_key")
+            ),
+            connector_config=data.get("connector", {}).get("config", {}),
+            connector_credentials=data.get("connector", {}).get("credentials", {}),
+            actions=data.get("actions", {}),
+            schedule=data.get("schedule", {}),
+        )
+
+    def to_dict(self) -> Dict[str, Any]:
+        """转换为字典（用于存储）"""
+        return {
+            "architecture_id": self.architecture_id,
+            "connector": {
+                "auth_type": self.connector_auth_type.value,
+                "config": self.connector_config,
+                "credentials": self.connector_credentials,
+            },
+            "actions": self.actions,
+            "schedule": self.schedule,
+        }