《alibaba.idle.isv.order.ship 接入实录：闲鱼订单发货回传的隐式约束》（附Python源码）

alibaba.idle.isv.order.ship 表面是“传订单号+运单号”，实际有一堆文档不写但线上必炸的隐式约束：

session 必须用卖家 token、运单号要和 lc_code 配套、发货人信息影响菜鸟回传、重复发货不是幂等成功、订单状态不对直接报 ORDER_INVALID。

这个接口不是“调一次就完事”，而是交易闭环里最容易被状态机/编码/授权三重卡住的一跳。

一、接口定位（先别调错）

⚠️ 订单创建用买家 token，发货/关单/退款用卖家 token。双 token 模型搞反，返回“非法授权”而不是“参数错误”，非常难排。

二、官方入参里“标红/半隐藏”的约束

logistics_company 中文名“可选”，但 lc_code 才是机器可校验的快递公司标识。

你传“顺丰”不传 SF 可能能发；传“SF Express”又不传 lc_code 就可能同步异常。

三、隐式约束清单（实测/踩坑向）

1️⃣ Token 身份约束

• 发货接口 → 卖家 session

• 订单创建 → 买家 session

• 用错 token：不是 ship fail，是 invalid session / no auth

2️⃣ 订单状态约束

• PAID → 可发

• PENDING → 未付款，发了也白发

• REFUNDING / RETURNED / CANCELLED → 平台可能拒收

• 已 SHIPPED 再发 → 不是幂等，是“重复发货”

3️⃣ lc_code 必须来自字典

alibaba.idle.logistics.companies.query
  └── company_list: [{ code: "SF", name: "顺丰", id: 1 }, ...]

4️⃣ 运单号语义约束

• 一个 ship_mail_no 不要跨订单复用

• 虚拟商品别调这个接口（用 virtual delivery）

• 无物流发货别硬传 000000000000

5️⃣ 发货人信息影响菜鸟

• 接口可能返回成功

• 但菜鸟物流详情页同步不完整

• 买家看到“已发货但无轨迹” → 投诉/退款风险

6️⃣ 重复调用不是安全幂等

• (biz_order_id, ship_mail_no, lc_code) 去重

• 已发货订单先查 alibaba.idle.isv.order.query 再决定要不要发

7️⃣ 时间戳/签名

• timestamp 用 GMT+8

• 和服务器误差 > 10 分钟直接拒

• 聚石塔内也要走标准 TOP 签名，不是内网免签

四、生产向封装（聚石塔内）

# xianyu_channel/order_ship.py
import time
from dataclasses import dataclass
from enum import Enum
from typing import Optional


class ShipBlockReason(str, Enum):
    NOT_PAID = "not_paid"
    ALREADY_SHIPPED = "already_shipped"
    REFUNDING = "refunding"
    CANCELLED = "cancelled"
    TOKEN_EXPIRED = "token_expired"
    BAD_LOGISTICS_CODE = "bad_logistics_code"
    DUPLICATE_WAYBILL = "duplicate_waybill"


@dataclass
class ShipRequest:
    seller_id: str
    biz_order_id: str
    ship_mail_no: str
    lc_code: str                 # SF / YTO / ZTO / JD ...
    logistics_company: str = ""   # 中文名，兜底用
    sender_name: str = ""
    sender_phone: str = ""
    sender_address: str = ""
    sender_divisionid: Optional[int] = None


class LogisticsCodeTable:
    """
    启动时从 alibaba.idle.logistics.companies.query 预热
    存 RDS：erp_company_alias -> lc_code
    """

    def __init__(self):
        # alias -> lc_code
        self.alias_map = {
            "顺丰": "SF",
            "顺丰速运": "SF",
            "sf": "SF",
            "圆通": "YTO",
            "yto": "YTO",
            "中通": "ZTO",
            "zto": "ZTO",
            "韵达": "YD",
            "yd": "YD",
            "京东物流": "JD",
            "jd": "JD",
            "ems": "EMS",
            "邮政ems": "EMS",
        }

    def resolve(self, erp_company: str) -> Optional[str]:
        return self.alias_map.get(erp_company.strip().lower()) \
            or self.alias_map.get(erp_company.strip())

# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
class XianyuOrderShipClient:
    def __init__(self, token_vault, logistics_table: LogisticsCodeTable,
                 order_query_client):
        self.vault = token_vault
        self.lc = logistics_table
        self.order_query = order_query_client
        self.shipped_log: dict[str, str] = {}   # biz_order_id -> ship_mail_no

    def _precheck(self, req: ShipRequest) -> Optional[ShipBlockReason]:
        token = self.vault.get_usable_token(req.seller_id)
        if not token:
            return ShipBlockReason.TOKEN_EXPIRED

        lc_code = self.lc.resolve(req.lc_code) or self.lc.resolve(req.logistics_company)
        if not lc_code:
            return ShipBlockReason.BAD_LOGISTICS_CODE

        # 重复运单号
        if req.ship_mail_no in self.shipped_log.values():
            return ShipBlockReason.DUPLICATE_WAYBILL

        # 查订单状态（避免盲发）
        order = self.order_query.query(req.seller_id, req.biz_order_id)
        status = (order.get("status") or "").upper()

        if status in ("PENDING",):
            return ShipBlockReason.NOT_PAID
        if status in ("SHIPPED", "SIGNED"):
            return ShipBlockReason.ALREADY_SHIPPED
        if status in ("REFUNDING", "RETURNED"):
            return ShipBlockReason.REFUNDING
        if status in ("CANCELLED",):
            return ShipBlockReason.CANCELLED

        return None

    def ship(self, req: ShipRequest) -> dict:
        block = self._precheck(req)
        if block:
            return {
                "ok": False,
                "blocked": True,
                "reason": block.value,
                "order_id": req.biz_order_id,
            }

        lc_code = self.lc.resolve(req.lc_code) or self.lc.resolve(req.logistics_company)

        params = {
            "biz_order_id": req.biz_order_id,
            "ship_mail_no": req.ship_mail_no,
            "lc_code": lc_code,
            "logistics_company": req.logistics_company or lc_code,
        }
        if req.sender_name:
            params["sender_name"] = req.sender_name
        if req.sender_phone:
            params["sender_phone"] = req.sender_phone
        if req.sender_address:
            params["sender_address"] = req.sender_address
        if req.sender_divisionid:
            params["sender_divisionid"] = req.sender_divisionid

        # ===== 真实调用（伪代码）=====
        # token = self.vault.get_usable_token(req.seller_id)
        # resp = top_client.execute(
        #     method="alibaba.idle.isv.order.ship",
        #     session=token,
        #     params=params,
        #     timestamp=beijing_now(),
        # )
        # 这里用模拟返回
        resp = {"result_success": True, "result_err_code": "", "result_err_msg": ""}

        if resp.get("result_success"):
            self.shipped_log[req.biz_order_id] = req.ship_mail_no
            return {
                "ok": True,
                "order_id": req.biz_order_id,
                "lc_code": lc_code,
                "waybill": req.ship_mail_no,
                "note": "发货回传成功，菜鸟轨迹依赖 sender_* 完整性",
            }

        return {
            "ok": False,
            "blocked": False,
            "err_code": resp.get("result_err_code"),
            "err_msg": resp.get("result_err_msg"),
        }

五、公网 ERP 侧该怎么用

# erp_core/ship_decision.py
def decide_ship(order):
    if order["status"] != "PAID":
        return {"action": "skip", "reason": "not_paid"}
    if not order.get("logistics_company"):
        return {"action": "skip", "reason": "no_carrier"}
    if order.get("refund_status") in ("WAIT_SELLER_AGREE", "SUCCESS"):
        return {"action": "hold", "reason": "refund_in_progress"}
    return {
        "action": "ship_via_tower",
        "biz_order_id": order["platform_order_id"],
        "erp_company": order["logistics_company"],
        "waybill": order["waybill_no"],
        "sender": order["warehouse_sender"],
    }

1. 解析卖家 token

2. 反查 lc_code

3. 查订单状态

4. 调 alibaba.idle.isv.order.ship

5. 把结果回传公网 ERP（脱敏后）

六、接入检查表（上线前过一遍）

• [ ] 用卖家 session 调 ship

• [ ] lc_code 来自 alibaba.idle.logistics.companies.query

• [ ] 发货前查 alibaba.idle.isv.order.query 看状态

• [ ] ship_mail_no 全局唯一

• [ ] 虚拟商品走 goosefish.virtual.delivery

• [ ] sender_phone/name/address 至少仓库实发信息完整

• [ ] 重复发货有本地去重

• [ ] 发货失败进本地消息表，不等人工手点

• [ ] 出塔事件不含买家明文手机/地址

七、和前几篇的收口

• 《聚石塔入塔》：ship 必须在塔内发，公网 ERP 只出“发哪个仓库/哪个快递”

• 《幂等消费》：ship 本身不幂等，要在业务层做 (order_id, waybill, lc_code) 去重

• 《六大坑》：时区影响 timestamp、编码影响 sender_address、消息丢失靠对账补发

• 《中台调度》：XianyuAdapter.ship() 内部必须知道“我只跑在聚石塔”

一句话：

order.ship 的难点不在“怎么发”，而在“什么时候能发、用谁的身份发、运单和编码是否自洽、发了之后菜鸟能不能追到轨迹”。