背景
在长时段AI数字人直播中，商品需要按预设顺序轮换讲解，讲过的不应重复出现，未讲的应按计划上架。人工盯屏手动切换费时费力，设计一套商品自动上下架系统可以大幅降低运营成本。
系统架构
核心组件包括：商品队列管理器、直播时间线引擎、平台API适配层和状态同步模块。
[商品数据库] → [队列管理器] → [时间线引擎] ↓ [API适配层] ↓ [直播平台] ↓ [状态同步 ← 用户行为事件]实现思路
商品队列管理器
维护一个优先级队列，每个商品包含字段：商品ID、上架状态、预计讲解时长（秒）、当前排位。
from dataclasses import dataclass from typing import Optional import heapq @dataclass class ProductItem: product_id: str title: str status: str # pending / live / finished duration: int # seconds priority: int def __lt__(self, other): return self.priority < other.priority class ProductQueue: def __init__(self): self.queue: list = [] def add_product(self, item: ProductItem): heapq.heappush(self.queue, item) def pop_next(self) -> Optional[ProductItem]: if self.queue: return heapq.heappop(self.queue) return None时间线引擎
根据每件商品的预计讲解时长，触发上下架事件。
import time import threading class TimelineEngine: def __init__(self, queue: ProductQueue, api_adapter): self.queue = queue self.api = api_adapter self.running = False self.current_product = None def start(self): self.running = True thread = threading.Thread(target=self._run_loop) thread.daemon = True thread.start() def _run_loop(self): while self.running: product = self.queue.pop_next() if product is None: break self.current_product = product self.api.launch_product(product.product_id) time.sleep(product.duration) self.api.end_product(product.product_id)API适配层
封装各平台的商品上下架API，提供统一接口。注意：以下为通用接口设计，不涉及任何真实平台API。
class PlatformAdapter: def launch_product(self, product_id: str): """通用上架接口""" raise NotImplementedError def end_product(self, product_id: str): """通用下架接口""" raise NotImplementedError关键设计点
- 异常重试机制：API调用失败时自动重试3次，间隔递增（1s/3s/5s），超过重试次数则记录日志并跳过该商品
- 人工干预接口：运营人员可以通过后台手动插播或跳过某件商品，操作后队列自动重排
- 时长弹性控制：如果某件商品在预设时长内产生较多互动和成交，自动延长曝光时间；反之提前结束
- 状态持久化：每完成一个商品上下架操作，写入数据库记录，直播中断后重启可从中断点继续
总结
商品自动上下架的核心是队列管理和时间线调度。本质上是生产者-消费者模式的一个应用：商品队列是缓冲区，时间线引擎是消费者。设计的关键在于异常处理和人工干预能力，完全自动化在直播场景中并不总是最优解。
