В Tirebase каталог — это порядка 300 тысяч SKU (шины, диски, аксессуары). Выгрузка на Яндекс.Маркет на первый взгляд выглядит тривиально: генерируешь YML-файл, загружаешь в личный кабинет — готово. Но когда цены меняются постоянно, а модерация отклоняет товары по неочевидным причинам, всё становится заметно сложнее.
Почему YML — это боль
YML (Yandex Market Language) — это XML-формат, который Яндекс использует для импорта товаров. На бумаге всё просто: описываешь <offer>, указываешь цену, картинки, характеристики. На практике — это минное поле.
Первая проблема: размер файла. Наивная генерация YML для 300K товаров давала файл больше гигабайта. Яндекс принимает файлы ограниченного размера (сжатые gzip). Я начал с lxml.etree и попытки собрать весь XML в памяти:
# Так делать НЕ НАДО -- жрёт много RAM
root = etree.Element("yml_catalog")
shop = etree.SubElement(root, "shop")
offers = etree.SubElement(shop, "offers")
for product in all_products: # ~300K итераций
offer = etree.SubElement(offers, "offer", id=str(product.id))
# ... заполняем поляРешение — потоковая генерация через xml.sax.saxutils.XMLGenerator. Пишем XML напрямую в файл, не держа дерево в памяти:
import gzip
from xml.sax.saxutils import XMLGenerator
def generate_yml_stream(products_iter, output_path: str):
with gzip.open(output_path, "wb") as f:
handler = XMLGenerator(f, encoding="UTF-8")
handler.startDocument()
handler.startElement("yml_catalog", {"date": datetime.now().isoformat()})
handler.startElement("shop", {})
handler.startElement("offers", {})
for product in products_iter:
write_offer(handler, product)
handler.endElement("offers")
handler.endElement("shop")
handler.endElement("yml_catalog")
handler.endDocument()Потребление памяти упало радикально — до десятков мегабайт вместо гигабайтов. Генерация 300K товаров занимает около минуты. Точные цифры зависят от длины описаний и количества картинок в каждом offer.
Правила валидации, которых нет в документации
Яндекс документирует основные требования к YML, но реальная валидация значительно строже. Вот что я выяснил методом проб и ошибок:
Картинки: документация говорит — URL картинки, формат JPEG/PNG. Не говорит, что URL не должен содержать кириллицу (даже URL-encoded), размер изображения должен быть минимум 300x300, а если сервер отдаёт 302-редирект — картинка будет отклонена молча.
Описания: HTML-теги в <description> разрешены, но <table> — нет. <br> — можно, <br/> — нет (хотя это валидный XML). Описание длиннее 3000 символов обрезается, но если обрезка попадает внутрь HTML-тега — весь товар отклоняется.
Я написал валидатор, который прогоняет каждый товар перед включением в фид:
class YandexOfferValidator:
MAX_DESCRIPTION_LENGTH = 2900 # с запасом
ALLOWED_HTML_TAGS = {"p", "br", "ul", "ol", "li", "b", "i", "strong", "em"}
MIN_IMAGE_SIZE = (300, 300)
def validate(self, product: Product) -> list[str]:
errors = []
if product.description:
clean_desc = self._strip_forbidden_tags(product.description)
if len(clean_desc) > self.MAX_DESCRIPTION_LENGTH:
errors.append(f"Description too long: {len(clean_desc)}")
for url in product.image_urls:
if not self._is_valid_image_url(url):
errors.append(f"Invalid image URL: {url}")
if product.price <= 0:
errors.append("Price must be positive")
if not product.category_id:
errors.append("Category required")
return errorsОбновление цен: от фидов к API
YML-фиды — это хорошо для полной синхронизации каталога, но цены у нас меняются постоянно. Перегенерировать весь фид каждый раз — неэффективно. Яндекс предоставляет API для точечных обновлений.
Проблема в том, что API обновления цен принимает ограниченное количество товаров за запрос и имеет rate limit. При нескольких тысячах ценовых изменений в день это может стать узким горлышком.
Я построил пайплайн с очередью и батчингом:
class PriceUpdatePipeline:
def __init__(self, yandex_client: YandexMarketClient):
self.client = yandex_client
self.queue: asyncio.Queue = asyncio.Queue()
self.batch_size = 500
self.rate_limit = AsyncLimiter(10, 60) # 10 req/min
async def enqueue(self, sku_id: str, new_price: Decimal):
await self.queue.put({"sku": sku_id, "price": float(new_price)})
async def process_loop(self):
while True:
batch = []
try:
while len(batch) < self.batch_size:
item = await asyncio.wait_for(
self.queue.get(), timeout=5.0
)
batch.append(item)
except asyncio.TimeoutError:
pass
if batch:
async with self.rate_limit:
await self.client.update_prices(batch)
logger.info(f"Updated {len(batch)} prices")Дедупликация — важный момент. Если цена товара менялась 3 раза за 5 минут, нет смысла отправлять все три обновления. Я добавил слой, который хранит последнее значение и отправляет только финальную цену:
class PriceDeduplicator:
def __init__(self):
self._pending: dict[str, Decimal] = {}
self._lock = asyncio.Lock()
async def set_price(self, sku_id: str, price: Decimal):
async with self._lock:
self._pending[sku_id] = price
async def flush(self) -> list[dict]:
async with self._lock:
batch = [
{"sku": sku, "price": float(p)}
for sku, p in self._pending.items()
]
self._pending.clear()
return batchЭто заметно сократило количество API-вызовов — точный процент зависит от дня, но в среднем отправляем значительно меньше запросов, чем без дедупликации.
Модерация: робот, который не объясняет
Самая болезненная часть — модерация. Яндекс может отклонить товар с причиной «Не соответствует требованиям», и всё. Никаких деталей. Я не нашёл в документации полного списка причин отклонения — пришлось собирать паттерны из опыта:
- Бренд в названии не совпадает с карточкой — если пишешь «Шина Michelin», а в карточке бренд «MICHELIN», товар может быть отклонён. Регистр имеет значение.
- Фото не соответствует товару — если на фото шина 205/55R16, а в карточке 195/65R15, робот это видит (да, у них есть CV-модель).
- Цена слишком низкая — если цена ниже рыночной на 40%+, товар уходит на ручную проверку, которая длится неделями.
Для борьбы с отклонениями я написал мониторинг, который каждый час проверяет статус модерации и алертит в Telegram:
async def check_moderation_status(self):
rejected = await self.client.get_offers(status="REJECTED")
if rejected:
message = f"Отклонено {len(rejected)} товаров:\n"
for offer in rejected[:20]:
message += f"• {offer['name']} — {offer['reason']}\n"
await self.telegram.send_alert(message)Архитектура пайплайна
В итоге система выглядит так:
- Полная синхронизация — раз в сутки, ночью. Генерируется полный YML-фид потоковым методом, загружается через API.
- Инкрементальные обновления — каждые 5 минут. Только изменившиеся товары отправляются батчами через API.
- Ценовой пайплайн — в реальном времени. Изменения цен из внутренней системы попадают в очередь и отправляются с дедупликацией.
- Мониторинг модерации — ежечасно. Отклонённые товары попадают в алерты и в очередь на исправление.
Весь пайплайн работает на Python 3.12 + asyncio, деплоится в Docker-контейнере. Полная синхронизация ~300K товаров занимает несколько минут.
Уроки
Интеграция с Яндекс.Маркетом научила меня нескольким вещам. Документация маркетплейсов — это отправная точка, а не истина. Реальные правила валидации узнаёшь только через отклонения. Потоковая обработка — не оптимизация, а необходимость, когда работаешь с сотнями тысяч записей. Дедупликация на уровне пайплайна экономит не только API-квоту, но и нервы.
Оговорюсь: часть описанных проблем с валидацией может быть уже исправлена на стороне Яндекса — они периодически обновляют требования. Но общий принцип остаётся: не доверяйте документации слепо, пишите свой валидатор и начинайте с мониторинга отклонений.
