feat: 手动代理节点支持、系统默认代理与跨格式流式 usage 提取

代理节点: - 支持手动添加代理节点(HTTP/HTTPS/SOCKS5)，含地址、认证信息和区域标签 - 新增手动节点的 CRUD API 和前端管理界面 - 提供商代理配置从 URL 字符串迁移至代理节点选择器(proxy_node_id) - 新增系统默认代理节点设置，未单独配置代理的提供商自动回退使用 - 删除节点时自动清除系统默认代理引用并失效缓存 - 健康检查跳过手动节点(无心跳，始终在线) 流式处理: - CLI handler 跨格式转换时委托基类解析 Provider 原始事件的 usage - StreamProcessor 新增 _extract_usage_from_converted_event 从转换后事件补充提取 usage - 支持 Claude/OpenAI/OpenAI Responses/Gemini 多种 usage 格式
2026-09-02 09:20:22 +08:00 · 2026-02-07 14:30:46 +08:00
parent 86dbbe83c2
commit db96c9a46e
30 changed files with 1182 additions and 243 deletions
--- a/src/api/admin/proxy_nodes/routes.py
+++ b/src/api/admin/proxy_nodes/routes.py
@@ -20,20 +20,30 @@ from src.api.base.context import ApiRequestContext
 from src.api.base.pipeline import ApiRequestPipeline
 from src.core.exceptions import InvalidRequestException, NotFoundException
 from src.database import get_db
-from src.models.database import ProxyNode, ProxyNodeStatus
+from src.models.database import ProxyNode, ProxyNodeStatus, SystemConfig

 router = APIRouter(prefix="/api/admin/proxy-nodes", tags=["Admin - Proxy Nodes"])
 pipeline = ApiRequestPipeline()


+def _mask_password(password: str | None) -> str | None:
+    """脱敏密码，仅显示前2位和后2位"""
+    if not password:
+        return None
+    if len(password) <= 4:
+        return "****"
+    return password[:2] + "****" + password[-2:]
+
+
 def _node_to_dict(node: ProxyNode) -> dict[str, Any]:
-    return {
+    d = {
        "id": node.id,
        "name": node.name,
        "ip": node.ip,
        "port": node.port,
        "region": node.region,
        "status": node.status.value if node.status else None,
+        "is_manual": bool(node.is_manual),
        "registered_by": node.registered_by,
        "last_heartbeat_at": node.last_heartbeat_at,
        "heartbeat_interval": node.heartbeat_interval,
@@ -43,6 +53,12 @@ def _node_to_dict(node: ProxyNode) -> dict[str, Any]:
        "created_at": node.created_at,
        "updated_at": node.updated_at,
    }
+    # 手动节点附带代理配置（密码脱敏）
+    if node.is_manual:
+        d["proxy_url"] = node.proxy_url
+        d["proxy_username"] = node.proxy_username
+        d["proxy_password"] = _mask_password(node.proxy_password)
+    return d


 class ProxyNodeRegisterRequest(BaseModel):
@@ -81,6 +97,58 @@ class ProxyNodeUnregisterRequest(BaseModel):
    node_id: str = Field(..., min_length=1, max_length=36, description="节点 ID")


+class ManualProxyNodeCreateRequest(BaseModel):
+    """手动创建代理节点"""
+
+    name: str = Field(..., min_length=1, max_length=100, description="节点名")
+    proxy_url: str = Field(
+        ..., min_length=1, max_length=500, description="代理 URL (http/https/socks5)"
+    )
+    username: str | None = Field(None, max_length=255, description="代理用户名")
+    password: str | None = Field(None, max_length=500, description="代理密码")
+    region: str | None = Field(None, max_length=100, description="区域标签")
+
+    @field_validator("proxy_url")
+    @classmethod
+    def validate_proxy_url(cls, v: str) -> str:
+        import re
+        from urllib.parse import urlparse
+
+        v = v.strip()
+        if not re.match(r"^(http|https|socks5)://", v, re.IGNORECASE):
+            raise ValueError("代理 URL 必须以 http://, https:// 或 socks5:// 开头")
+        parsed = urlparse(v)
+        if not parsed.hostname:
+            raise ValueError("代理 URL 必须包含有效的 host")
+        return v
+
+
+class ManualProxyNodeUpdateRequest(BaseModel):
+    """更新手动代理节点"""
+
+    name: str | None = Field(None, min_length=1, max_length=100, description="节点名")
+    proxy_url: str | None = Field(None, min_length=1, max_length=500, description="代理 URL")
+    username: str | None = Field(None, max_length=255, description="代理用户名")
+    password: str | None = Field(None, max_length=500, description="代理密码")
+    region: str | None = Field(None, max_length=100, description="区域标签")
+
+    @field_validator("proxy_url")
+    @classmethod
+    def validate_proxy_url(cls, v: str | None) -> str | None:
+        if v is None:
+            return None
+        import re
+        from urllib.parse import urlparse
+
+        v = v.strip()
+        if not re.match(r"^(http|https|socks5)://", v, re.IGNORECASE):
+            raise ValueError("代理 URL 必须以 http://, https:// 或 socks5:// 开头")
+        parsed = urlparse(v)
+        if not parsed.hostname:
+            raise ValueError("代理 URL 必须包含有效的 host")
+        return v
+
+
@router.post("/register")
 async def register_proxy_node(request: Request, db: Session = Depends(get_db)) -> Any:
    adapter = AdminRegisterProxyNodeAdapter()
@@ -111,6 +179,20 @@ async def list_proxy_nodes(
    return await pipeline.run(adapter=adapter, http_request=request, db=db, mode=adapter.mode)


+@router.post("/manual")
+async def create_manual_proxy_node(request: Request, db: Session = Depends(get_db)) -> Any:
+    adapter = AdminCreateManualProxyNodeAdapter()
+    return await pipeline.run(adapter=adapter, http_request=request, db=db, mode=adapter.mode)
+
+
+@router.patch("/{node_id}")
+async def update_manual_proxy_node(
+    node_id: str, request: Request, db: Session = Depends(get_db)
+) -> Any:
+    adapter = AdminUpdateManualProxyNodeAdapter(node_id=node_id)
+    return await pipeline.run(adapter=adapter, http_request=request, db=db, mode=adapter.mode)
+
+
@router.delete("/{node_id}")
 async def delete_proxy_node(node_id: str, request: Request, db: Session = Depends(get_db)) -> Any:
    adapter = AdminDeleteProxyNodeAdapter(node_id=node_id)
@@ -298,7 +380,151 @@ class AdminDeleteProxyNodeAdapter(AdminApiAdapter):
            proxy_node_port=node.port,
        )

+        # 若该节点是系统默认代理，自动清除引用
+        was_system_proxy = False
+        sys_cfg = (
+            context.db.query(SystemConfig)
+            .filter(SystemConfig.key == "system_proxy_node_id")
+            .first()
+        )
+        if sys_cfg and sys_cfg.value == self.node_id:
+            sys_cfg.value = None
+            was_system_proxy = True
+
        context.db.delete(node)
        context.db.commit()

-        return {"message": "deleted", "node_id": self.node_id}
+        if was_system_proxy:
+            from src.clients.http_client import invalidate_system_proxy_cache
+
+            invalidate_system_proxy_cache()
+
+        msg = "deleted"
+        if was_system_proxy:
+            msg = "deleted, system default proxy cleared"
+        return {"message": msg, "node_id": self.node_id, "cleared_system_proxy": was_system_proxy}
+
+
+def _parse_host_port(proxy_url: str) -> tuple[str, int]:
+    """从代理 URL 中解析 host 和 port（含协议前缀，避免唯一约束冲突）"""
+    from urllib.parse import urlparse
+
+    parsed = urlparse(proxy_url)
+    host = parsed.hostname or "manual"
+    default_ports = {"https": 443, "socks5": 1080}
+    port = parsed.port or default_ports.get((parsed.scheme or "").lower(), 80)
+    # 添加协议前缀区分同 host:port 不同协议的场景
+    scheme = (parsed.scheme or "http").lower()
+    if scheme != "http":
+        host = f"{scheme}://{host}"
+    return host, port
+
+
+@dataclass
+class AdminCreateManualProxyNodeAdapter(AdminApiAdapter):
+    name: str = "admin_create_manual_proxy_node"
+
+    async def handle(self, context: ApiRequestContext) -> Any:
+        payload = context.ensure_json_body()
+        try:
+            req = ManualProxyNodeCreateRequest.model_validate(payload)
+        except ValidationError as exc:
+            raise InvalidRequestException("输入验证失败: " + _format_validation_error(exc))
+
+        host, port = _parse_host_port(req.proxy_url)
+        now = datetime.now(timezone.utc)
+
+        node = ProxyNode(
+            id=str(uuid.uuid4()),
+            name=req.name,
+            ip=host,
+            port=port,
+            region=req.region,
+            is_manual=True,
+            proxy_url=req.proxy_url,
+            proxy_username=req.username,
+            proxy_password=req.password,
+            status=ProxyNodeStatus.ONLINE,
+            registered_by=context.user.id if context.user else None,
+            last_heartbeat_at=None,
+            heartbeat_interval=0,
+            active_connections=0,
+            total_requests=0,
+            avg_latency_ms=None,
+            created_at=now,
+            updated_at=now,
+        )
+        # 检查是否已存在同地址的节点
+        existing = (
+            context.db.query(ProxyNode).filter(ProxyNode.ip == host, ProxyNode.port == port).first()
+        )
+        if existing:
+            raise InvalidRequestException(
+                f"已存在相同地址的代理节点: {existing.name} ({existing.ip}:{existing.port})"
+            )
+
+        context.db.add(node)
+        context.db.commit()
+        context.db.refresh(node)
+
+        context.add_audit_metadata(
+            action="proxy_node_manual_create",
+            proxy_node_id=node.id,
+        )
+
+        return {"node_id": node.id, "node": _node_to_dict(node)}
+
+
+@dataclass
+class AdminUpdateManualProxyNodeAdapter(AdminApiAdapter):
+    name: str = "admin_update_manual_proxy_node"
+    node_id: str = ""
+
+    async def handle(self, context: ApiRequestContext) -> Any:
+        node = context.db.query(ProxyNode).filter(ProxyNode.id == self.node_id).first()
+        if not node:
+            raise NotFoundException(f"ProxyNode {self.node_id} 不存在", "proxy_node")
+        if not node.is_manual:
+            raise InvalidRequestException("只能编辑手动添加的代理节点")
+
+        payload = context.ensure_json_body()
+        try:
+            req = ManualProxyNodeUpdateRequest.model_validate(payload)
+        except ValidationError as exc:
+            raise InvalidRequestException("输入验证失败: " + _format_validation_error(exc))
+
+        if req.name is not None:
+            node.name = req.name
+        if req.proxy_url is not None:
+            node.proxy_url = req.proxy_url
+            host, port = _parse_host_port(req.proxy_url)
+            # 检查新地址是否与其他节点冲突
+            existing = (
+                context.db.query(ProxyNode)
+                .filter(ProxyNode.ip == host, ProxyNode.port == port, ProxyNode.id != node.id)
+                .first()
+            )
+            if existing:
+                raise InvalidRequestException(
+                    f"已存在相同地址的代理节点: {existing.name} ({existing.ip}:{existing.port})"
+                )
+            node.ip = host
+            node.port = port
+        if req.username is not None:
+            node.proxy_username = req.username
+        # password: None=不发送(保留原值), ""=清空, 非空=更新
+        if req.password is not None:
+            node.proxy_password = req.password or None
+        if req.region is not None:
+            node.region = req.region
+
+        node.updated_at = datetime.now(timezone.utc)
+        context.db.commit()
+        context.db.refresh(node)
+
+        context.add_audit_metadata(
+            action="proxy_node_manual_update",
+            proxy_node_id=node.id,
+        )
+
+        return {"node_id": node.id, "node": _node_to_dict(node)}
--- a/src/api/handlers/base/stream_processor.py
+++ b/src/api/handlers/base/stream_processor.py
@@ -276,6 +276,73 @@ class StreamProcessor:
            if finish_reason is not None:
                ctx.has_completion = True

+    def _extract_usage_from_converted_event(
+        self,
+        ctx: StreamContext,
+        evt: dict[str, Any],
+        event_type: str,
+    ) -> None:
+        """
+        从转换后的事件中提取 usage 信息（补充 Provider 事件解析）。
+
+        支持多种格式：
+        - Claude: message_delta.usage, message_start.message.usage
+        - OpenAI: chunk.usage / response.completed.response.usage
+        - Gemini: usageMetadata
+        """
+        usage: dict[str, Any] | None = None
+
+        # Claude 格式: message_delta 或 message_start
+        if event_type == "message_delta":
+            usage = evt.get("usage")
+        elif event_type == "message_start":
+            message = evt.get("message", {})
+            if isinstance(message, dict):
+                usage = message.get("usage")
+        # OpenAI Responses API 格式: response.completed 中 usage 嵌套在 response 对象内
+        elif event_type == "response.completed":
+            resp_obj = evt.get("response")
+            if isinstance(resp_obj, dict):
+                usage = resp_obj.get("usage")
+            # 兼容: 部分实现可能在顶层也有 usage
+            if not usage:
+                usage = evt.get("usage")
+        # OpenAI Chat 格式: 直接在 chunk 中
+        elif "usage" in evt:
+            usage = evt.get("usage")
+        # Gemini 格式: usageMetadata
+        elif "usageMetadata" in evt:
+            meta = evt.get("usageMetadata", {})
+            if isinstance(meta, dict):
+                usage = {
+                    "input_tokens": meta.get("promptTokenCount", 0),
+                    "output_tokens": meta.get("candidatesTokenCount", 0),
+                    "cache_read_tokens": meta.get("cachedContentTokenCount", 0),
+                    "cache_creation_tokens": 0,
+                }
+
+        if usage and isinstance(usage, dict):
+            new_input = usage.get("input_tokens", 0) or 0
+            new_output = usage.get("output_tokens", 0) or 0
+            new_cached = usage.get("cache_read_tokens") or usage.get("cache_read_input_tokens") or 0
+            new_cache_creation = (
+                usage.get("cache_creation_tokens") or usage.get("cache_creation_input_tokens") or 0
+            )
+
+            if new_input > ctx.input_tokens:
+                ctx.input_tokens = new_input
+                logger.debug("[{}] 从转换后事件更新 input_tokens: {}", self.request_id, new_input)
+            if new_output > ctx.output_tokens:
+                ctx.output_tokens = new_output
+                logger.debug("[{}] 从转换后事件更新 output_tokens: {}", self.request_id, new_output)
+            if new_cached > ctx.cached_tokens:
+                ctx.cached_tokens = new_cached
+            if new_cache_creation > ctx.cache_creation_tokens:
+                ctx.cache_creation_tokens = new_cache_creation
+
+            if any([new_input, new_output, new_cached, new_cache_creation]):
+                ctx.final_usage = usage
+
    async def prefetch_and_check_error(
        self,
        byte_iterator: Any,
@@ -748,6 +815,18 @@ class StreamProcessor:
                            ctx.data_count += 1
                            if ctx.record_parsed_chunks:
                                ctx.parsed_chunks.append(evt)
+                            event_type = evt.get("type", "")
+                            if event_type in ("message_stop", "response.completed"):
+                                ctx.has_completion = True
+                            elif "choices" in evt:
+                                choices = evt.get("choices", [])
+                                for choice in choices:
+                                    if isinstance(choice, dict) and choice.get("finish_reason"):
+                                        ctx.has_completion = True
+                                        break
+
+                            # 从转换后的事件中补充 usage 信息
+                            self._extract_usage_from_converted_event(ctx, evt, event_type)

                        # 统一使用 SSE 格式输出（Gemini streamGenerateContent 也使用 SSE）
                        # 参考: https://ai.google.dev/api/generate-content
--- a/src/api/handlers/claude_cli/handler.py
+++ b/src/api/handlers/claude_cli/handler.py
@@ -87,7 +87,18 @@ class ClaudeCliMessageHandler(CliMessageHandlerBase):
        - content_block_delta: 文本增量
        - message_delta: 消息增量，包含最终 usage
        - message_stop: 消息结束
+
+        跨格式转换时（如 provider=openai:cli），原始事件数据是 Provider 格式而非 Claude 格式。
+        此时委托基类方法通过 Provider 格式解析器提取 usage。
        """
+        # 跨格式转换时：原始事件是 Provider 格式，
+        # 基类 _process_event_data 会自动选择正确的 Provider 解析器提取 usage/text
+        if ctx.provider_api_format and ctx.provider_api_format != ctx.client_api_format:
+            super()._process_event_data(ctx, event_type, data)
+            return
+
+        # 以下是同格式（claude:cli / claude:chat）的处理逻辑
+
        # 处理 message_start 事件
        if event_type == "message_start":
            message = data.get("message", {})
--- a/src/api/handlers/gemini_cli/handler.py
+++ b/src/api/handlers/gemini_cli/handler.py
@@ -208,7 +208,18 @@ class GeminiCliMessageHandler(CliMessageHandlerBase):

        注意: Gemini 流解析器会将每个 JSON 对象作为一个"事件"传递
        event_type 在这里可能为空或是自定义的标记
+
+        跨格式转换时（如 provider=claude:chat），原始事件数据是 Provider 格式而非 Gemini 格式。
+        此时委托基类方法通过 Provider 格式解析器提取 usage。
        """
+        # 跨格式转换时：原始事件是 Provider 格式，
+        # 基类 _process_event_data 会自动选择正确的 Provider 解析器提取 usage/text
+        if ctx.provider_api_format and ctx.provider_api_format != ctx.client_api_format:
+            super()._process_event_data(ctx, _event_type, data)
+            return
+
+        # 以下是同格式（gemini:cli / gemini:chat）的处理逻辑
+
        # 提取候选响应
        candidates = data.get("candidates", [])
        if candidates:
--- a/src/api/handlers/openai_cli/handler.py
+++ b/src/api/handlers/openai_cli/handler.py
@@ -84,7 +84,18 @@ class OpenAICliMessageHandler(CliMessageHandlerBase):
        事件类型：
        - response.output_text.delta: 文本增量
        - response.completed: 响应完成（包含 usage）
+
+        跨格式转换时（如 provider=claude:chat），原始事件数据是 Provider 格式而非 OpenAI CLI 格式。
+        此时先调用基类方法通过 Provider 格式解析器提取 usage，再执行 OpenAI CLI 特定的处理逻辑。
        """
+        # 跨格式转换时：原始事件是 Provider 格式（如 Claude），
+        # 基类 _process_event_data 会自动选择正确的 Provider 解析器提取 usage/text
+        if ctx.provider_api_format and ctx.provider_api_format != ctx.client_api_format:
+            super()._process_event_data(ctx, event_type, data)
+            return
+
+        # 以下是同格式（openai:cli）的处理逻辑
+
        # 提取 response_id
        if not ctx.response_id:
            response_obj = data.get("response")