cmautobuy/client/src/http_admin_gateway.py

"""使用 Python 标准库调用 Admin 登记、领取和提交接口。"""

import hashlib
import json
import re
import socket
from http.client import RemoteDisconnected
from typing import Any, Callable, Mapping, 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 (
    AdminGateway,
    AdminTask,
    AdminGatewayError,
    ClaimCapabilities,
    ClientInfo,
    RegistrationReceipt,
    SpecResolutionMatch,
    SpecResolutionReceipt,
    SubmissionReceipt,
)
from .task_models import TaskType


DEFAULT_ADMIN_BASE_URL = "https://buy.833729.com"
PURCHASE_SPEC_RESOLUTION_TIMEOUT_SECONDS = 132.0


class HttpAdminGateway(AdminGateway):
    """通过 HTTP 登记、领取和提交任务；令牌只保存在内存。"""

    def __init__(
        self,
        base_url: str = DEFAULT_ADMIN_BASE_URL,
        token: str = "",
        timeout_seconds: float = 3.0,
        opener: Optional[Callable] = None,
        client_id: str = "",
    ):
        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 (
            parsed.username is not None
            or parsed.password is not None
            or parsed.query
            or parsed.fragment
        ):
            raise ValueError(
                "Admin 服务地址不能包含账号、密码、查询参数或片段"
            )
        if timeout_seconds <= 0:
            raise ValueError("请求超时必须大于 0 秒")

        self._base_url = normalized_url
        self._token = token.strip()
        self._timeout_seconds = timeout_seconds
        self._client_id = client_id.strip()
        # 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())
        self._client_id = client.client_id.strip()
        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"],
        )

    def claim_next(
        self,
        client: ClientInfo,
        capabilities: ClaimCapabilities,
    ) -> Optional[AdminTask]:
        """按调用方已安全确认的能力领取一个任务。"""

        request_id = str(uuid4())
        self._client_id = client.client_id.strip()
        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/tasks/claim",
            data=json.dumps(payload, ensure_ascii=False).encode("utf-8"),
            headers=headers,
            method="POST",
        )

        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 == 204:
            return None
        if status != 200:
            raise AdminGatewayError(
                "ADMIN_UNEXPECTED_RESPONSE",
                f"Admin 返回了未预期的状态码 {status}",
                status >= 500,
                request_id,
            )
        return self._parse_claim_response(body, request_id)

    def submit_result(
        self,
        task_id: str,
        idempotency_key: str,
        result: Mapping[str, Any],
    ) -> SubmissionReceipt:
        """幂等提交成功结果。"""

        return self._submit(task_id, idempotency_key, result, "result")

    def submit_failure(
        self,
        task_id: str,
        idempotency_key: str,
        failure: Mapping[str, Any],
    ) -> SubmissionReceipt:
        """幂等提交失败或人工处理结果。"""

        return self._submit(task_id, idempotency_key, failure, "failure")

    def resolve_purchase_spec(
        self,
        task_id: str,
        idempotency_key: str,
        observation: Mapping[str, Any],
    ) -> SpecResolutionReceipt:
        """发送一次已持久化的规格解析命令并严格校验响应。"""

        if not task_id.strip() or not idempotency_key.strip():
            raise ValueError("规格解析的 task_id 和 Idempotency-Key 不能为空")
        if not self._client_id:
            raise AdminGatewayError(
                "CLIENT_ID_MISSING", "规格解析前必须配置 Client 设备号", False
            )
        request_id = str(uuid4())
        headers = {
            "Content-Type": "application/json; charset=utf-8",
            "Accept": "application/json",
            "Idempotency-Key": idempotency_key.strip(),
            "X-Request-Id": request_id,
            "X-Client-Id": self._client_id,
        }
        if self._token:
            headers["Authorization"] = f"Bearer {self._token}"
        request = Request(
            f"{self._base_url}/api/v1/client/tasks/{task_id.strip()}/spec-resolution",
            data=json.dumps(observation, ensure_ascii=False).encode("utf-8"),
            headers=headers,
            method="POST",
        )
        try:
            with self._opener(
                request,
                timeout=PURCHASE_SPEC_RESOLUTION_TIMEOUT_SECONDS,
            ) as response:
                status = getattr(response, "status", None) or 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)
            code = (
                "ADMIN_TIMEOUT"
                if isinstance(reason, (socket.timeout, TimeoutError))
                else "ADMIN_UNAVAILABLE"
            )
            raise AdminGatewayError(
                code,
                "Admin 规格解析超时" if code == "ADMIN_TIMEOUT" else "无法连接 Admin 规格解析服务",
                True,
                request_id,
            ) from exc
        if status != 200:
            raise AdminGatewayError(
                "ADMIN_UNEXPECTED_RESPONSE",
                f"Admin 规格解析返回了未预期的状态码 {status}",
                status >= 500,
                request_id,
            )
        return self._parse_spec_resolution_response(body, request_id)

    @classmethod
    def _parse_spec_resolution_response(
        cls, body: bytes, request_id: str
    ) -> SpecResolutionReceipt:
        data = cls._decode_json(body, request_id)
        schema_version = data.get("schema_version")
        resolution_id = data.get("resolution_id")
        outcome = data.get("outcome")
        source = data.get("source")
        snapshot_hash = data.get("candidate_snapshot_hash")
        confidence = data.get("confidence_bps")
        reason = data.get("reason")
        resolved_at = data.get("resolved_at")
        match_data = data.get("match")
        valid = (
            schema_version == 1
            and isinstance(resolution_id, str)
            and bool(resolution_id.strip())
            and outcome in {"matched", "uncertain", "rejected", "failed"}
            and (source is None or source in {"rule", "ai", "reused"})
            and isinstance(snapshot_hash, str)
            and re.fullmatch(r"[0-9a-f]{64}", snapshot_hash) is not None
            and (confidence is None or (
                isinstance(confidence, int)
                and not isinstance(confidence, bool)
                and 0 <= confidence <= 10000
            ))
            and isinstance(reason, str)
            and len(reason) <= 500
            and isinstance(resolved_at, str)
            and bool(resolved_at.strip())
        )
        match = None
        if outcome == "matched" and isinstance(match_data, Mapping):
            candidate_id = match_data.get("candidate_id")
            raw_text = match_data.get("raw_text")
            options = match_data.get("options")
            if (
                isinstance(candidate_id, str)
                and bool(candidate_id.strip())
                and isinstance(raw_text, str)
                and bool(raw_text)
                and isinstance(options, Mapping)
                and set(options) == {"color", "size"}
                and all(
                    isinstance(value, str) and bool(value)
                    for value in options.values()
                )
                and options.get("size") == raw_text
            ):
                match = SpecResolutionMatch(
                    candidate_id,
                    raw_text,
                    {str(key): str(value) for key, value in options.items()},
                )
        if (
            not valid
            or (outcome == "matched") != (match is not None)
            or (outcome != "matched" and match_data is not None)
        ):
            raise AdminGatewayError(
                "ADMIN_INVALID_RESPONSE",
                "Admin 规格解析响应字段不完整或不合法",
                False,
                request_id,
            )
        return SpecResolutionReceipt(
            1,
            resolution_id,
            outcome,
            source,
            snapshot_hash,
            match,
            confidence,
            reason,
            resolved_at,
        )

    def _submit(
        self,
        task_id: str,
        idempotency_key: str,
        payload: Mapping[str, Any],
        endpoint: str,
    ) -> SubmissionReceipt:
        if not task_id.strip():
            raise ValueError("task_id 不能为空")
        if not idempotency_key.strip():
            raise ValueError("Idempotency-Key 不能为空")
        if not self._client_id:
            raise AdminGatewayError(
                "CLIENT_ID_MISSING",
                "提交结果前必须先配置 Client 设备号",
                False,
            )
        request_id = str(uuid4())
        headers = {
            "Content-Type": "application/json; charset=utf-8",
            "Accept": "application/json",
            "Idempotency-Key": idempotency_key.strip(),
            "X-Request-Id": request_id,
            "X-Client-Id": self._client_id,
        }
        if self._token:
            headers["Authorization"] = f"Bearer {self._token}"
        request = Request(
            f"{self._base_url}/api/v1/client/tasks/{task_id.strip()}/{endpoint}",
            data=json.dumps(payload, ensure_ascii=False).encode("utf-8"),
            headers=headers,
            method="POST",
        )
        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)
            code = "ADMIN_TIMEOUT" if isinstance(reason, (socket.timeout, TimeoutError)) else "ADMIN_UNAVAILABLE"
            message = "Admin 提交请求超时，请稍后重试" if code == "ADMIN_TIMEOUT" else "无法连接 Admin，结果已保存在本地"
            raise AdminGatewayError(code, message, True, request_id) from exc
        if status not in (200, 201):
            raise AdminGatewayError(
                "ADMIN_UNEXPECTED_RESPONSE",
                f"Admin 返回了未预期的状态码 {status}",
                status >= 500,
                request_id,
            )
        data = self._decode_json(body, request_id)
        accepted = data.get("accepted")
        result_id = data.get("result_id")
        accepted_at = data.get("accepted_at")
        legacy_failure = (
            endpoint == "failure"
            and accepted is True
            and isinstance(accepted_at, str)
            and bool(accepted_at.strip())
            and isinstance(data.get("task_status"), str)
            and bool(data["task_status"].strip())
            and not (isinstance(result_id, str) and result_id.strip())
        )
        if legacy_failure:
            digest = hashlib.sha256(
                idempotency_key.encode("utf-8")
            ).hexdigest()[:16]
            return SubmissionReceipt(
                True,
                f"legacy-failure-{digest}",
                accepted_at,
            )
        if (
            accepted is not True
            or not isinstance(result_id, str)
            or not result_id.strip()
            or not isinstance(accepted_at, str)
            or not accepted_at.strip()
        ):
            raise AdminGatewayError(
                "ADMIN_INVALID_RESPONSE",
                "Admin 返回成功状态，但响应格式异常，数据可能已接收，将安全重试",
                True,
                request_id,
            )
        return SubmissionReceipt(True, result_id, accepted_at)

    @classmethod
    def _parse_claim_response(
        cls,
        body: bytes,
        request_id: str,
    ) -> AdminTask:
        data = cls._decode_json(body, request_id)
        task = data.get("task")
        if not isinstance(task, Mapping):
            raise AdminGatewayError(
                "ADMIN_INVALID_RESPONSE",
                "Admin 领取响应缺少 task 对象",
                False,
                request_id,
            )

        task_id = task.get("id")
        raw_type = task.get("type")
        version = task.get("version")
        priority = task.get("priority")
        execution_mode = task.get("execution_mode", "dry_run")
        task_payload = task.get("payload")
        created_at = task.get("created_at")
        updated_at = task.get("updated_at")
        valid = (
            isinstance(task_id, str)
            and bool(task_id.strip())
            and isinstance(raw_type, str)
            and isinstance(version, int)
            and not isinstance(version, bool)
            and version > 0
            and isinstance(priority, int)
            and not isinstance(priority, bool)
            and isinstance(execution_mode, str)
            and isinstance(task_payload, Mapping)
            and isinstance(created_at, str)
            and isinstance(updated_at, str)
        )
        if not valid:
            raise AdminGatewayError(
                "ADMIN_INVALID_RESPONSE",
                "Admin 领取响应的任务字段不完整",
                False,
                request_id,
            )

        try:
            task_type = TaskType(raw_type)
        except ValueError as exc:
            raise AdminGatewayError(
                "ADMIN_INVALID_RESPONSE",
                f"Admin 返回了不支持的任务类型 {raw_type}",
                False,
                request_id,
            ) from exc

        if execution_mode not in {"dry_run", "live"} or (
            task_type is not TaskType.PURCHASE and execution_mode != "dry_run"
        ):
            raise AdminGatewayError(
                "ADMIN_INVALID_RESPONSE",
                "Admin 返回了不支持的任务执行模式",
                False,
                request_id,
            )

        cls._validate_claim_payload(task_type, task_payload, request_id)

        return AdminTask(
            task_id=task_id,
            task_type=task_type,
            version=version,
            priority=priority,
            execution_mode=execution_mode,
            payload=dict(task_payload),
            created_at=created_at,
            updated_at=updated_at,
        )

    @staticmethod
    def _validate_claim_payload(
        task_type: TaskType,
        payload: Mapping[str, object],
        request_id: str,
    ) -> None:
        """在采购任务进入本地执行前校验全部安全字段。"""

        goods_url = payload.get("goods_url")
        if not isinstance(goods_url, str) or not goods_url.strip():
            raise AdminGatewayError(
                "ADMIN_INVALID_RESPONSE",
                "Admin 任务 payload.goods_url 不能为空",
                False,
                request_id,
            )
        if task_type is not TaskType.PURCHASE:
            return
        goods_id = payload.get("goods_id")
        options = payload.get("options")
        quantity = payload.get("quantity")
        max_price_cent = payload.get("max_price_cent")
        valid_options = (
            isinstance(options, Mapping)
            and bool(options)
            and all(
                isinstance(key, str)
                and bool(key.strip())
                and isinstance(value, str)
                and bool(value.strip())
                for key, value in options.items()
            )
        )
        valid = (
            isinstance(goods_id, str)
            and bool(goods_id.strip())
            and valid_options
            and isinstance(quantity, int)
            and not isinstance(quantity, bool)
            and quantity > 0
            and isinstance(max_price_cent, int)
            and not isinstance(max_price_cent, bool)
            and max_price_cent > 0
        )
        if not valid:
            raise AdminGatewayError(
                "ADMIN_INVALID_RESPONSE",
                (
                    "Admin 采购任务缺少有效的 goods_id、动态 options、"
                    "quantity 或订单总价上限 max_price_cent"
                ),
                False,
                request_id,
            )

    @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,
        operation: 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 f"Admin 拒绝了{operation}请求")
            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 {operation}失败，状态码 {error.code}"
            retryable = error.code >= 500
            response_request_id = request_id
            extra = {}

        raise AdminGatewayError(
            code,
            message,
            retryable,
            response_request_id,
            extra,
        ) from error