KirillandClaude Fable 5.1
4534ab2683
- 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>
341 lines
13 KiB
Python
341 lines
13 KiB
Python
import asyncio
|
||
import html
|
||
import logging
|
||
from asyncio import Queue
|
||
from urllib.parse import quote
|
||
from datetime import datetime
|
||
|
||
from aiogram import Router, types
|
||
from aiogram.utils.keyboard import InlineKeyboardBuilder
|
||
|
||
from app import config
|
||
from app.misc import bot
|
||
from app.models import Review, Settings
|
||
from app.services.sheets import add_rows_to_sheet
|
||
|
||
logger = logging.getLogger(__name__)
|
||
router = Router()
|
||
|
||
PLATFORM_NAMES = {"ozon": "Ozon", "wb": "Wildberries", "ym": "Яндекс Маркет"}
|
||
PLATFORM_TAGS = {"ozon": "🟦 #Ozon", "wb": "🟪 #Wildberries", "ym": "🟨 #ЯндексМаркет"}
|
||
|
||
send_queue = Queue()
|
||
sheet_queue = Queue()
|
||
|
||
|
||
def add_to_message_queue(review: Review) -> None:
|
||
sheet_queue.put_nowait(review)
|
||
# Отправляем в Telegram и отзывы с текстом, и пустые (только оценка)
|
||
send_queue.put_nowait(review)
|
||
|
||
|
||
def render_review_to_sheet(review: Review) -> list[str]:
|
||
"""Безопасный рендер: None-поля не вызывают падение."""
|
||
if review.published_at:
|
||
formatted_date = review.published_at.astimezone(config.TIMEZONE).strftime(
|
||
"%d.%m.%Y %H:%M:%S"
|
||
)
|
||
else:
|
||
formatted_date = "-"
|
||
|
||
bables_str = "-"
|
||
if review.bables and isinstance(review.bables, list):
|
||
bables_str = ", ".join(str(x).strip() for x in review.bables if x is not None).strip() or "-"
|
||
elif review.bables:
|
||
bables_str = str(review.bables)
|
||
|
||
tone_str = (
|
||
{"negative": "негативная", "neutral": "нейтральная", "positive": "положительная"}.get(
|
||
review.tone, "-"
|
||
)
|
||
if review.tone
|
||
else "-"
|
||
)
|
||
critical_str = "критичный" if review.is_critical else "-"
|
||
|
||
# Порядок столбцов таблицы (12 колонок): 1–8 данные, 9 тональность, 10 причина, 11 критичность, 12 Тэги
|
||
return [
|
||
formatted_date,
|
||
PLATFORM_NAMES.get(review.platform, review.platform),
|
||
review.product_name or "-",
|
||
review.product_category or "-",
|
||
review.product_subcategory or "-",
|
||
review.product_id or "-",
|
||
review.rating if review.rating is not None else "-",
|
||
review.text or "-",
|
||
tone_str,
|
||
review.tone_reason or "-",
|
||
critical_str,
|
||
bables_str, # 12-й столбец — «Тэги» (WB bables)
|
||
]
|
||
|
||
|
||
_SHEETS_REQUEST_TIMEOUT = 60 # сек — защита от зависания при неотвечающем Google API
|
||
|
||
|
||
def _requeue_batch(batch: list[Review]) -> None:
|
||
"""Возвращает отзывы в очередь при ошибке."""
|
||
for review in batch:
|
||
try:
|
||
sheet_queue.put_nowait(review)
|
||
except asyncio.QueueFull:
|
||
logger.exception(
|
||
"SHEETS_QUEUE_FULL: очередь переполнена при возврате отзывов, потерян отзыв id=%s",
|
||
getattr(review, "id", "?") if review else "?",
|
||
)
|
||
|
||
|
||
async def sheet_queue_worker() -> None:
|
||
"""Send queued reviews to Google Sheet. Puts items back on failure so nothing is lost.
|
||
|
||
Возможные причины сбоев (ищи в логах SHEETS_API_ERROR / SHEETS_QUEUE_ERROR):
|
||
- 429: превышен лимит Google Sheets API (60 запросов/мин на пользователя)
|
||
- 503/504: временная недоступность Google API
|
||
- Сетевые ошибки: timeout, DNS, обрыв соединения
|
||
- 401/403: проблемы с service account / доступом к таблице
|
||
"""
|
||
retry_delay = 5
|
||
consecutive_failures = 0
|
||
loop = asyncio.get_running_loop()
|
||
|
||
while True:
|
||
if sheet_queue.empty():
|
||
await asyncio.sleep(1)
|
||
continue
|
||
|
||
batch: list[Review] = []
|
||
rows: list[list] = []
|
||
|
||
try:
|
||
for _ in range(sheet_queue.qsize()):
|
||
review = sheet_queue.get_nowait()
|
||
batch.append(review)
|
||
rows.append(render_review_to_sheet(review))
|
||
except Exception as e:
|
||
logger.exception(
|
||
"SHEETS_RENDER_ERROR: ошибка при подготовке данных для таблицы: %s. "
|
||
"Возвращаю %d отзывов в очередь.",
|
||
e,
|
||
len(batch),
|
||
)
|
||
_requeue_batch(batch)
|
||
await asyncio.sleep(retry_delay)
|
||
retry_delay = min(60, retry_delay * 2)
|
||
continue
|
||
|
||
try:
|
||
await asyncio.wait_for(
|
||
loop.run_in_executor(
|
||
None,
|
||
add_rows_to_sheet,
|
||
config.REVIEWS_SHEET_LINK,
|
||
rows,
|
||
),
|
||
timeout=_SHEETS_REQUEST_TIMEOUT,
|
||
)
|
||
if consecutive_failures > 0:
|
||
logger.info(
|
||
"SHEETS_RECOVERY: запись в Google Sheets восстановлена после %d неудачных попыток, "
|
||
"записано %d строк",
|
||
consecutive_failures,
|
||
len(rows),
|
||
)
|
||
consecutive_failures = 0
|
||
retry_delay = 5
|
||
except asyncio.TimeoutError:
|
||
consecutive_failures += 1
|
||
logger.exception(
|
||
"SHEETS_TIMEOUT: запрос к Google Sheets завис более %d сек. "
|
||
"Попытка=%d, возвращаю в очередь %d отзывов.",
|
||
_SHEETS_REQUEST_TIMEOUT,
|
||
consecutive_failures,
|
||
len(batch),
|
||
)
|
||
_requeue_batch(batch)
|
||
await asyncio.sleep(retry_delay)
|
||
retry_delay = min(60, retry_delay * 2)
|
||
except Exception as e:
|
||
consecutive_failures += 1
|
||
logger.exception(
|
||
"SHEETS_QUEUE_ERROR: ошибка при записи в Google Sheets: %s. "
|
||
"Попытка=%d, перезапуск через %d сек. Возвращаю в очередь %d отзывов.",
|
||
e,
|
||
consecutive_failures,
|
||
retry_delay,
|
||
len(batch),
|
||
)
|
||
_requeue_batch(batch)
|
||
await asyncio.sleep(retry_delay)
|
||
retry_delay = min(60, retry_delay * 2)
|
||
|
||
|
||
def _is_empty_review(review: Review) -> bool:
|
||
"""Отзыв без текста — только оценка."""
|
||
return not (review.text and review.text.strip())
|
||
|
||
|
||
def _get_thread_id_for_review(review: Review) -> int:
|
||
"""Топик для отзыва: для пустых — отдельные топики 1–3 и 4–5 звёзд, для остальных — хорошие/плохие."""
|
||
if _is_empty_review(review):
|
||
if review.rating <= 3:
|
||
return config.EMPTY_LOW_RATING_THREAD_ID or config.BAD_REVIEWS_THREAD_ID
|
||
return config.EMPTY_HIGH_RATING_THREAD_ID or config.GOOD_REVIEWS_THREAD_ID
|
||
return (
|
||
config.GOOD_REVIEWS_THREAD_ID
|
||
if review.is_good
|
||
else config.BAD_REVIEWS_THREAD_ID
|
||
)
|
||
|
||
|
||
async def send_review(review: Review) -> None:
|
||
escaped_text = html.escape(review.text) if review.text else "—"
|
||
escaped_product_id = html.escape(review.product_id)
|
||
|
||
formatted_date = review.published_at.astimezone(config.TIMEZONE).strftime(
|
||
"%d.%m.%Y %H:%M:%S МСК"
|
||
)
|
||
if review.platform == "ozon":
|
||
product_url = f"https://www.ozon.ru/product/{review.product_id}/"
|
||
elif review.platform == "ym":
|
||
# У Маркета нет публичной ссылки по SKU продавца — ведём в поиск по нему
|
||
product_url = f"https://market.yandex.ru/search?text={quote(review.product_id)}"
|
||
else:
|
||
product_url = (
|
||
f"https://www.wildberries.ru/catalog/{review.product_id}/detail.aspx"
|
||
)
|
||
|
||
now = datetime.now(config.TIMEZONE)
|
||
today = now.date()
|
||
|
||
count = await Review.filter(
|
||
is_good=review.is_good,
|
||
published_at__gte=datetime.combine(
|
||
today, datetime.min.time(), tzinfo=config.TIMEZONE
|
||
),
|
||
message_id__isnull=False,
|
||
).count()
|
||
|
||
category_hashtag = (review.product_category or "неизвестная категория").lower().replace(" ", "_")
|
||
|
||
message = (
|
||
(
|
||
"🚨 <b>Требует обработки</b> #требует_обработки\n"
|
||
if review.is_critical
|
||
else ""
|
||
)
|
||
+ f"{'✅ Положительный' if review.is_good else '⚠️ Отрицательный'} отзыв №{count + 1}\n"
|
||
f"📅 Дата: {formatted_date}\n"
|
||
f"📦 Товар: <code>{escaped_product_id}</code> #{category_hashtag}\n"
|
||
f"{PLATFORM_TAGS.get(review.platform, '#' + review.platform)}\n"
|
||
f"🔗 <a href='{product_url}'>Ссылка</a>\n"
|
||
f"⭐ Оценка: <code>{review.rating}/5</code>\n"
|
||
f"💬 Отзыв: {escaped_text}\n\n"
|
||
)
|
||
|
||
if review.response_text:
|
||
message += f"\nАвтоответ: {html.escape(review.response_text)}\n"
|
||
|
||
if review.tone:
|
||
tone_name = {
|
||
"negative": "Негативная",
|
||
"neutral": "Нейтральная",
|
||
"positive": "Позитивная",
|
||
}.get(review.tone, "неизвестно")
|
||
message += f"📊 Анализ отзыва:\nТональность: <b>#{tone_name}</b>\n"
|
||
if review.tone_reason:
|
||
message += f"Причина: <b>#{review.tone_reason}</b>\n"
|
||
|
||
thread_id = _get_thread_id_for_review(review)
|
||
|
||
markup = InlineKeyboardBuilder()
|
||
if not review.response_text:
|
||
if _is_empty_review(review):
|
||
# Пустой отзыв (только оценка): одна кнопка шаблона по рейтингу + написать свой ответ
|
||
settings = await Settings.get_from_context(review=review)
|
||
is_low_rating = review.rating is not None and review.rating <= 3
|
||
if is_low_rating:
|
||
template_text = (
|
||
settings.template_empty_low_rating
|
||
or config.TEMPLATE_EMPTY_LOW_RATING
|
||
)
|
||
markup.row(
|
||
types.InlineKeyboardButton(
|
||
text=f"Ответить (1–3 ★): {template_text}",
|
||
callback_data=f"quick_reply:{review.id}:low",
|
||
),
|
||
)
|
||
else:
|
||
template_text = (
|
||
settings.template_empty_high_rating
|
||
or config.TEMPLATE_EMPTY_HIGH_RATING
|
||
)
|
||
markup.row(
|
||
types.InlineKeyboardButton(
|
||
text=f"Ответить (4–5 ★): {template_text}",
|
||
callback_data=f"quick_reply:{review.id}:high",
|
||
),
|
||
)
|
||
markup.row(
|
||
types.InlineKeyboardButton(
|
||
text="Написать свой ответ",
|
||
callback_data=f"custom_reply:{review.id}",
|
||
),
|
||
)
|
||
else:
|
||
markup.row(
|
||
types.InlineKeyboardButton(
|
||
text="Сгенерировать ответ",
|
||
callback_data=f"generate_answer:{review.id}",
|
||
),
|
||
types.InlineKeyboardButton(
|
||
text="Написать свой ответ",
|
||
callback_data=f"custom_reply:{review.id}",
|
||
),
|
||
)
|
||
|
||
if review.response_id:
|
||
markup.row(
|
||
types.InlineKeyboardButton(
|
||
text="Удалить ответ",
|
||
callback_data=f"{review.platform}_delete:{review.response_id}",
|
||
)
|
||
)
|
||
|
||
# Send images if they exist
|
||
if review.images:
|
||
media = []
|
||
for image_url in review.images:
|
||
media.append(types.InputMediaPhoto(media=image_url))
|
||
if media:
|
||
try:
|
||
await bot.send_media_group(
|
||
chat_id=config.TG_CHAT_ID,
|
||
message_thread_id=thread_id,
|
||
media=media,
|
||
)
|
||
except Exception as e:
|
||
logger.error(f"Error sending review images: {e}")
|
||
message += "\n⚠️ Не удалось загрузить изображения\n"
|
||
|
||
sent = await bot.send_message(
|
||
chat_id=config.TG_CHAT_ID,
|
||
text=message,
|
||
message_thread_id=thread_id,
|
||
parse_mode="HTML",
|
||
reply_markup=markup.as_markup(),
|
||
)
|
||
review.chat_id = sent.chat.id
|
||
review.message_id = sent.message_id
|
||
await review.save()
|
||
|
||
|
||
async def send_queue_worker() -> None:
|
||
while True:
|
||
review = await send_queue.get()
|
||
try:
|
||
await send_review(review)
|
||
except Exception as e:
|
||
logger.error(f"Error sending message: {e=} {review.id=}")
|
||
await asyncio.sleep(3)
|
||
send_queue.task_done()
|