gy9vin
0df3018703
Реализована отказоустойчивая система отправки чеков в налоговую: - Добавлен NalogoQueueService для фоновой обработки очереди чеков - При недоступности nalog.ru (503) чеки сохраняются в Redis - Автоматическая повторная отправка с настраиваемым интервалом - Защита от DDoS: задержка между чеками (NALOGO_QUEUE_RECEIPT_DELAY) - Уведомления админам в топик при проблемах и успешной разгрузке Изменения в файлах: - app/services/nalogo_queue_service.py: новый фоновый сервис - app/services/nalogo_service.py: методы очереди, определение 503 - app/utils/cache.py: lpush/rpop/llen/lrange для Redis List - app/handlers/admin/monitoring.py: статистика чеков в админке - app/config.py: NALOGO_QUEUE_* и ADMIN_NOTIFICATIONS_NALOG_TOPIC_ID - main.py: интеграция запуска/остановки сервиса Новые ENV переменные: - ADMIN_NOTIFICATIONS_NALOG_TOPIC_ID - NALOGO_QUEUE_CHECK_INTERVAL (300с) - NALOGO_QUEUE_RECEIPT_DELAY (3с) - NALOGO_QUEUE_MAX_ATTEMPTS (10)
286 lines
12 KiB
Python
286 lines
12 KiB
Python
"""Фоновый сервис для обработки очереди чеков NaloGO.
|
||
|
||
При временной недоступности сервиса nalog.ru (503), чеки сохраняются в Redis
|
||
и отправляются позже этим сервисом.
|
||
"""
|
||
|
||
import asyncio
|
||
import logging
|
||
from datetime import datetime, timedelta
|
||
from typing import Optional
|
||
|
||
from aiogram import Bot
|
||
|
||
from app.config import settings
|
||
from app.services.nalogo_service import NaloGoService
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class NalogoQueueService:
|
||
"""Сервис фоновой обработки очереди чеков NaloGO."""
|
||
|
||
def __init__(self, nalogo_service: Optional[NaloGoService] = None):
|
||
self._nalogo_service = nalogo_service
|
||
self._bot: Optional[Bot] = None
|
||
self._task: Optional[asyncio.Task] = None
|
||
self._running = False
|
||
self._last_notification_time: Optional[datetime] = None
|
||
self._notification_cooldown = timedelta(hours=1) # Не чаще раза в час
|
||
self._had_pending_receipts = False # Флаг для отслеживания успешной разгрузки
|
||
|
||
def set_nalogo_service(self, service: NaloGoService) -> None:
|
||
"""Установить сервис NaloGO."""
|
||
self._nalogo_service = service
|
||
|
||
def set_bot(self, bot: Bot) -> None:
|
||
"""Установить бота для отправки уведомлений."""
|
||
self._bot = bot
|
||
|
||
def is_running(self) -> bool:
|
||
"""Проверка, запущен ли сервис."""
|
||
return self._running and self._task is not None and not self._task.done()
|
||
|
||
@property
|
||
def _check_interval(self) -> int:
|
||
"""Интервал проверки очереди в секундах."""
|
||
return getattr(settings, "NALOGO_QUEUE_CHECK_INTERVAL", 300)
|
||
|
||
@property
|
||
def _receipt_delay(self) -> int:
|
||
"""Задержка между отправкой чеков в секундах."""
|
||
return getattr(settings, "NALOGO_QUEUE_RECEIPT_DELAY", 3)
|
||
|
||
@property
|
||
def _max_attempts(self) -> int:
|
||
"""Максимальное количество попыток отправки чека."""
|
||
return getattr(settings, "NALOGO_QUEUE_MAX_ATTEMPTS", 10)
|
||
|
||
async def start(self) -> None:
|
||
"""Запустить фоновую обработку очереди."""
|
||
if not self._nalogo_service or not self._nalogo_service.configured:
|
||
logger.info("NaloGO не настроен, сервис очереди чеков не запущен")
|
||
return
|
||
|
||
if self.is_running():
|
||
logger.warning("Сервис очереди чеков уже запущен")
|
||
return
|
||
|
||
self._running = True
|
||
self._task = asyncio.create_task(self._process_queue_loop())
|
||
logger.info(
|
||
f"Сервис очереди чеков NaloGO запущен "
|
||
f"(интервал: {self._check_interval}с, задержка между чеками: {self._receipt_delay}с)"
|
||
)
|
||
|
||
async def stop(self) -> None:
|
||
"""Остановить фоновую обработку."""
|
||
self._running = False
|
||
if self._task and not self._task.done():
|
||
self._task.cancel()
|
||
try:
|
||
await self._task
|
||
except asyncio.CancelledError:
|
||
pass
|
||
self._task = None
|
||
logger.info("Сервис очереди чеков NaloGO остановлен")
|
||
|
||
async def _send_admin_notification(self, message: str, skip_cooldown: bool = False) -> None:
|
||
"""Отправить уведомление админам о чеках."""
|
||
if not self._bot:
|
||
return
|
||
|
||
chat_id = settings.get_admin_notifications_chat_id()
|
||
if not chat_id:
|
||
return
|
||
|
||
topic_id = settings.ADMIN_NOTIFICATIONS_NALOG_TOPIC_ID
|
||
|
||
# Проверяем cooldown (можно пропустить для важных уведомлений)
|
||
if not skip_cooldown:
|
||
now = datetime.now()
|
||
if self._last_notification_time:
|
||
if now - self._last_notification_time < self._notification_cooldown:
|
||
logger.debug("Уведомление о чеках пропущено (cooldown)")
|
||
return
|
||
|
||
try:
|
||
await self._bot.send_message(
|
||
chat_id=chat_id,
|
||
message_thread_id=topic_id,
|
||
text=message,
|
||
parse_mode="HTML",
|
||
)
|
||
self._last_notification_time = datetime.now()
|
||
logger.info("Отправлено уведомление о чеках NaloGO")
|
||
except Exception as error:
|
||
logger.error(f"Ошибка отправки уведомления о чеках: {error}")
|
||
|
||
async def _process_queue_loop(self) -> None:
|
||
"""Основной цикл обработки очереди."""
|
||
while self._running:
|
||
try:
|
||
await self._process_pending_receipts()
|
||
except Exception as error:
|
||
logger.error(f"Ошибка в цикле обработки очереди чеков: {error}")
|
||
|
||
await asyncio.sleep(self._check_interval)
|
||
|
||
async def _process_pending_receipts(self) -> None:
|
||
"""Обработать все ожидающие чеки в очереди."""
|
||
if not self._nalogo_service:
|
||
return
|
||
|
||
queue_length = await self._nalogo_service.get_queue_length()
|
||
if queue_length == 0:
|
||
return
|
||
|
||
logger.info(f"Начинаем обработку очереди чеков: {queue_length} шт.")
|
||
self._had_pending_receipts = True
|
||
|
||
processed = 0
|
||
failed = 0
|
||
skipped = 0
|
||
total_processed_amount = 0.0
|
||
service_unavailable = False
|
||
|
||
while True:
|
||
receipt_data = await self._nalogo_service.pop_receipt_from_queue()
|
||
if not receipt_data:
|
||
break
|
||
|
||
attempts = receipt_data.get("attempts", 0)
|
||
payment_id = receipt_data.get("payment_id", "unknown")
|
||
amount = receipt_data.get("amount", 0)
|
||
|
||
# Проверяем количество попыток
|
||
if attempts >= self._max_attempts:
|
||
logger.error(
|
||
f"Чек {payment_id} превысил лимит попыток ({self._max_attempts}), "
|
||
f"удален из очереди"
|
||
)
|
||
skipped += 1
|
||
continue
|
||
|
||
# Пытаемся отправить чек
|
||
try:
|
||
receipt_uuid = await self._nalogo_service.create_receipt(
|
||
name=receipt_data.get("name", "Интернет-сервис - Пополнение баланса"),
|
||
amount=amount,
|
||
quantity=receipt_data.get("quantity", 1),
|
||
client_info=receipt_data.get("client_info"),
|
||
payment_id=payment_id,
|
||
queue_on_failure=False, # Не добавлять в очередь повторно автоматически
|
||
)
|
||
|
||
if receipt_uuid:
|
||
processed += 1
|
||
total_processed_amount += amount
|
||
logger.info(
|
||
f"Чек из очереди успешно создан: {receipt_uuid} "
|
||
f"(payment_id={payment_id}, попытка {attempts + 1})"
|
||
)
|
||
else:
|
||
# Вернуть в очередь с увеличенным счетчиком попыток
|
||
await self._nalogo_service.requeue_receipt(receipt_data)
|
||
failed += 1
|
||
service_unavailable = True
|
||
logger.warning(
|
||
f"Не удалось создать чек из очереди (payment_id={payment_id}), "
|
||
f"возвращен в очередь (попытка {attempts + 1}/{self._max_attempts})"
|
||
)
|
||
# Если сервис недоступен, прекращаем попытки до следующего цикла
|
||
break
|
||
|
||
except Exception as error:
|
||
await self._nalogo_service.requeue_receipt(receipt_data)
|
||
failed += 1
|
||
logger.error(
|
||
f"Ошибка при создании чека из очереди (payment_id={payment_id}): {error}"
|
||
)
|
||
# Прекращаем попытки при ошибке
|
||
break
|
||
|
||
# Задержка между чеками чтобы не долбить API
|
||
await asyncio.sleep(self._receipt_delay)
|
||
|
||
if processed > 0 or failed > 0 or skipped > 0:
|
||
logger.info(
|
||
f"Обработка очереди завершена: "
|
||
f"успешно={processed}, неудачно={failed}, пропущено={skipped}"
|
||
)
|
||
|
||
# Проверяем остаток в очереди
|
||
remaining = await self._nalogo_service.get_queue_length()
|
||
|
||
# Отправляем уведомление если есть проблемы
|
||
if service_unavailable or failed > 0:
|
||
if remaining > 0:
|
||
queued = await self._nalogo_service.get_queued_receipts()
|
||
total_queued_amount = sum(r.get("amount", 0) for r in queued)
|
||
|
||
message = (
|
||
f"<b>⚠️ Проблема с отправкой чеков NaloGO</b>\n\n"
|
||
f"Сервис nalog.ru временно недоступен.\n\n"
|
||
f"📋 <b>В очереди:</b> {remaining} чек(ов)\n"
|
||
f"💰 <b>На сумму:</b> {total_queued_amount:,.2f} ₽\n\n"
|
||
f"Чеки будут отправлены автоматически когда сервис восстановится."
|
||
)
|
||
await self._send_admin_notification(message)
|
||
|
||
# Уведомление об успешной разгрузке очереди
|
||
elif remaining == 0 and self._had_pending_receipts and processed > 0:
|
||
self._had_pending_receipts = False
|
||
message = (
|
||
f"<b>✅ Очередь чеков NaloGO разгружена</b>\n\n"
|
||
f"Все отложенные чеки успешно отправлены!\n\n"
|
||
f"📋 <b>Отправлено:</b> {processed} чек(ов)\n"
|
||
f"💰 <b>На сумму:</b> {total_processed_amount:,.2f} ₽"
|
||
)
|
||
await self._send_admin_notification(message, skip_cooldown=True)
|
||
|
||
async def force_process(self) -> dict:
|
||
"""Принудительно обработать очередь (для ручного запуска)."""
|
||
if not self._nalogo_service:
|
||
return {"error": "NaloGO сервис не настроен"}
|
||
|
||
queue_length = await self._nalogo_service.get_queue_length()
|
||
if queue_length == 0:
|
||
return {"message": "Очередь пуста", "processed": 0}
|
||
|
||
await self._process_pending_receipts()
|
||
new_length = await self._nalogo_service.get_queue_length()
|
||
|
||
return {
|
||
"message": "Обработка завершена",
|
||
"was_in_queue": queue_length,
|
||
"remaining": new_length,
|
||
"processed": queue_length - new_length,
|
||
}
|
||
|
||
async def get_status(self) -> dict:
|
||
"""Получить статус сервиса и очереди."""
|
||
queue_length = 0
|
||
total_amount = 0.0
|
||
queued_receipts = []
|
||
|
||
if self._nalogo_service:
|
||
queue_length = await self._nalogo_service.get_queue_length()
|
||
if queue_length > 0:
|
||
queued_receipts = await self._nalogo_service.get_queued_receipts()
|
||
total_amount = sum(r.get("amount", 0) for r in queued_receipts)
|
||
|
||
return {
|
||
"running": self.is_running(),
|
||
"check_interval_seconds": self._check_interval,
|
||
"receipt_delay_seconds": self._receipt_delay,
|
||
"queue_length": queue_length,
|
||
"total_amount": total_amount,
|
||
"max_attempts": self._max_attempts,
|
||
"queued_receipts": queued_receipts[:10], # Показываем только первые 10
|
||
}
|
||
|
||
|
||
# Глобальный экземпляр сервиса
|
||
nalogo_queue_service = NalogoQueueService()
|