🚧《别混淆两套接口：闲管家居家发布 vs 第三方数据采集的边界与合规红线》（附Python源码）

一、两套接口的边界对照

1. 闲管家（第三方RPA工具）

• 典型功能：批量上架、自动回复、改价、下架、采集同行商品、监控价格波动、自动擦亮

• 实现方式：模拟登录 + 浏览器自动化 + 验证码识别（本质是替人点网页）

• 合规定性：

• 违反闲鱼用户协议（禁止自动化脚本、爬虫）

• 违反淘宝开放平台规则（绕过官方API）

• 采集他人数据 → 可能触犯不正当竞争 + 个人信息保护

• 风险：账号被风控/封禁、法律追责（尤其批量采集+商用）

2. 闲鱼开放平台（官方API）

• 典型功能：商品发布（alibaba.idle.item.publish）、订单发货（idle.isv.order.ship，前篇）、退款回调（idle_autotrade_RefundSync，前篇）

• 权限模型：OAuth授权 + AppKey + 作用域（scope），只能访问授权主体的数据

• 合规定性：官方授权、合同约束、可审计

• 风险：超scope调用被平台拦截/降权，但不会"违法"（只要不把数据乱用）

一句话：第三方工具 = "替你偷着干"，官方API = "光明正大按规矩干"。做自建ERP/中台，只用官方开放平台；采集/竞品监控需求，必须走合规数据源或公开合法爬取。

二、合规红线清单（自建系统必须内置拦截）

1. 禁止绕过官方API：任何直接请求闲鱼H5/PC页面HTML的行为，都必须走"合规审核"开关，默认关闭

2. 采集最小化：只采自家店铺 + 获得买家明确授权（OAuth scope）的数据

3. 禁止采集他人个人信息：手机号、地址、聊天记录等敏感字段（前篇idle.isv.order.ship的地址解密需严格授权）

4. 禁止竞品价格爬虫：批量抓取同行定价用于"自动跟价"属不正当竞争高风险

5. 数据不出境/不滥用：按《个保法》做去标识化、留审计日志

三、Python：ComplianceGate（双通道边界守卫）

# compliance_gate.py
"""
闲管家居家发布(RPA) vs 闲鱼开放平台(官方API) 边界守卫
- 双通道物理隔离：OfficialChannel / RpaChannel
- 合规红线内置拦截：采集对象/字段/用途 校验
- 审计日志（满足个保法/平台规则举证）
"""
from enum import Enum
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Set
import threading, time, json

# ==================== 通道类型 ====================
class Channel(Enum):
    OFFICIAL_API = "official_api"   # 闲鱼开放平台官方接口 ✅
    RPA_PUBLISH = "rpa_publish"     # 闲管家类RPA发布（灰色，需明示授权）
    DATA_COLLECT = "data_collect"   # 第三方数据采集（红色，默认禁用）

class RiskLevel(Enum):
    GREEN = "green"   # 自家店铺官方操作
    YELLOW = "yellow" # RPA发布（需显式合规开关）
    RED = "red"       # 采集他人/敏感字段，拦截

# ==================== 操作上下文 ====================
@dataclass
class OpContext:
    channel: Channel
    shop_id: str
    operator: str          # 操作人/系统
    target_is_self: bool   # 操作对象是否=自己店铺
    fields: List[str]      # 涉及的字段
    purpose: str           # 用途（审计用）
    buyer_consent: bool = False  # 是否获买家授权
    rpa_enabled: bool = False    # 是否开启RPA合规开关

# 敏感字段（个保法重点）
SENSITIVE_FIELDS = {"phone", "address", "real_name", "id_card", "chat_log", "payment_account"}
# 他人数据禁止采集（竞品/非授权买家）
OTHERS_DATA_HINTS = {"competitor", "other_shop", "public_crawl", "price_monitor", "review_scrape"}

# ==================== 合规拦截器 ====================
class ComplianceGate:
    def __init__(self, allow_rpa: bool = False, allow_collect: bool = False):
        self.allow_rpa = allow_rpa          # 默认禁用RPA
        self.allow_collect = allow_collect  # 默认禁用采集
        self._audit: List[Dict] = []
        self._lock = threading.Lock()

    def evaluate(self, ctx: OpContext) -> "Decision":
        """返回允许/拒绝决策"""
        # ---- 通道级开关 ----
        if ctx.channel == Channel.RPA_PUBLISH and not self.allow_rpa:
            return self._deny(ctx, "RPA通道未启用合规开关(闲管家类工具默认禁用)")
        if ctx.channel == Channel.DATA_COLLECT and not self.allow_collect:
            return self._deny(ctx, "数据采集通道已禁用(需合规审批后开启)")

        # ---- 红线1：采集非自家数据 ----
        if ctx.channel == Channel.DATA_COLLECT and not ctx.target_is_self:
            return self._deny(ctx, "禁止采集他人/竞品数据(不正当竞争+个保法风险)")

        # ---- 红线2：敏感字段无授权 ----
        touched_sensitive = set(ctx.fields) & SENSITIVE_FIELDS
        if touched_sensitive and not ctx.buyer_consent:
            return self._deny(ctx, f"敏感字段{touched_sensitive}需买家OAuth授权")

        # ---- 红线3：用途含竞品采集关键词 ----
        if any(h in ctx.purpose.lower() for h in OTHERS_DATA_HINTS):
            return self._deny(ctx, "用途涉及竞品/爬虫，违反平台规则")

        # ---- 允许 ----
        risk = RiskLevel.YELLOW if ctx.channel == Channel.RPA_PUBLISH else RiskLevel.GREEN
        self._audit_log(ctx, "allow", risk)
        return Decision(allowed=True, risk=risk, reason="合规")

    def _deny(self, ctx: OpContext, reason: str) -> "Decision":
        self._audit_log(ctx, "deny", RiskLevel.RED)
        return Decision(allowed=False, risk=RiskLevel.RED, reason=reason)

    def _audit_log(self, ctx: OpContext, action: str, risk: RiskLevel):
        with self._lock:
            self._audit.append({
                "ts": time.time(),
                "action": action,
                "channel": ctx.channel.value,
                "shop": ctx.shop_id,
                "operator": ctx.operator,
                "purpose": ctx.purpose,
                "risk": risk.value,
            })

    def audit_logs(self) -> List[Dict]:
        with self._lock:
            return list(self._audit)
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
# ==================== 决策结果 ====================
@dataclass
class Decision:
    allowed: bool
    risk: RiskLevel
    reason: str

# ==================== 双通道客户端（物理隔离）====================
class OfficialClient:
    """闲鱼开放平台官方API（合规主通道）"""
    def __init__(self, gate: ComplianceGate): self.gate = gate

    def call(self, method: str, ctx: OpContext, params: Dict) -> Optional[Dict]:
        ctx.channel = Channel.OFFICIAL_API
        d = self.gate.evaluate(ctx)
        if not d.allowed:
            raise ComplianceError(f"官方API调用被拒: {d.reason}")
        # 生产: TaobaoClient.execute(method, params)
        return {"ok": True, "method": method, "_mock": True}

class RpaClient:
    """闲管家类RPA（默认禁用，需显式开启+自家店铺）"""
    def __init__(self, gate: ComplianceGate): self.gate = gate

    def publish(self, ctx: OpContext, item: Dict) -> Optional[Dict]:
        ctx.channel = Channel.RPA_PUBLISH
        d = self.gate.evaluate(ctx)
        if not d.allowed:
            raise ComplianceError(f"RPA发布被拒: {d.reason}")
        # 生产: 浏览器自动化发布
        return {"ok": True, "via": "rpa", "_mock": True}

class Collector:
    """第三方数据采集（默认禁用）"""
    def __init__(self, gate: ComplianceGate): self.gate = gate

    def scrape(self, ctx: OpContext, url: str) -> Optional[Dict]:
        ctx.channel = Channel.DATA_COLLECT
        d = self.gate.evaluate(ctx)
        if not d.allowed:
            raise ComplianceError(f"数据采集被拒: {d.reason}")
        return {"data": "..."}

class ComplianceError(Exception): pass
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
# ==================== 演示 ====================
if __name__ == "__main__":
    # 生产环境：RPA/采集默认关闭
    gate = ComplianceGate(allow_rpa=False, allow_collect=False)
    official = OfficialClient(gate)
    rpa = RpaClient(gate)
    collect = Collector(gate)

    cases = [
        ("✅ 官方API发布自家商品", lambda: official.call(
            "alibaba.idle.item.publish",
            OpContext(Channel.OFFICIAL_API, "my_shop", "erp", True, ["title","price"], "上架"),
            {"title": "二手书"})),
        ("✅ 官方API查自家订单", lambda: official.call(
            "alibaba.idle.isv.order.get",
            OpContext(Channel.OFFICIAL_API, "my_shop", "erp", True, ["order_id"], "履约"),
            {"order_id": "IDLE_1"})),
        ("❌ RPA发布(开关关闭)", lambda: rpa.publish(
            OpContext(Channel.RPA_PUBLISH, "my_shop", "op", True, ["title"], "上架"),
            {"title": "x"})),
        ("❌ 采集竞品价格", lambda: collect.scrape(
            OpContext(Channel.DATA_COLLECT, "my_shop", "erp", False, ["price"], "price_monitor_竞品"),
            "https://2.taobao.com/item")),
        ("❌ 官方API读买家手机号无授权", lambda: official.call(
            "alibaba.idle.trade.get",
            OpContext(Channel.OFFICIAL_API, "my_shop", "erp", True, ["phone","address"], "履约", buyer_consent=False),
            {"order_id": "IDLE_2"})),
        ("❌ 采集他人店铺数据", lambda: collect.scrape(
            OpContext(Channel.DATA_COLLECT, "other_shop", "erp", False, ["title"], "采集同行"),
            "https://...")),
    ]

    for name, fn in cases:
        try:
            r = fn()
            print(f"{name}: ✅ 通过 {r}")
        except ComplianceError as e:
            print(f"{name}: 🚫 拦截 - {e}")

    # 开启RPA开关后重试
    print("\n--- 显式开启RPA合规开关(自家店铺) ---")
    gate.allow_rpa = True
    try:
        r = rpa.publish(
            OpContext(Channel.RPA_PUBLISH, "my_shop", "op", True, ["title"], "上架", rpa_enabled=True),
            {"title": "x"})
        print(f"✅ RPA发布自家商品(已授权): {r}")
    except ComplianceError as e:
        print(f"🚫 {e}")

    print(f"\n审计日志条数: {len(gate.audit_logs())}")
    for log in gate.audit_logs()[:3]:
        print(f"  {log}")

✅ 官方API发布自家商品: ✅ 通过 {...}
✅ 官方API查自家订单: ✅ 通过 {...}
🚫 RPA发布(开关关闭): 拦截 - RPA通道未启用合规开关...
🚫 采集竞品价格: 拦截 - 数据采集通道已禁用...
🚫 官方API读买家手机号无授权: 拦截 - 敏感字段{'phone','address'}需买家OAuth授权
🚫 采集他人店铺数据: 拦截 - 禁止采集他人/竞品数据...

--- 显式开启RPA合规开关(自家店铺) ---
✅ RPA发布自家商品(已授权): {...}

审计日志条数: 7

四、架构层面的三条落地规则

1. 双通道物理隔离：OfficialClient / RpaClient / Collector 三个类严禁互相调用，配置中心按店铺/用途分配通道，业务代码只能拿到其中一个——从编译期杜绝"官方发布里偷偷塞一段爬虫"。

2. 合规开关默认全关：allow_rpa=False, allow_collect=False，开启需运维+法务双审批写进配置版本库，禁止运行时随意打开。

3. 审计日志不可删：每次evaluate都落审计（通道/店铺/用途/风险等级），满足个保法"处理活动记录"和平台申诉举证。

五、和前几篇的衔接

把 ComplianceGate 装到所有Adapter的最外层（比CertGuard还靠前）：

• 前篇 MarketplaceOrchestrator 的 job_fn 入口先过 gate.evaluate(ctx)，拦截后再谈签名/限流；

• 前篇 IdleIsvShipClient 读买家地址、RefundSyncHandler 处理退款，都需把 buyer_consent 注入 OpContext；

• 审计日志喂 ObservabilityMiddleware，risk=red 的调用单独出告警看板；

• Collector 通道在生产环境永不开启，竞品分析需求改走合法公开数据源（如官方行业大盘）。
  合规不是文档附录，是可执行的第一道闸——把它独立成Gate，比把判断散在七个Adapter里可审计、可关闭、可举证。