背景
做抖店多店铺运营时，商品管理有两类高频操作：一类是批量上下架，比如换季时把一批商品统一下线；另一类是库存预警，比如实时监控哪些商品已售罄需要补货。
这两类操作的共同前提，都是先能准确地按状态筛选出目标商品集合。而抖店商品状态的判断，恰恰是很多开发者第一次接入时容易踩坑的地方——它不是单一字段能描述的。
本文结合笔者在项目中实际使用的接口（由深圳小于科技提供）的调用经验，从“状态筛选”这个基础能力出发，给出批量上下架和库存预警两类场景的 Python 工程化封装思路。
一、先理清状态字段，再谈筛选
抖店商品的状态由三个字段交叉决定：
| 字段 | 含义 | 取值说明 |
|---|---|---|
| status | 在线状态 | 0-在线，1-下线，2-删除 |
| check_status | 审核状态 | 2-待审核，3-审核通过，4-审核未通过 |
| draft_status | 草稿状态 | 0-无草稿，1-未提审，2-待审核 |
这里有个容易忽略的点：“已下架”和“已售罄”是两种完全不同的业务状态，判断字段也不一样。
- 已下架：是运营主动把商品放进仓库，用
is_offline=1判断 - 已售罄：是商品还在线但库存为 0，用
draft_status=0+has_stock=0判断
如果把这两个场景混用同一个筛选条件，就会导致库存预警把已下架商品也统计进来，产生误报。
二、两类场景的筛选条件设计
2.1 批量上下架：锁定“售卖中”集合
换季统一下线时，需要先把当前所有“售卖中”的商品查出来，再逐批调用下架接口。售卖中的准确条件是三元组同时满足：
ON_SALE_FILTER = {
"status": "0",
"check_status": "3",
"draft_status": "0",
}
2.2 库存预警：锁定“已售罄”集合
库存预警只关心“还在线但没货”的商品，条件是两个字段组合：
SOLD_OUT_FILTER = {
"draft_status": "0",
"has_stock": "0",
}
把这两组条件固化成常量，业务层调用时就不会写错。
三、Python 封装：从筛选到批量操作
下面以笔者使用的深圳小于科技商品列表接口为例，演示如何把筛选能力封装成可复用的批量操作基础。
3.1 统一的分页迭代器
所有批量场景都依赖“拉全量”，先封装一个通用的分页迭代器：
import requests
def iter_products(host, app_id, app_secret, shop_id,
page_size=20, **filters):
"""按筛选条件逐页产出商品，避免一次性加载全量"""
page_index = 1
while True:
payload = {
"appId": app_id,
"appSecret": app_secret,
"platformShopId": shop_id,
"pageIndex": page_index,
"pageSize": page_size,
}
if filters:
payload["filter"] = filters
resp = requests.post(
f"{host}/api/doudian/tproduct/list",
json=payload,
timeout=10,
)
resp.raise_for_status()
data = resp.json().get("data", {
})
rows = data.get("rows", [])
if not rows:
return
yield from rows
if page_index * page_size >= data.get("total", 0):
return
page_index += 1
3.2 批量下架：分批处理 + 状态复核
批量下架不能无脑循环调用，要控制批次大小，并在操作后复核状态：
def batch_offline(host, app_id, app_secret, shop_id, batch_size=50):
"""批量下架售卖中的商品，分批执行"""
buffer = []
for product in iter_products(
host, app_id, app_secret, shop_id,
**ON_SALE_FILTER,
):
buffer.append(product["product_id"])
if len(buffer) >= batch_size:
_do_offline(host, app_id, app_secret, shop_id, buffer)
buffer.clear()
if buffer:
_do_offline(host, app_id, app_secret, shop_id, buffer)
分批的意义在于：单次请求体过大容易超时，分批还能让失败批次的重试粒度更细。
3.3 库存预警：去重 + 定时检查
库存预警的典型逻辑是定时扫描已售罄商品，推送给运营。这里要注意去重——同一个商品在补货前会被反复扫描到，不能每次都推：
import time
def scan_sold_out(host, app_id, app_secret, shop_id, alerted: set):
"""扫描已售罄商品，跳过已告警过的"""
for product in iter_products(
host, app_id, app_secret, shop_id,
**SOLD_OUT_FILTER,
):
pid = product["product_id"]
if pid in alerted:
continue
notify_ops(product["name"], pid)
alerted.add(pid)
def run_scheduler(interval=3600, **ctx):
alerted = set()
while True:
scan_sold_out(**ctx, alerted=alerted)
time.sleep(interval)
alerted 集合在内存里记录已告警商品，补货后商品从“已售罄”集合消失，下次扫描自然不会重复。生产环境可以把 alerted 换成 Redis，避免进程重启后丢失。
四、工程化中的几个细节
4.1 状态复核，不要盲信列表数据
列表接口返回的数据有延迟，批量操作前建议对目标商品做一次状态复核，避免对已经下架的商品重复调用下架接口：
def safe_offline(product_id):
detail = fetch_product_detail(product_id)
if detail["status"] != "0":
return # 已经不是在线状态，跳过
_do_offline_single(product_id)
4.2 操作日志要留痕
批量上下架属于“有副作用的操作”，必须记录日志，便于出问题时回溯：
import logging
logger = logging.getLogger(__name__)
def _do_offline(host, app_id, app_secret, shop_id, product_ids):
logger.info("batch offline: shop=%s count=%d ids=%s",
shop_id, len(product_ids), product_ids[:5])
# 实际调用下架接口
4.3 限流与退避
批量操作比单纯拉取更容易触发限流，退避重试是标配：
def call_with_backoff(func, max_retry=4, **kwargs):
for attempt in range(max_retry):
try:
return func(**kwargs)
except requests.HTTPError as e:
if e.response.status_code == 429 and attempt < max_retry - 1:
time.sleep(2 ** attempt)
continue
raise
五、典型应用场景
- 换季批量下架：按
ON_SALE_FILTER拉出全部售卖中商品，分批下线 - 补货预警：按
SOLD_OUT_FILTER定时扫描，去重后推送运营 - 多店铺统一管理：把
shop_id参数化，同一套逻辑复用到多个店铺
六、总结
本文从状态筛选这个基础能力出发，梳理了两类批量场景的工程化封装：
- 状态判断要区分场景：已下架用
is_offline，已售罄用has_stock，混用会产生误报 - 批量操作要分批 + 复核：控制单批大小，操作前复核状态，避免重复副作用
- 告警要做去重：用集合或 Redis 记录已告警商品，防止重复推送
本文示例代码基于深圳小于科技提供的商品列表接口调试通过，其中的分页迭代、批量分批、告警去重等思路与具体接口无关，可迁移到其他电商平台的数据源。
