class YandexMarketAdapter(BaseAdapter):
"""Адаптер для Yandex.Market Partner API."""
marketplace = Marketplace.YANDEX_MARKET
def __init__(self, credentials: dict):
self.oauth_token = credentials["oauth_token"]
self.campaign_id = credentials["campaign_id"]
self.business_id = credentials["business_id"]
self.base_url = credentials.get("base_url", "https://api.partner.market.yandex.ru")
self.headers = {
"Authorization": f"Bearer {self.oauth_token}",
"Content-Type": "application/json"
}
async def get_card(self, sku: str) -> Optional[CardData]:
"""Получение карточки по артикулу."""
url = f"{self.base_url}/businesses/{self.business_id}/offer-cards"
payload = {
"offerIds": [sku]
}
async with aiohttp.ClientSession() as session:
async with session.post(url, json=payload, headers=self.headers) as resp:
if resp.status != 200:
return None
data = await resp.json()
cards = data.get("result", {}).get("offerCards", [])
if not cards:
return None
return self._map_card_data(cards[0], sku)
def _map_card_data(self, raw: dict, sku: str) -> CardData:
"""Маппинг данных YM в CardData."""
mapping = raw.get("mapping", {})
card = mapping.get("marketSku", {})
# Извлечение параметров
attributes = {}
for param in card.get("parameterValues", []):
param_id = param.get("parameterId")
value = param.get("value", {})
attributes[param_id] = value.get("value") or value.get("optionId")
return CardData(
sku=sku,
marketplace=Marketplace.YANDEX_MARKET,
nm_id=str(card.get("marketSku")),
title=raw.get("offer", {}).get("name"),
description=raw.get("offer", {}).get("description"),
attributes=attributes,
category=card.get("categoryName"),
category_id=card.get("categoryId"),
brand=raw.get("offer", {}).get("vendor"),
photos=[img for img in raw.get("offer", {}).get("pictures", [])],
seo_tags=raw.get("offer", {}).get("tags", []),
raw_data=raw
)
async def update_card(self, sku: str, content: CardContent) -> PublishResult:
"""Обновление контента карточки."""
current = await self.get_card(sku)
if not current:
return PublishResult(
success=False,
marketplace=self.marketplace,
sku=sku,
error_code="CARD_NOT_FOUND",
error_message=f"Карточка {sku} не найдена"
)
url = f"{self.base_url}/businesses/{self.business_id}/offer-cards/update"
# Формируем offer для обновления
offer = {
"offerId": sku
}
if content.title:
offer["name"] = content.title[:150]
if content.description:
offer["description"] = content.description[:3000]
if content.seo_tags:
offer["tags"] = content.seo_tags
payload = {"offerCards": [{"offer": offer}]}
async with aiohttp.ClientSession() as session:
async with session.post(url, json=payload, headers=self.headers) as resp:
data = await resp.json()
if resp.status == 200:
return PublishResult(
success=True,
marketplace=self.marketplace,
sku=sku,
nm_id=current.nm_id,
raw_response=data
)
else:
errors = data.get("errors", [])
error_msg = errors[0].get("message") if errors else str(data)
return PublishResult(
success=False,
marketplace=self.marketplace,
sku=sku,
nm_id=current.nm_id,
error_code=str(data.get("status", "UNKNOWN")),
error_message=error_msg,
raw_response=data
)
async def get_categories(self) -> List[dict]:
"""Получение дерева категорий."""
url = f"{self.base_url}/categories/tree"
async with aiohttp.ClientSession() as session:
async with session.post(url, json={}, headers=self.headers) as resp:
if resp.status != 200:
return []
data = await resp.json()
return data.get("result", {}).get("children", [])
async def validate_content(self, content: CardContent, category_id: int) -> List[str]:
"""Валидация контента для YM."""
errors = []
if content.title and len(content.title) > 150:
errors.append(f"Title превышает лимит 150 символов ({len(content.title)})")
if content.description and len(content.description) > 3000:
errors.append(f"Description превышает лимит 3000 символов ({len(content.description)})")
return errors