diff --git a/handlers/admin/sender/sender_handler.py b/handlers/admin/sender/sender_handler.py index f80048fa..894e361d 100644 --- a/handlers/admin/sender/sender_handler.py +++ b/handlers/admin/sender/sender_handler.py @@ -16,6 +16,22 @@ from .sender_states import AdminSender from .sender_utils import get_recipients, parse_message_buttons +def _broadcast_progress_text(completed: int, total: int, sent: int, failed: int) -> str: + """Формирует текст статус-бара рассылки.""" + if total <= 0: + pct = 0 + bar_filled = 0 + else: + pct = min(100, int(100 * completed / total)) + bar_filled = min(10, int(10 * completed / total)) + bar = "█" * bar_filled + "░" * (10 - bar_filled) + return ( + f"📤 Рассылка...\n\n" + f"[{bar}] {pct}% ({completed}/{total})\n" + f"✅ {sent} ❌ {failed}" + ) + + router = Router() @@ -170,16 +186,39 @@ async def handle_broadcast_confirm(callback_query: CallbackQuery, state: FSMCont await state.clear() return - await callback_query.message.edit_text(f"📤 Рассылка начата!\n👥 Количество получателей: {total_users}") + status_message = callback_query.message + total_users_for_bar = len(tg_ids) + await status_message.edit_text( + _broadcast_progress_text(0, total_users_for_bar, 0, 0), + ) messages = [] for tg_id in tg_ids: message_data = {"tg_id": tg_id, "text": text_message, "photo": photo, "keyboard": keyboard} messages.append(message_data) - broadcast_service = BroadcastService(bot=callback_query.bot, session=session, messages_per_second=35) + bot = callback_query.bot - stats = await broadcast_service.broadcast(messages, workers=5) + async def on_progress(completed: int, total: int, sent: int, failed: int) -> None: + text = _broadcast_progress_text(completed, total, sent, failed) + try: + await bot.edit_message_text( + chat_id=status_message.chat.id, + message_id=status_message.message_id, + text=text, + ) + except TelegramBadRequest as e: + if "message is not modified" not in str(e).lower(): + logger.debug(f"[Sender] Обновление прогресса: {e}") + + broadcast_service = BroadcastService(bot=bot, session=session, messages_per_second=35) + + stats = await broadcast_service.broadcast( + messages, + workers=5, + on_progress=on_progress, + progress_interval=2.0, + ) duration_minutes = int(stats["total_duration"] // 60) duration_seconds = int(stats["total_duration"] % 60) diff --git a/handlers/admin/sender/sender_service.py b/handlers/admin/sender/sender_service.py index cefef810..30c61544 100644 --- a/handlers/admin/sender/sender_service.py +++ b/handlers/admin/sender/sender_service.py @@ -1,6 +1,7 @@ import asyncio import time +from collections.abc import Awaitable, Callable from collections import deque from typing import Any @@ -183,7 +184,32 @@ class BroadcastService: if self._session is not None: await self._session.rollback() - async def broadcast(self, messages: list[dict], workers: int = 20) -> dict: + async def _progress_loop( + self, + total: int, + on_progress: Callable[[int, int, int, int], Awaitable[None]], + interval: float, + ) -> None: + """Периодически вызывает on_progress(completed, total, sent, failed).""" + while self.is_running: + await asyncio.sleep(interval) + if not self.is_running: + break + completed = len(self.results) + sent = self.total_sent + failed = completed - sent + try: + await on_progress(completed, total, sent, failed) + except Exception as e: + logger.debug(f"[Broadcast] Ошибка обновления прогресса: {e}") + + async def broadcast( + self, + messages: list[dict], + workers: int = 20, + on_progress: Callable[[int, int, int, int], Awaitable[None]] | None = None, + progress_interval: float = 2.0, + ) -> dict: self.is_running = True self.start_time = time.time() self.results = [] @@ -201,6 +227,13 @@ class BroadcastService: logger.info(f"📤 Начата рассылка на {len(messages)} пользователей с {workers} воркерами") + total = len(messages) + progress_task = None + if on_progress and total > 0: + progress_task = asyncio.create_task( + self._progress_loop(total, on_progress, progress_interval), + ) + worker_tasks = [asyncio.create_task(self._worker()) for _ in range(workers)] delayed_task = asyncio.create_task(self._process_delayed_messages()) @@ -213,6 +246,23 @@ class BroadcastService: self.is_running = False + if progress_task is not None: + progress_task.cancel() + try: + await progress_task + except asyncio.CancelledError: + pass + completed = len(self.results) + try: + await on_progress( + completed, + total, + self.total_sent, + completed - self.total_sent, + ) + except Exception as e: + logger.debug(f"[Broadcast] Финальное обновление прогресса: {e}") + for task in worker_tasks: task.cancel() delayed_task.cancel() diff --git a/handlers/notifications/notify_utils.py b/handlers/notifications/notify_utils.py index 58fb07f9..59c4f940 100644 --- a/handlers/notifications/notify_utils.py +++ b/handlers/notifications/notify_utils.py @@ -17,7 +17,7 @@ from aiogram.exceptions import ( from aiogram.types import BufferedInputFile, InlineKeyboardMarkup from sqlalchemy.ext.asyncio import AsyncSession -from database import create_blocked_user +from database import async_session_maker, create_blocked_user from handlers.tariffs.tariff_display import get_key_tariff_display from handlers.utils import format_hours, format_minutes, get_russian_month from logger import logger @@ -147,7 +147,7 @@ class FastNotificationSender: await asyncio.sleep(0.1) async def _save_blocked_users(self): - if not self.blocked_users or not self.session: + if not self.blocked_users: return try: from sqlalchemy.dialects.postgresql import insert @@ -155,12 +155,12 @@ class FastNotificationSender: values = [{"tg_id": tg_id} for tg_id in self.blocked_users] stmt = insert(BlockedUser).values(values).on_conflict_do_nothing(index_elements=[BlockedUser.tg_id]) - await self.session.execute(stmt) - await self.session.commit() + async with async_session_maker() as session: + await session.execute(stmt) + await session.commit() logger.info(f"📝 Добавлено {len(self.blocked_users)} пользователей в blocked_users") except Exception as e: logger.error(f"❌ Ошибка при сохранении заблокированных пользователей: {e}") - await self.session.rollback() async def send_all(self, messages: list[dict], workers: int = 15) -> list[bool]: if not messages: