"""使用 Python 标准库调用 Admin 登记接口。""" import json import socket from http.client import RemoteDisconnected from typing import Callable, Optional from urllib.error import HTTPError, URLError from urllib.parse import urlparse from urllib.request import ProxyHandler, Request, build_opener from uuid import uuid4 from .admin_gateway import ( AdminGatewayError, ClaimCapabilities, ClientInfo, ClientRegistrationGateway, RegistrationReceipt, ) DEFAULT_ADMIN_BASE_URL = "http://127.0.0.1:8080" class HttpAdminGateway(ClientRegistrationGateway): """通过 HTTP 登记 Client;访问令牌只保存在内存。""" def __init__( self, base_url: str = DEFAULT_ADMIN_BASE_URL, token: str = "", timeout_seconds: float = 3.0, opener: Optional[Callable] = None, ): normalized_url = base_url.strip().rstrip("/") parsed = urlparse(normalized_url) if parsed.scheme not in {"http", "https"} or not parsed.netloc: raise ValueError("Admin 服务地址必须是有效的 http 或 https 地址") if timeout_seconds <= 0: raise ValueError("请求超时必须大于 0 秒") self._base_url = normalized_url self._token = token.strip() self._timeout_seconds = timeout_seconds # Admin 通常运行在本机或局域网。明确禁用环境代理,避免 # HTTP_PROXY 把 127.0.0.1 请求错误转发到代理服务器。 self._opener = opener or build_opener(ProxyHandler({})).open def register_client( self, client: ClientInfo, capabilities: ClaimCapabilities ) -> RegistrationReceipt: """调用独立登记接口,不领取或修改任务。""" request_id = str(uuid4()) payload = { "client": {"name": client.name.strip()}, "supported_types": [ task_type.value for task_type in capabilities.supported_types ], "capabilities": { "purchase_mode": capabilities.purchase_mode, "schema_versions": list(capabilities.schema_versions), }, } if capabilities.device is not None: payload["device"] = { "address": capabilities.device.address.strip(), "platform": capabilities.device.platform, "pdd_package": capabilities.device.pdd_package.strip(), } headers = { "Content-Type": "application/json; charset=utf-8", "Accept": "application/json", "X-Client-Id": client.client_id.strip(), "X-Request-Id": request_id, } if self._token: headers["Authorization"] = f"Bearer {self._token}" request = Request( f"{self._base_url}/api/v1/client/registration", data=json.dumps(payload, ensure_ascii=False).encode("utf-8"), headers=headers, method="PUT", ) try: with self._opener(request, timeout=self._timeout_seconds) as response: status = getattr(response, "status", None) if status is None: status = response.getcode() body = response.read() except HTTPError as exc: self._raise_http_error(exc, request_id) except ( URLError, RemoteDisconnected, ConnectionError, socket.timeout, TimeoutError, ) as exc: reason = getattr(exc, "reason", exc) if isinstance(reason, (socket.timeout, TimeoutError)): raise AdminGatewayError( "ADMIN_TIMEOUT", "Admin 请求超时,请稍后重试", True, request_id ) from exc raise AdminGatewayError( "ADMIN_UNAVAILABLE", "无法连接 Admin,请检查服务地址", True, request_id ) from exc if status != 200: raise AdminGatewayError( "ADMIN_UNEXPECTED_RESPONSE", f"Admin 返回了未预期的状态码 {status}", status >= 500, request_id, ) data = self._decode_json(body, request_id) if ( data.get("registered") is not True or data.get("client_id") != client.client_id.strip() or not isinstance(data.get("registered_at"), str) or not data["registered_at"].strip() ): raise AdminGatewayError( "ADMIN_INVALID_RESPONSE", "Admin 登记响应缺少必要字段", False, request_id, ) return RegistrationReceipt( registered=True, client_id=data["client_id"], registered_at=data["registered_at"], ) @staticmethod def _decode_json(body: bytes, request_id: str) -> dict: try: data = json.loads(body.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise AdminGatewayError( "ADMIN_INVALID_RESPONSE", "Admin 返回的不是有效 JSON", False, request_id, ) from exc if not isinstance(data, dict): raise AdminGatewayError( "ADMIN_INVALID_RESPONSE", "Admin 返回的 JSON 不是对象", False, request_id, ) return data @classmethod def _raise_http_error(cls, error: HTTPError, request_id: str) -> None: try: body = error.read() except OSError: body = b"" try: data = cls._decode_json(body, request_id) details = data.get("error", {}) if not isinstance(details, dict): raise ValueError code = str(details.get("code") or "ADMIN_HTTP_ERROR") message = str(details.get("message") or "Admin 拒绝了登记请求") retryable = bool(details.get("retryable", error.code >= 500)) response_request_id = str(details.get("request_id") or request_id) extra = details.get("details") if not isinstance(extra, dict): extra = {} except (AdminGatewayError, ValueError): code = "ADMIN_HTTP_ERROR" message = f"Admin 登记失败,状态码 {error.code}" retryable = error.code >= 500 response_request_id = request_id extra = {} raise AdminGatewayError( code, message, retryable, response_request_id, extra, ) from error