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)}