ozon-seller-kit/docs/ozon-seller-api/09-stocks.md

# 库存管理 API

> 官方文档：https://docs.ozon.ru/api/seller/zh/#operation/ProductAPI_ProductsStocksV2

---

## 接口信息

| 项 | 值 |
|---|---|
| 方法 | POST |
| 路径 | `/v2/products/stocks` |
| 鉴权 | 需要 `Client-Id` + `Api-Key` |
| 用途 | **设置/更新商品库存**（必须操作，否则商品不可购买） |

---

## 1. 重要约束

⚠️ **商品导入成功后不会自动上架**，必须设置库存才能开售：

```
POST /v3/product/import → status=imported（商品已创建）
    ↓
POST /v2/products/stocks → 设置库存（商品可购买）
```

未设置库存的商品：
- ✅ 在卖家后台可见
- ❌ 前台不展示
- ❌ 无法购买

---

## 2. 请求

### 请求体

```json
{
  "stocks": [
    {
      "offer_id": "MY-THERMOS-001",
      "product_id": 987654321,
      "stock": 100,
      "warehouse_id": 12345678
    },
    {
      "offer_id": "MY-THERMOS-002",
      "stock": 50,
      "warehouse_id": 12345678
    }
  ]
}
```

### 参数说明

| 字段 | 类型 | 必填 | 说明 |
|---|---|---|---|
| stocks | array | ✅ 是 | 库存数组（最多 **100** 个） |
| stocks[].offer_id | string | 条件 | 自己的货号（与 product_id 二选一） |
| stocks[].product_id | integer | 条件 | Ozon 商品 ID（与 offer_id 二选一） |
| stocks[].stock | integer | ✅ 是 | 库存数量。`0` = 无库存（下架） |
| stocks[].warehouse_id | integer | ✅ 是 | 仓库 ID（见下节） |

⚠️ **必须提供 offer_id 或 product_id**（建议用 offer_id，更稳定）。

---

## 3. 仓库 ID（warehouse_id）

### 获取仓库 ID

**接口**：`POST /v1/warehouse/list`

```json
{}
```

**响应**：

```json
{
  "result": [
    {
      "warehouse_id": 12345678,
      "name": "FBS 仓库-莫斯科",
      "can_print_act_in_advance": true,
      "is_rfbs": false,
      "has_postings_limit": false,
      "postings_limit": 0,
      "status": "working"
    }
  ]
}
```

| 字段 | 说明 |
|---|---|
| warehouse_id | **仓库 ID**（设置库存时用） |
| name | 仓库名称 |
| is_rfbs | 是否 rFBS 仓库（Ozon 代发货） |
| status | 状态：`working`（运行中）/ `disabled`（禁用） |

### 仓库类型

| 类型 | 说明 | warehouse_id |
|---|---|---|
| **FBS** | 自发货（Fulfillment by Seller） | 从 `/v1/warehouse/list` 获取 |
| **FBO** | Ozon 发货（Fulfillment by Ozon） | 从 `/v1/warehouse/list` 获取 |
| **rFBS** | Ozon 代发货（类似 FBO，但库存在卖家处） | `is_rfbs=true` |

**推荐**：新商户优先用 **FBS**（自发货），灵活且门槛低。

---

## 4. 响应

### 成功响应（200）

```json
{
  "result": [
    {
      "errors": [],
      "offer_id": "MY-THERMOS-001",
      "product_id": 987654321,
      "updated": true,
      "warehouse_id": 12345678
    },
    {
      "errors": [
        {
          "code": "PRODUCT_NOT_FOUND",
          "message": "Product not found"
        }
      ],
      "offer_id": "MY-THERMOS-999",
      "product_id": 0,
      "updated": false,
      "warehouse_id": 12345678
    }
  ]
}
```

### 字段说明

| 字段 | 类型 | 说明 |
|---|---|---|
| updated | boolean | 是否更新成功 |
| errors | array | 错误数组（失败时） |
| offer_id | string | 货号（回显） |
| product_id | integer | Ozon 商品 ID（回显） |
| warehouse_id | integer | 仓库 ID（回显） |

---

## 5. 常见错误

| code | message | 原因 | 解决 |
|---|---|---|---|
| `PRODUCT_NOT_FOUND` | 商品不存在 | offer_id/product_id 错误或商品已删除 | 检查 ID 是否正确 |
| `WAREHOUSE_NOT_FOUND` | 仓库不存在 | warehouse_id 错误 | 调 `/v1/warehouse/list` 获取正确 ID |
| `INVALID_STOCK` | 库存值错误 | stock < 0 | 库存必须 ≥ 0 |
| `PRODUCT_ARCHIVED` | 商品已归档 | 商品处于归档状态 | 先恢复商品再设库存 |

---

## 6. 使用场景

### 场景 1：发布后设置初始库存

```python
async def publish_and_set_stock(product: Product, shop: Shop):
    """发布商品 + 设置库存（完整流程）"""
    # 1. 导入商品
    resp = await ozon_client.import_products(
        shop.client_id_dec,
        shop.api_key_dec,
        [build_import_item(product)]
    )
    task_id = resp["result"]["task_id"]
    
    # 2. 轮询直到成功
    result = await wait_for_import(task_id)
    if not result["success"]:
        raise Exception(f"发布失败: {result['errors']}")
    
    product_id = result["product_id"]
    
    # 3. 获取仓库 ID
    warehouses = await ozon_client.get_warehouses(
        shop.client_id_dec,
        shop.api_key_dec
    )
    warehouse_id = warehouses[0]["warehouse_id"]  # 取第一个
    
    # 4. 设置库存
    stock_resp = await ozon_client.update_stocks(
        shop.client_id_dec,
        shop.api_key_dec,
        [{
            "product_id": product_id,
            "stock": 100,  # 初始库存
            "warehouse_id": warehouse_id
        }]
    )
    
    return stock_resp
```

### 场景 2：批量更新库存

```python
async def batch_update_stocks(updates: list[dict]):
    """批量更新库存（最多 100 个）"""
    # updates = [
    #   {"offer_id": "PROD-001", "stock": 50},
    #   {"offer_id": "PROD-002", "stock": 0},  # 0 = 下架
    # ]
    
    warehouse_id = await get_default_warehouse_id()
    
    stocks = [
        {
            "offer_id": u["offer_id"],
            "stock": u["stock"],
            "warehouse_id": warehouse_id
        }
        for u in updates
    ]
    
    resp = await ozon_client.update_stocks(
        client_id, api_key, stocks
    )
    
    # 检查失败项
    failed = [
        item for item in resp["result"]
        if not item["updated"]
    ]
    
    return {
        "success": len(resp["result"]) - len(failed),
        "failed": failed
    }
```

### 场景 3：库存为 0 时下架

```python
async def out_of_stock(offer_id: str):
    """库存售罄，设为 0（自动下架）"""
    await ozon_client.update_stocks(
        client_id, api_key,
        [{
            "offer_id": offer_id,
            "stock": 0,  # 库存为 0 → 前台不展示
            "warehouse_id": warehouse_id
        }]
    )
```

### 场景 4：补货后上架

```python
async def restock(offer_id: str, quantity: int):
    """补货后重新上架"""
    await ozon_client.update_stocks(
        client_id, api_key,
        [{
            "offer_id": offer_id,
            "stock": quantity,  # 设置新库存 → 自动上架
            "warehouse_id": warehouse_id
        }]
    )
```

---

## 7. 查询当前库存

### 接口信息

| 项 | 值 |
|---|---|
| 方法 | POST |
| 路径 | `/v3/product/info/stocks` |
| 用途 | 查询商品当前库存 |

### 请求

```json
{
  "filter": {
    "offer_id": ["MY-THERMOS-001"],
    "product_id": [987654321],
    "visibility": "ALL"
  },
  "last_id": "",
  "limit": 100
}
```

### 响应

```json
{
  "result": {
    "items": [
      {
        "offer_id": "MY-THERMOS-001",
        "product_id": 987654321,
        "stocks": [
          {
            "type": "fbs",
            "present": 100,
            "reserved": 5,
            "warehouse_id": 12345678,
            "warehouse_name": "FBS 仓库-莫斯科"
          }
        ]
      }
    ],
    "last_id": "",
    "total": 1
  }
}
```

| 字段 | 说明 |
|---|---|
| stocks[].present | 可用库存 |
| stocks[].reserved | 已预订数量（订单未完成） |
| stocks[].type | 仓库类型：`fbs` / `fbo` / `rfbs` |

---

## 8. 库存同步策略

### 策略 A：实时同步（推荐）

```python
# 本地库存变化时立即更新 Ozon
async def on_local_stock_change(product_id: str, new_stock: int):
    product = await get_product(db, product_id)
    
    if not product.ozon_product_id:
        return  # 未发布到 Ozon
    
    shop = await get_default_shop(db)
    warehouse_id = await get_default_warehouse_id()
    
    await ozon_client.update_stocks(
        shop.client_id_dec,
        shop.api_key_dec,
        [{
            "offer_id": product.offer_id,
            "stock": new_stock,
            "warehouse_id": warehouse_id
        }]
    )
```

### 策略 B：定时同步

```python
# 每天凌晨同步一次（防止偏差累积）
async def daily_sync_stocks():
    """定时任务：同步本地库存到 Ozon"""
    products = await db.execute(
        select(Product)
        .where(
            Product.stage == "published",
            Product.ozon_product_id.isnot(None)
        )
    )
    products = products.scalars().all()
    
    warehouse_id = await get_default_warehouse_id()
    
    # 批量更新（100 个一批）
    for i in range(0, len(products), 100):
        batch = products[i:i+100]
        stocks = [
            {
                "offer_id": p.offer_id,
                "stock": p.local_stock,  # 假设有 local_stock 字段
                "warehouse_id": warehouse_id
            }
            for p in batch
        ]
        
        await ozon_client.update_stocks(
            client_id, api_key, stocks
        )
```

### 策略 C：反向同步（从 Ozon 读回）

```python
# 定期从 Ozon 读回库存（多渠道销售时需要）
async def sync_stocks_from_ozon():
    """从 Ozon 同步库存到本地"""
    resp = await ozon_client.get_product_stocks(
        filter={"visibility": "VISIBLE"},
        limit=1000
    )
    
    for item in resp["result"]["items"]:
        offer_id = item["offer_id"]
        ozon_stock = item["stocks"][0]["present"]
        
        # 更新本地库存
        await db.execute(
            update(Product)
            .where(Product.offer_id == offer_id)
            .values(local_stock=ozon_stock)
        )
    
    await db.commit()
```

---

## 9. V2 项目集成

### API 层（待实现）

```python
# server/api/products.py

@router.post("/products/{product_id}/set-stock")
async def set_stock(
    product_id: str,
    stock: int = Body(..., ge=0),
    shop_id: str = Body(...),
    db: AsyncSession = Depends(get_db)
):
    """设置商品库存"""
    product = await get_product(db, product_id)
    
    if not product.ozon_product_id and not product.offer_id:
        raise HTTPException(400, "商品尚未发布到 Ozon")
    
    shop = await get_shop(db, shop_id)
    
    # 获取仓库 ID（缓存）
    warehouse_id = await get_or_cache_warehouse_id(shop)
    
    # 调用 Ozon API
    resp = await ozon_client.update_stocks(
        shop.client_id_dec,
        shop.api_key_dec,
        [{
            "offer_id": product.offer_id,
            "stock": stock,
            "warehouse_id": warehouse_id
        }]
    )
    
    result = resp["result"][0]
    
    if not result["updated"]:
        raise HTTPException(500, f"更新失败: {result['errors']}")
    
    # 更新本地记录
    product.local_stock = stock
    await db.commit()
    
    return {"success": True, "stock": stock}


@router.get("/shops/{shop_id}/warehouses")
async def get_warehouses(
    shop_id: str,
    db: AsyncSession = Depends(get_db)
):
    """获取店铺的仓库列表"""
    shop = await get_shop(db, shop_id)
    
    warehouses = await ozon_client.get_warehouses(
        shop.client_id_dec,
        shop.api_key_dec
    )
    
    return {"result": warehouses}
```

### 前端（待实现）

```tsx
// studio/src/pages/product/components/StockPanel.tsx

import { InputNumber, Button, message } from 'antd';

export function StockPanel({ productId, shopId }) {
  const [stock, setStock] = useState(0);
  const [saving, setSaving] = useState(false);
  
  const handleSave = async () => {
    setSaving(true);
    try {
      await fetch(`/api/products/${productId}/set-stock`, {
        method: 'POST',
        body: JSON.stringify({ stock, shop_id: shopId })
      });
      
      message.success(`库存已设置为 ${stock}`);
    } catch (err) {
      message.error(`设置失败: ${err.message}`);
    } finally {
      setSaving(false);
    }
  };
  
  return (
    <div>
      <InputNumber
        min={0}
        value={stock}
        onChange={setStock}
        placeholder="库存数量"
      />
      <Button
        type="primary"
        loading={saving}
        onClick={handleSave}
      >
        设置库存
      </Button>
      <div style={{ marginTop: 8, fontSize: 12, color: '#888' }}>
        💡 库存为 0 时商品自动下架
      </div>
    </div>
  );
}
```

---

## 10. 最佳实践

### 1. 发布流程中必须设库存

```
✅ 正确：
POST /v3/product/import → 轮询成功 → POST /v2/products/stocks

❌ 错误：
POST /v3/product/import → 轮询成功 → 结束（商品不可购买）
```

### 2. 缓存仓库 ID

```python
# 仓库 ID 不常变，启动时拉取并缓存
_warehouse_cache = {}

async def get_warehouse_id(shop_id: str):
    if shop_id not in _warehouse_cache:
        warehouses = await ozon_client.get_warehouses(...)
        _warehouse_cache[shop_id] = warehouses[0]["warehouse_id"]
    
    return _warehouse_cache[shop_id]
```

### 3. 库存为 0 的处理

```python
# 库存为 0 → 自动下架，但商品仍在后台
# 补货后再设置库存 → 自动上架
# 不需要删除商品，只需更新库存
```

### 4. 批量操作

```python
# 单次最多 100 个，超过需分批
async def update_large_batch(stocks: list):
    results = []
    for i in range(0, len(stocks), 100):
        batch = stocks[i:i+100]
        resp = await ozon_client.update_stocks(client_id, api_key, batch)
        results.extend(resp["result"])
    
    return results
```

---

## 相关文档

- [04-product-import.md](./04-product-import.md) —— 创建商品（发布前置）
- [05-product-info.md](./05-product-info.md) —— 查询商品信息（含库存）
- [10-prices.md](./10-prices.md) —— 价格更新
- [docs/v2/ozon-publish.md](../v2/ozon-publish.md) §5 —— 发布链路（含库存设置）