import asyncio import logging import time from datetime import datetime 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__) 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 WB products list") sheet = read_sheet(config.PRODUCTS_WB_SHEET_LINK) return { str(product.get("SKU") or product.get("Артикул WB")): product for product in sheet } class WildberriesClient: def __init__(self): self.base_url = config.WB_BASE_URL self.headers = { "Authorization": config.WB_API_KEY, "Content-Type": "application/json", } async def get_reviews( self, date_from: int, date_to: int, limit: int = 30 ) -> list[dict[str, Any]]: url = f"{self.base_url}/api/v1/feedbacks" all_feedbacks = [] for is_answered in [True, False]: params = { "isAnswered": is_answered, "take": limit, "skip": 0, "order": "dateDesc", "dateFrom": date_from, "dateTo": date_to, } try: async with httpx.AsyncClient() as client: response = await client.get( url, headers=self.headers, params=params ) response.raise_for_status() data = response.json() feedbacks = data.get("data", {}).get("feedbacks", []) all_feedbacks.extend(feedbacks) except httpx.HTTPError as e: logger.error(f"Error fetching Wildberries reviews: {e}") return all_feedbacks async def post_comment(self, review_id: str, text: str) -> dict[str, Any]: url = f"{self.base_url}/api/v1/feedbacks/answer" data = {"id": review_id, "text": text} try: async with httpx.AsyncClient() as client: response = await client.post(url, headers=self.headers, json=data) response.raise_for_status() if response.status_code == 204: return {"success": True} return { "success": False, "error": f"Status code: {response.status_code}", } except httpx.HTTPError as e: logger.error(f"Error posting comment to Wildberries: {e}") return {"success": False, "error": str(e)} async def delete_comment(self, reply_id: str) -> dict[str, Any]: url = f"https://www.wildberries.ru/api/comments/replies/{reply_id}" try: async with httpx.AsyncClient() as client: response = await client.delete(url, headers=self.headers) if response.status_code == 204: return {"success": True} return { "success": False, "error": f"Status code: {response.status_code}", } except httpx.HTTPError as e: logger.error(f"Error deleting comment from Wildberries: {e}") return {"success": False, "error": str(e)} @staticmethod def format_datetime(iso_datetime: str) -> str: """Format ISO datetime to a human-readable format.""" try: dt = datetime.strptime(iso_datetime, "%Y-%m-%dT%H:%M:%SZ") dt = dt.replace(tzinfo=None) # Make naive before applying timezone dt = dt.astimezone(config.TIMEZONE) return dt.strftime("%d.%m.%Y %H:%M:%S") except ValueError: return iso_datetime async def process_new_reviews(self) -> dict[str, int]: gpt_client = GPTClient() current_time = int(datetime.now(config.TIMEZONE).timestamp()) try: latest_review = ( await Review.filter(platform="wb").order_by("-published_at").first() ) start_time = ( int(latest_review.published_at.timestamp()) if latest_review else int(datetime(2000, 1, 1, tzinfo=config.TIMEZONE).timestamp()) ) except Exception as e: logger.error(f"Error getting latest review: {e}") start_time = int(datetime(2000, 1, 1, tzinfo=config.TIMEZONE).timestamp()) reviews = await self.get_reviews(start_time, current_time) if not reviews: logger.info("No reviews fetched from Wildberries") return {"processed": 0, "new": 0} processed = 0 for feedback in reviews: review_id = feedback.get("id") existing = await Review.filter( external_id=review_id, platform="wb" ).exists() if existing: continue logger.info(f"WB: processing review {feedback}") text = ( (feedback.get("text", "") or "") + "\n" + (feedback.get("pros", "") or "") + "\n" + (feedback.get("cons", "") or "") ).strip() try: published_at = datetime.strptime( feedback.get("createdDate"), "%Y-%m-%dT%H:%M:%SZ" ) published_at = published_at.replace( tzinfo=None ) # Make naive before applying timezone published_at = published_at.astimezone(config.TIMEZONE) except (ValueError, TypeError): published_at = datetime.now(config.TIMEZONE) product_id = str(feedback.get("productDetails", {}).get("nmId", "")) if not product_id: logger.error(f"WB: no product id in review {feedback}") continue product = get_products_dict(get_cache_key()).get(product_id) if product: product.setdefault("Название", product.get("Наименование")) product.setdefault("Категория", product.get("Категория продавца")) product.setdefault("Подкатегория", "неизвестно") else: logger.warning(f"WB: no product found for id {product_id}") product = { "Название": feedback.get("productDetails", {}).get( "productName", "" ), "Категория": "неизвестно", "Подкатегория": "неизвестно", } # Extract image URLs from the feedback images = [] if feedback.get("photoLinks"): for photo in feedback["photoLinks"]: if photo.get("fullSize"): images.append(photo["fullSize"]) bables_raw = feedback.get("bables") bables = ( list(bables_raw) if isinstance(bables_raw, list) else ([bables_raw] if bables_raw is not None else None) ) review_data = { "external_id": review_id, "platform": "wb", "product_id": product_id, "product_name": product.get("Название", ""), "product_category": product.get("Категория", ""), "product_subcategory": product.get("Подкатегория", ""), "rating": feedback.get("productValuation", 0), "is_good": feedback.get("productValuation", 0) > config.NEGATIVE_RATING, "text": text, "published_at": published_at, "processed": False, "images": images if images else None, "bables": bables, } 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 result.get("success"): review_obj.response_text = response 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 result.get("success"): review_obj.response_text = template 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)}