🛡️《拼多多二手商品接口边界：平台对二手品类管控规则与ERP合规适配》（附Python源码）

一、拼多多二手管控的三重边界

关键认知：拼多多的成色不是闲鱼那种自由字段，而是被绑死在"类目属性体系"里——standard_id 是属性值ID，必须先查类目属性拿到合法取值。这和京东POP的"成色=某个attrId的valueId"思路一致（前篇 JdConditionResolver），但拼多多额外加了规格互斥这条。

二、standard_id：新旧程度属性

不同类目下具体取值可能略有差异，生产必须调 pdd.goods.cat.rule.get / pdd.goods.relation.cat.info.get 动态解析。源码里用 StandardIdResolver 模拟这套动态查表。

• 标题必须含成色关键词（二手、95新、9成新、官翻等）

• 详情必须如实描述瑕疵

• 同商品所有SKU的 standard_id 必须同属"全新组"或"非全新组"，不可混搭

三、规格互斥：多SKU的隐形地雷

• SKU-A：全新（standard_id=1001）

• SKU-B：95新（standard_id=1002）

违规 ❌:  Goods(商品X)
             ├─ SKU1: standard_id=1001 (全新)     ← 混搭=违规
             └─ SKU2: standard_id=1002 (95新)

合规 ✅:  Goods(商品X-全新)  SKU: standard_id=1001
         Goods(商品X-二手)   SKU: standard_id=1002  ← 按成色拆商品

四、Python：PddSecondHandGuard（合规适配层）

# pdd_secondhand_guard.py
"""
拼多多二手品类管控合规适配
- CatResolver: 类目是否为二手类目树 (cat_id 层级校验)
- StandardIdResolver: standard_id 动态解析 (1001全新/1002-1004非全新)
- TitleValidator: 标题成色词 vs standard_id 一致性
- SkuMutexChecker: 多SKU规格互斥 (禁止同时全新+非全新)
- GoodsComplianceGate: 发布前静态校验 (组合以上)
- GoodsPayloadBuilder: 组装 goods.add/edit 请求
复用前几篇: ConditionIntMapper(成色统一) / ComplianceGate / MasterSku
"""
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Set

# ==================== 常量 ====================
SECOND_HAND_ROOT_CATS = {50024728}   # 二手/闲置 一级类目 (示例, 生产以平台返回为准)
FRESH_STANDARD_ID = 1001             # 全新
NON_FRESH_STANDARD_IDS = {1002, 1003, 1004}

CONDITION_KEYWORDS = [
    "二手", "非全新", "准新", "99新", "95新", "9成新", "9新", "85新", "8成新",
    "官翻", "官换新", "资源机", "激活机", "拆封机", "样机", "展机",
]

# ==================== 类目解析器 ====================
class CatResolver:
    """判断 cat_id 是否在二手类目树下"""
    def __init__(self, root_cats: Set[int] = None):
        self.roots = root_cats or SECOND_HAND_ROOT_CATS
        # 缓存: cat_id -> 是否为二手类目
        self._cache: Dict[int, bool] = {}

    def is_second_hand(self, cat_id: int, parent_chain: List[int] = None) -> bool:
        if cat_id in self._cache:
            return self._cache[cat_id]
        # 方式1: 直接是根
        if cat_id in self.roots:
            self._cache[cat_id] = True
            return True
        # 方式2: 父类目链包含根 (生产调 cat.rule.get 拿层级)
        if parent_chain:
            if any(p in self.roots for p in parent_chain):
                self._cache[cat_id] = True
                return True
        self._cache[cat_id] = False
        return False

    def validate_cat(self, cat_id: int, parent_chain: List[int] = None) -> Optional[str]:
        if not self.is_second_hand(cat_id, parent_chain):
            return (f"cat_id={cat_id} 不在二手/闲置类目树下"
                    f"(须在 {self.roots} 及其子树下发布)")
        return None

# ==================== standard_id 解析器 ====================
class StandardIdResolver:
    """standard_id 动态解析: 统一成色 <-> 拼多多 vid"""
    # 生产: 调 pdd.goods.cat.rule.get 拿类目下属性, 再查属性值
    # 模拟: 类目 -> {成色文本: standard_id}
    CATE_STANDARD: Dict[int, Dict[str, int]] = {
        50024728: {"全新": 1001, "99新": 1002, "95新": 1002, "9成新": 1003,
                   "8成新": 1004, "官翻": 1002, "资源机": 1003, "激活机": 1004},
    }
    # 默认取值 (类目未配置时)
    DEFAULT = {"全新": 1001}

    def resolve(self, cat_id: int, condition_text: str) -> int:
        table = self.CATE_STANDARD.get(cat_id, self.DEFAULT)
        vid = table.get(condition_text)
        if vid is None:
            # 未知成色 -> 归为非全新(保守, 但要求调用方显式声明)
            raise PddMappingError(
                f"类目{cat_id}不支持成色'{condition_text}', 可选{list(table.keys())}")
        return vid

    def is_fresh(self, standard_id: int) -> bool:
        return standard_id == FRESH_STANDARD_ID

    def from_condition_int(self, cat_id: int, condition_int: int,
                           condition_text: str = "") -> int:
        """统一成色int -> standard_id (复用前篇 ConditionIntMapper 思路)"""
        text = condition_text or _int_to_text(condition_int)
        return self.resolve(cat_id, text)

def _int_to_text(v: int) -> str:
    from jd_pop_secondhand import ConditionIntMapper  # 复用前篇映射
    return ConditionIntMapper().to_text(v)

# ==================== 标题校验 ====================
class TitleValidator:
    def validate(self, title: str, standard_id: int) -> List[str]:
        issues = []
        is_fresh = standard_id == FRESH_STANDARD_ID
        has_kw = any(kw in title for kw in CONDITION_KEYWORDS)

        if is_fresh:
            # 全新商品标题不应含"二手/非全新"类词 (除非类目本身二手但标全新)
            bad = [kw for kw in ["二手", "非全新", "官翻", "资源机", "激活机"]
                    if kw in title]
            if bad:
                issues.append(f"standard_id=1001(全新)但标题含'{bad}', 语义冲突")
        else:
            # 非全新必须标题含成色词
            if not has_kw:
                issues.append("非全新商品标题须含成色词(二手/95新/官翻/激活机...)")
        return issues
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
# ==================== SKU 规格互斥检查 ====================
@dataclass
class SkuInput:
    sku_id: str
    standard_id: int
    stock: int = 0

class SkuMutexChecker:
    """同一商品下 SKU 不可同时全新(1001)+非全新"""
    def check(self, skus: List[SkuInput]) -> List[str]:
        issues = []
        if not skus:
            return issues
        has_fresh = any(s.standard_id == FRESH_STANDARD_ID for s in skus)
        has_non_fresh = any(s.standard_id in NON_FRESH_STANDARD_IDS for s in skus)
        if has_fresh and has_non_fresh:
            issues.append(
                f"多规格违规: 同时存在全新(standard_id=1001)与非全新SKU, "
                f"须拆分为两个商品")
        return issues

# ==================== 发布前合规门禁 ====================
@dataclass
class GoodsDraft:
    cat_id: int
    parent_chain: List[int] = field(default_factory=list)
    goods_name: str = ""
    standard_id: int = 0            # 主商品新旧程度
    condition_text: str = ""
    condition_int: int = 0
    sku_list: List[SkuInput] = field(default_factory=list)
    detail_desc: str = ""

class GoodsComplianceGate:
    """组合校验: 类目 + 属性 + 标题 + 规格互斥"""
    def __init__(self, cat_resolver: CatResolver = None,
                 std_resolver: StandardIdResolver = None):
        self.cat = cat_resolver or CatResolver()
        self.std = std_resolver or StandardIdResolver()
        self.title = TitleValidator()
        self.mutex = SkuMutexChecker()

    def validate(self, draft: GoodsDraft) -> List[str]:
        issues: List[str] = []

        # 1. 类目围栏
        c = self.cat.validate_cat(draft.cat_id, draft.parent_chain)
        if c: issues.append(c)

        # 2. standard_id 解析 (若未显式设置, 从成色推导)
        sid = draft.standard_id
        if sid == 0:
            try:
                sid = self.std.resolve(draft.cat_id, draft.condition_text)
                draft.standard_id = sid
            except PddMappingError as e:
                issues.append(str(e))

        # 3. 标题 vs standard_id 一致性
        if sid:
            issues += self.title.validate(draft.goods_name, sid)

        # 4. SKU 规格互斥 (用每个SKU自己的standard_id)
        skus = draft.sku_list or [SkuInput("main", sid)]
        issues += self.mutex.check(skus)

        # 5. 非全新必须详情描述瑕疵
        if sid and not self.std.is_fresh(sid):
            if draft.condition_text and draft.condition_text not in draft.detail_desc:
                issues.append("非全新商品详情页须如实描述成色/瑕疵")

        return issues
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
# ==================== 请求组装 ====================
class GoodsPayloadBuilder:
    """校验通过后组装 goods.add / goods.edit 请求"""
    def build(self, draft: GoodsDraft) -> Dict:
        issues = GoodsComplianceGate().validate(draft)
        if issues:
            raise PddValidationError(issues)

        properties = [{"template_id": 1, "value": str(draft.standard_id)}]
        sku_list = [
            {"sku_id": s.sku_id, "stock": s.stock, "standard_id": s.standard_id}
            for s in draft.sku_list
        ] or [{"sku_id": "0", "stock": 0, "standard_id": draft.standard_id}]

        return {
            "type": "pdd.goods.add",
            "cat_id": draft.cat_id,
            "goods_name": draft.goods_name,
            "goods_properties": properties,   # 新旧程度等属性
            "sku_list": sku_list,
            "detail_desc": draft.detail_desc,
        }

# ==================== 异常 ====================
class PddMappingError(Exception): pass
class PddValidationError(Exception):
    def __init__(self, issues: List[str]):
        self.issues = issues
        super().__init__("; ".join(issues))

# ==================== 演示 ====================
if __name__ == "__main__":
    gate = GoodsComplianceGate()
    builder = GoodsPayloadBuilder()

    cases = [
        ("✅ 合规: 95新单SKU", GoodsDraft(
            cat_id=50024728, goods_name="95新 Apple iPhone 13 128G 黑色",
            condition_text="95新", standard_id=0, detail_desc="95新: 细微划痕,功能正常")),
        ("✅ 合规: 全新(标题无二手词)", GoodsDraft(
            cat_id=50024728, goods_name="全新 Apple AirPods Pro",
            condition_text="全新", standard_id=1001, detail_desc="全新未拆封")),
        ("❌ 跨类目: 不在二手树", GoodsDraft(
            cat_id=1, goods_name="二手手机",   # 假设cat_id=1非二手根
            condition_text="95新", standard_id=0)),
        ("❌ 标题无成色词(非全新)", GoodsDraft(
            cat_id=50024728, goods_name="Apple iPhone 13 128G",
            condition_text="95新", standard_id=0, detail_desc="细节描述")),
        ("❌ 规格互斥: 全新+95新", GoodsDraft(
            cat_id=50024728, goods_name="Apple iPhone 13",
            condition_text="95新", standard_id=1002,
            sku_list=[SkuInput("s1", 1001), SkuInput("s2", 1002)])),
        ("❌ 全新标题却写二手", GoodsDraft(
            cat_id=50024728, goods_name="二手全新 AirPods",
            condition_text="全新", standard_id=1001)),
    ]

    for name, draft in cases:
        try:
            payload = builder.build(draft)
            print(f"{name}\n  ✅ 通过 → {payload['type']} cat={payload['cat_id']} "
                  f"props={payload['goods_properties']}")
        except PddValidationError as e:
            print(f"{name}\n  🚫 拦截:")
            for i in e.issues:
                print(f"    · {i}")

    print("\n=== standard_id 动态解析 ===")
    resolver = StandardIdResolver()
    for cond in ["全新", "95新", "官翻", "激活机", "8成新"]:
        print(f"  {cond} → standard_id={resolver.resolve(50024728, cond)}")

✅ 合规: 95新单SKU
  ✅ 通过 → pdd.goods.add cat=50024728 props=[{'template_id': 1, 'value': '1002'}]

✅ 合规: 全新(标题无二手词)
  ✅ 通过 → pdd.goods.add cat=50024728 props=[{'template_id': 1, 'value': '1001'}]

❌ 跨类目: 不在二手树
  🚫 拦截:
    · cat_id=1 不在二手/闲置类目树下(须在 {50024728} 及其子树下发布)

❌ 标题无成色词(非全新)
  🚫 拦截:
    · 非全新商品标题须含成色词(二手/95新/官翻/激活机...)
    · 非全新商品详情页须如实描述成色/瑕疵

❌ 规格互斥: 全新+95新
  🚫 拦截:
    · 多规格违规: 同时存在全新(standard_id=1001)与非全新SKU, 须拆分为两个商品

❌ 全新标题却写二手
  🚫 拦截:
    · standard_id=1001(全新)但标题含'['二手']', 语义冲突

=== standard_id 动态解析 ===
  全新 → 1001
  95新 → 1002
  官翻 → 1002
  激活机 → 1004
  8成新 → 1004

五、五个合规铁律

1. 类目是第一道闸门：cat_id 必须在二手/闲置树下，否则后面全白搭。生产必须缓存类目树（调 cat.rule.get），不能硬编码。

2. standard_id 必须显式设置：默认推导只用于兜底，ERP应把成色→standard_id的映射做成配置表，GoodsComplianceGate 在校验时强制填充。

3. 标题是声明不是营销：非全新商品的成色词是合规声明，平台会校验标题与属性一致性——"标题写95新、属性标全新"=违规。

4. 多规格互斥要拆商品：有"全新+二手"混合需求的场景（如手机），必须在ERP侧拆成两个 master_sku，别指望一个商品挂两个SKU蒙混过关。

5. 详情如实描述是底线：非全新商品详情页必须描述瑕疵，否则属虚假宣传——这条接入前篇 ComplianceGate 的内容合规维度统一审计。

六、和前几篇的衔接

把 PddSecondHandGuard 作为 commerce-mesh/adapters/pdd/ 的发布前门禁：

• 成色中枢：StandardIdResolver 与闲鱼 stuff_status int / 抖音枚举 / 京东 attrId×valueId / 转转 quality 共享统一 condition_int，写回时各平台展开；

• 规格互斥检查 SkuMutexChecker 接入前篇 StockEngine——拼多多二手SKU的库存扣减必须按"成色分组"，混组会导致平台侧超卖判定异常；

• 类目校验 CatResolver 的结果缓存进前篇 TwoLevelCache（L1 60s + L2 5min），类目树变更走主动失效；

• 违规预扫描：GoodsComplianceGate.validate() 做成每日定时任务，扫全店草稿+在售，输出"违规清单"给运营（呼应前篇 PublishVerifier 的批量扫描思路）；

• 属性动态解析：生产接入 pdd.goods.cat.rule.get + pdd.goods.relation.cat.info.get，把 standard_id 映射表从硬编码升级为接口预热+YAML兜底双通道。
  拼多多二手对接的本质，是把"平台管控规则"翻译成"代码里的不可绕过校验"——规则在接口之外，但合规必须在发布之前。