《二手ERP对接闲鱼API：聚石塔强制入塔后的架构重构实录》（附Python源码）
0. 先拍结论
闲鱼 ISV 不是“能不能调 API”的问题，而是“在哪里调、数据存哪里、消息从哪里来”的问题。聚石塔强制入塔后，旧架构（公网 ERP 直连闲鱼 / 第三方聚合 API / 人工补单）直接失效；新架构必须是：公网 ERP 控制台 → 奇门/网关 → 聚石塔内「闲鱼通道服务」→ TOP 网关 → 闲鱼订单/会员/R2 数据必须落聚石塔 RDS，塔内发起写操作，塔外只能通过奇门标准接口拿结果。
一、强制入塔到底强制了什么
维度 | 旧架构（塔外） | 新架构（入塔后） |
|---|---|---|
商品/订单/退款写操作 | 第三方只读 / 人工在闲鱼 APP 补 | 必须在聚石塔内调用 alibaba.idle.isv.* |
用户ID / 订单信息存储 | 自建公网 MySQL | 聚石塔 RDS（MySQL） |
消息来源 | 抓包 / 轮询 / 聚合 API | TMC 长连接 / 聚石塔 RocketMQ（交易、商品、退款） |
塔外交互 | 直接 HTTP 回传 | 走奇门自定义 API，审批通过才行 |
Token | 放公网配置中心 | 按 seller/店铺隔离，塔内 KMS/配置中心托管 |
官方口径很硬：
闲鱼合作方服务端：用户 id、订单信息等必须存储在聚石塔内数据库，服务均部署在聚石塔内。
聚石塔规则：ERP/订单管理/仓储/CRM 类应用必须入塔；高风险写 API 必须从塔内发起；R2 数据禁止塔外二次开放。
交易/商品/退款类消息：TMC 免费但存 24h；核心电商消息必须走 MQ/ONS 订阅到聚石塔 RocketMQ。
二、重构前后拓扑
❌ 旧架构：公网 ERP 直连
运营后台(公网) │ ├──▶ 第三方聚合API（只能读：选品/比价/监控） ├──▶ 闲鱼APP/闲管家（写操作靠人） └──▶ 定时任务轮询（易被限流/漏单） 问题： - 发不了商品 / 处理不了退款闭环 - 订单状态靠人眼对齐 - 聚石塔强制入塔后：alibaba.idle.isv.* 直接报“非聚石塔调用”
✅ 新架构：塔内通道 + 塔外控制台
┌──────────────────────────────┐ │ 公网 ERP（控制台/报表/人工） │ └──────────┬───────────────────┘ │ 奇门标准接口 / 内网隧道（审批） ▼ ┌────────────────────────────────────────────┐ │ 聚石塔 ECS │ │ ┌──────────────────────────────────────┐ │ │ │ XianyuChannelService（闲鱼通道服务） │ │ │ │ - TokenManager(按seller隔离) │ │ │ │ - IdleItemClient / IdleOrderClient │ │ │ │ - RefundSyncHandler │ │ │ │ - TMC/RocketMQ Consumer │ │ │ │ - PII脱敏 / 审计 / 本地消息表 │ │ │ └──────────────┬───────────────────────┘ │ │ │ HTTPS │ │ ▼ │ │ gw.api.taobao.com / open.goofish.com│ │ ▲ │ │ idle_autotrade_OrderStateSync / RefundSync│ └────┬────────────────────────────────┬──────┘ │ │ ┌────▼─────┐ ┌───────▼────────┐ │ RDS MySQL│ │ RocketMQ 实例 │ │ 订单/用户│ │ 交易/商品/退款 │ └──────────┘ └─────────────────┘
核心原则：
写操作全进塔：发商品、改价、发货、退款处理，不在公网 ERP 做。
公网 ERP 只做“业务决策”：审单规则、库存分配、WMS 指令、财务报表。
塔内外边界用奇门：不能自己写个公网 HTTP 把订单吐出去。
三、入塔后必须重画的几个边界
1. Token 不再放公网
闲鱼走淘宝 OAuth2：
authorize拿 code（一次性、5 分钟过期）oauth.taobao.com/token换 SessionKey / accessTokenSessionKey 是“哪个卖家授权查哪个卖家”
入塔后：
公网ERP: 只存 shop_id / seller_nick 聚石塔内: 存 access_token / refresh_token / expire_at（加密）
2. 消息不再是“webhook 打公网”
闲鱼正向/逆向消息：
正向：
idle_autotrade_OrderStateSync逆向：
idle_autotrade_RefundSync（状态快照，不是事件流）
入塔后消费方式：
轻量：TMC 长连接（存 24h）
生产：MQ/ONS → 聚石塔 RocketMQ 4.x 实例
3. 订单/会员数据不能出塔
聚石塔规则明确：
R2 数据必须在聚石塔内完成，禁止通过自有接口二次开放。
所以公网 ERP 看到的是：
{
"order_id": "XY-xxx",
"buyer_mask": "u_9f2a****",
"phone_mask": "138****1234",
"address_mask": "浙江 杭州 **** 街道"
}原始手机号、明文地址、用户 uid 只在塔内 RDS。
四、重构后的核心代码（生产向，不是 demo）
1. 聚石塔内：闲鱼通道服务骨架
# xianyu_channel/service.py 运行在聚石塔 ECS 内
from dataclasses import dataclass
from typing import Optional
@dataclass
class SellerSession:
seller_id: str
shop_id: str
access_token: str # 加密存储，运行时解密
refresh_token: str
expires_at: int # 秒级时间戳
token_type: str = "session_key"
class TokenVault:
"""
聚石塔内 Token 保险库
- 公网 ERP 永远拿不到 access_token
- 只返回「能否调用 / 还剩多久 / 是否需要重新授权」
"""
def __init__(self):
# 实际用 KMS + RDS；这里用内存模拟
self._store: dict[str, SellerSession] = {}
def save(self, sess: SellerSession):
# TODO: 写入 RDS + KMS 加密
self._store[sess.seller_id] = sess
def get_usable_token(self, seller_id: str) -> Optional[str]:
sess = self._store.get(seller_id)
if not sess:
return None
if sess.expires_at - 300 < __import__("time").time():
# 距过期小于5分钟：订购类可刷，自研类只能重新授权
return None
return sess.access_token
def needs_reauth(self, seller_id: str) -> bool:
return self.get_usable_token(seller_id) is None
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
class XianyuChannelService:
"""
聚石塔内的闲鱼通道服务
公网 ERP 不允许直接调 TOP，只能调本服务的奇门接口
"""
def __init__(self, token_vault: TokenVault):
self.vault = token_vault
def publish_item(self, seller_id: str, item: dict) -> dict:
token = self.vault.get_usable_token(seller_id)
if not token:
return {"ok": False, "code": "TOKEN_EXPIRED", "need_reauth": True}
# 实际: requests.post(gw.api.taobao.com, method=alibaba.idle.isv.item.publish, session=token)
# 入塔后写操作只能在这里发生
return {
"ok": True,
"platform": "xianyu",
"seller_id": seller_id,
"called_from": "jushuitan-ecs",
"method": "alibaba.idle.isv.item.publish",
}
def ship_order(self, seller_id: str, order_id: str, company_code: str, waybill: str) -> dict:
token = self.vault.get_usable_token(seller_id)
if not token:
return {"ok": False, "code": "TOKEN_EXPIRED"}
# alibaba.idle.isv.order.ship
return {
"ok": True,
"order_id": order_id,
"logistics": f"{company_code}:{waybill}",
"note": "发货指令由聚石塔内发起，公网ERP只下业务决策",
}2. 消息消费：正向 + 逆向都按“快照 upsert”
# xianyu_channel/consumers.py import time class OrderStateSyncConsumer: """ idle_autotrade_OrderStateSync 正向订单状态变更：PENDING/PAID/PICKING/SHIPPED/SIGNED """ def __init__(self, erp_gateway): self.erp = erp_gateway # 奇门/内网：把统一事件发给公网ERP self.seen_msg_ids = set() def handle(self, msg: dict) -> str: msg_id = msg["msg_id"] if msg_id in self.seen_msg_ids: return "dup_ack" # 消息中间件 at-least-once self.seen_msg_ids.add(msg_id) order_id = msg["order_id"] status = msg["status"] modified = int(msg["modified"]) # 业务幂等：order_id + status + modified self.erp.upsert_order_status( order_id=order_id, status=status, modified=modified, source="xianyu:OrderStateSync", ) return "ok" # 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex class RefundSyncConsumer: """ idle_autotrade_RefundSync 逆向：退款/退货/平台介入 官方口径：退款消息是交易状态「最新版本」，不是事件计数 → 同 refund_id 推 5 次，以 modified 最大那次覆盖 """ def __init__(self, erp_gateway): self.erp = erp_gateway def handle(self, msg: dict) -> str: refund_id = msg["refund_id"] modified = int(msg["modified"]) # 状态机只前移，不回退 self.erp.apply_refund_snapshot( refund_id=refund_id, order_id=msg["order_id"], refund_status=msg["refund_status"], # WAIT_SELLER_AGREE/SUCCESS/FAILED/CLOSED refund_fee=msg["refund_fee"], modified=modified, source="xianyu:RefundSync", ) return "ok"
3. 公网 ERP 侧：只收“统一事件”，不碰 TOP
# erp_core/unified_event_handler.py
from enum import Enum
class UnifiedOrderStatus(str, Enum):
PENDING = "PENDING"
PAID = "PAID"
PICKING = "PICKING"
SHIPPED = "SHIPPED"
SIGNED = "SIGNED"
REFUNDING = "REFUNDING"
RETURNED = "RETURNED"
CANCELLED = "CANCELLED"
class ErpEventHandler:
"""
公网 ERP 收到聚石塔通道服务发来的统一事件
不再关心闲鱼/TOP/签名/Token
"""
def upsert_order_status(self, order_id, status, modified, source):
# 状态机只前移
print(f"[ERP] order={order_id} <- {status} @ {modified} from {source}")
def apply_refund_snapshot(self, refund_id, order_id, refund_status, refund_fee, modified, source):
print(f"[ERP] refund={refund_id} order={order_id} "
f"refund_status={refund_status} fee={refund_fee} modified={modified}")
if refund_status == "SUCCESS":
# 触发：释放库存 / 财务挂账 / 通知WMS拦截发货
pass4. 塔内外边界：奇门出口（而不是裸 HTTP）
# xianyu_channel/qimen_exit.py
class QimenExitGate:
"""
聚石塔规则：
塔内应用不得通过自定义接口与塔外系统交互；
确需出塔 → 奇门标准接口 + 平台审批
"""
ALLOWED_OUTBOUND_FIELDS = {
"order_id", "status", "sku", "qty",
"buyer_mask", "phone_mask", "address_mask",
"logistics_company", "waybill_no",
"refund_status", "refund_fee",
}
def export_to_public_erp(self, row: dict) -> dict:
leaked = set(row.keys()) - self.ALLOWED_OUTBOUND_FIELDS
if leaked:
raise PermissionError(f"禁止出塔字段: {leaked}")
return {k: row[k] for k in row if k in self.ALLOWED_OUTBOUND_FIELDS}五、迁移清单（踩坑顺序）
先判要不要入塔
只做选品/比价 → 不用入塔
要发商品 / 接单 / 发货 / 退款 → 必须入塔
应用入口收口
闲鱼：open.goofish.com
TOP 授权：oauth.taobao.com
聚石塔：console.cloud.tmall.com
买资源
ECS（跑通道服务）
RDS MySQL（订单/用户）
RocketMQ 4.x（交易/商品/退款消息）
可选：QPS 资源包、消息积压报警
代码切边
把所有
requests.post("gw.api.taobao.com", session=...)从公网 ERP 删掉挪到聚石塔
XianyuChannelService公网 ERP 只调自己内网/奇门接口
消息切推不拉
关掉“每 30s 拉一次订单”
订阅
idle_autotrade_OrderStateSync/RefundSync保留“主动查询兜底”但对账驱动，不当主链路
合规收口
PII 出塔前脱敏
Token 不出塔
审计日志落 RDS
塔外交互走奇门并报备
六、和前面系列的关系（收口）
前篇《能推不拉》：闲鱼入塔后推得更彻底——TMC/RocketMQ 是主，轮询只是兜底
前篇《幂等消费》：
msg_id去重不够，还要order_id+status+modified/refund_id+modified前篇《对账机制》：入塔后更要对账，因为消息可能重复/乱序/延迟
前篇《中台调度》：
XianyuAdapter内部必须知道“我这段代码只能在聚石塔跑”前篇《六大坑》：时区/税价/超卖/编码/映射/消息丢失——入塔后消息可靠性变好，但边界合规变成第 7 个坑
七、一句话收口
聚石塔强制入塔不是“部署方式变了”，而是二手 ERP 的闲鱼通道从「外挂脚本」升级成「合规交易系统」：写操作进塔、数据进塔、消息进塔；公网 ERP 只做业务大脑，不做淘系数据搬运工。
如果你愿意，下一篇可以直接写：
《聚石塔内 XianyuChannelService 的奇门出入参规范 + RocketMQ 消费位点恢复 + 公网 ERP 的“最小可信事件集”》
把“塔内/塔外”的字段边界直接写成 schema。
