# -*- coding: utf-8 -*-
"""品源协议一致性测试套件（conformance suite）· 纯标准库 / Python 3.6+ / 零第三方依赖。

这是什么
========
一套**黑盒**检查：只通过 HTTP 打 ``/api/v1`` 的公开面，逐条核对协议自己的承诺 ——
自描述端点、鉴权、信封共同核心键、分页游标、错误模型、写口幂等、跨商家聚合、
**ACT 目录口（`CID-MER-CAT`，G 组）** ——
每条给 PASS / FAIL / SKIP + 一句人话，最后给汇总与**非零退出码**。

"协议"不是文档里的形容词，而是能被机器否证的一组断言。这份套件就是那组断言。

谁该用它
========
* **第三方实现方**：声称"兼容品源协议"之前先自己跑一遍；红了就别对外说兼容。
* **品源自己**：出库口一改就当回归门禁跑（本地 dev 在 http://127.0.0.1:8972）。
* **验收方**：拿它当"这份实现到底像不像"的机器判据，而不是靠读对方的对接说明。

它**不**读数据库、**不** import 品源代码 —— 换一套语言重写的品源照样能测。

怎么跑
======
::

    # 1) 本地只读（默认 base 就是本地 dev；默认跑法不写任何数据）
    python3 tools/conformance.py --base http://127.0.0.1:8972 --key <渠道密钥>

    # 2) 指定远端（base 不是 localhost/127.0.0.1 时必须显式加 --i-know-its-remote）
    python3 tools/conformance.py --base https://pinyuan.sxai77.cn --key <密钥> --i-know-its-remote

    # 3) 带写口（会真的扣库存；必须 --write + 指定测试用 store/uid）
    python3 tools/conformance.py --base http://127.0.0.1:8972 --key <密钥> \\
        --write --store st0001 --uid <商品 uid>

退出码：``0`` = 无 FAIL；``1`` = 有 FAIL；``2`` = 被用法/环境拒（例如远端没加确认旗标）。

三条安全纪律（写进代码，不靠自觉）
==================================
1. **默认只读**：``POST /channel/stock_report`` 只在 ``--write`` 下才发；否则 E 组直接 SKIP。
2. **默认不连生产**：base 的 host 不是 localhost/127.0.0.1 时先打醒目警告，
   没有 ``--i-know-its-remote`` 就退出码 2 收工。
3. **不引入第二份事实源**：字段词汇表（37 个字段名）**不硬编码**，一律从 ``/schema`` 现读；
   本文件硬编码的只有**协议形状**（信封共同核心键、列表键名、三档门槛的取值集合、硬上限 500）——
   这些正是被检验的契约本身，不写在这里就没有"标准"可言。
4. **限流不算不合规**（R-95，2026-10-03）：实现方给自己的口设额度是**正当**的，
   套件密集打撞上 429 不能判 FAIL —— 那会把合规实现判红，而红线正是这份套件的产出。
   撞上 429 先按 body 的 ``retry_after`` 有界重试（``Suite.RETRY_429_TIMES`` /
   ``RETRY_429_MAX_WAIT``）；仍是 429 就按"**这一条没测到**"记 SKIP，既不是 FAIL 也不是 PASS。
   （教训来自 F1 的一条假 FAIL：真因是我们自己的聚合口额度只有 6 次/分。）

可测性
======
传输层可注入，签名与 ``tools/pyclient.py`` 一致：
``transport(method, url, headers, body_bytes) -> (status, bytes)``。
注意传输层**只回状态码与字节**（不回响应头），所以"是不是 HTML 错误页"只能由响应体判定；
这也让套件可以脱离网络、用假传输层做单元测试。
"""

import argparse
import datetime
import json
import os
import sys
import time
import urllib.error
import urllib.parse
import ssl
import urllib.request

# ── 协议常量：被检验的契约本身 ────────────────────────────────────────────────
API_PREFIX = "/api/v1"
PATH_SCHEMA = API_PREFIX + "/schema"
PATH_PRODUCTS = API_PREFIX + "/channel/products"
PATH_STORES = API_PREFIX + "/channel/stores"
PATH_MOVES = API_PREFIX + "/channel/moves"
PATH_FIELD_RULES = API_PREFIX + "/channel/field-rules"
PATH_STOCK_REPORT = API_PREFIX + "/channel/stock_report"
PATH_AGGREGATE = API_PREFIX + "/channel/aggregate"
PATH_CATALOG = API_PREFIX + "/channel/catalog"
PATH_PACKS = API_PREFIX + "/channel/packs"
#: 能力发现口**不在 `/api/v1` 下** —— 这是**有意的**（RFC 8615 的约定位置）。
#: 买方在还不知道我们的 API 长什么样时，第一件事就是探这个路径。
PATH_WELLKNOWN = "/.well-known/rec-agent"
PATH_NOPE = API_PREFIX + "/channel/__nope__"

DEFAULT_BASE = "http://127.0.0.1:8972"
LOCAL_HOSTS = ("127.0.0.1", "localhost", "::1")
REMOTE_FLAG = "--i-know-its-remote"

# 商品信封的**共同核心键**（子集）。
#
# ⚠️ 为什么叫"核心"而不是"七键"：**各端点键面并不统一** ——
#   /channel/products 顶层 8 键（多一个 stores=店码数组）
#   /channel/stores    只有 6 键（stores/total/filters_applied/generated_at/next_cursor/schema_version）
#   /channel/moves     只有 4 键（store/generated_at/items/next_since）
# 逐端点真实键面以 docs/PROTOCOL.md §2.1 为准。
#
# 本常量只用于**子集检查**（"这些键必须在"），不做精确相等：
# 协议对消费方的要求是**容忍未知键**（COMPATIBILITY.md §2 第 1 条），
# 所以子集断言既不会漏掉"核心键被删"，也不会因为品源合法的加法变更而误报。
#
# 另：**不把 /schema 的 v5.envelope_keys 当权威** —— 那是 v5 的历史沿革记录，
# 今天读起来像现状（见 server/api/views/common.py 该条的措辞已标注）。
ENVELOPE_CORE_KEYS = ("schema_version", "generated_at", "store", "filters_applied",
                      "excluded", "items", "next_cursor")
# 三档门槛的合法取值（"必填/推荐/可选"）。
FIELD_LEVELS = ("required", "recommended", "optional")
HARD_PAGE_MAX = 500          # 服务端硬上限（CHANNEL_MAX_PAGE）
WALK_MAX_PAGES = 200         # 翻页上限保护：绝不死循环
BAD_CURSOR = "__conformance_bad__"
WRONG_KEY = "pyk_conformance_definitely_not_a_real_key"
MAX_STORE_CANDIDATES = 20    # 写口选店时最多试几家，避免在大库上打太多请求
#: ACT `CID-MER-CAT` 的**四项必备**（键名逐字取自 ACT，不是我们的叫法）——
#: 出处：https://www.act-protocol.com/documentation/commerce
ACT_REQUIRED_FIELDS = ("id", "name", "category", "price")
#: ACT 的**推荐**项。★ 它们**不判 FAIL**：ACT 原文用的是「**宜**进一步包含」，
#: 而"宜"不是"必须" —— 把 SHOULD 当 MUST 判红，会把合规实现判成不合规
#: （这条套件的规矩是"红了就别对外说兼容"，一条假红会直接误导第三方）。
#: 对推荐项只判一件事：**自述与事实不许矛盾**（见 G6）。
ACT_RECOMMENDED_FIELDS = ("availability", "price_valid_until", "quoted_at",
                          "lead_time", "merchant", "detail_url")


def _https_opener():
    """https 用的 opener：**有 certifi 就用 certifi 的 CA**。

    为什么必须处理：macOS 的 framework python 常常没有系统证书链，直连 https 会
    `CERTIFICATE_VERIFY_FAILED` —— 而第三方十有八九是在自己的 Mac 上跑本套件，
    第一次跑就看到"品源的实例 8 条全红"，会以为是标准坏了（实测踩过）。
    `SSL_CERT_FILE` 环境变量仍由 stdlib 认，这里只是把最常见的那个补上。
    """
    ctx = None
    try:
        import certifi
        ctx = ssl.create_default_context(cafile=certifi.where())
    except Exception:                       # 没装 certifi → 用系统默认
        ctx = ssl.create_default_context()
    return urllib.request.build_opener(urllib.request.HTTPSHandler(context=ctx))


_OPENER = _https_opener()


def urllib_transport(method, url, headers, body_bytes, timeout=10.0):
    """默认传输实现（urllib）。非 2xx **也算正常响应**，原样交回上层判分。

    传输层不吞、不解释状态码 —— 否则 401/404/409 的具体文案就丢了，
    而"错误模型对不对"正是本套件要测的东西。
    """
    req = urllib.request.Request(url, data=body_bytes, headers=headers or {}, method=method)
    try:
        resp = _OPENER.open(req, timeout=timeout)
    except urllib.error.HTTPError as exc:
        try:
            return exc.code, exc.read()
        finally:
            exc.close()
    try:
        return resp.getcode(), resp.read()
    finally:
        resp.close()


class Response(object):
    """一次 HTTP 往返的解读结果。

    传输层只给 ``(status, bytes)``，其余在这里派生。``status == 0`` = 连都没连上
    （DNS/连接被拒/超时），用 ``error`` 记住原因 —— 这样"目标没起"表现为一条干净的
    FAIL，而不是一坨 traceback。
    """

    def __init__(self, status, raw, url, request_headers=None, error=None):
        self.status = status
        self.url = url
        self.request_headers = dict(request_headers or {})
        self.error = error
        if raw is None:
            self.text = ""
        elif isinstance(raw, (bytes, bytearray)):
            self.text = bytes(raw).decode("utf-8", "replace")
        else:
            self.text = str(raw)
        try:
            self.payload = json.loads(self.text)
            self.json_ok = True
        except ValueError:
            self.payload = None
            self.json_ok = False

    @property
    def looks_html(self):
        """是不是一块 HTML 错误页。响应头拿不到，只能看体 —— 够用且不看实现细节。"""
        head = self.text.lstrip()[:400].lower()
        return (head.startswith("<!doctype") or head.startswith("<html")
                or "<html" in head or head.startswith("<head"))

    @property
    def detail(self):
        if isinstance(self.payload, dict):
            for key in ("detail", "note"):
                value = self.payload.get(key)
                if value:
                    return value if isinstance(value, str) else json.dumps(
                        value, ensure_ascii=False)
        return self.text.strip()[:200]

    @property
    def retry_after(self):
        """429 响应里建议的等待秒数；拿不到就 ``None``。

        ★ 为什么只读 body 不读响应头（R-95，2026-10-03）：传输层只回
        ``(status, bytes)``（`Response` 的类注释就写着这件事），响应头根本没被带上来。
        品源把 `retry_after` 放在 **body** 里（见 `channel.py` 的 429 分支），
        所以这里读得到；别的实现若只给 `Retry-After` 头，这里就是 ``None`` ——
        那只会让重试更保守，不会让判据出错。
        """
        payload = self.payload
        if isinstance(payload, dict):
            raw = payload.get("retry_after")
            if isinstance(raw, bool):       # bool 是 int 的子类，必须先挡掉
                return None
            if isinstance(raw, (int, float)):
                return float(raw)
        return None


def status_text(resp):
    """给人类看的状态描述：网络层失败时把原因带出来。"""
    if resp.error:
        return "HTTP %s（请求失败：%s）" % (resp.status, resp.error)
    return "HTTP %s" % resp.status


def item_skus(item):
    """条目里的规格清单；不是列表就当没有（协议口径：空数组=单规格）。"""
    skus = item.get("skus") if isinstance(item, dict) else None
    return skus if isinstance(skus, list) else []


class Reporter(object):
    """收集并即时打印每条检查结果。"""

    def __init__(self, stream=None):
        self.stream = stream if stream is not None else sys.stdout
        self.rows = []

    def section(self, title):
        self.stream.write("\n== %s ==\n" % title)
        self.stream.flush()

    def add(self, status, check_id, message):
        self.rows.append((status, check_id, message))
        self.stream.write("%-4s %-4s %s\n" % (status, check_id, message))
        self.stream.flush()

    def count(self, status):
        return len([row for row in self.rows if row[0] == status])

    def failures(self):
        return [row for row in self.rows if row[0] == "FAIL"]

    def skips(self):
        return [row for row in self.rows if row[0] == "SKIP"]


class Suite(object):
    """一致性套件本体。构造后调 ``run(reporter)``。

    * ``base``      : ``http://127.0.0.1:8972`` 这类，带不带 ``/`` 都行。
    * ``key``       : 渠道密钥（``X-Channel-Key``）。不给则所有需要鉴权的检查 SKIP。
    * ``transport`` : 可注入传输（写单元测试用），默认 urllib。
    * ``write``     : True 才跑写口幂等（会真扣库存）。
    * ``store``/``uid`` : 写口指定测试对象；不给则自动从可见商品里挑。
    """

    #: 撞上 429 时最多重试几次（理由见 `_request` 里那段"限流不是不合规"）。
    RETRY_429_TIMES = 2
    #: 单次重试最多愿意等多少秒 —— 超过就**不猜也不硬等**，把 429 如实上报给检查。
    #: 取 15 是权衡：品源聚合口的窗口是 60 秒，等满一个窗口会让套件明显变慢；
    #: 而"等 15 秒还不行"多半说明对方额度是给人的日常调用用的、不适合被套件密集打。
    RETRY_429_MAX_WAIT = 15.0

    def __init__(self, base, key=None, transport=None, timeout=10.0, page_size=100,
                 write=False, store=None, uid=None, walk_max_pages=WALK_MAX_PAGES):
        if not base or not str(base).strip():
            raise ValueError("base 不能为空（例：http://127.0.0.1:8972）")
        self.base = str(base).strip().rstrip("/")
        self.key = (key or "").strip()
        self.timeout = timeout
        self.page_size = max(1, min(int(page_size), HARD_PAGE_MAX))
        self.write = bool(write)
        self.store = (store or "").strip() or None
        self.uid = (uid or "").strip() or None
        self.walk_max_pages = int(walk_max_pages)
        self.transport = transport if transport is not None else self._urllib
        self._cache = {}
        self._schema = None
        self._store_codes = None
        # B 组跑完后填：True=钥可用 / False=钥被拒 / None=没给钥。
        # 用途：没钥或钥被拒时，后面所有需要鉴权的检查一律 SKIP ——
        # 否则一把坏钥会带来十几条 FAIL，把"凭据不对"这一个根因伪装成"协议实现崩了"。
        self._keyed_ok = None

    # ── 传输与 URL ──────────────────────────────────────────────────────────

    def _urllib(self, method, url, headers, body_bytes):
        return urllib_transport(method, url, headers, body_bytes, self.timeout)

    def _url(self, path, params=None):
        url = self.base + path
        clean = []
        for key, value in (params or {}).items():
            if value is None:
                continue
            clean.append((key, value if isinstance(value, str) else str(value)))
        if clean:
            url += "?" + urllib.parse.urlencode(clean)
        return url

    def _request(self, method, path, params=None, body=None, auth=True,
                 fresh=False, extra_headers=None):
        """发一次请求。

        GET 结果按 (method, url, auth, body) 缓存 —— 套件里同一个 URL 会被多条检查
        用到（如 /schema），没必要重复打；``fresh=True`` 用于**写后读回**这种必须看
        新值的地方。

        ## 429 的处理（R-95，2026-10-03）

        **限流不是不合规。** 一个实现完全可以给自己的口设额度，套件密集打就会撞上。
        修这条之前 F1 报过一条**假 FAIL**（"聚合位返回了意外状态 HTTP 429"），
        真因是品源自己的聚合口额度只有 6 次/分、而前面的检查已经把共用额度用掉了
        —— 于是**合规的实现被自己的套件判成不合规**。第三方照这份套件跑，
        遇到任何会限流的实现都会得到同样的假红，而套件的规矩是"红了就别对外说兼容"。

        所以这里先**有界重试**（多数 429 是瞬时的，按 body 的 `retry_after` 等）：
        等得太久（超过 `RETRY_429_MAX_WAIT`）或重试完仍是 429，就把这个 429 如实
        返回给调用方 —— 由各条检查按"**这个口没被评估到**"处理（SKIP），
        **不判 FAIL**。判据要能区分"它做错了"和"我没测到"。
        """
        url = self._url(path, params)
        cache_key = (method, url, bool(auth), json.dumps(body, sort_keys=True) if body else "")
        if method == "GET" and not fresh and cache_key in self._cache:
            return self._cache[cache_key]
        headers = {"Accept": "application/json"}
        if auth and self.key:
            headers["X-Channel-Key"] = self.key
        data = None
        if body is not None:
            data = json.dumps(body, ensure_ascii=False).encode("utf-8")
            headers["Content-Type"] = "application/json"
        if extra_headers:
            headers.update(extra_headers)
        attempt = 0
        while True:
            try:
                status, raw = self.transport(method, url, headers, data)
                resp = Response(status, raw, url, headers)
            except Exception as exc:  # noqa: BLE001 —— 目标没起/超时也要出一条 FAIL，不是崩栈
                resp = Response(0, None, url, headers, error=str(exc) or exc.__class__.__name__)
            if resp.status != 429 or attempt >= self.RETRY_429_TIMES:
                break
            wait = resp.retry_after
            if wait is None or wait > self.RETRY_429_MAX_WAIT:
                # 拿不到等待时间、或对方要我们等太久 → 不猜、不硬等，如实往上报。
                break
            attempt += 1
            sys.stderr.write("    （429：等 %.0f 秒后重试第 %d/%d 次）\n"
                             % (wait, attempt, self.RETRY_429_TIMES))
            sys.stderr.flush()
            time.sleep(max(0.0, wait))
        if method == "GET":
            self._cache[cache_key] = resp
        return resp

    # ── 现读的事实（不硬编码业务词汇） ──────────────────────────────────────

    def _first_store_code(self):
        codes = self._visible_store_codes()
        if self.store:
            return self.store
        return codes[0] if codes else None

    def _visible_store_codes(self):
        """从 /channel/stores 现读店码 —— 店码是环境事实，不该写死在套件里。"""
        if self._store_codes is not None:
            return self._store_codes
        if self.store:
            self._store_codes = [self.store]
            return self._store_codes
        codes = []
        cursor = None
        for _ in range(5):                      # 店目录也可能多页；取几页够挑测试对象
            params = {"page_size": 200}
            if cursor:
                params["cursor"] = cursor
            resp = self._request("GET", PATH_STORES, params=params)
            if resp.status != 200 or not isinstance(resp.payload, dict):
                break
            page = resp.payload.get("stores")
            if not isinstance(page, list):
                break
            for st in page:
                if isinstance(st, dict) and st.get("code"):
                    codes.append(str(st["code"]))
            cursor = resp.payload.get("next_cursor")
            if not cursor:
                break
        self._store_codes = codes
        return codes

    def _product_page(self, store, page_size, cursor=None, fresh=False):
        params = {"page_size": page_size}
        if store:
            params["store"] = store
        if cursor:
            params["cursor"] = cursor
        return self._request("GET", PATH_PRODUCTS, params=params, fresh=fresh)

    def _read_stock(self, store, uid, fresh=False):
        """从出库口读某商品当前库存。读不到返回 None（调用方转 SKIP，不猜）。"""
        resp = self._product_page(store, HARD_PAGE_MAX, fresh=fresh)
        if resp.status != 200 or not isinstance(resp.payload, dict):
            return None
        for item in (resp.payload.get("items") or []):
            if isinstance(item, dict) and str(item.get("uid") or "") == uid:
                stock = item.get("stock")
                return stock if isinstance(stock, int) else None
        return None

    # ── A. 自描述端点 ───────────────────────────────────────────────────────

    def check_schema(self, rep):
        resp = self._request("GET", PATH_SCHEMA, auth=False)
        self._schema = resp.payload if resp.json_ok else None
        if resp.status == 200:
            rep.add("PASS", "A1", "/schema 不带密钥即可访问（HTTP 200）—— 对接前自检无需凭据")
        else:
            rep.add("FAIL", "A1", "/schema 应公开可访问（期望 200），实际 %s" % status_text(resp))

        if not isinstance(self._schema, dict):
            reason = "响应不是 JSON 对象（%s）" % status_text(resp)
            rep.add("FAIL", "A2", "schema_version 读不到：%s" % reason)
            rep.add("FAIL", "A3", "fields 读不到：%s" % reason)
            rep.add("FAIL", "A4", "字段 key 唯一性无从校验：%s" % reason)
            return

        version = self._schema.get("schema_version")
        # bool 是 int 的子类，别让 True 冒充"整数版本号"
        if isinstance(version, bool) or not isinstance(version, int):
            rep.add("FAIL", "A2", "schema_version 应为整数，实际 %r（%s）"
                    % (version, type(version).__name__))
        else:
            rep.add("PASS", "A2", "schema_version 是整数：%d" % version)

        fields = self._schema.get("fields")
        if not isinstance(fields, list) or not fields:
            rep.add("FAIL", "A3", "fields 应为非空列表，实际 %s" % type(fields).__name__)
            rep.add("FAIL", "A4", "fields 非列表，无法校验 key 唯一性")
            return

        problems = []
        for entry in fields:
            if not isinstance(entry, dict):
                problems.append("存在非对象条目 %r" % (entry,))
                continue
            for need in ("key", "ftype", "level"):
                if not entry.get(need):
                    problems.append("条目缺 %s（key=%r）" % (need, entry.get("key")))
            if entry.get("level") not in FIELD_LEVELS:
                problems.append("level=%r 不在 %s 内（key=%r）"
                                % (entry.get("level"), "/".join(FIELD_LEVELS), entry.get("key")))
        if problems:
            rep.add("FAIL", "A3", "fields 形状不合规：%s" % "；".join(problems[:3]))
        else:
            rep.add("PASS", "A3", "fields 共 %d 条，每条含 key/ftype/level，"
                                  "level 均落在 required/recommended/optional 内" % len(fields))

        counter = {}
        unhashable = 0
        for entry in fields:
            if not isinstance(entry, dict):
                continue
            try:
                k = entry.get("key")
                counter[k] = counter.get(k, 0) + 1
            except TypeError:
                # key 是列表/字典这类不可哈希值 —— 形状本来就错了，
                # 不能让它把整套件掀翻（第三方实现什么都能返回）。
                unhashable += 1
        if unhashable:
            rep.add("FAIL", "A4", "有 %d 条字段的 key 不是标量（无法做唯一性判断）" % unhashable)
            return
        dups = [k for k, n in counter.items() if n > 1]
        if dups:
            rep.add("FAIL", "A4", "字段 key 有重复：%s" % "、".join(str(d) for d in dups[:5]))
        else:
            rep.add("PASS", "A4", "%d 个字段的 key 无重复" % len(counter))

    def check_wellknown(self, rep):
        """能力发现口 `/.well-known/rec-agent`（2026-10-15 加）。

        ## 为什么它是**可选**的

        发现口不是协议强制的：一个实现完全可以让对方来问人要对接文档。
        所以 **404 → SKIP，不是 FAIL** —— 与 F1（聚合位）、G 组（ACT 目录口）同一条口径：
        **"选择不做"与"做错了"必须分开判**。

        ## A6 才是这一组的重点：**自述与事实不许矛盾**

        发现口最容易出的一种错是"声称兼容某协议，但那个端点根本不存在" ——
        对方照着自述写对接代码，跑到一半 404。所以 A6 拿它**声明过的端点**
        真去打一次：**404 = 自述撒谎（FAIL）**；`200`/`401` 都算"口存在"
        （`401` 只说明需要钥，不说明没有这个口）。

        这条刻意**不看实现自己的内部字段**，只看它对外声明了什么 ——
        所以对任何实现都成立。
        """
        resp = self._request("GET", PATH_WELLKNOWN, auth=False, fresh=True)
        if resp.status == 404:
            rep.add("SKIP", "A5", "/.well-known/rec-agent 不存在 —— 该实现没有能力发现口。"
                                  "**这是它的选择，不构成不合规**"
                                  "（品源自 2026-10-15 起提供）")
            rep.add("SKIP", "A6", "没有发现口，自述与事实的一致性无从判定")
            return
        if resp.status == 429:
            rep.add("SKIP", "A5", "/.well-known/rec-agent 返回 429 —— 限流不构成不合规，"
                                  "但这条形状断言因此没跑成（故 SKIP 而非 PASS）")
            rep.add("SKIP", "A6", "A5 未跑成")
            return
        if resp.status != 200:
            rep.add("FAIL", "A5", "发现口返回了意外状态 %s；应是 200，"
                                  "或 404 表示该实现没做这个口" % status_text(resp))
            rep.add("SKIP", "A6", "A5 未跑成")
            return

        body = resp.payload if resp.json_ok else None
        if not isinstance(body, dict):
            rep.add("FAIL", "A5", "发现口返回 200 但不是 JSON 对象 —— "
                                  "**发现口尤其不能吐 HTML**：对方本就在猜路径，"
                                  "HTML 会让它把'没有发现口'误判成'这个域名坏了'")
            rep.add("SKIP", "A6", "A5 未跑成")
            return

        missing = [k for k in ("protocol", "role", "capabilities", "auth") if k not in body]
        caps = body.get("capabilities")
        problems = []
        if missing:
            problems.append("缺键 %s" % "、".join(missing))
        # ⚠️ **"键在"不等于"能用"**（与目录口 G2/G3 同一条分工）：
        #    `protocol: null` / `role: ""` 过得"键存在"那一条，却让消费方什么都读不出。
        for key in ("protocol", "role"):
            value = body.get(key)
            if not isinstance(value, str) or not value.strip():
                problems.append("%s 不是非空字符串（%r）" % (key, value))
        if not isinstance(caps, list) or not caps:
            problems.append("capabilities 不是非空数组（对方读不到'你能做什么'）")
        auth = body.get("auth")
        if not isinstance(auth, dict) or not auth.get("header"):
            problems.append("auth.header 缺失（对方不知道用哪个请求头鉴权）")
        if problems:
            rep.add("FAIL", "A5", "发现口形状不符：%s" % "；".join(problems))
        else:
            rep.add("PASS", "A5", "发现口形状正确（protocol/role/capabilities/auth 齐全，"
                                  "能力 %d 项）" % len(caps))

        # ── A6 · 声明过的端点必须真的存在 ──
        act = body.get("act")
        if not isinstance(act, dict):
            rep.add("SKIP", "A6", "该实现的发现口没有 `act` 段 —— 它没声称 ACT 兼容，"
                                  "所以没有'声明与事实'可对（不判 FAIL）")
            return
        endpoint = act.get("endpoint")
        if not isinstance(endpoint, str) or not endpoint.strip():
            rep.add("SKIP", "A6", "`act` 段没给 `endpoint`，无从核对")
            return
        probe = self._request("GET", endpoint.strip(), fresh=True)
        if probe.status == 404:
            rep.add("FAIL", "A6", "发现口声称 ACT 目录口在 `%s`，但那个路径 404 —— "
                                  "**自述撒了谎**，对方会照着它写对接代码然后跑挂" % endpoint)
        elif probe.status in (200, 401):
            rep.add("PASS", "A6", "声明过的 ACT 目录口 `%s` 真的存在（HTTP %d；"
                                  "401 只说明需要钥）" % (endpoint, probe.status))
        elif probe.status == 429:
            rep.add("SKIP", "A6", "探测 `%s` 撞上限流 —— 这条没验到" % endpoint)
        else:
            rep.add("FAIL", "A6", "探测声明过的 ACT 目录口 `%s` 返回 %s —— "
                                  "既不是存在（200/401）也不是不存在（404）"
                    % (endpoint, status_text(probe)))

    # ── B. 鉴权 ─────────────────────────────────────────────────────────────

    def check_auth(self, rep):
        anon = self._request("GET", PATH_PRODUCTS, auth=False, fresh=True)
        if anon.status in (401, 403):
            rep.add("PASS", "B1", "不带密钥访问 /channel/products 被拒（HTTP %d）" % anon.status)
        else:
            rep.add("FAIL", "B1", "不带密钥访问 /channel/products 期望 401/403，实际 %s"
                    % status_text(anon))

        wrong = self._request("GET", PATH_PRODUCTS, fresh=True,
                              extra_headers={"X-Channel-Key": WRONG_KEY})
        if wrong.status in (401, 403):
            rep.add("PASS", "B2", "错误密钥被拒（HTTP %d）" % wrong.status)
        else:
            rep.add("FAIL", "B2", "错误密钥期望 401/403，实际 %s" % status_text(wrong))

        if not self.key:
            self._keyed_ok = None
            rep.add("SKIP", "B3", "未提供 --key，无法验证正确密钥能通过")
            return
        good = self._request("GET", PATH_PRODUCTS, fresh=True)
        if good.status == 200:
            self._keyed_ok = True
            rep.add("PASS", "B3", "正确密钥通过（HTTP 200）")
        else:
            self._keyed_ok = False
            rep.add("FAIL", "B3", "正确密钥期望 200，实际 %s —— %s"
                    % (status_text(good), good.detail))

    def _keyed_block_reason(self):
        """需要鉴权的检查为什么跑不了；跑得了则返回 None。"""
        if self._keyed_ok is True:
            return None
        if self._keyed_ok is None:
            return "未提供 --key（渠道读口需要 X-Channel-Key）"
        return "B3 中正确密钥未通过鉴权，后续需要鉴权的检查无从进行"

    # ── C. 信封与分页 ───────────────────────────────────────────────────────

    def check_envelope(self, rep):
        block = self._keyed_block_reason()
        if block:
            for cid in ("C1", "C2", "C3", "C4", "C5", "C6", "C7", "C8"):
                rep.add("SKIP", cid, "无法检查信封/分页：%s" % block)
            return
        store = self._first_store_code()
        env = self._product_page(store, self.page_size)
        payload = env.payload if env.json_ok else None

        # C1 共同核心键齐全（子集检查；逐端点精确键面由服务端守卫守，见 docs/PROTOCOL.md §2.1）
        if not isinstance(payload, dict):
            rep.add("FAIL", "C1", "/channel/products 应返回 200 JSON 信封，实际 %s"
                    % status_text(env))
        else:
            missing = [k for k in ENVELOPE_CORE_KEYS if k not in payload]
            if missing:
                rep.add("FAIL", "C1", "信封缺共同核心键：%s（实际键面：%s）"
                        % ("、".join(missing), "、".join(sorted(str(k) for k in payload))))
            else:
                rep.add("PASS", "C1", "信封含共同核心键（子集检查）：%s"
                        % "/".join(ENVELOPE_CORE_KEYS))

        # C2 键类型
        if not isinstance(payload, dict):
            rep.add("FAIL", "C2", "信封不是 JSON 对象，无法校验 items/next_cursor 类型")
        else:
            items = payload.get("items")
            nxt = payload.get("next_cursor")
            problems = []
            if not isinstance(items, list):
                problems.append("items 应为列表，实际 %s" % type(items).__name__)
            if nxt is not None and not isinstance(nxt, str):
                problems.append("next_cursor 应为字符串或 null，实际 %s" % type(nxt).__name__)
            if problems:
                rep.add("FAIL", "C2", "信封键类型不符：%s" % "；".join(problems))
            else:
                rep.add("PASS", "C2", "items 是列表（本页 %d 条）、next_cursor 是字符串或 null"
                        % len(items))

        # C3 版本一致
        if not isinstance(payload, dict):
            rep.add("SKIP", "C3", "信封不可读，无法与 /schema 比对 schema_version")
        elif not isinstance(self._schema, dict):
            rep.add("SKIP", "C3", "A 组没取到 /schema，没有可比对的版本号")
        else:
            v_env = payload.get("schema_version")
            v_schema = self._schema.get("schema_version")
            if v_env == v_schema:
                rep.add("PASS", "C3", "信封 schema_version=%r 与 /schema 报的一致" % (v_env,))
            else:
                rep.add("FAIL", "C3", "信封 schema_version=%r 与 /schema 的 %r 不一致（同一次会话两个版本）"
                        % (v_env, v_schema))

        # C4 硬上限
        big = self._product_page(store, 9999, fresh=True)
        body = big.payload if big.json_ok else None
        if big.status != 200 or not isinstance(body, dict) or not isinstance(body.get("items"), list):
            rep.add("FAIL", "C4", "page_size=9999 应被夹到硬上限并正常返回，实际 %s"
                    % status_text(big))
        else:
            n = len(body["items"])
            if n > HARD_PAGE_MAX:
                rep.add("FAIL", "C4", "page_size=9999 单页返回 %d 条，超过硬上限 %d"
                        % (n, HARD_PAGE_MAX))
            elif n >= HARD_PAGE_MAX:
                rep.add("PASS", "C4", "page_size=9999 被夹到硬上限（本页 %d 条 ≤ %d）"
                        % (n, HARD_PAGE_MAX))
            else:
                rep.add("PASS", "C4", "page_size=9999 单页 %d 条 ≤ %d（本库可见商品不足上限，"
                                      "属弱断言、未真正触顶）" % (n, HARD_PAGE_MAX))

        # C5 非法游标
        bad = self._request("GET", PATH_PRODUCTS, fresh=True,
                            params={"store": store, "cursor": BAD_CURSOR})
        if bad.status == 400:
            rep.add("PASS", "C5", "非法 cursor 被拒（HTTP 400）—— %s" % bad.detail)
        elif 500 <= bad.status < 600:
            rep.add("FAIL", "C5", "非法游标应被拒，不该炸：实际 HTTP %d（%s）"
                    % (bad.status, bad.detail))
        elif 400 <= bad.status < 500:
            rep.add("PASS", "C5", "非法 cursor 被拒（HTTP %d；协议约定为 400）" % bad.status)
        else:
            rep.add("FAIL", "C5", "非法 cursor 未被拒绝（%s），返回了数据" % status_text(bad))

        # C6 游标收敛
        cursor = None
        pages = 0
        total = 0
        seen = set()
        failure = None
        while True:
            resp = self._product_page(store, self.page_size, cursor=cursor, fresh=True)
            body = resp.payload if resp.json_ok else None
            if resp.status != 200 or not isinstance(body, dict):
                failure = "第 %d 页返回 %s，游标没能走完" % (pages + 1, status_text(resp))
                break
            page_items = body.get("items")
            if not isinstance(page_items, list):
                failure = "第 %d 页信封里没有 items 列表键（信封漂移）" % (pages + 1)
                break
            pages += 1
            total += len(page_items)
            nxt = body.get("next_cursor")
            if nxt is None or nxt == "":
                break
            # 先判类型再入集合：游标若不是字符串（数字/列表/对象），
            # 直接拿去做成员测试会抛 TypeError 把整套件掀翻 —— 形状错要报 FAIL，不是崩掉。
            if not isinstance(nxt, str):
                failure = ("第 %d 页的 next_cursor 类型非法（%s），无法继续翻页"
                           % (pages, type(nxt).__name__))
                break
            if nxt in seen:
                failure = "游标回环：第 %d 页又给出已见过的 next_cursor，翻页不会终止" % pages
                break
            seen.add(nxt)
            if pages >= self.walk_max_pages:
                failure = ("已翻 %d 页仍未结束（触发上限保护）；可能是游标不收敛，"
                           "也可能是数据集确实很大" % pages)
                break
            cursor = nxt
        if failure:
            rep.add("FAIL", "C6", "游标收敛：%s" % failure)
        else:
            rep.add("PASS", "C6", "游标收敛：共 %d 页、累计 %d 条，正常终止" % (pages, total))

        # C7 店目录列表键
        stores = self._request("GET", PATH_STORES, fresh=True,
                               params={"page_size": min(self.page_size, 200)})
        sbody = stores.payload if stores.json_ok else None
        if stores.status != 200 or not isinstance(sbody, dict):
            rep.add("FAIL", "C7", "/channel/stores 应返回 200 JSON 信封，实际 %s"
                    % status_text(stores))
        elif isinstance(sbody.get("stores"), list):
            extra = "（另有非契约的 items 键）" if "items" in sbody else ""
            rep.add("PASS", "C7", "店目录列表键是 stores（不是 items），本页 %d 家%s"
                    % (len(sbody["stores"]), extra))
        elif "items" in sbody:
            rep.add("FAIL", "C7", "店目录列表键成了 items —— 契约是 stores（历史上真踩过的坑）")
        else:
            rep.add("FAIL", "C7", "店目录信封里找不到 stores 列表键（实际键面：%s）"
                    % "、".join(sorted(str(k) for k in sbody)))

        # C8 流水下一页起点键
        if not store:
            rep.add("SKIP", "C8", "没有可用店码，无法调用 /channel/moves?store=")
        else:
            mv = self._request("GET", PATH_MOVES, fresh=True,
                               params={"store": store, "limit": 1})
            mbody = mv.payload if mv.json_ok else None
            if mv.status != 200 or not isinstance(mbody, dict):
                rep.add("FAIL", "C8", "/channel/moves?store=%s 应返回 200 JSON 信封，实际 %s"
                        % (store, status_text(mv)))
            elif "next_since" in mbody:
                rep.add("PASS", "C8", "moves 的下一页起点键是 next_since（limit=1 实测 next_since=%r）"
                        % (mbody.get("next_since"),))
            elif "next_cursor" in mbody:
                rep.add("FAIL", "C8", "moves 的下一页起点键成了 next_cursor —— 契约是 next_since")
            else:
                rep.add("FAIL", "C8", "moves 信封缺 next_since 键（实际键面：%s）"
                        % "、".join(sorted(str(k) for k in mbody)))

    # ── D. 错误模型 ─────────────────────────────────────────────────────────

    def check_errors(self, rep):
        nope = self._request("GET", PATH_NOPE, fresh=True)
        if nope.status != 404:
            rep.add("FAIL", "D1", "不存在的路径期望 404，实际 %s" % status_text(nope))
        elif nope.json_ok:
            rep.add("PASS", "D1", "不存在的路径 404 且错误体是 JSON")
        elif nope.looks_html:
            # 这条红的时候，最常见的原因是**被测实例跑在 DEBUG=True**：
            # Django 的基类处理器在 DEBUG 模式下直接走 technical_404_response，
            # 根本不会调到自定义 handler404（品源自己就是这样，见 pinyuan/urls.py）。
            # 但"一致性"是按协议判的，不是按部署模式判的 —— 所以仍然记 FAIL，
            # 只把病因指出来，省掉一轮排查。
            hint = ""
            if "DEBUG = True" in nope.text or "technical 404" in nope.text:
                hint = ("；**检出该实例 DEBUG=True** —— Django 在 DEBUG 模式下绕过自定义 "
                        "handler404，请以 DEBUG=False 的实例复测（品源 pinyuan/urls.py 已按 "
                        "/api/ 前缀回 JSON）")
            rep.add("FAIL", "D1", "不存在的路径 404 但吐了 HTML 错误页 —— "
                                  "错误模型应为 JSON/非 HTML，消费方解析会直接崩%s" % hint)
        else:
            rep.add("PASS", "D1", "不存在的路径 404 且错误体非 HTML（%r）"
                    % nope.text.strip()[:60])

        block = self._keyed_block_reason()
        if block:
            rep.add("SKIP", "D2", "无法校验非法参数的错误响应：%s" % block)
            rep.add("SKIP", "D3", "无法校验错误体：%s" % block)
            return

        store = self._first_store_code()
        bad_param = self._request("GET", PATH_PRODUCTS, fresh=True,
                                  params={"store": store, "price_max": "abc"})
        if 500 <= bad_param.status < 600:
            rep.add("FAIL", "D2", "非法参数 price_max=abc 触发了 5xx（HTTP %d）—— "
                                  "参数校验应在 4xx 收口" % bad_param.status)
        elif 400 <= bad_param.status < 500:
            rep.add("PASS", "D2", "非法参数 price_max=abc 被拒（HTTP %d）" % bad_param.status)
        elif bad_param.status == 200:
            rep.add("FAIL", "D2", "非法参数 price_max=abc 被接受（HTTP 200，应按 4xx 拒绝）")
        else:
            rep.add("FAIL", "D2", "非法参数 price_max=abc 返回意外状态 %s"
                    % status_text(bad_param))

        if not (400 <= bad_param.status < 500):
            rep.add("SKIP", "D3", "D2 没有产生错误响应，没有错误体可校验")
        elif not isinstance(bad_param.payload, dict):
            rep.add("FAIL", "D3", "错误体不是 JSON 对象，读不到 detail（原文：%r）"
                    % bad_param.text.strip()[:80])
        else:
            detail = bad_param.payload.get("detail")
            note = bad_param.payload.get("note")
            if isinstance(detail, str) and detail.strip():
                rep.add("PASS", "D3", "错误体含给人看的 detail：%s" % detail.strip()[:70])
            elif note:
                rep.add("PASS", "D3", "错误体没有 detail，有 note 兜底：%s"
                        % str(note).strip()[:70])
            else:
                rep.add("FAIL", "D3", "错误体既无 detail 也无 note（实际键面：%s）"
                        % "、".join(sorted(str(k) for k in bad_param.payload)))

    # ── E. 写口幂等（仅 --write） ────────────────────────────────────────────

    def _pick_write_target(self):
        """挑一个测试商品：优先无 SKU 的（单规格不需要 sku_code，路径最短）。

        返回 (item, store_code, 说明)；挑不到时 item 为 None。
        """
        candidates = [self.store] if self.store else self._visible_store_codes()
        if not candidates:
            return None, None, "没有可见店码（/channel/stores 为空或不可用）"
        for code in candidates[:MAX_STORE_CANDIDATES]:
            resp = self._product_page(code, HARD_PAGE_MAX, fresh=True)
            if resp.status != 200 or not isinstance(resp.payload, dict):
                continue
            items = [it for it in (resp.payload.get("items") or []) if isinstance(it, dict)]
            if self.uid:
                for item in items:
                    if str(item.get("uid") or "") == self.uid:
                        return item, code, "命中 --uid"
                continue
            for item in items:
                if not item_skus(item):
                    return item, code, "自动挑中单规格商品"
            if items:
                return items[0], code, "自动挑中首个商品（多规格，将带 sku_code）"
        if self.uid:
            return None, None, "在可见店里找不到 --uid=%s 这件商品" % self.uid
        return None, None, "可见店里一件商品都没有"

    def check_write(self, rep):
        if not self.key:
            for cid in ("E1", "E2", "E3"):
                rep.add("SKIP", cid, "未提供 --key，无法验证写口")
            return
        block = self._keyed_block_reason()
        if block:
            for cid in ("E1", "E2", "E3"):
                rep.add("SKIP", cid, "无法验证写口：%s" % block)
            return
        item, store_code, why = self._pick_write_target()
        if item is None:
            for cid in ("E1", "E2", "E3"):
                rep.add("SKIP", cid, "无法选定测试商品：%s" % why)
            return
        uid = str(item.get("uid") or "")
        stock_before = item.get("stock")
        skus = item_skus(item)
        sku_code = None
        if skus and isinstance(skus[0], dict):
            sku_code = str(skus[0].get("code") or "") or None
        if not uid:
            for cid in ("E1", "E2", "E3"):
                rep.add("SKIP", cid, "选中的商品没有 uid，写口认品需要它")
            return
        if not isinstance(stock_before, int):
            for cid in ("E1", "E2", "E3"):
                rep.add("SKIP", cid, "选中的商品没有可读的 stock（实际 %r），无法验证扣减量"
                        % (stock_before,))
            return

        # request_id 每次运行随机 —— 否则第二次跑会撞上上次的回执，
        # 首次上报就变成"回放"，E3 会因为"没扣"而假红。
        request_id = "conformance-" + os.urandom(8).hex()
        body = {"uid": uid, "sold": 1, "request_id": request_id}
        if sku_code:
            body["sku_code"] = sku_code
        rep.stream.write("     · 写口目标：store=%s uid=%s sku=%s 库存=%d request_id=%s\n"
                         % (store_code, uid, sku_code or "-", stock_before, request_id))

        first = self._request("POST", PATH_STOCK_REPORT, body=body)
        if first.status in (401, 403):
            for cid in ("E1", "E2", "E3"):
                rep.add("SKIP", cid, "密钥没有写权或写口鉴权未通过（HTTP %d）：%s"
                        % (first.status, first.detail))
            return
        if not (200 <= first.status < 300):
            rep.add("FAIL", "E1", "首次 stock_report 期望 2xx，实际 %s —— %s"
                    % (status_text(first), first.detail))
            rep.add("SKIP", "E2", "首次上报未成功，幂等回放无从谈起")
            rep.add("SKIP", "E3", "首次上报未成功，无法断言扣减量")
            return
        rep.add("PASS", "E1", "首次 stock_report 成功（HTTP %d，sold=1，request_id=%s）"
                % (first.status, request_id))

        after_first = self._read_stock(store_code, uid, fresh=True)
        second = self._request("POST", PATH_STOCK_REPORT, body=body, fresh=True)
        if not (200 <= second.status < 300):
            rep.add("FAIL", "E2", "同一 request_id 重复上报期望 2xx（回放），实际 %s —— %s"
                    % (status_text(second), second.detail))
        else:
            dup_flag = second.payload.get("duplicate") if isinstance(second.payload, dict) else None
            same_body = isinstance(second.payload, dict) and second.payload == first.payload
            if dup_flag is True:
                rep.add("PASS", "E2", "同一 request_id 重复上报回放首次结果（HTTP %d，duplicate=true）"
                        % second.status)
            elif same_body:
                rep.add("PASS", "E2", "同一 request_id 重复上报回放首次结果（HTTP %d，"
                                      "响应体与首次一致；未带 duplicate 标记）" % second.status)
            else:
                rep.add("FAIL", "E2", "同一 request_id 重复上报没有回放首次结果"
                                      "（HTTP %d，duplicate=%r，响应体与首次不同：%r）"
                        % (second.status, dup_flag, second.payload))

        after_second = self._read_stock(store_code, uid, fresh=True)
        if after_first is None or after_second is None:
            rep.add("SKIP", "E3", "写后从出库口读不到该商品的库存，无法断言扣减量"
                                  "（--store 是否对？商品是否已下架？）")
            return
        delta = stock_before - after_second
        if delta == 1 and after_second == after_first:
            rep.add("PASS", "E3", "库存只扣一次：%d → %d（首次报后 %d，两次上报总变化 %d）"
                    % (stock_before, after_second, after_first, delta))
        else:
            rep.add("FAIL", "E3", "库存变化不符：%d → %d（首次报后 %d，两次上报总变化 %d，"
                                  "期望恰好 1）—— 幂等没生效，重复上报重复扣了"
                    % (stock_before, after_second, after_first, delta))

    # ── F. 聚合位（尚未实现，应 SKIP 而不是 FAIL） ──────────────────────────

    def check_aggregate(self, rep):
        # R-89：聚合**已实现**，而且它**要求参数**（`metric`/`key`）——
        # 裸调它是 400（"你没说要聚合什么"），那是**正确的客户端错误语义**。
        # ⚠️ 第一版这里拿裸路径打：实现启用后落到下面的 `else` → 报一条
        # **假 FAIL**（"聚合位返回了意外状态 400"）—— 而我们自己的参考实现会被
        # 自己的套件判成不合规。所以：**按文档 §8 的示例带上参数打**。
        resp = self._request("GET", PATH_AGGREGATE,
                             params={"metric": "price_dist", "key": "category:粮油米面"},
                             fresh=True)
        # 401 = 预留位**也走渠道鉴权**（品源的设计：预留位不许是敞口，与其余渠道口逐字一致）。
        # 这不是"不合规"，而是"**没带钥就判不了**"—— 与 D2/D3 同一个处理方式。
        # （第一版这里直接判 FAIL，于是**我们自己的参考实现被自己的套件判成不合规**，
        #   而真相是"套件少带了一把钥"。第三方跑出这个 FAIL 会以为标准是坏的。）
        if resp.status == 401:
            if self.key:
                rep.add("FAIL", "F1", "带钥访问 /channel/aggregate 仍返回 401 —— 钥无效，"
                                      "或该口没有与其余渠道口一致的鉴权")
            else:
                rep.add("SKIP", "F1", "聚合口要求渠道鉴权（与其余渠道口一致），"
                                      "未提供 --key 无法判定；带钥时应回 200 或 400")
            return
        if resp.status in (404, 501):
            # ★ 品源自己的聚合口自 v14/R-89 起**已经实现**（200），501 只可能来自
            #   "别的实现还没做这个能力" —— 那是它的选择，不构成不合规，仍然 SKIP。
            #   （2026-09-30 审查：这段文案原来写"协议已预留、实现未启用"，读起来像是
            #    套件在说"标准就是这个状态"，容易让人以为品源也没做。）
            rep.add("SKIP", "F1", "/channel/aggregate 返回 HTTP %d —— 该实现在这个口上"
                              "尚未实现（品源自 v14/R-89 起已返回 200）" % resp.status)
        elif resp.status == 400:
            # 400 = 参数不合法/缺 key（**这是客户端错误，不是"没实现"**）。
            # 已实现的实现方对**带示例参数**的调用不该回 400 —— 那才判 FAIL。
            rep.add("SKIP", "F1", "/channel/aggregate 返回 400：参数不合法或该实现要求"
                              "不同的 key 形状（见 §8.4）；**HTTP 400 本身不构成不合规**")
        elif resp.status == 429:
            # ★ R-95（2026-10-03）：**限流不是不合规，而这个口没有被评估到。**
            #
            # 加这条之前这里是落到最下面的 `else` → 报 FAIL
            # （"聚合位返回了意外状态 HTTP 429"）。实测那次真因是：品源的聚合口
            # 额度只有 6 次/分，而前面的检查已经把这个钥的额度用掉了 ——
            # **合规的实现被自己的套件判成不合规**。第三方照这份套件跑，遇到任何
            # 会限流的实现都会得到同样的假红，而套件的规矩是"红了就别对外说兼容"：
            # 一条假红会直接误导别人。
            #
            # 判定用 SKIP 而不是 PASS：SKIP 的语义是"**这一条没测到**"，
            # 而 429 恰恰意味着形状断言一条都没跑。PASS 会把"没测"伪装成"测过了"。
            extra = resp.retry_after
            rep.add("SKIP", "F1", "/channel/aggregate 返回 429（该实现对本口限流%s）："
                              "**限流不构成不合规**，但这条形状断言因此没跑成 —— "
                              "故 SKIP 而不是 FAIL，也不是 PASS"
                              % ("，建议等 %.0f 秒" % extra if extra is not None else ""))
        elif resp.status == 200:
            body = resp.payload if resp.json_ok else None
            if not isinstance(body, dict):
                rep.add("FAIL", "F1", "聚合位返回 200 但不是 JSON 对象")
                return
            missing = [k for k in ("n_stores", "sample_ok") if k not in body]
            if missing:
                rep.add("FAIL", "F1", "聚合位已启用但形状不符，缺键：%s（实际键面：%s）"
                        % ("、".join(missing), "、".join(sorted(str(k) for k in body))))
            else:
                # ── 形状断言（2026-09-29 加）────────────────────────────────
                # 不只判"键在不在"，还要判**两个数自不自洽**。ADR-11（不许"警告但照出"）
                # 说的就是这条：`sample_ok=false` 却带着一个 `dist`，等于把个体数据
                # 塞给了一个"看起来不能信"的字段里 —— 而警告是最容易被忽略的东西。
                ns, sk, dist = body.get("n_stores"), body.get("sample_ok"), body.get("dist")
                bad = None
                if not isinstance(ns, int) or isinstance(ns, bool) or ns < 0:
                    bad = "`n_stores` 应为 ≥0 的整数，实际 %r" % (ns,)
                elif not isinstance(sk, bool):
                    bad = "`sample_ok` 应为布尔，实际 %r" % (sk,)
                elif sk and not isinstance(dist, dict):
                    bad = "`sample_ok=true` 却没有 `dist`（应给完整分布），实际 %r" % (dist,)
                elif (not sk) and dist is not None:
                    bad = ("`sample_ok=false` 却带回了 `dist`=%r —— **这就是"
                           "「警告但照出」**（ADR-11：要么给结果，要么什么都不给）" % (dist,))
                if bad:
                    rep.add("FAIL", "F1", "聚合位形状自洽性不符：%s" % bad)
                else:
                    rep.add("PASS", "F1", "聚合位已启用且形状正确（n_stores=%r，sample_ok=%r）"
                            % (ns, sk))
        else:
            rep.add("FAIL", "F1", "聚合位返回了意外状态 %s；未实现应回 404/501，"
                                  "已实现应回 200（样本不足时 dist=null）" % status_text(resp))

    # ── G. ACT 目录口（CID-MER-CAT，2026-10-15 加） ─────────────────────────

    def check_catalog(self, rep):
        """ACT `CID-MER-CAT` 兼容目录口：**必备四项必须齐**，推荐项只判自述一致性。

        ## 三条口径（都不许自己发明）

        1. **该实现可以选择不做这个口。** `CID-MER-CAT` 是 ACT 的一个组件，而 ACT 把
           「商户或平台**已具备**对外提供结构化商品或服务信息的能力」列为**前置条件** ——
           没这个能力就谈不上这个组件。所以 **404/501 → SKIP，不是 FAIL**。
           不写清这条，套件就会把"选择不做"判成"不合规"。
        2. **推荐项不判 FAIL。** ACT 原文是「**宜**进一步包含」——"宜"不是"必须"。
           把 SHOULD 当 MUST 判红，就是在拿这条套件去判红合规实现。
        3. **空目录是 SKIP 不是 PASS。** 0 条商品时形状一条都没验到；
           "没测到"与"通过了"必须分开（同 C 组遇到空库的处置）。

        FAIL 只留给两种事：**必备项缺了/不可用**，或**自述与事实矛盾**。
        """
        params = {"store": self.store} if self.store else None
        resp = self._request("GET", PATH_CATALOG, params=params, fresh=True)

        # 与 F1 逐字同一套处理：401 是"**没带钥就判不了**"，不是"不合规"。
        if resp.status == 401:
            if self.key:
                rep.add("FAIL", "G1", "带钥访问 /channel/catalog 仍返回 401 —— 钥无效，"
                                      "或该口没有与其余渠道口一致的鉴权")
            else:
                rep.add("SKIP", "G1", "ACT 目录口要求渠道鉴权（与其余渠道口一致），"
                                      "未提供 --key 无法判定；带钥时应回 200")
            return
        if resp.status in (404, 501):
            rep.add("SKIP", "G1", "/channel/catalog 返回 HTTP %d —— 该实现没有提供 ACT "
                                  "目录口。**这是它的选择，不构成不合规**"
                                  "（品源自 2026-10-15 起返回 200）" % resp.status)
            return
        if resp.status == 429:
            extra = resp.retry_after
            rep.add("SKIP", "G1", "/channel/catalog 返回 429（该实现对本口限流%s）："
                                  "**限流不构成不合规**，但形状断言因此没跑成 —— "
                                  "故 SKIP 而不是 FAIL，也不是 PASS"
                                  % ("，建议等 %.0f 秒" % extra if extra is not None else ""))
            return
        if resp.status != 200:
            rep.add("FAIL", "G1", "ACT 目录口返回了意外状态 %s；"
                                  "未实现应回 404/501，已实现应回 200" % status_text(resp))
            return

        body = resp.payload if resp.json_ok else None
        if not isinstance(body, dict):
            rep.add("FAIL", "G1", "ACT 目录口返回 200 但不是 JSON 对象 —— "
                                  "该实现可能根本没有这个口（别拿别的路由顶上）")
            return

        # ── G1 · 协议自述块 ──
        act = body.get("act")
        if not isinstance(act, dict):
            rep.add("FAIL", "G1", "响应缺少 `act` 自述块 —— 消费方无从知道"
                                  "「这是 ACT 哪个组件、必备哪几项」")
        else:
            problems = []
            if act.get("component") != "CID-MER-CAT":
                problems.append("component 应为 CID-MER-CAT，实际 %r" % act.get("component"))
            if "act-protocol.com" not in str(act.get("source") or ""):
                problems.append("source 未指向 act-protocol.com（ACT 要求引用时附出处）")
            declared = act.get("required")
            if not isinstance(declared, list) or tuple(declared) != ACT_REQUIRED_FIELDS:
                problems.append("required 应为 %s，实际 %r"
                                % ("、".join(ACT_REQUIRED_FIELDS), declared))
            if problems:
                rep.add("FAIL", "G1", "act 自述块与 ACT 不符：%s" % "；".join(problems))
            else:
                rep.add("PASS", "G1", "act 自述块齐全（组件名 / 出处 / 必备四项声明都对）")

        items = body.get("items")
        if not isinstance(items, list):
            rep.add("FAIL", "G2", "items 不是数组（实际 %s）" % type(items).__name__)
            return
        if not items:
            # ★ 空目录 = 「**没数据可判**」，不是「不合规」。
            rep.add("SKIP", "G2", "该实例目录为空（0 条商品）—— 必备四项形状**没被验到**。"
                                  "空库不等于不合规，但也不等于通过")
            rep.add("SKIP", "G3", "目录为空，必备①③的取值无法判定")
            rep.add("SKIP", "G4", "目录为空，必备④的标价与货币单位无法判定")
            rep.add("SKIP", "G5", "目录为空，catalog_ready 无法判定")
            rep.add("SKIP", "G6", "目录为空，自述与事实的一致性无法判定")
            rep.add("SKIP", "G7", "目录为空，进价是否外泄无法判定")
            rep.add("PASS", "G8", "该口要求渠道鉴权，且带钥返回 200（与其余渠道口一致）")
            return

        # ── G2 · 必备四项齐全（逐条） ──
        bad = []
        for index, item in enumerate(items):
            if not isinstance(item, dict):
                bad.append("第 %d 条不是对象" % index)
                continue
            missing = [k for k in ACT_REQUIRED_FIELDS if k not in item]
            if missing:
                bad.append("第 %d 条缺 %s" % (index, "、".join(missing)))
        if bad:
            rep.add("FAIL", "G2", "ACT 必备四项缺失（%d/%d 条）：%s"
                    % (len(bad), len(items), "；".join(bad[:5])))
        else:
            rep.add("PASS", "G2", "全部 %d 条都带齐必备四项（%s）"
                    % (len(items), "、".join(ACT_REQUIRED_FIELDS)))

        # ── G3 · 必备①标识 与 必备③类目 的**取值**可用 ──
        #   「键在」不等于「能用」：`id: ""` 与 `category: null` 都过得 G2 的子集检查，
        #   却让买方 Agent 无法标识商品 / 无法做意图匹配。
        bad = []
        for index, item in enumerate(items):
            if not isinstance(item, dict):
                continue
            # 目录口**允许**把缺必备项的行照出（`catalog_ready=false` + 缺项说明）；
            # 买方 Agent 只对 ready 的行做支付前确认。G3/G4 判的是"可用于交易的行"，
            # 把非 ready 行也拿去判会把"诚实披露"错判成"不合规"。
            # 自洽性由 G5 单独守。
            if item.get("catalog_ready") is False:
                continue
            if not item.get("id"):
                bad.append("第 %d 条 id 为空" % index)
            if not item.get("name"):
                bad.append("第 %d 条 name 为空" % index)
            cat = item.get("category")
            if not isinstance(cat, dict) or not cat.get("code"):
                bad.append("第 %d 条 category 不可用于意图匹配" % index)
        if bad:
            rep.add("FAIL", "G3", "必备①③取值不可用（ACT 要求标识可唯一解析、类目能支持"
                                  "与用户意图约束匹配）：%s" % "；".join(bad[:5]))
        else:
            rep.add("PASS", "G3", "标识与类目都可用于机器匹配")

        # ── G4 · 必备④「标价**及**货币单位」两半都要在 ──
        bad = []
        for index, item in enumerate(items):
            if not isinstance(item, dict):
                continue
            if item.get("catalog_ready") is False:
                continue          # 理由同 G3：非 ready 行不参与"可用于交易"的判定
            price = item.get("price")
            if not isinstance(price, dict):
                bad.append("第 %d 条没有 price 对象（须含标价与货币单位）" % index)
                continue
            amount = price.get("amount")
            try:
                amount_ok = float(str(amount)) > 0
            except (TypeError, ValueError):
                amount_ok = False
            if not amount_ok:
                bad.append("第 %d 条标价不可解析（%r）" % (index, amount))
            cur = price.get("currency")
            # ISO 4217 是三位大写字母。长度/大小写一起判，别只判"非空"。
            if not (isinstance(cur, str) and len(cur) == 3
                    and cur.isalpha() and cur.isupper()):
                bad.append("第 %d 条货币单位不是 ISO 4217 三字母（%r）" % (index, cur))
        if bad:
            rep.add("FAIL", "G4", "必备④「明确的标价及货币单位」不达标：%s" % "；".join(bad[:5]))
        else:
            rep.add("PASS", "G4", "标价与货币单位两半都在，且金额可解析")

        # ── G5 · `catalog_ready` 与 `missing_required` 自洽 ──
        #   这条守的是一句很实际的话：**买方 Agent 只能对 ready 的行做支付前确认**
        #   （ACT 原文）。两个字段互相矛盾时，Agent 会照着矛盾去做错事。
        bad = []
        for index, item in enumerate(items):
            if not isinstance(item, dict):
                continue
            if "catalog_ready" not in item:
                bad.append("第 %d 条缺 catalog_ready（买方无从判断能否用于支付前确认）" % index)
                continue
            ready = item.get("catalog_ready")
            if not isinstance(ready, bool):
                bad.append("第 %d 条 catalog_ready 不是布尔（%r）" % (index, ready))
                continue
            missing = item.get("missing_required")
            if not isinstance(missing, list):
                bad.append("第 %d 条缺 missing_required 数组" % index)
                continue
            unknown = [k for k in missing if k not in ACT_REQUIRED_FIELDS]
            if unknown:
                bad.append("第 %d 条 missing_required 含非必备项 %s"
                           % (index, "、".join(str(k) for k in unknown)))
            if ready and missing:
                bad.append("第 %d 条 ready=true 却报了缺项 %s"
                           % (index, "、".join(str(k) for k in missing)))
            if not ready and not missing:
                bad.append("第 %d 条 ready=false 却没说缺什么" % index)
            # ★ v15（2026-10-15）：**已过期的标价不得声称 ready**。
            #
            # 依据是 ACT 对必备项④的措辞：「应能支撑金额比较、规则前置检验和
            # **支付前确认**」—— 一个已经作废的标价在**定义上**就不满足它。
            #
            # 这条刻意**只看 ACT 侧的 `price_valid_until`**，不看品源自己的
            # `price_state`（那是本仓的叫法）—— 所以对第三方实现同样成立。
            #
            # 解析不了的日期**不判**：那是"没验到"，不是"做错了"
            # （同 429／空库的处置 —— 别把"我没测到"伪装成"它不合规"）。
            pvu = item.get("price_valid_until")
            if ready and isinstance(pvu, str) and pvu:
                try:
                    expired = datetime.date.fromisoformat(pvu) < datetime.date.today()
                except ValueError:
                    expired = False
                if expired:
                    bad.append("第 %d 条价格有效期 %s 已过（今天 %s），却仍报 ready=true"
                               % (index, pvu, datetime.date.today().isoformat()))
        if bad:
            rep.add("FAIL", "G5", "catalog_ready 与 missing_required 不自洽：%s"
                    % "；".join(bad[:5]))
        else:
            rep.add("PASS", "G5", "catalog_ready 与 missing_required 自洽"
                                  "（缺项只列必备项、方向不矛盾）")

        # ── G6 · 自述与事实**不许矛盾**（推荐项只判这一件事） ──
        noted = act.get("not_provided") if isinstance(act, dict) else None
        if noted is None:
            rep.add("SKIP", "G6", "该实现没有 `act.not_provided` 自述块 —— "
                                  "推荐项是否获提供**没被验到**"
                                  "（ACT 只把推荐项写作「宜」，故不判 FAIL）")
        elif not isinstance(noted, dict):
            rep.add("FAIL", "G6", "`act.not_provided` 不是对象（实际 %s）—— "
                                  "自述块形状不对，消费方读不了" % type(noted).__name__)
        else:
            bad = []
            for key in ACT_RECOMMENDED_FIELDS:
                provided = any(isinstance(i, dict) and i.get(key) is not None for i in items)
                declared = key in noted
                if declared and provided:
                    bad.append("%s 在 not_provided 里说没有，items 里却给了值" % key)
                if declared and not str(noted.get(key) or "").strip():
                    bad.append("%s 声明「未提供」却没写原因" % key)
            if bad:
                rep.add("FAIL", "G6", "自述与事实矛盾：%s" % "；".join(bad[:4]))
            else:
                rep.add("PASS", "G6", "推荐项的自述与事实一致（声明了就没有、且写了原因）")

        # ── G7 · 进价**一个字节都不带** ──
        #   本项目协议 §2.9 写着 `cost`「**有字段，但读不到**」——新开的口必须重新
        #   证明一遍（复评 F2 那次进价是从分享链接外泄的）。
        #   判据扫**键名**：值可能是数字也可能是字符串，扫值会漏。
        leaked = sorted({k for i in items if isinstance(i, dict)
                         for k in i if k == "cost" or k.startswith("cost_")})
        if leaked:
            rep.add("FAIL", "G7", "目录里出现了进价字段 %s —— ACT 要的是**标价**，"
                                  "而本项目协议 §2.9 承诺 cost 读不到" % "、".join(leaked))
        else:
            rep.add("PASS", "G7", "目录不含进价字段（与 §2.9 的承诺一致）")

        # ── G8 · 鉴权与其余渠道口一致 ──
        #   用**错钥**打一次：必须 401（不是 200、也不是 403/500）。
        wrong = self._request("GET", PATH_CATALOG, auth=False, fresh=True,
                              extra_headers={"X-Channel-Key": WRONG_KEY})
        if wrong.status == 401:
            rep.add("PASS", "G8", "错钥回 401（与其余渠道口同形）")
        elif wrong.status == 429:
            rep.add("SKIP", "G8", "错钥探测撞上限流 429 —— 这条没验到（限流不构成不合规）")
        else:
            rep.add("FAIL", "G8", "错钥访问返回 %s，应与其余渠道口一致回 401"
                    % status_text(wrong))

    # ── H. 插件包（Pack，2026-10-28）────────────────────────────────────────

    def check_packs(self, rep):
        """插件字段是**增量键**：老消费方不读也不能被它搞坏。

        这条只验"键在、类型对、字段能归到某个包"，**不验业务口径**
        （某个包要求哪些字段是插件自己的事，套件不该记住它）。
        """
        block = self._keyed_block_reason()
        if block:
            for cid in ("H1", "H2", "H3"):
                rep.add("SKIP", cid, "无法检查插件包：%s" % block)
            return
        store = self._first_store_code()

        # H1 · /channel/packs 可读，带 store 时出字段
        resp = self._request("GET", PATH_PACKS, params={"store": store} if store else None)
        body = resp.payload if resp.json_ok else None
        if resp.status != 200 or not isinstance(body, dict) \
                or not isinstance(body.get("packs"), list):
            rep.add("FAIL", "H1", "/channel/packs 应返回 200 且含 packs 列表，实际 %s"
                    % status_text(resp))
        else:
            bad = [p.get("code") for p in body["packs"]
                   if not isinstance(p, dict) or not p.get("code")
                   or not isinstance(p.get("fields"), list)]
            if bad:
                rep.add("FAIL", "H1", "这些包缺 code 或 fields 不是列表：%r" % bad[:5])
            else:
                rep.add("PASS", "H1", "/channel/packs 可读（%d 个包，字段定义齐全）"
                        % len(body["packs"]))

        # H2 · 商品信封里的 packs / pack_fields 类型正确（键必须**恒在**）
        env = self._product_page(store, self.page_size)
        payload = env.payload if env.json_ok else None
        items = payload.get("items") if isinstance(payload, dict) else None
        if not isinstance(items, list):
            rep.add("SKIP", "H2", "商品信封不可读，无法检查插件键")
        else:
            problems = []
            for i, it in enumerate(items[:50]):
                if not isinstance(it, dict):
                    continue
                if not isinstance(it.get("packs"), list):
                    problems.append("第 %d 条缺 packs 列表" % i)
                if not isinstance(it.get("pack_fields"), dict):
                    problems.append("第 %d 条缺 pack_fields 对象" % i)
                if not isinstance(it.get("pack_field_meta"), dict):
                    problems.append("第 %d 条缺 pack_field_meta 对象" % i)
            if problems:
                rep.add("FAIL", "H2", "插件键类型不符：%s" % "；".join(problems[:5]))
            else:
                rep.add("PASS", "H2", "商品条目带 packs/pack_fields/pack_field_meta（%d 条）"
                        % len(items[:50]))

        # H3 · field-rules 的 pack 段。`effective` 只在传了 category/kind 时出
        # （不带参数时响应必须与老版本逐字节一致），所以这里显式带上。
        params = {"kind": "product"}
        if store:
            params["store"] = store
        rules = self._request("GET", PATH_FIELD_RULES, params=params)
        rb = rules.payload if rules.json_ok else None
        if rules.status != 200 or not isinstance(rb, dict):
            rep.add("FAIL", "H3", "/channel/field-rules 应返回 200 JSON，实际 %s"
                    % status_text(rules))
        else:
            missing = [k for k in ("base", "effective", "packs", "requested_packs")
                       if k not in rb]
            if missing:
                rep.add("FAIL", "H3", "field-rules 缺键：%s" % "、".join(missing))
            else:
                rep.add("PASS", "H3",
                        "field-rules 含 base/effective/packs/requested_packs（请求包 %s）"
                        % (",".join(rb.get("requested_packs") or []) or "无"))

    # ── 跑完全部 ────────────────────────────────────────────────────────────

    def run(self, rep):
        rep.section("A. 自描述端点")
        self.check_schema(rep)
        self.check_wellknown(rep)
        rep.section("B. 鉴权")
        self.check_auth(rep)
        rep.section("C. 信封与分页")
        self.check_envelope(rep)
        rep.section("D. 错误模型")
        self.check_errors(rep)
        if self.write:
            rep.section("E. 写口幂等（--write）")
            self.check_write(rep)
        else:
            rep.section("E. 写口幂等（只读模式：未加 --write）")
            rep.add("SKIP", "E1", "只读模式：未加 --write，不触碰任何写口")
            rep.add("SKIP", "E2", "只读模式：未加 --write")
            rep.add("SKIP", "E3", "只读模式：未加 --write")
        rep.section("F. 跨商家聚合")
        self.check_aggregate(rep)
        rep.section("G. ACT 目录口（CID-MER-CAT）")
        self.check_catalog(rep)
        rep.section("H. 插件包（Pack）")
        self.check_packs(rep)
        return rep


def parse_args(argv):
    parser = argparse.ArgumentParser(
        prog="conformance.py",
        description="品源协议一致性测试套件（只读优先；写口需显式 --write）")
    parser.add_argument("--base", default=DEFAULT_BASE,
                        help="目标实例根地址（默认 %s，本地 dev）" % DEFAULT_BASE)
    parser.add_argument("--key", default=os.environ.get("PY_CHANNEL_KEY", ""),
                        help="渠道密钥 X-Channel-Key（也可用环境变量 PY_CHANNEL_KEY）")
    parser.add_argument("--write", action="store_true",
                        help="跑写口幂等检查（会真扣库存；需配合 --store/--uid）")
    parser.add_argument("--store", default="", help="写口测试用店码（如 st0001）")
    parser.add_argument("--uid", default="", help="写口测试用商品 uid")
    parser.add_argument("--timeout", type=float, default=10.0, help="单请求超时秒数（默认 10）")
    parser.add_argument("--page-size", type=int, default=100,
                        help="检查用的分页大小（默认 100，硬上限 500）")
    parser.add_argument("--" + REMOTE_FLAG.lstrip("-"), dest="i_know_its_remote",
                        action="store_true",
                        help="确认目标不是生产实例（base 非本地时必需）")
    return parser.parse_args(argv)


def is_remote(base):
    host = urllib.parse.urlsplit(base).hostname or ""
    return host not in LOCAL_HOSTS


def mask_key(key):
    key = key or ""
    if not key:
        return "(未提供)"
    if len(key) <= 10:
        return key[:3] + "…"
    return key[:10] + "…(%d 字符)" % len(key)


def main(argv=None):
    args = parse_args(argv)
    base = (args.base or "").strip()
    if not base:
        sys.stderr.write("--base 不能为空\n")
        return 2

    if is_remote(base):
        line = "!" * 74
        sys.stderr.write("\n".join([
            line,
            "!!  警告：目标不是本地实例 —— %s" % base,
            "!!  一致性套件会在该实例上发真实请求；加 --write 时还会真的扣库存。",
            "!!  确认这不是生产环境后，重新执行并加上 %s" % REMOTE_FLAG,
            line, "",
        ]))
        if not args.i_know_its_remote:
            sys.stderr.write("已中止（退出码 2）：远端目标需要 %s 显式确认。\n" % REMOTE_FLAG)
            return 2
        sys.stderr.write("已确认远端目标（%s），继续。\n\n" % REMOTE_FLAG)

    sys.stdout.write("品源协议一致性测试套件\n")
    sys.stdout.write("目标 base : %s\n" % base)
    sys.stdout.write("渠道密钥  : %s\n" % mask_key(args.key))
    sys.stdout.write("模式      : %s\n" % ("读写（--write，会改动库存）" if args.write
                                          else "只读（默认，不改动任何数据）"))

    rep = Reporter()
    suite = Suite(base, key=args.key, timeout=args.timeout, page_size=args.page_size,
                  write=args.write, store=args.store, uid=args.uid)

    # ── 连通性自检（v12/R-72）────────────────────────────────────────────
    # 连不上/证书问题时，**先告诉人怎么修**，而不是打印 22 行 FAIL ——
    # 那种输出会被读成"这个实现不合规"，而其实一个请求都没发出去。
    probe = suite._request("GET", PATH_SCHEMA, auth=False)
    if probe.status == 0:
        err = (probe.error or "")
        sys.stdout.write("\n!! 连不上目标，套件没有开始判分（这不是协议问题）\n")
        sys.stdout.write("   目标：%s\n   错误：%s\n\n" % (base, err))
        if "CERTIFICATE_VERIFY" in err.upper():
            sys.stdout.write("这看起来是 **HTTPS 证书链**问题（macOS 的 framework python 常见）：\n"
                             "  ① `pip install certifi`（装完本套件会自动用它）；或\n"
                             "  ② `export SSL_CERT_FILE=$(python3 -c \"import certifi;print(certifi.where())\")\n"
                             "  ③ 或先用 http 试：`--base http://<host>:8970`\n")
        else:
            sys.stdout.write("  ① 确认地址与端口（生产 8970 / 本地 dev 8972）；\n"
                             "  ② 确认服务在跑、网络可达（curl <base>/api/v1/schema 先试一下）。\n")
        return 3
    suite.run(rep)

    n_pass = rep.count("PASS")
    n_fail = rep.count("FAIL")
    n_skip = rep.count("SKIP")
    sys.stdout.write("\n" + "-" * 74 + "\n")
    sys.stdout.write("汇总：共 %d 条 —— 通过 %d / 失败 %d / 跳过 %d\n"
                     % (n_pass + n_fail + n_skip, n_pass, n_fail, n_skip))
    if n_fail:
        sys.stdout.write("失败项：%s\n" % "、".join(row[1] for row in rep.failures()))
    if n_skip:
        sys.stdout.write("跳过项：%s\n" % "、".join(row[1] for row in rep.skips()))
    sys.stdout.write("结论：%s\n" % ("不一致（有 FAIL）" if n_fail else "全部通过（无 FAIL）"))
    return 1 if n_fail else 0


if __name__ == "__main__":
    sys.exit(main())