ksenia_mikhailova
/
ozon_api_helpers
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

first commit

This commit is contained in:
Kseninia Mikhaylova 2025-06-17 14:38:44 +03:00
commit cfd3d9e81c
6 changed files with 392 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
.gitignore vendored Normal file
View File

@ -0,0 +1,3 @@
__pycache__
.vscode
result/

97
bundles.py Normal file
View File

@ -0,0 +1,97 @@
import requests
import time
import os
import csv
import pandas as pd
from ozon_data import headers
# URL для первого и второго запроса
url_list = "https://api-seller.ozon.ru/v2/supply-order/list"
url_get = "https://api-seller.ozon.ru/v2/supply-order/get"
url_bundle = "https://api-seller.ozon.ru/v1/supply-order/bundle"
order_state = [
"ORDER_STATE_DATA_FILLING",
]
try:
response_list = requests.post(
url_list,
headers=headers,
json={
"filter": {"states": order_state},
"paging": {"from_supply_order_id": 0, "limit": 100},
},
)
response_list.raise_for_status()
data_list = response_list.json()
# print("Результат первого запроса:")
# print(data_list)
order_ids = data_list.get("supply_order_id", [])
if not order_ids:
print(data_list)
raise ValueError(
"Не удалось получить supply_order_id из ответа первого запроса"
)
# Второй запрос
response_get = requests.post(
url_get, headers=headers, json={"order_ids": order_ids}
)
response_get.raise_for_status() # Проверка на ошибки
data_get = response_get.json()
# print("Результат второго запроса:")
print(data_get)
orders = data_get.get("orders", [])
result = []
for order in orders:
all_bundle_ids = [supply.get("bundle_id") for supply in order.get("supplies", [])]
b_response = requests.post(
url=url_bundle,
json={"bundle_ids": all_bundle_ids, "limit": 100},
headers=headers,
)
b_response.raise_for_status()
b_result = b_response.json()
for b in b_result.get('items'):
result.append({
"supply_order_number": order.get("supply_order_number"),
"creation_date": order.get("creation_date"),
**b,
})
time.sleep(0.3)
df = pd.DataFrame(result)
# print("Результат:", result)
output_dir = "result"
if not os.path.exists(output_dir):
os.makedirs(output_dir)
# Получаем имя текущего скрипта без расширения
script_name = os.path.splitext(os.path.basename(__file__))[0]
filename = '_'.join([script_name, *order_state])
csv_filename = os.path.join(
output_dir, f"{filename}_output.csv"
)
xls_filename = os.path.join(
output_dir, f"{filename}_output.xlsx"
)
df.to_csv(csv_filename, encoding="utf-8")
df.to_excel(xls_filename)
print(f"Данные успешно сохранены в {csv_filename}")
except requests.exceptions.RequestException as e:
print(f"Ошибка при выполнении запроса: {e}")
except ValueError as e:
print(f"Ошибка обработки данных: {e}")

6
ozon_data.py Normal file
View File

@ -0,0 +1,6 @@
# Заголовки для авторизации и передачи данных
headers = {
"Content-Type": "application/json",
"Client-Id": "1355708",
"Api-Key": "21a9f824-9bc3-422c-8fd1-86d6a6a13d54",
}

80
products.py Normal file
View File

@ -0,0 +1,80 @@
import requests
import time
import os
import ast
import pandas as pd
from ozon_data import headers
# URL для первого и второго запроса
url_list = "https://api-seller.ozon.ru/v4/product/info/stocks"
url_detail = "https://api-seller.ozon.ru/v4/product/info/attributes"
product_type = "INVISIBLE"
try:
response_list = requests.post(
url_list,
headers=headers,
json={
"filter": {
"visibility": product_type,
},
"limit": 100,
},
)
response_list.raise_for_status()
data_list = response_list.json()
df_list = pd.DataFrame(data_list["items"])
df_list['stocks'] = df_list['stocks'].apply(
lambda x: ast.literal_eval(x.replace("'", '"')) if isinstance(x, str) else x
)
stocks_df = pd.json_normalize(df_list['stocks'].str[0])
df_list = pd.concat([df_list, stocks_df], axis=1)
response_detail = requests.post(
url_detail,
headers=headers,
json={
"filter": {
"product_id": df_list["product_id"].to_list(),
},
"limit": 1000
},
)
response_detail.raise_for_status()
data_detail = response_detail.json()
df_detail = pd.DataFrame(data_detail["result"])
# Объединяем по id и product_id
df = pd.merge(
df_detail,
df_list,
left_on="id",
right_on="product_id",
how="inner" # или 'left', 'right', 'outer' в зависимости от задачи
)
# print("Результат:", result)
output_dir = "result"
if not os.path.exists(output_dir):
os.makedirs(output_dir)
# Получаем имя текущего скрипта без расширения
script_name = os.path.splitext(os.path.basename(__file__))[0]
filename = "_".join([script_name, product_type])
csv_filename = os.path.join(output_dir, f"{filename}_output.csv")
xls_filename = os.path.join(output_dir, f"{filename}_output.xlsx")
df.to_csv(csv_filename, encoding="utf-8")
df.to_excel(xls_filename)
print(f"Данные успешно сохранены в {csv_filename}")
except requests.exceptions.RequestException as e:
print(f"Ошибка при выполнении запроса: {e}")
print(e)
except ValueError as e:
print(f"Ошибка обработки данных: {e}")

110
stat.py Normal file
View File

@ -0,0 +1,110 @@
import requests
import time
import os
import pandas as pd
from datetime import datetime, timedelta
# Импортируем заголовки (Client-Id и Api-Key), должны быть в файле ozon_data.py
from ozon_data import headers
# URL эндпоинтов
url_analytics = "https://api-seller.ozon.ru/v1/analytics/data"
url_product_list = "https://api-seller.ozon.ru/v3/product/info/list"
# Автоматически определяем даты: за последние 7 дней
date_to = datetime.today().date()
date_from = date_to - timedelta(days=7)
# Преобразуем даты в строку для запроса
date_from_str = str(date_from)
date_to_str = str(date_to)
# Тело запроса
payload = {
"date_from": date_from_str,
"date_to": date_to_str,
"dimension": ["sku"],
"metrics": ["ordered_units", "hits_view_pdp"],
"limit": 1000
}
try:
response = requests.post(url_analytics, headers=headers, json=payload)
response.raise_for_status()
data = response.json()
# Парсим результат
result_data = []
for item in data.get("result", {}).get("data", []):
dimensions = item.get("dimensions", [])
metrics = item.get("metrics", [])
row = {
"sku": dimensions[0].get("id") if len(dimensions) > 0 else None,
"ordered_units": metrics[0] if len(metrics) > 0 else 0,
"hits_view_pdp": metrics[1] if len(metrics) > 1 else 0,
}
result_data.append(row)
df = pd.DataFrame(result_data)
# --- Шаг 2: Запрашиваем названия товаров через /v3/product/info/list ---
skus = df["sku"].dropna().astype(int).unique().tolist()
# Разбиваем на чанки по 1000 штук (если больше 1000)
def chunked(lst, n):
"""Разбивает список на части по n элементов"""
return [lst[i:i + n] for i in range(0, len(lst), n)]
for chunk in chunked(skus, 1000): # Ozon позволяет до 1000 SKU за раз
payload_products = {
"sku": chunk
}
try:
response_products = requests.post(url_product_list, headers=headers, json=payload_products)
response_products.raise_for_status()
data_products = response_products.json()
if "items" not in data_products:
print("Ошибка: В ответе нет ключа 'items'")
continue
for item in data_products.get("items", []):
el_sku = item["sources"][0]["sku"]
if el_sku and "name" in item:
df.loc[df['sku'] == str(el_sku), 'name'] = item['name']
else:
print("Ошибка в элементе товара — нет 'sku' или 'name':")
except requests.exceptions.RequestException as e:
print(f"Ошибка запроса к /v3/product/info/list: {e}")
if e.response:
print("Текст ответа:", e.response.text)
# Добавляем колонку: конверсия заказов к показам * 100
df["conversion_ordered_to_views (%)"] = (
df["ordered_units"] / df["hits_view_pdp"] * 100
).round(2)
# Создаем папку для вывода
output_dir = "result"
if not os.path.exists(output_dir):
os.makedirs(output_dir)
# Имя файла на основе скрипта
script_name = os.path.splitext(os.path.basename(__file__))[0]
filename = f"{script_name}_analytics_conversion_{date_from_str}_{date_to_str}"
csv_filename = os.path.join(output_dir, f"{filename}.csv")
xls_filename = os.path.join(output_dir, f"{filename}.xlsx")
# Сохраняем
df.to_csv(csv_filename, index=False, encoding="utf-8-sig")
df.to_excel(xls_filename, index=False)
print(f"Данные успешно сохранены в {csv_filename}")
except requests.exceptions.RequestException as e:
print(f"Ошибка при выполнении запроса: {e}")
except Exception as e:
print(f"Ошибка обработки данных: {e}")

96
total_month.py Normal file
View File

@ -0,0 +1,96 @@
import os
import requests
import pandas as pd
from datetime import datetime
from dateutil.relativedelta import relativedelta
from ozon_data import headers
# URL API
url = "https://api-seller.ozon.ru/v2/posting/fbo/list"
# Параметры для месяца
year = 2025
# Укажите нужный месяц (1-12)
month = 1
# Определение первого и последнего дня месяца
first_day = datetime(year, month, 1) # Первый день месяца
last_day = first_day + relativedelta(
months=1, days=-1
) # Последний день текущего месяца
# Формируем строки дат в ISO формате
since = first_day.isoformat() + "Z" # Формат: 2025-01-01T00:00:00Z
to = last_day.isoformat() + "Z" # Формат: 2025-01-31T23:59:59.999Z
# Тело запроса
payload = {
"dir": "ASC",
"filter": {
"since": since,
# "status": "delivered",
"to": to,
}, # Укажите статус, если нужно
"limit": 1000,
"offset": 0,
"with": {"analytics_data": True, "financial_data": True},
}
print(payload)
try:
# Выполнение запроса
response = requests.post(url, headers=headers, json=payload)
response.raise_for_status()
data_raw = response.json()
# Составляем таблицу
data = []
for order in data_raw["result"]:
for product in order["products"]:
product_id = product["sku"]
financial = [
i
for i in order["financial_data"]["products"]
if i["product_id"] == product_id
]
financial = financial[0]
row = {
"order_id": order["order_id"],
"order_number": order["order_number"],
"status": order["status"],
"created_at": order["created_at"],
"product_sku": product["sku"],
"product_name": product["name"],
"offer_id": product["offer_id"],
"quantity": product["quantity"],
"price": financial["price"],
"commission_amount": financial["commission_amount"],
"payout": financial["payout"],
"warehouse": order["analytics_data"]["warehouse_name"],
}
data.append(row)
# Преобразуем в DataFrame
df = pd.DataFrame(data)
output_dir = "result"
if not os.path.exists(output_dir):
os.makedirs(output_dir)
# Получаем имя текущего скрипта без расширения
script_name = os.path.splitext(os.path.basename(__file__))[0]
# Указываем полный путь для файла
csv_filename = os.path.join(
output_dir, f"{'_'.join([script_name, str(month), str(year)])}_output.csv"
)
xls_filename = os.path.join(
output_dir, f"{'_'.join([script_name, str(month), str(year)])}_output.xlsx"
)
df.to_csv(csv_filename, encoding="utf-8")
df.to_excel(xls_filename)
print(f"Таблица успешно экспортирована в {csv_filename}")
except requests.exceptions.RequestException as e:
print(f"Ошибка запроса: {e}")
except Exception as e:
print(f"Произошла ошибка: {e}")