Интеграция с Ozon Seller API: полное руководство по автоматизации продаж

 1{
 2  "result": {
 3    "items": [
 4      {
 5        "id": 123456789,
 6        "name": "Наушники беспроводные BT 5.0",
 7        "offer_id": "BT-HEADPHONES-001",
 8        "barcode": "4601234567890",
 9        "category_id": 17028922,
10        "images": [
11          "https://cdn.ozon.ru/s1/product/1.jpg"
12        ],
13        "attributes": [
14          {
15            "id": 85,
16            "values": [{"value": "MyBrand"}]
17          }
18        ],
19        "price": {
20          "price": "2990.00",
21          "old_price": "3990.00",
22          "premium_price": "2690.00"
23        },
24        "stocks": {
25          "present": 150,
26          "reserved": 12
27        }
28      }
29    ],
30    "total": 2100,
31    "last_id": "abc123"
32  }
33}

 1import httpx
 2import asyncio
 3from typing import Optional
 4from datetime import datetime
 5
 6class OzonSellerAPI:
 7    """Клиент для работы с Ozon Seller API"""
 8    
 9    BASE_URL = "https://api-seller.ozon.ru"
10    
11    def __init__(self, client_id: str, api_key: str):
12        self.client_id = client_id
13        self.api_key = api_key
14        self.headers = {
15            "Client-Id": client_id,
16            "Api-Key": api_key,
17            "Content-Type": "application/json"
18        }
19    
20    async def get_product_list(
21        self, limit: int = 100, last_id: str = ""
22    ) -> dict:
23        """Получить список товаров"""
24        url = f"{self.BASE_URL}/v2/product/list"
25        payload = {"filter": {}, "last_id": last_id, "limit": limit}
26        async with httpx.AsyncClient() as client:
27            response = await client.post(
28                url, json=payload, headers=self.headers
29            )
30            response.raise_for_status()
31            return response.json()
32    
33    async def get_product_info(self, product_id: int) -> dict:
34        """Получить детальную информацию о товаре"""
35        url = f"{self.BASE_URL}/v2/product/info"
36        payload = {"product_id": product_id}
37        async with httpx.AsyncClient() as client:
38            response = await client.post(
39                url, json=payload, headers=self.headers
40            )
41            response.raise_for_status()
42            return response.json()
43    
44    async def update_prices(self, prices: list[dict]) -> dict:
45        """Обновить цены на товары
46        prices: [{"product_id": 123, "price": "1990.00", 
47                   "old_price": "2490.00", "min_price": "1590.00"}]
48        """
49        url = f"{self.BASE_URL}/v1/product/import/prices"
50        async with httpx.AsyncClient() as client:
51            response = await client.post(
52                url, json={"prices": prices}, headers=self.headers
53            )
54            response.raise_for_status()
55            return response.json()
56    
57    async def update_stocks(self, stocks: list[dict]) -> dict:
58        """Обновить остатки
59        stocks: [{"product_id": 123, "stock": 100, 
60                   "warehouse_id": 12345}]
61        """
62        url = f"{self.BASE_URL}/v2/products/stocks"
63        async with httpx.AsyncClient() as client:
64            response = await client.post(
65                url, json={"stocks": stocks}, headers=self.headers
66            )
67            response.raise_for_status()
68            return response.json()
69    
70    async def get_orders(
71        self, date_from: str, date_to: str, status: str = ""
72    ) -> dict:
73        """Получить список заказов за период"""
74        url = f"{self.BASE_URL}/v3/posting/fbs/list"
75        payload = {
76            "dir": "ASC",
77            "filter": {
78                "date": {"from": date_from, "to": date_to},
79            },
80            "limit": 50,
81            "offset": 0,
82            "with": {"analytics_data": True, "financial_data": True}
83        }
84        if status:
85            payload["filter"]["status"] = status
86        async with httpx.AsyncClient() as client:
87            response = await client.post(
88                url, json=payload, headers=self.headers
89            )
90            response.raise_for_status()
91            return response.json()
92    
93    async def get_analytics(
94        self, date_from: str, date_to: str, 
95        metrics: list[str], dimensions: list[str]
96    ) -> dict:
97        """Получить аналитические данные"""
98        url = f"{self.BASE_URL}/v1/analytics/data"
99        payload = {
100            "date_from": date_from,
101            "date_to": date_to,
102            "metrics": metrics,
103            "dimension": dimensions,
104            "limit": 1000,
105            "offset": 0
106        }
107        async with httpx.AsyncClient() as client:
108            response = await client.post(
109                url, json=payload, headers=self.headers
110            )
111            response.raise_for_status()
112            return response.json()
113
114
115# Пример использования
116async def main():
117    ozon = OzonSellerAPI(
118        client_id="your-client-id",
119        api_key="your-api-key"
120    )
121    
122    # Получить список товаров
123    products = await ozon.get_product_list(limit=50)
124    print(f"Товаров: {products['result']['total']}")
125    
126    # Получить аналитику продаж
127    analytics = await ozon.get_analytics(
128        date_from="2026-08-01",
129        date_to="2026-08-25",
130        metrics=["revenue", "ordered_units", "session_view"],
131        dimensions=["sku"]
132    )
133    print(f"Данных аналитики: {len(analytics['result']['data'])}")
134
135asyncio.run(main())

 1async def sync_stocks_from_1c(
 2    ozon_api: OzonSellerAPI,
 3    warehouse_id: int
 4):
 5    """Синхронизация остатков из 1С на Ozon"""
 6    
 7    # Получаем остатки из 1С через промежуточную таблицу
 8    stocks_1c = await get_stocks_from_1c_database()
 9    
10    # Ozon принимает батчами до 100 позиций
11    batch_size = 100
12    for i in range(0, len(stocks_1c), batch_size):
13        batch = stocks_1c[i:i + batch_size]
14        stocks_payload = [
15            {
16                "product_id": item["ozon_id"],
17                "stock": item["quantity"],
18                "warehouse_id": warehouse_id
19            }
20            for item in batch
21        ]
22        await ozon_api.update_stocks(stocks_payload)
23        await asyncio.sleep(1)  # Соблюдение лимитов
24    
25    print(f"Синхронизировано {len(stocks_1c)} товаров с 1С")

 1async def auto_participate_in_promos(
 2    ozon_api: OzonSellerAPI,
 3    min_margin: float = 0.15
 4):
 5    """Автоматическое подключение товаров к акциям Ozon"""
 6    
 7    # Получаем доступные акции
 8    url = f"{ozon_api.BASE_URL}/v1/actions"
 9    async with httpx.AsyncClient() as client:
10        response = await client.post(
11            url, json={}, headers=ozon_api.headers
12        )
13        actions = response.json()
14    
15    for action in actions.get("result", []):
16        action_id = action["id"]
17        action_title = action["title"]
18        
19        # Получаем товары, которые можно подключить
20        url = f"{ozon_api.BASE_URL}/v1/actions/candidates"
21        async with httpx.AsyncClient() as client:
22            response = await client.post(
23                url, 
24                json={"action_id": action_id, "limit": 100},
25                headers=ozon_api.headers
26            )
27            candidates = response.json()
28        
29        # Фильтруем по маржинальности
30        products_to_add = []
31        for product in candidates.get("result", []):
32            max_action_price = product.get("max_action_price", 0)
33            cost_price = await get_cost_price(product["id"])
34            
35            if cost_price > 0:
36                margin = (max_action_price - cost_price) / max_action_price
37                if margin >= min_margin:
38                    products_to_add.append({
39                        "product_id": product["id"],
40                        "action_price": str(max_action_price)
41                    })
42        
43        # Подключаем товары к акции
44        if products_to_add:
45            url = f"{ozon_api.BASE_URL}/v1/actions/products/activate"
46            async with httpx.AsyncClient() as client:
47                await client.post(
48                    url,
49                    json={
50                        "action_id": action_id,
51                        "products": products_to_add
52                    },
53                    headers=ozon_api.headers
54                )
55            print(
56                f"Акция '{action_title}': "
57                f"подключено {len(products_to_add)} товаров"
58            )

 1async def manage_fbo_shipments(
 2    ozon_api: OzonSellerAPI,
 3    products_to_ship: list[dict]
 4):
 5    """Управление поставками на склад Ozon (FBO)"""
 6    
 7    for product in products_to_ship:
 8        # Проверяем остатки на складе Ozon
 9        stock_info = await ozon_api.get_product_info(product["id"])
10        ozon_stock = stock_info["result"]["stocks"][0]["present"]
11        
12        # Если остаток менее 20 штук — готовим отгрузку
13        if ozon_stock < 20:
14            shipment_qty = 50 - ozon_stock
15            print(
16                f"Товар {product['name']}: "
17                f"отправить {shipment_qty} шт. на склад Ozon"
18            )
19            await create_shipment_document(product, shipment_qty)

 1async def process_fbs_orders(ozon_api: OzonSellerAPI):
 2    """Обработка заказов FBS — от подтверждения до отгрузки"""
 3    
 4    orders = await ozon_api.get_orders(
 5        date_from="2026-08-25T00:00:00",
 6        date_to="2026-08-25T23:59:59",
 7        status="awaiting_packaging"
 8    )
 9    
10    for order in orders.get("result", {}).get("postings", []):
11        posting_number = order["posting_number"]
12        
13        # Подтверждаем заказ
14        await confirm_fbs_order(ozon_api, posting_number)
15        
16        # Формируем этикетку
17        label = await get_shipping_label(ozon_api, posting_number)
18        
19        # Уведомляем склад
20        await notify_warehouse(posting_number, order["products"])
21        
22        print(f"Заказ {posting_number} готов к отгрузке")