# -*- coding: utf-8 -*- """品源渠道 API · 官方纯标准库客户端（Python 3.6+，零第三方依赖）。 这是**参考实现**。规范正文（规范性文档）与兼容性承诺在： ``../docs/PROTOCOL.md`` / ``../docs/COMPATIBILITY.md``； 自证兼容请跑 ``conformance.py``；许可见 ``LICENSE`` （**允许任何人免费实现兼容的客户端或服务端，含商业用途**）。 为什么要有这份 client ===================== 品源对外的渠道读/写口是 ``/api/v1/channel/*``，用请求头 ``X-Channel-Key`` 鉴权。 在此之前，每个消费方都各自手抄了一遍「拼 URL + 解析信封」： ``tools/litmus.sh``（品源自带的端到端探针）， 以及另一个项目里的导购适配器（一代导购已按 R-79 迁出数据库仓库， 现归档在 ``品源导购/legacy-一代/``；二代就是那个项目本体）。 各抄各的后果不是"代码重复"这么轻，而是**实测踩过的三类事故**： 1. **字段名读错**：信封里商品/店铺的组织键是 ``org_uid``，有消费方写成 ``org_id`` （取出来永远是 None），于是"按组织聚合/连锁对齐"整条失效却没人报错。 2. **认款键用错**：跨店认「同款商品」的键是 ``barcode_norm``（只留数字的规范化条码） 或跨店唯一的 ``qualified_code``；有消费方拿了 ``fields.barcode`` 这份**展示文本** （带「（演示）」之类的后缀），于是跨店对不上、最后爆在财务对账上。 3. **店目录加分页后静默少数据**：``/api/v1/channel/stores`` 默认 200 条/页、带 ``next_cursor``；三个消费方都只读了首页，跨店推荐直接少掉一大半店铺 —— **不报错、只是少**，属于线上最难查的一类问题。 三件事同一个根因：**"字段在哪、游标怎么走"这份知识被复制到了 N 个地方**， 每一份都会独立漂移。所以把它收口成一个模块，让消费方 import 而不是手抄。 纪律（不是建议，是要求） ======================== * **禁止**再自己拼 ``/api/v1/channel/...`` 的 URL —— 一律走本模块的方法。 * **禁止**再自己从信封里摸字段位置（``item["fields"]["price"]``、``env["stores"]``、 ``env["items"]``、``next_cursor`` 的键名……）—— 一律走 :func:`item_fields` / :func:`item_store_code` / :func:`item_uid` / :func:`item_qualified_code`， 以及 :meth:`Client.walk_products` / :meth:`Client.walk_stores`。 * 分页**必须走完游标**：单页口（:meth:`Client.products` / :meth:`Client.stores`） 只回一页；除非你明确只要第一页，否则用 ``walk_*``。 信封键面（本模块已经吸收的差异，消费方不用再记） ================================================ * 商品信封的列表键是 ``items``；**店目录的列表键是 ``stores``**（不是 ``items``）。 * 商品/店目录的游标键都是 ``next_cursor``；流水 ``/channel/moves`` 的下一页起点键是 ``next_since``（:meth:`Client.moves` 原样返回，未做 walker）。 * 商品条目的属性字段在 ``item["fields"]``（字符串键面）；同键面的类型化副本在 ``item["fields_structured"]``，字段来源/核验在 ``item["field_meta"]``。 :func:`item_fields` 只认 ``fields`` —— 它就是出库合同里"属性都在这里"的那一份。 用法 ==== :: import pyclient c = pyclient.Client.from_env() # 读 PY_BASE / PY_CHANNEL_KEY stores, pages = c.walk_stores() # 走完游标，不会静默少店 for st in stores: print(st["code"], st.get("org_uid")) items, pages = c.walk_products(store="st0001", min_level="recommended") for it in items: f = pyclient.item_fields(it) # 属性字段收口在这里 print(pyclient.item_store_code(it), it["code"], f.get("price")) c.stock_report(uid="<商品 uid>", sold=1, request_id="订单号-唯一串") # 幂等口 Python 3.6 兼容 =============== 生产服务器是 python3.6.8。本模块**刻意**不用 f-string、dataclass、海象运算符 ``:=``、 ``subprocess.run(capture_output=/text=)`` 等 3.6 之后的语法/参数；格式化一律用 ``%`` 或 ``str.format``。改动时请保持这条纪律（自查：``ast.parse(src, feature_version=(3, 6))`` + grep f-string）。 HTTPS 提示：部分 macOS 的 framework python 没有系统证书链，直连 https 会报 CERTIFICATE_VERIFY_FAILED；这时设置 ``SSL_CERT_FILE`` 环境变量，或注入自定义 ``transport``（见下）即可，不必改本模块。 """ import json import os import urllib.error import urllib.parse 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_FIELD_RULES = API_PREFIX + "/channel/field-rules" PATH_MOVES = API_PREFIX + "/channel/moves" PATH_STOCK_REPORT = API_PREFIX + "/channel/stock_report" PATH_STOCK_RETURN = API_PREFIX + "/channel/stock_return" PATH_CONSUMER_ORDERS = API_PREFIX + "/channel/consumer/orders" PATH_METRICS = API_PREFIX + "/channel/metrics" #: ACT `CID-MER-CAT` 兼容目录口（2026-10-15 新增，schema v15）。 #: 协议出处：https://www.act-protocol.com/documentation/commerce PATH_CATALOG = API_PREFIX + "/channel/catalog" PATH_AGGREGATE = API_PREFIX + "/channel/aggregate" #: 平台共享类目树（跨店对齐的唯一事实源；店内类目 uid 只在店内唯一）。 PATH_CATEGORIES = API_PREFIX + "/channel/categories" #: 跨店批量取数：一次请求查多家（上限 20 家/次），解决 1+N 次往返。 PATH_PRODUCTS_BATCH = API_PREFIX + "/channel/products-batch" DEFAULT_PAGE_SIZE = 100 # 与出库口默认一致；构造可覆盖 MAX_PAGE_SIZE = 500 # 服务端硬上限（CHANNEL_MAX_PAGE），传更大的一律截到这里 DEFAULT_MAX_PAGES = 2000 # 翻页防不收敛兜底：累计页数到顶就抛错，绝不死循环 class PyError(Exception): """渠道 API 调用失败（非 2xx；翻页类错误 status 为 None）。 属性： status : HTTP 状态码（int）；客户端侧错误为 None。 detail : 给人看的文案，尽量取响应体的 ``detail`` / ``note``。 payload : 响应体原样解码后的对象（通常是 dict；解不出 JSON 时 None）。 url : 出错的完整 URL。 """ HEAD = "品源渠道 API 错误" def __init__(self, status=None, detail=None, payload=None, url=None): self.status = status self.detail = detail self.payload = payload self.url = url Exception.__init__(self, self.__str__()) def __str__(self): parts = [self.HEAD] if self.status is not None: parts.append("HTTP %s" % self.status) if self.url: parts.append(self.url) out = " ".join(parts) if self.detail: out += "：" + str(self.detail) return out class PyConfigError(PyError): """客户端配置有问题（base/key 缺失、环境变量非法等），不是服务端返回的错。""" HEAD = "品源渠道客户端配置错误" class PyPaginationError(PyError): """游标不收敛：连续多页都返回 next_cursor，已中止以免死循环。""" HEAD = "品源渠道 API 翻页未收敛" def clamp_page_size(value): """把 page_size 夹到 [1, MAX_PAGE_SIZE]；转不成整数就原样返回（交给服务端报 400）。""" try: page = int(value) except (TypeError, ValueError): return value if page < 1: return 1 if page > MAX_PAGE_SIZE: return MAX_PAGE_SIZE return page def urllib_transport(method, url, headers, body_bytes, timeout=10): """默认传输实现：urllib。 签名与可注入传输一致：``transport(method, url, headers, body_bytes) -> (status, bytes)``。 **非 2xx 也算"正常响应"**，把状态码与响应体原样交回上层统一抛 PyError —— 传输层不吞、不解释（否则 401/404/409 的具体文案就丢了）。 """ req = urllib.request.Request(url, data=body_bytes, headers=headers or {}, method=method) try: resp = urllib.request.urlopen(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() # ── 信封访问助手：把"字段在哪"收口一次，消费方不许再自己摸 ────────────────────── # 这几条是设计上唯一的字段位置说明。商品条目的属性在 item["fields"]（字符串键面）； # 类型化副本在 item["fields_structured"]（同键面、taglist/list 已数组化）， # 字段来源/核验在 item["field_meta"]。三者都在**条目内部**，不在信封顶层。 def item_fields(item): """商品条目里的 ``fields`` 字典（字符串键面）；缺失/类型不对时返回 ``{}``，不抛。""" if not isinstance(item, dict): return {} fields = item.get("fields") if isinstance(fields, dict): return fields return {} def item_store_code(item): """该条**自带**的店码（``store_code``）。 跨店取数时认款/认店一律读它，**不要靠外层 store 推断** —— 跨店信封的 ``store`` 可能是 None，而每条 item 的 ``store_code`` 永远是自己那家店。 """ if not isinstance(item, dict): return "" return str(item.get("store_code") or "") def item_uid(item): """商品全局唯一 id（``uid``，跨实例唯一）；缺失时返回空串。 跨系统引用/下单认品用 uid；``code`` 只在店内唯一，跨店会串品。 """ if not isinstance(item, dict): return "" return str(item.get("uid") or "") def item_qualified_code(item): """带店前缀的商品码 ``<店码>:<店内短码>``（跨店去重/缓存的键）；缺失时返回空串。""" if not isinstance(item, dict): return "" return str(item.get("qualified_code") or "") def _decode_body(raw): """返回 (payload, text)：payload=解码后的对象或 None；text=原始文本，供报错取文案。""" if raw is None: return None, "" if isinstance(raw, (bytes, bytearray)): text = bytes(raw).decode("utf-8", "replace") else: text = str(raw) try: return json.loads(text), text except ValueError: return None, text def _detail_from(payload, text): """优先取响应体的 detail / note 作为给人看的文案，其次退回原始文本。""" if isinstance(payload, dict): for key in ("detail", "note"): value = payload.get(key) if value: if isinstance(value, str): return value return json.dumps(value, ensure_ascii=False) if text: return text.strip()[:300] return None class Client(object): """品源渠道 API 客户端。 构造:: Client(base, key, timeout=10, page_size=100, transport=None) * ``base`` : 形如 ``https://pinyuan.sxai77.cn`` 或 ``http://127.0.0.1:8972`` （只要 scheme://host[:port]；带不带尾部 ``/`` 都行， 误带 ``/api/v1`` 也会被容错剥掉）。 * ``key`` : 渠道密钥（``X-Channel-Key``）。``schema()`` 是公开端点，不需要它。 * ``timeout`` : 秒；只作用于默认 urllib 传输。 * ``page_size`` : ``walk_*`` 每页条数（默认 100，硬上限 500）。 * ``transport`` : 可注入传输 ``transport(method, url, headers, body_bytes) -> (status, bytes)``；默认用 urllib。注入是为了可测。 """ MAX_PAGES = DEFAULT_MAX_PAGES # 类属性；测试可调小（见 test-pyclient.py） def __init__(self, base, key, timeout=10, page_size=DEFAULT_PAGE_SIZE, transport=None): if not base or not str(base).strip(): raise PyConfigError(detail="base 不能为空（例：https://pinyuan.sxai77.cn）") try: page = int(page_size) except (TypeError, ValueError): raise PyConfigError(detail="page_size 需为整数，收到 %r" % (page_size,)) self.base = str(base).strip() self.key = ("" if key is None else str(key)).strip() self.timeout = timeout self.page_size = clamp_page_size(page) self._transport = transport if transport is not None else self._urllib # ── 构造入口 ──────────────────────────────────────────────────────────── @classmethod def from_env(cls, prefix="PY_"): """从环境变量构造：``BASE`` / ``CHANNEL_KEY`` （可选 ``TIMEOUT`` / ``PAGE_SIZE``）。缺了就报清楚是哪几个。""" base = (os.environ.get(prefix + "BASE") or "").strip() key = (os.environ.get(prefix + "CHANNEL_KEY") or "").strip() missing = [] if not base: missing.append(prefix + "BASE") if not key: missing.append(prefix + "CHANNEL_KEY") if missing: raise PyConfigError(detail=( "环境变量未设置：%s。示例：export %sBASE=https://pinyuan.sxai77.cn ；" "export %sCHANNEL_KEY=<渠道密钥>" % ("、".join(missing), prefix, prefix))) kwargs = {} raw_timeout = (os.environ.get(prefix + "TIMEOUT") or "").strip() if raw_timeout: try: kwargs["timeout"] = float(raw_timeout) except ValueError: raise PyConfigError(detail="%sTIMEOUT 需为数值，收到 %r" % (prefix, raw_timeout)) raw_page = (os.environ.get(prefix + "PAGE_SIZE") or "").strip() if raw_page: try: kwargs["page_size"] = int(raw_page) except ValueError: raise PyConfigError(detail="%sPAGE_SIZE 需为整数，收到 %r" % (prefix, raw_page)) return cls(base, key, **kwargs) # ── 传输 ──────────────────────────────────────────────────────────────── def _urllib(self, method, url, headers, body_bytes): return urllib_transport(method, url, headers, body_bytes, self.timeout) def _build_url(self, path, params=None): base = self.base.rstrip("/") # 容错：base 误带了 /api/v1 时不要拼成 /api/v1/api/v1/... if base.endswith(API_PREFIX): base = base[: -len(API_PREFIX)] url = base + path clean = {} for key, value in (params or {}).items(): if value is None: continue clean[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): url = self._build_url(path, params) 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" status, raw = self._transport(method, url, headers, data) payload, text = _decode_body(raw) try: status = int(status) except (TypeError, ValueError): status = 0 if status < 200 or status >= 300: raise PyError(status=status, detail=_detail_from(payload, text), payload=payload, url=url) if payload is None: raise PyError(status=status, detail="响应不是合法 JSON：%s" % (text.strip()[:200] or "(空响应体)"), payload=None, url=url) return payload # ── 单页口 ────────────────────────────────────────────────────────────── def schema(self): """``GET /api/v1/schema`` —— 公开自描述端点，不带密钥（对接前自检用）。""" return self._request("GET", PATH_SCHEMA, auth=False) def products(self, **params): """``GET /api/v1/channel/products`` —— **只回一页**。 要全量请用 :meth:`walk_products`。``params`` 直通查询串 （store / cursor / q / tag / scene / crowd / origin / min_level / min_completeness / category / category_uid / category_template / price_max / near_expiry / page_size / fields……）。 """ return self._request("GET", PATH_PRODUCTS, params=self._page_params(params)) def stores(self, **params): """``GET /api/v1/channel/stores`` —— **只回一页**。 要全量请用 :meth:`walk_stores`（默认 200/页，只读首页就是"静默少店"）。 """ return self._request("GET", PATH_STORES, params=self._page_params(params)) def field_rules(self): """``GET /api/v1/channel/field-rules`` —— 按类目的必填/推荐/可选三档门槛。 消费侧的完整度判据读它，**不要自己维护一套**（自己维护必然和品源漂移）。 """ return self._request("GET", PATH_FIELD_RULES) def moves(self, **params): """``GET /api/v1/channel/moves`` —— 库存流水对账口（单页）。 必备 ``store=<店码>``；``since``/``limit`` 可选。下一页起点在 ``next_since`` （**不是** ``next_cursor``），本方法不代翻页。 """ return self._request("GET", PATH_MOVES, params=params) def catalog(self, **params): """``GET /api/v1/channel/catalog`` —— **ACT `CID-MER-CAT` 兼容目录口**（单页）。 与 :meth:`products` 的区别不是"字段多一点少一点"，而是**形状是协议定的**： 它出 ACT 要的最低信息（必备 `id`/`name`/`category`/`price`， 推荐 `availability`/`quoted_at`/`price_valid_until`/`merchant`）， **不是**本库 45 字段的全量。 ⚠️ 读它的**两条口径**（写在这里，因为不问就会读错）： · **只对 `catalog_ready: true` 的行做支付前确认。** ACT 原文说 「当目录结果无法满足最低信息要求时，买方智能体**不宜**直接将该结果 用于购物车确认或支付前衔接」。缺必备项的行**照出**（带 `missing_required`）， 由消费方决定怎么处理。 · **`price_state == "expired"` 的行不算可用的标价** —— 该行的 `catalog_ready` 会是 `false` 且 `missing_required` 含 `price`。 `unknown`（商家没填有效期）**不阻塞**。 完整的自述（必备/推荐/未提供逐条原因）在响应的 ``act`` 段里。 """ return self._request("GET", PATH_CATALOG, params=self._page_params(params)) def aggregate(self, metric, key): """``GET /api/v1/channel/aggregate`` —— 跨商家聚合查询（**只出统计不出个体**）。 三条硬约束由服务端执行（k-匿名 / 只出统计 / 商家 opt-in）， 客户端不需要也不应该复制这套判断。样本不足回 **200 + `sample_ok: false`** （**不是错误**）；参数不合法才 400。 """ return self._request("GET", PATH_AGGREGATE, params={"metric": metric, "key": key}) def categories(self): """``GET /api/v1/channel/categories`` —— 平台共享类目树（不分页）。 商品信封里挂的是 ``category_template_uid``；要知道"这个节点叫什么、 在哪一层"只能查这里。跨店对齐**用这里的 uid** —— 店内类目 uid 只在店内唯一。 """ return self._request("GET", PATH_CATEGORIES) def products_batch(self, stores, **params): """``GET /api/v1/channel/products-batch`` —— 跨店批量取数（单次最多 20 家）。 ``stores`` 是店码列表（或已逗号连接的字符串）。响应按店分桶： ``results[店码] = {ok, items|…}``，失败店进 ``failed`` 数组并带原因 —— **一家失败不影响其余，读的时候必须先看 ok / failed，不要默认全成功**。 20 家上限是服务端故意设的（防超时与下游打满），分批查是使用方的责任。 """ if isinstance(stores, (list, tuple)): stores = ",".join(str(s).strip() for s in stores if str(s).strip()) p = {"stores": stores} p.update(params or {}) return self._request("GET", PATH_PRODUCTS_BATCH, params=p) def stock_report(self, uid=None, store=None, code=None, sold=1, sku_code=None, request_id=None): """``POST /api/v1/channel/stock_report`` —— 报销量扣库存。 认品二选一：``uid``（推荐，跨店不串品）或 ``store`` + ``code``（兼容老消费者）。 带 ``request_id``（≤64 字符）= **幂等口**：同(钥,店,request_id)重复上报回放首次 结果、不重复扣；不带则重复报=重复扣。未给的键不会出现在 body 里。 """ body = {"sold": sold} if uid is not None: body["uid"] = uid if store is not None: body["store"] = store if code is not None: body["code"] = code if sku_code is not None: body["sku_code"] = sku_code if request_id is not None: body["request_id"] = request_id return self._request("POST", PATH_STOCK_REPORT, body=body) def stock_return(self, *, uid=None, store=None, code=None, sku_code=None, qty, amount="0", return_type="refund_only", order_no="", refund_no, occurred_at=None, request_id=None): """退货/仅退款（2026-10-29）：return 才回补库存，refund_only 只冲净销量。""" body = {"qty": qty, "amount": amount, "return_type": return_type, "order_no": order_no, "refund_no": refund_no} if uid is not None: body["uid"] = uid if store is not None: body["store"] = store if code is not None: body["code"] = code if sku_code is not None: body["sku_code"] = sku_code if occurred_at is not None: body["occurred_at"] = occurred_at if request_id is not None: body["request_id"] = request_id return self._request("POST", PATH_STOCK_RETURN, body=body) def consumer_orders(self, *, store, orders, request_id): """消费方订单/退款台账（2026-10-30）。字段白名单见 PROTOCOL §5.3。""" return self._request("POST", PATH_CONSUMER_ORDERS, body={ "request_id": request_id, "store": store, "orders": orders}) # ── 走完游标的口 ──────────────────────────────────────────────────────── def walk_products(self, **params): """走完游标取全量商品，返回 ``(items, pages)``。 内部反复带 ``cursor=`` 直到没有 next_cursor。 **不许只读首页**：那正是"分批后消费方静默少数据"的成因。 """ return self._walk(PATH_PRODUCTS, "items", "next_cursor", params) def walk_stores(self, **params): """走完游标取全量店铺，返回 ``(stores, pages)``。 注意店目录的列表键是 ``stores`` 而不是 ``items``（本模块已吸收这处差异）。 """ return self._walk(PATH_STORES, "stores", "next_cursor", params) def _page_params(self, params, default=None): out = dict(params or {}) if out.get("page_size") is None and default is not None: out["page_size"] = default if out.get("page_size") is not None: out["page_size"] = clamp_page_size(out["page_size"]) return out def _walk(self, path, items_key, cursor_key, params): items = [] pages = 0 cursor = None base_params = dict(params or {}) while True: page_params = dict(base_params) if cursor: page_params["cursor"] = cursor page_params = self._page_params(page_params, default=self.page_size) url = self._build_url(path, page_params) env = self._request("GET", path, params=page_params) pages += 1 if not isinstance(env, dict): raise PyError(status=None, detail="信封不是 JSON 对象，无法翻页", payload=env, url=url) page_items = env.get(items_key) # 列表键缺失/类型不对 = 信封漂移。**拒绝静默返回空** —— # 历史上"加分页后消费方少数据"就是静默造成的，这里必须响。 if not isinstance(page_items, list): raise PyError( status=None, detail="信封里没有 %r 列表键（实际键面：%s）；拒绝静默返回空" % (items_key, "、".join(sorted(str(k) for k in env.keys()))), payload=env, url=url) items.extend(page_items) # 注意：即使本页 0 条，只要还有 next_cursor 就必须继续翻 —— 别再只看本页条数。 cursor = env.get(cursor_key) if not cursor: return items, pages if pages >= self.MAX_PAGES: raise PyPaginationError( detail="连续 %d 页都返回 %s，疑似游标不收敛；已中止以免死循环" "（如确需放宽，调大 Client.MAX_PAGES）" % (pages, cursor_key), payload=env, url=url) # 信封访问助手：模块级同名函数的便捷入口，语义完全一致。 item_fields = staticmethod(item_fields) item_store_code = staticmethod(item_store_code) item_uid = staticmethod(item_uid) item_qualified_code = staticmethod(item_qualified_code)