Добавить поддержку Яндекс Маркета (goods-feedback API)
- app/services/ym.py: клиент Partner API — getGoodsFeedbacks с пагинацией и фильтром по дате, updateGoodsFeedbackComment (ответ), deleteGoodsFeedbackComment (удаление), обработка новых отзывов по аналогии с WB/Ozon - воркер ym_worker, включается при заданных YM_API_KEY и YM_BUSINESS_ID - кнопки отправки/удаления ответа, платформа в сообщении и в таблице, статистика /stat и /allstat - переменные YM_API_KEY, YM_BUSINESS_ID, YM_CHECK_INTERVAL, PRODUCTS_YM_SHEET_LINK - README и .env.example Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Claude Fable 5.1
@@ -0,0 +1,330 @@
|
||||
import asyncio
|
||||
import logging
|
||||
import time
|
||||
from datetime import datetime, timedelta
|
||||
from functools import lru_cache
|
||||
from typing import Any
|
||||
|
||||
import httpx
|
||||
|
||||
from app import config
|
||||
from app.handlers.reviews import add_to_message_queue
|
||||
from app.models import Review, Settings
|
||||
from app.services.gpt import GPTClient
|
||||
from app.services.sheets import read_sheet
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# API отдаёт отзывы максимум за 6 месяцев; берём чуть меньше, чтобы не упереться в границу
|
||||
_MAX_LOOKBACK = timedelta(days=180 - 1)
|
||||
# Защита от бесконечной пагинации при первом запуске
|
||||
_MAX_PAGES = 40
|
||||
|
||||
|
||||
def get_cache_key() -> int:
|
||||
return int(time.time()) // config.PRODUCTS_CACHE_TTL
|
||||
|
||||
|
||||
@lru_cache(maxsize=1)
|
||||
def get_products_dict(_: int) -> dict[str, dict[str, Any]]:
|
||||
logger.info("Requesting Yandex Market products list")
|
||||
if not config.PRODUCTS_YM_SHEET_LINK:
|
||||
return {}
|
||||
sheet = read_sheet(config.PRODUCTS_YM_SHEET_LINK)
|
||||
return {
|
||||
str(
|
||||
product.get("SKU")
|
||||
or product.get("Ваш SKU")
|
||||
or product.get("offerId")
|
||||
or product.get("Артикул")
|
||||
): product
|
||||
for product in sheet
|
||||
}
|
||||
|
||||
|
||||
def _parse_datetime(value: str | None) -> datetime:
|
||||
"""createdAt приходит в ISO 8601 (с Z или смещением)."""
|
||||
if not value:
|
||||
return datetime.now(config.TIMEZONE)
|
||||
try:
|
||||
dt = datetime.fromisoformat(value.replace("Z", "+00:00"))
|
||||
if dt.tzinfo is None:
|
||||
dt = dt.replace(tzinfo=config.TIMEZONE)
|
||||
return dt.astimezone(config.TIMEZONE)
|
||||
except ValueError:
|
||||
return datetime.now(config.TIMEZONE)
|
||||
|
||||
|
||||
class YandexMarketClient:
|
||||
def __init__(self):
|
||||
self.base_url = config.YM_BASE_URL
|
||||
self.business_id = config.YM_BUSINESS_ID
|
||||
self.headers = {
|
||||
"Api-Key": config.YM_API_KEY,
|
||||
"Content-Type": "application/json",
|
||||
}
|
||||
|
||||
@staticmethod
|
||||
def _error_from_response(response: httpx.Response) -> str | None:
|
||||
"""Возвращает текст ошибки, если API ответил status=ERROR."""
|
||||
try:
|
||||
data = response.json()
|
||||
except ValueError:
|
||||
return None
|
||||
if data.get("status") == "ERROR":
|
||||
errors = data.get("errors") or []
|
||||
return "; ".join(
|
||||
f"{e.get('code')}: {e.get('message')}" for e in errors
|
||||
) or "unknown error"
|
||||
return None
|
||||
|
||||
async def get_reviews(
|
||||
self, date_from: datetime | None = None, limit: int = 50
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Все отзывы, начиная с date_from (не включительно), с обходом страниц."""
|
||||
url = f"{self.base_url}/v2/businesses/{self.business_id}/goods-feedback"
|
||||
body: dict[str, Any] = {"reactionStatus": "ALL"}
|
||||
if date_from is not None:
|
||||
body["dateTimeFrom"] = date_from.astimezone(config.TIMEZONE).isoformat(
|
||||
timespec="seconds"
|
||||
)
|
||||
|
||||
all_feedbacks: list[dict[str, Any]] = []
|
||||
page_token: str | None = None
|
||||
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=30) as client:
|
||||
for _ in range(_MAX_PAGES):
|
||||
params: dict[str, Any] = {"limit": limit}
|
||||
if page_token:
|
||||
params["page_token"] = page_token
|
||||
response = await client.post(
|
||||
url, headers=self.headers, params=params, json=body
|
||||
)
|
||||
response.raise_for_status()
|
||||
data = response.json()
|
||||
if data.get("status") == "ERROR":
|
||||
logger.error(
|
||||
f"Yandex Market returned error: {data.get('errors')}"
|
||||
)
|
||||
break
|
||||
result = data.get("result", {}) or {}
|
||||
all_feedbacks.extend(result.get("feedbacks", []) or [])
|
||||
page_token = (result.get("paging") or {}).get("nextPageToken")
|
||||
if not page_token:
|
||||
break
|
||||
except httpx.HTTPError as e:
|
||||
logger.error(f"Error fetching Yandex Market reviews: {e}")
|
||||
|
||||
return all_feedbacks
|
||||
|
||||
async def post_comment(self, review_id: str, text: str) -> dict[str, Any]:
|
||||
"""Создаёт комментарий продавца к отзыву. Возвращает {"comment_id": ...} или {"error": ...}."""
|
||||
url = (
|
||||
f"{self.base_url}/v2/businesses/{self.business_id}"
|
||||
f"/goods-feedback/comments/update"
|
||||
)
|
||||
text = (text or "").strip()
|
||||
if not text:
|
||||
return {"error": "empty comment text"}
|
||||
data = {"feedbackId": int(review_id), "comment": {"text": text[:4096]}}
|
||||
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=30) as client:
|
||||
response = await client.post(url, headers=self.headers, json=data)
|
||||
error = self._error_from_response(response)
|
||||
if error:
|
||||
logger.error(f"Error posting comment to Yandex Market: {error}")
|
||||
return {"error": error}
|
||||
response.raise_for_status()
|
||||
result = response.json().get("result", {}) or {}
|
||||
comment_id = result.get("id")
|
||||
if comment_id is None:
|
||||
return {"error": "no comment id in response"}
|
||||
return {"comment_id": str(comment_id), "status": result.get("status")}
|
||||
except httpx.HTTPError as e:
|
||||
logger.error(f"Error posting comment to Yandex Market: {e}")
|
||||
return {"error": str(e)}
|
||||
|
||||
async def delete_comment(self, comment_id: str) -> dict[str, Any]:
|
||||
url = (
|
||||
f"{self.base_url}/v2/businesses/{self.business_id}"
|
||||
f"/goods-feedback/comments/delete"
|
||||
)
|
||||
data = {"id": int(comment_id)}
|
||||
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=30) as client:
|
||||
response = await client.post(url, headers=self.headers, json=data)
|
||||
error = self._error_from_response(response)
|
||||
if error:
|
||||
logger.error(f"Error deleting comment from Yandex Market: {error}")
|
||||
return {"error": error}
|
||||
response.raise_for_status()
|
||||
return {"success": True}
|
||||
except httpx.HTTPError as e:
|
||||
logger.error(f"Error deleting comment from Yandex Market: {e}")
|
||||
return {"error": str(e)}
|
||||
|
||||
@staticmethod
|
||||
def format_datetime(iso_datetime: str) -> str:
|
||||
return _parse_datetime(iso_datetime).strftime("%d.%m.%Y %H:%M:%S") + " МСК"
|
||||
|
||||
@staticmethod
|
||||
def _review_text(feedback: dict[str, Any]) -> str:
|
||||
description = feedback.get("description") or {}
|
||||
parts = []
|
||||
if description.get("comment"):
|
||||
parts.append(str(description["comment"]).strip())
|
||||
if description.get("advantages"):
|
||||
parts.append("Достоинства: " + str(description["advantages"]).strip())
|
||||
if description.get("disadvantages"):
|
||||
parts.append("Недостатки: " + str(description["disadvantages"]).strip())
|
||||
return "\n".join(parts).strip()
|
||||
|
||||
async def process_new_reviews(self) -> dict[str, int]:
|
||||
gpt_client = GPTClient()
|
||||
|
||||
now = datetime.now(config.TIMEZONE)
|
||||
date_from: datetime | None = None
|
||||
try:
|
||||
latest_review = (
|
||||
await Review.filter(platform="ym").order_by("-published_at").first()
|
||||
)
|
||||
if latest_review:
|
||||
date_from = latest_review.published_at.astimezone(config.TIMEZONE)
|
||||
except Exception as e:
|
||||
logger.error(f"Error getting latest review: {e}")
|
||||
|
||||
min_from = now - _MAX_LOOKBACK
|
||||
if date_from is None or date_from < min_from:
|
||||
date_from = min_from
|
||||
|
||||
reviews = await self.get_reviews(date_from)
|
||||
if not reviews:
|
||||
logger.info("No reviews fetched from Yandex Market")
|
||||
return {"processed": 0, "new": 0}
|
||||
|
||||
# API отдаёт от новых к старым; обрабатываем от старых к новым
|
||||
reviews.sort(key=lambda r: r.get("createdAt", ""))
|
||||
|
||||
processed = 0
|
||||
for feedback in reviews:
|
||||
review_id = str(feedback.get("feedbackId", "") or feedback.get("id", ""))
|
||||
if not review_id:
|
||||
logger.error(f"YM: no feedback id in review {feedback}")
|
||||
continue
|
||||
|
||||
existing = await Review.filter(
|
||||
external_id=review_id, platform="ym"
|
||||
).exists()
|
||||
if existing:
|
||||
continue
|
||||
|
||||
logger.info(f"YM: processing review {feedback}")
|
||||
|
||||
text = self._review_text(feedback)
|
||||
published_at = _parse_datetime(feedback.get("createdAt"))
|
||||
|
||||
identifiers = feedback.get("identifiers") or {}
|
||||
product_id = str(
|
||||
identifiers.get("offerId") or identifiers.get("modelId") or ""
|
||||
)
|
||||
if not product_id:
|
||||
logger.error(f"YM: no product id in review {feedback}")
|
||||
continue
|
||||
|
||||
product = get_products_dict(get_cache_key()).get(product_id)
|
||||
if product:
|
||||
product.setdefault(
|
||||
"Название",
|
||||
product.get("Наименование") or product.get("Название товара"),
|
||||
)
|
||||
product.setdefault("Категория", product.get("Категория продавца"))
|
||||
product.setdefault("Подкатегория", "неизвестно")
|
||||
else:
|
||||
logger.warning(f"YM: no product found for id {product_id}")
|
||||
product = {
|
||||
"Название": "неизвестно",
|
||||
"Категория": "неизвестно",
|
||||
"Подкатегория": "неизвестно",
|
||||
}
|
||||
|
||||
media = feedback.get("media") or {}
|
||||
images = [url for url in (media.get("photos") or []) if url]
|
||||
|
||||
statistics = feedback.get("statistics") or {}
|
||||
rating = int(statistics.get("rating") or 0)
|
||||
|
||||
review_data = {
|
||||
"external_id": review_id,
|
||||
"platform": "ym",
|
||||
"product_id": product_id,
|
||||
"product_name": product.get("Название") or "неизвестно",
|
||||
"product_category": product.get("Категория") or "неизвестно",
|
||||
"product_subcategory": product.get("Подкатегория") or "неизвестно",
|
||||
"rating": rating,
|
||||
"is_good": rating > config.NEGATIVE_RATING,
|
||||
"text": text,
|
||||
"published_at": published_at,
|
||||
"processed": False,
|
||||
"images": images if images else None,
|
||||
}
|
||||
|
||||
review_obj = await Review.create(**review_data)
|
||||
|
||||
settings = await Settings.get_from_context(review=review_obj)
|
||||
|
||||
if not review_obj.text and review_obj.is_good and not settings.auto_response_empty_enabled:
|
||||
add_to_message_queue(review_obj)
|
||||
await asyncio.sleep(0.1)
|
||||
processed += 1
|
||||
continue
|
||||
|
||||
if settings.analysis_enabled and review_obj.text:
|
||||
try:
|
||||
analysis = await gpt_client.analyze_review(settings, review_obj)
|
||||
|
||||
await review_obj.update_from_dict(
|
||||
gpt_client.extract_analysis_data(analysis)
|
||||
)
|
||||
await review_obj.save()
|
||||
except Exception as e:
|
||||
logger.error(f"Error analyzing review {review_id}: {e}")
|
||||
|
||||
if settings.auto_response_enabled and review_obj.text:
|
||||
try:
|
||||
response = await gpt_client.generate_answer(settings, review_obj)
|
||||
if response:
|
||||
result = await self.post_comment(review_id, response)
|
||||
if "error" not in result:
|
||||
review_obj.response_text = response
|
||||
review_obj.response_id = result.get("comment_id")
|
||||
review_obj.responded_at = datetime.now(config.TIMEZONE)
|
||||
await review_obj.save()
|
||||
except Exception as e:
|
||||
logger.error(f"Error auto-responding to review {review_id}: {e}")
|
||||
|
||||
if settings.auto_response_empty_enabled and not review_obj.text:
|
||||
try:
|
||||
template = (
|
||||
(settings.template_empty_high_rating or config.TEMPLATE_EMPTY_HIGH_RATING)
|
||||
if review_obj.rating > config.NEGATIVE_RATING
|
||||
else (settings.template_empty_low_rating or config.TEMPLATE_EMPTY_LOW_RATING)
|
||||
)
|
||||
result = await self.post_comment(review_id, template)
|
||||
if "error" not in result:
|
||||
review_obj.response_text = template
|
||||
review_obj.response_id = result.get("comment_id")
|
||||
review_obj.responded_at = datetime.now(config.TIMEZONE)
|
||||
await review_obj.save()
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
f"Error auto-responding (template) to review {review_id}: {e}"
|
||||
)
|
||||
|
||||
add_to_message_queue(review_obj)
|
||||
processed += 1
|
||||
await asyncio.sleep(0.1)
|
||||
|
||||
return {"processed": processed, "new": len(reviews)}
|
||||
Reference in New Issue
Block a user