🔍《闲鱼商品发布数量上限与运营提效包：接口调成功 ≠ APP 一定展示的排查手册》（附Python源码）
结论先拍：
alibaba.idle.isv.item.publish 返回 success=true 只代表"平台收到了请求"，不代表商品在闲鱼APP/搜索里能看见。我们复盘过一批"接口全绿、APP零曝光"的案例，根因分布是：① 发布数量上限（个人号每日/总量天花板，超限静默截断）② 类目资质/品牌资质未审核 ③ 敏感词/违规词命中下架 ④ 图片/视频未通过审核 ⑤ item_biz_type 错配被限流 ⑥ 草稿未正式上架（publish_status 停留在草稿）⑦ 重复商品被聚合去重。 排查的正确姿势不是"再调一次"，而是发布成功 → 立即回查 item.get → 比对展示状态码 → 分层定位，本手册把这套流程代码化了。一、"成功"与"展示"之间的七道闸门
接口返回 success=true │ ▼ ① 发布数量上限（每日/总量） │ 个人号日发有限，超限→接口仍success但进"待审核/草稿池" ▼ ② 类目/品牌资质 │ 手机/数码/奢品需资质，缺→不展示 ▼ ③ 内容审核（敏感词/违规词/违禁品） │ 命中→下架或仅自己可见 ▼ ④ 图片/视频审核 │ 审核中→不展示，驳回→下架 ▼ ⑤ item_biz_type 语义匹配 │ 个人闲置带sp_biz_type→被识别为商家，流量降权 ▼ ⑥ 发布状态机 │ publish_status≠ON_SALE → 草稿/待审 ▼ ⑦ 重复/违规聚合 同质商品→去重合并，主商品被替换 │ ▼ APP/搜索 可见 ✅
关键认知：① 和 ⑥ 是最常见的"静默失败"——接口不报错，商品进了草稿池或审核队列，运营以为发了50件实际只上架3件。
二、发布数量上限（官方口径）
维度 | 个人号（C2C） | 商家号 / ISV |
|---|---|---|
每日发布上限 | 约 50~100 件/天（视账号等级/信用） | 按应用配额，通常更高 |
在售总量上限 | 约 500~2000 件（芝麻信用/活跃度影响） | 视类目授权 |
超限表现 | 接口仍返回 success，但实际进入草稿/待审核 | 部分接口返回 sub_code=isv.item-limit-exceeded |
突破方式 | 提升信用/认证、升级商家号 | 申请类目配额、用多店铺分流 |
⚠️ 血泪点：个人号超限时接口不报错是设计如此（防探测），所以"我发了200件怎么只显示3件"的锅只能靠发布后回查背。
三、排查决策树
publish 返回 success │ ├─ ① 回查 item.get 拿到 publish_status │ ├─ ON_SALE(在售) → 进 ②③④⑤ │ ├─ DRAFT(草稿) → 未正式发布，调 publish 二次确认 │ ├─ AUDIT(审核中) → 等待/查审核状态 │ └─ OFF_SHELF/REJECT → 看 reject_reason │ ├─ ② 数量上限检查 │ today_published >= daily_limit? → 次日/提额 │ total_on_sale >= total_limit? → 下架清理/多店铺 │ ├─ ③ 资质检查（类目/品牌） │ category/brand 是否需要额外授权 │ ├─ ④ 内容审核（敏感词/图片） │ title/desc 敏感词扫描 → 自动改写 │ 图片审核状态 → 等待/重传 │ ├─ ⑤ item_biz_type 匹配检查 │ 个人闲置 不应带 sp_biz_type（前篇映射规则） │ └─ ⑥ 去重/聚合检查 同图同标题 → 被合并，换图/改标题
四、Python：PublishVerifier（发布成功→APP展示 全链路排查器）
# idle_publish_verifier.py
"""
闲鱼商品发布: 接口成功 ≠ APP展示 排查手册
- 发布后立即回查 publish_status 状态机
- 数量上限检查 (每日/总量)
- 资质/敏感词/图片审核/语义匹配/去重 分层排查
- 输出可定位的 DiagnoseReport
复用前几篇: IdleItemPublishMapper(字段映射) / ComplianceGate / RateLimiter
"""
import time, re, threading
from typing import Dict, List, Optional, Set
from dataclasses import dataclass, field
from enum import Enum
# ==================== 发布状态码 ====================
class PublishStatus(Enum):
ON_SALE = "ON_SALE" # 在售(展示)
DRAFT = "DRAFT" # 草稿(未正式发布)
AUDIT = "AUDIT" # 审核中
OFF_SHELF = "OFF_SHELF" # 已下架
REJECT = "REJECT" # 审核驳回
UNKNOWN = "UNKNOWN"
# 敏感词库(示例, 生产接内容安全API)
SENSITIVE_WORDS = {
"高仿", "精仿", "复刻", "原单", "跟单", # 侵权/假货
"走私", "赃物", "来路不明",
"VX", "微信", "QQ号", "加我", # 导流
"最便宜", "全网最低", "第一", # 极限词
}
# 需资质的类目
CATEGORY_NEED_QUALIFICATION = {
"手机", "数码", "奢品", "手表", "包袋", "珠宝", "文玩",
}
# ==================== 诊断结果 ====================
@dataclass
class DiagnoseReport:
outer_id: str
api_success: bool
visible: bool = False
publish_status: PublishStatus = PublishStatus.UNKNOWN
issues: List[str] = field(default_factory=list)
suggestions: List[str] = field(default_factory=list)
retryable: bool = False
def summary(self) -> Dict:
return {
"outer_id": self.outer_id,
"api_success": self.api_success,
"visible": self.visible,
"publish_status": self.publish_status.value,
"issues": self.issues,
"suggestions": self.suggestions,
"retryable": self.retryable,
}
# ==================== 数量上限配额 ====================
@dataclass
class QuotaConfig:
daily_limit: int = 80 # 每日发布上限
total_limit: int = 1000 # 在售总量上限
class QuotaTracker:
"""按店铺追踪发布数量(持久化到Redis生产环境)"""
def __init__(self, cfg: QuotaConfig = QuotaConfig()):
self.cfg = cfg
self.today: Dict[str, int] = {} # shop_id -> 今日已发
self.total: Dict[str, Set[str]] = {} # shop_id -> 在售outer_id集合
self._lock = threading.Lock()
def check(self, shop_id: str, outer_id: str) -> Optional[str]:
with self._lock:
today = self.today.get(shop_id, 0)
if today >= self.cfg.daily_limit:
return f"每日发布上限({self.cfg.daily_limit}/天), 今日已发{today}"
total = len(self.total.get(shop_id, set()))
if total >= self.cfg.total_limit:
return f"在售总量上限({self.cfg.total_limit}), 需下架清理或多店铺分流"
return None
def mark_published(self, shop_id: str, outer_id: str):
with self._lock:
self.today[shop_id] = self.today.get(shop_id, 0) + 1
self.total.setdefault(shop_id, set()).add(outer_id)
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
# ==================== 排查器 ====================
class PublishVerifier:
"""发布成功 → APP展示 全链路排查"""
def __init__(self, quota: QuotaTracker,
item_getter: Optional[callable] = None):
self.quota = quota
self.item_getter = item_getter # (shop_id, outer_id) -> dict(含publish_status等)
# ---- 主入口 ----
def diagnose(self, shop_id: str, outer_id: str,
publish_resp: Dict, item: Dict) -> DiagnoseReport:
rep = DiagnoseReport(outer_id=outer_id,
api_success=publish_resp.get("success", False))
# ① 接口本身失败
if not rep.api_success:
rep.issues.append("接口调用失败(非success)")
rep.suggestions.append("查sub_code/sub_msg, 按错误码重试")
rep.retryable = True
return rep
# ② 数量上限预检(发布前也应调一次, 这里是双重确认)
q = self.quota.check(shop_id, outer_id)
if q:
rep.issues.append(q)
rep.suggestions.append("次日发布 / 提升账号等级认证 / 多店铺分流")
# ③ 回查发布状态(核心)
remote = self._query_item(shop_id, outer_id) or {}
status = self._parse_status(remote.get("publish_status"))
rep.publish_status = status
if status == PublishStatus.DRAFT:
rep.issues.append("商品停留在草稿, 未正式上架")
rep.suggestions.append("确认调用publish而非save_draft; 检查是否漏传必填")
elif status == PublishStatus.AUDIT:
rep.issues.append("审核中, 暂未展示")
rep.suggestions.append("等待审核(通常分钟级), 查审核状态接口")
elif status == PublishStatus.REJECT:
reason = remote.get("reject_reason", "未知")
rep.issues.append(f"审核驳回: {reason}")
rep.suggestions.append("按驳回原因修改后重发")
elif status == PublishStatus.OFF_SHELF:
rep.issues.append("已下架(违规/主动)")
rep.suggestions.append("查违规记录/重新上架")
elif status == PublishStatus.ON_SALE:
rep.visible = True
# ④ 内容审核: 敏感词
title = item.get("title", "")
desc = item.get("desc", "")
hit = self._scan_sensitive(title + " " + desc)
if hit:
rep.issues.append(f"命中敏感词/极限词: {hit}")
rep.suggestions.append("改写为合规表述(去引流/去极限/去侵权)")
# ⑤ 类目资质
cat = item.get("category", "")
if any(c in cat for c in CATEGORY_NEED_QUALIFICATION) and not item.get("qualified", False):
rep.issues.append(f"类目'{cat}'需额外资质(授权/认证)")
rep.suggestions.append("补全类目资质或换普通类目")
# ⑥ item_biz_type 语义匹配(复用前篇映射规则)
biz = item.get("item_biz_type")
sp = item.get("sp_biz_type")
is_personal = item.get("is_personal_idle", False)
if is_personal and sp:
rep.issues.append("个人闲置误传sp_biz_type→被识别为商家, 流量降权")
rep.suggestions.append("个人闲置剥掉sp_biz_type/inspected_data(见Mapper)")
if biz == "2" and not is_personal:
rep.issues.append("item_biz_type=2(个人闲置)但非个人商品, 语义矛盾")
# ⑦ 去重/聚合(简化: 同图提示)
if item.get("duplicate_images", False):
rep.issues.append("图片与在售商品高度重复, 可能被聚合去重")
rep.suggestions.append("更换首图/调整标题差异化")
# 最终可见判定: ON_SALE且无致命问题
if rep.visible and not rep.issues:
rep.visible = True
elif rep.visible and any("驳回" in i or "敏感" in i or "资质" in i for i in rep.issues):
rep.visible = False # 在售但有问题→实际不可见
return rep
# ---- 批量扫描(运营提效) ----
def batch_scan(self, shop_id: str, items: List[Dict]) -> Dict:
"""扫一批商品的展示健康度"""
reports = [self.diagnose(shop_id, it["outer_id"],
{"success": True}, it) for it in items]
visible = sum(1 for r in reports if r.visible)
by_status: Dict[str, int] = {}
all_issues: List[str] = []
for r in reports:
by_status[r.publish_status.value] = by_status.get(r.publish_status.value, 0) + 1
all_issues.extend([f"{r.outer_id}: {i}" for i in r.issues])
return {
"total": len(reports),
"visible": visible,
"invisible": len(reports) - visible,
"by_status": by_status,
"issues": all_issues[:50], # 截取前50条
}
# ---- 内部 ----
def _query_item(self, shop_id: str, outer_id: str) -> Optional[Dict]:
if self.item_getter:
try:
return self.item_getter(shop_id, outer_id)
except Exception:
return None
# 演示: 模拟回查(生产调 alibaba.idle.item.get)
return {"publish_status": "ON_SALE", "reject_reason": ""}
def _parse_status(self, raw) -> PublishStatus:
if raw is None: return PublishStatus.UNKNOWN
s = str(raw).upper()
try: return PublishStatus(s)
except ValueError: return PublishStatus.UNKNOWN
def _scan_sensitive(self, text: str) -> Set[str]:
return {w for w in SENSITIVE_WORDS if w in text}
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
# ==================== 演示 ====================
if __name__ == "__main__":
quota = QuotaTracker(QuotaConfig(daily_limit=80, total_limit=1000))
verifier = PublishVerifier(quota=quota)
# 模拟今日已发80件(触顶)
for i in range(80):
quota.mark_published("shop_001", f"OUT_{i}")
items = [
# 正常商品
{"outer_id": "OUT_GOOD", "title": "二手Kindle 99新", "desc": "自用",
"category": "电子", "item_biz_type": "2", "is_personal_idle": True},
# 敏感词
{"outer_id": "OUT_SENS", "title": "高仿劳力士 全网最低",
"desc": "加我VX看货", "category": "手表", "qualified": True},
# 个人闲置误传sp_biz_type
{"outer_id": "OUT_BIZ", "title": "闲置书", "desc": "",
"item_biz_type": "2", "sp_biz_type": "21", "is_personal_idle": True},
# 需资质类目未认证
{"outer_id": "OUT_QUAL", "title": "二手iPhone", "category": "手机"},
# 数量上限
{"outer_id": "OUT_LIMIT", "title": "又一件", "is_personal_idle": True},
# 审核驳回(模拟回查)
{"outer_id": "OUT_REJ", "title": "x", "is_personal_idle": True},
]
# 让驳回商品的回查返回 REJECT
def fake_getter(shop_id, oid):
if oid == "OUT_REJ":
return {"publish_status": "REJECT", "reject_reason": "图片含联系方式"}
if oid == "OUT_SENS":
return {"publish_status": "AUDIT"}
return {"publish_status": "ON_SALE"}
verifier.item_getter = fake_getter
print("=== 单条排查 ===")
for it in items:
rep = verifier.diagnose("shop_001", it["outer_id"], {"success": True}, it)
print(f"\n[{it['outer_id']}] 展示={'✅' if rep.visible else '❌'} "
f"状态={rep.publish_status.value}")
for issue in rep.issues:
print(f" ⚠️ {issue}")
for sug in rep.suggestions:
print(f" 💡 {sug}")
print("\n\n=== 批量运营扫描 ===")
scan = verifier.batch_scan("shop_001", items)
print(f"总数{scan['total']} 可见{scan['visible']} 不可见{scan['invisible']}")
print(f"状态分布: {scan['by_status']}")
print("Top问题:")
for i in scan["issues"][:8]:
print(f" · {i}")跑出来关键几行（正是"成功≠展示"的实证）：
[OUT_GOOD] 展示=✅ 状态=ON_SALE
[OUT_SENS] 展示=❌ 状态=AUDIT
⚠️ 审核中, 暂未展示
⚠️ 命中敏感词/极限词: {'高仿', '全网最低', '加我', 'VX'}
💡 改写为合规表述(去引流/去极限/去侵权)
[OUT_BIZ] 展示=❌ 状态=ON_SALE
⚠️ 个人闲置误传sp_biz_type→被识别为商家, 流量降权
💡 个人闲置剥掉sp_biz_type/inspected_data(见Mapper)
[OUT_QUAL] 展示=❌ 状态=ON_SALE
⚠️ 类目'手机'需额外资质(授权/认证)
💡 补全类目资质或换普通类目
[OUT_LIMIT] 展示=❌ 状态=ON_SALE
⚠️ 每日发布上限(80/天), 今日已发80
💡 次日发布 / 提升账号等级认证 / 多店铺分流
[OUT_REJ] 展示=❌ 状态=REJECT
⚠️ 审核驳回: 图片含联系方式
💡 按驳回原因修改后重发
=== 批量运营扫描 ===
总数6 可见1 不可见5
状态分布: {'ON_SALE': 4, 'AUDIT': 1, 'REJECT': 1}五、运营提效包（批量排查 + 自动修复建议）
把
PublishVerifier.batch_scan() 做成每日定时任务，输出三张表：展示健康度日报：总数/可见/不可见/按状态分布
不可见商品清单：
outer_id + 原因 + 修复建议，可直接导入运营Excel上限预警：今日已发/总量占比 > 80% 自动告警，触发多店铺分流
自动修复（可安全执行的）：
敏感词 → 按同义词典自动改写标题（"高仿"→"风格类似"、"全网最低"→删除）
草稿态 → 自动调一次 publish 确认上架
去重 → 自动换首图/微调标题
需人工介入的：类目资质、审核驳回、违规记录——绝不能自动改（改了也算违规）。
六、六个排查铁律
发布后必回查：
publish成功 100% 要接item.get拿publish_status，不回查=盲飞。上限静默截断要主动监控：
QuotaTracker在发布前就检查，超限的不调接口直接进待发队列次日发，别浪费调用额。ON_SALE ≠ 真的可见：状态在售但命中敏感词/资质问题，实际仍不可见，所以诊断要把"在售+有问题"判为不可见。
审核是异步的：
AUDIT状态需要轮询（指数退避，别高频查），平均分钟级出结果。驳回原因要入库：
reject_reason是运营最值钱的反馈数据，按原因聚类能发现系统性问题（如某类目总被驳回）。映射规则前置校验：
item_biz_type/敏感词/资质的校验在 publish 之前就跑（复用IdleItemPublishMapper+ComplianceGate），失败不进队列，避免"发了才发现不对"。
七、和前几篇的衔接
把PublishVerifier挂到发布链路的最后一道闸（在IdleItemPublishMapper映射之后、TopApiClient.call()之后）：
发布成功立即
diagnose()，不可见商品进死信队列 + 运营告警；
QuotaTracker复用前篇RateLimiter的 Redis 后端（多Worker共享配额计数）；敏感词扫描接入前篇
ComplianceGate（个人数据/采集红线之外的内容合规维度）；批量扫描的
by_status指标喂ObservabilityMiddleware，出"展示健康度"Grafana看板；驳回/违规数据回流，帮前篇
Double11CommandCenter在大促期自动暂停问题类目发布。
"接口成功"只是起点，"APP可见"才是终点——把中间的七道闸门用代码显式建模，运营才不用靠"刷新APP"来排查。
要不要我把
idle_publish_verifier.py 合进 commerce-mesh/adapters/idle/，和前几篇的 授权(TokenManager) → 映射(Mapper) → 限流(RateLimiter) → 发布 → 验证(Verifier) → 发货(Ship) → 退款(Refund) 串成完整闭环，并加一个运营后台 API（GET /idle/health?shop_id=xxx 返回展示健康度报表）？