diff --git a/app/handlers/admin/messages.py b/app/handlers/admin/messages.py
index aa0af353..1ee8c7ac 100644
--- a/app/handlers/admin/messages.py
+++ b/app/handlers/admin/messages.py
@@ -1146,7 +1146,10 @@ async def confirm_broadcast(callback: types.CallbackQuery, db_user: User, state:
admin_language: str = db_user.language
await safe_edit_or_send_text(
- callback, '📨 Начинаю рассылку...\n\n⏳ Это может занять несколько минут.', reply_markup=None, parse_mode='HTML'
+ callback,
+ '📨 Подготовка рассылки...\n\n⏳ Загружаю список получателей...',
+ reply_markup=None,
+ parse_mode='HTML',
)
# Загружаем пользователей и сразу извлекаем telegram_id в список
@@ -1193,93 +1196,198 @@ async def confirm_broadcast(callback: types.CallbackQuery, db_user: User, state:
broadcast_keyboard = create_broadcast_keyboard(selected_buttons, admin_language)
- # Ограничение на количество одновременных отправок и базовая задержка между сообщениями,
- # чтобы избежать перегрузки бота и лимитов Telegram при больших рассылках
- max_concurrent_sends = 5
- per_message_delay = 0.05
- semaphore = asyncio.Semaphore(max_concurrent_sends)
+ # =========================================================================
+ # Rate limiting: Telegram допускает ~30 msg/sec для бота.
+ # Используем batch_size=25 + 1 сек задержка между батчами = ~25 msg/sec
+ # с запасом, чтобы не получать FloodWait.
+ # Semaphore=25 — все сообщения батча отправляются параллельно.
+ # =========================================================================
+ _BATCH_SIZE = 25
+ _BATCH_DELAY = 1.0 # секунда между батчами
+ _MAX_SEND_RETRIES = 3
+ # Обновляем прогресс каждые N батчей (не каждое сообщение — иначе FloodWait на edit_text)
+ _PROGRESS_UPDATE_INTERVAL = max(1, 500 // _BATCH_SIZE) # ~каждые 500 сообщений
+ # Минимальный интервал между обновлениями прогресса (секунды)
+ _PROGRESS_MIN_INTERVAL = 5.0
- async def send_single_broadcast(telegram_id: int) -> tuple[bool, int]:
- """Отправляет одно сообщение рассылки с семафором ограничения."""
- async with semaphore:
- for attempt in range(3):
- try:
- if has_media and media_file_id:
- if media_type == 'photo':
- await callback.bot.send_photo(
- chat_id=telegram_id,
- photo=media_file_id,
- caption=message_text,
- parse_mode='HTML',
- reply_markup=broadcast_keyboard,
- )
- elif media_type == 'video':
- await callback.bot.send_video(
- chat_id=telegram_id,
- video=media_file_id,
- caption=message_text,
- parse_mode='HTML',
- reply_markup=broadcast_keyboard,
- )
- elif media_type == 'document':
- await callback.bot.send_document(
- chat_id=telegram_id,
- document=media_file_id,
- caption=message_text,
- parse_mode='HTML',
- reply_markup=broadcast_keyboard,
- )
+ # Глобальная пауза при FloodWait — тормозим ВСЕ отправки, а не один слот семафора
+ flood_wait_until: float = 0.0
+
+ async def send_single_broadcast(telegram_id: int) -> bool:
+ """Отправляет одно сообщение. Возвращает True при успехе."""
+ nonlocal flood_wait_until
+
+ for attempt in range(_MAX_SEND_RETRIES):
+ # Глобальная пауза при FloodWait
+ now = asyncio.get_event_loop().time()
+ if flood_wait_until > now:
+ await asyncio.sleep(flood_wait_until - now)
+
+ try:
+ if has_media and media_file_id:
+ send_method = {
+ 'photo': callback.bot.send_photo,
+ 'video': callback.bot.send_video,
+ 'document': callback.bot.send_document,
+ }.get(media_type)
+ if send_method:
+ media_kwarg = {
+ 'photo': 'photo',
+ 'video': 'video',
+ 'document': 'document',
+ }[media_type]
+ await send_method(
+ chat_id=telegram_id,
+ **{media_kwarg: media_file_id},
+ caption=message_text,
+ parse_mode='HTML',
+ reply_markup=broadcast_keyboard,
+ )
else:
+ # Неизвестный media_type — отправляем как текст
await callback.bot.send_message(
chat_id=telegram_id,
text=message_text,
parse_mode='HTML',
reply_markup=broadcast_keyboard,
)
+ else:
+ await callback.bot.send_message(
+ chat_id=telegram_id,
+ text=message_text,
+ parse_mode='HTML',
+ reply_markup=broadcast_keyboard,
+ )
+ return True
- await asyncio.sleep(per_message_delay)
- return True, telegram_id
- except TelegramRetryAfter as e:
- retry_delay = min(e.retry_after + 1, 30)
- logger.warning(f'Превышен лимит Telegram для {telegram_id}, ожидание {retry_delay} сек.')
- await asyncio.sleep(retry_delay)
- except TelegramForbiddenError:
- # Пользователь мог удалить бота или запретить сообщения
- logger.info(f'Рассылка недоступна для пользователя {telegram_id}: Forbidden')
- return False, telegram_id
- except TelegramBadRequest as e:
- logger.error(f'Некорректный запрос при рассылке пользователю {telegram_id}: {e}')
- return False, telegram_id
- except Exception as e:
- logger.error(f'Ошибка отправки рассылки пользователю {telegram_id} (попытка {attempt + 1}/3): {e}')
+ except TelegramRetryAfter as e:
+ # Глобальная пауза — тормозим все корутины
+ wait_seconds = e.retry_after + 1
+ flood_wait_until = asyncio.get_event_loop().time() + wait_seconds
+ logger.warning(
+ 'FloodWait: Telegram просит подождать %d сек (пользователь %d, попытка %d/%d)',
+ e.retry_after,
+ telegram_id,
+ attempt + 1,
+ _MAX_SEND_RETRIES,
+ )
+ await asyncio.sleep(wait_seconds)
+
+ except TelegramForbiddenError:
+ return False
+
+ except TelegramBadRequest as e:
+ logger.debug('BadRequest при рассылке пользователю %d: %s', telegram_id, e)
+ return False
+
+ except Exception as e:
+ logger.error(
+ 'Ошибка отправки пользователю %d (попытка %d/%d): %s',
+ telegram_id,
+ attempt + 1,
+ _MAX_SEND_RETRIES,
+ e,
+ )
+ if attempt < _MAX_SEND_RETRIES - 1:
await asyncio.sleep(0.5 * (attempt + 1))
- return False, telegram_id
+ return False
- # Отправляем сообщения пакетами для эффективности
- batch_size = 50
- for i in range(0, len(recipient_telegram_ids), batch_size):
- batch = recipient_telegram_ids[i : i + batch_size]
- tasks = [send_single_broadcast(tid) for tid in batch]
- results = await asyncio.gather(*tasks, return_exceptions=True)
+ # =========================================================================
+ # Прогресс-бар в реальном времени (как в сканере заблокированных)
+ # =========================================================================
+ total_recipients = len(recipient_telegram_ids)
+ last_progress_update: float = 0.0
+ # ID сообщения, которое обновляем (может быть заменено при ошибке)
+ progress_message = callback.message
+
+ def _build_progress_text(
+ current_sent: int,
+ current_failed: int,
+ total: int,
+ phase: str = 'sending',
+ ) -> str:
+ processed = current_sent + current_failed
+ percent = round(processed / total * 100, 1) if total > 0 else 0
+ bar_length = 20
+ filled = int(bar_length * processed / total) if total > 0 else 0
+ bar = '█' * filled + '░' * (bar_length - filled)
+
+ if phase == 'sending':
+ return (
+ f'📨 Рассылка в процессе...\n\n'
+ f'[{bar}] {percent}%\n\n'
+ f'📊 Прогресс:\n'
+ f'• Отправлено: {current_sent}\n'
+ f'• Ошибок: {current_failed}\n'
+ f'• Обработано: {processed}/{total}\n\n'
+ f'⏳ Не закрывайте диалог — рассылка продолжается...'
+ )
+ return ''
+
+ async def _update_progress_message(current_sent: int, current_failed: int) -> None:
+ """Безопасно обновляет сообщение с прогрессом."""
+ nonlocal last_progress_update, progress_message
+ now = asyncio.get_event_loop().time()
+ if now - last_progress_update < _PROGRESS_MIN_INTERVAL:
+ return
+ last_progress_update = now
+
+ text = _build_progress_text(current_sent, current_failed, total_recipients)
+ try:
+ await progress_message.edit_text(text, parse_mode='HTML')
+ except TelegramRetryAfter as e:
+ # Не паникуем — пропускаем обновление прогресса
+ logger.debug('FloodWait при обновлении прогресса, пропускаем: %d сек', e.retry_after)
+ except TelegramBadRequest:
+ # Сообщение удалено или контент не изменился — отправляем новое
+ try:
+ progress_message = await callback.bot.send_message(
+ chat_id=callback.message.chat.id,
+ text=text,
+ parse_mode='HTML',
+ )
+ except Exception:
+ pass
+ except Exception:
+ pass # Не ломаем рассылку из-за ошибок обновления прогресса
+
+ # Первое обновление прогресса
+ await _update_progress_message(0, 0)
+
+ # =========================================================================
+ # Основной цикл рассылки — батчами по _BATCH_SIZE
+ # =========================================================================
+ for batch_idx, i in enumerate(range(0, total_recipients, _BATCH_SIZE)):
+ batch = recipient_telegram_ids[i : i + _BATCH_SIZE]
+
+ # Отправляем батч параллельно
+ results = await asyncio.gather(
+ *[send_single_broadcast(tid) for tid in batch],
+ return_exceptions=True,
+ )
for result in results:
- if isinstance(result, tuple): # (success, telegram_id)
- success, _ = result
- if success:
+ if isinstance(result, bool):
+ if result:
sent_count += 1
else:
failed_count += 1
elif isinstance(result, Exception):
failed_count += 1
+ logger.error('Необработанное исключение в рассылке: %s', result)
- # Небольшая задержка между пакетами для снижения нагрузки на API
- await asyncio.sleep(0.25)
+ # Обновляем прогресс каждые _PROGRESS_UPDATE_INTERVAL батчей
+ if batch_idx % _PROGRESS_UPDATE_INTERVAL == 0:
+ await _update_progress_message(sent_count, failed_count)
+
+ # Задержка между батчами для соблюдения rate limits
+ await asyncio.sleep(_BATCH_DELAY)
# Учитываем пропущенных email-only пользователей
- skipped_email_users = total_users_count - len(recipient_telegram_ids)
+ skipped_email_users = total_users_count - total_recipients
if skipped_email_users > 0:
- logger.info(f'Пропущено {skipped_email_users} email-only пользователей при рассылке')
+ logger.info('Пропущено %d email-only пользователей при рассылке', skipped_email_users)
status = 'completed' if failed_count == 0 else 'partial'
@@ -1291,31 +1399,25 @@ async def confirm_broadcast(callback: types.CallbackQuery, db_user: User, state:
status=status,
)
- media_info = ''
- if has_media:
- media_info = f'\n🖼️ Медиафайл: {media_type}'
+ success_rate = round(sent_count / total_users_count * 100, 1) if total_users_count else 0
+ media_info = f'\n🖼️ Медиафайл: {media_type}' if has_media else ''
- # Используем заранее сохранённое имя админа
- result_text = f"""
-✅ Рассылка завершена!
+ result_text = (
+ f'✅ Рассылка завершена!\n\n'
+ f'📊 Результат:\n'
+ f'• Отправлено: {sent_count}\n'
+ f'• Не доставлено: {failed_count}\n'
+ f'• Всего пользователей: {total_users_count}\n'
+ f'• Успешность: {success_rate}%{media_info}\n\n'
+ f'Администратор: {admin_name}'
+ )
-📊 Результат:
-- Отправлено: {sent_count}
-- Не доставлено: {failed_count}
-- Всего пользователей: {total_users_count}
-- Успешность: {round(sent_count / total_users_count * 100, 1) if total_users_count else 0}%{media_info}
-
-Администратор: {admin_name}
-"""
+ back_keyboard = types.InlineKeyboardMarkup(
+ inline_keyboard=[[types.InlineKeyboardButton(text='📨 К рассылкам', callback_data='admin_messages')]]
+ )
try:
- await callback.message.edit_text(
- result_text,
- reply_markup=types.InlineKeyboardMarkup(
- inline_keyboard=[[types.InlineKeyboardButton(text='📨 К рассылкам', callback_data='admin_messages')]]
- ),
- parse_mode='HTML',
- )
+ await progress_message.edit_text(result_text, reply_markup=back_keyboard, parse_mode='HTML')
except TelegramBadRequest as e:
error_msg = str(e).lower()
if (
@@ -1323,15 +1425,10 @@ async def confirm_broadcast(callback: types.CallbackQuery, db_user: User, state:
or 'there is no text' in error_msg
or "message can't be edited" in error_msg
):
- # Сообщение удалено или это медиа - отправляем новое
await callback.bot.send_message(
chat_id=callback.message.chat.id,
text=result_text,
- reply_markup=types.InlineKeyboardMarkup(
- inline_keyboard=[
- [types.InlineKeyboardButton(text='📨 К рассылкам', callback_data='admin_messages')]
- ]
- ),
+ reply_markup=back_keyboard,
parse_mode='HTML',
)
else:
@@ -1339,7 +1436,12 @@ async def confirm_broadcast(callback: types.CallbackQuery, db_user: User, state:
await state.clear()
logger.info(
- f'Рассылка выполнена админом {admin_telegram_id}: {sent_count}/{total_users_count} (медиа: {has_media})'
+ 'Рассылка завершена админом %s: sent=%d, failed=%d, total=%d (медиа: %s)',
+ admin_telegram_id,
+ sent_count,
+ failed_count,
+ total_users_count,
+ has_media,
)
diff --git a/app/services/broadcast_service.py b/app/services/broadcast_service.py
index f3ab9da3..91310361 100644
--- a/app/services/broadcast_service.py
+++ b/app/services/broadcast_service.py
@@ -7,6 +7,7 @@ from datetime import datetime
from typing import TYPE_CHECKING
from aiogram import Bot
+from aiogram.exceptions import TelegramBadRequest, TelegramForbiddenError, TelegramRetryAfter
from aiogram.types import InlineKeyboardMarkup
from sqlalchemy.exc import InterfaceError, SQLAlchemyError
@@ -27,8 +28,18 @@ logger = logging.getLogger(__name__)
VALID_MEDIA_TYPES = {'photo', 'video', 'document'}
-LARGE_BROADCAST_THRESHOLD = 20_000
-PROGRESS_UPDATE_STEP = 5_000
+
+# =========================================================================
+# Telegram rate limits: ~30 msg/sec для бота.
+# batch_size=25 + 1 sec delay = ~25 msg/sec с запасом.
+# =========================================================================
+_TG_BATCH_SIZE = 25
+_TG_BATCH_DELAY = 1.0 # секунда между батчами
+_TG_MAX_RETRIES = 3 # retry при FloodWait / transient errors
+
+# Прогресс обновляется каждые ~500 сообщений ИЛИ раз в 5 секунд (что наступит раньше)
+_PROGRESS_UPDATE_MESSAGES = 500
+_PROGRESS_MIN_INTERVAL_SEC = 5.0
# Email broadcast rate limiting: max 8 emails per second
EMAIL_RATE_LIMIT = 8
@@ -61,6 +72,14 @@ class EmailBroadcastConfig:
initiator_name: str | None = None
+@dataclass(slots=True)
+class _EmailRecipient:
+ """Скалярные данные получателя email (без ORM)."""
+
+ email: str
+ user_name: str
+
+
@dataclass(slots=True)
class _BroadcastTask:
task: asyncio.Task
@@ -136,7 +155,8 @@ class BroadcastService:
broadcast.failed_count = 0
await session.commit()
- recipients = await self._fetch_recipients(config.target)
+ # _fetch_recipients теперь возвращает list[int] (telegram_id), а не ORM-объекты
+ recipient_ids: list[int] = await self._fetch_recipients(config.target)
async with AsyncSessionLocal() as session:
broadcast = await session.get(BroadcastHistory, broadcast_id)
@@ -144,45 +164,35 @@ class BroadcastService:
logger.error('Запись рассылки %s удалена до запуска', broadcast_id)
return
- broadcast.total_count = len(recipients)
+ broadcast.total_count = len(recipient_ids)
await session.commit()
if cancel_event.is_set():
await self._mark_cancelled(broadcast_id, sent_count, failed_count)
return
- if not recipients:
+ if not recipient_ids:
logger.info('Рассылка %s: получатели не найдены', broadcast_id)
await self._mark_finished(broadcast_id, sent_count, failed_count, cancelled=False)
return
keyboard = self._build_keyboard(config.selected_buttons)
- if len(recipients) > LARGE_BROADCAST_THRESHOLD:
- logger.info('Запускаем стабильный режим рассылки для %s получателей', len(recipients))
- (
- sent_count,
- failed_count,
- cancelled_during_run,
- ) = await self._run_resilient_broadcast(
- broadcast_id,
- recipients,
- config,
- keyboard,
- cancel_event,
- )
- else:
- (
- sent_count,
- failed_count,
- cancelled_during_run,
- ) = await self._run_standard_broadcast(
- broadcast_id,
- recipients,
- config,
- keyboard,
- cancel_event,
- )
+ logger.info(
+ 'Рассылка %s: начинаем отправку %d получателям (batch=%d, delay=%.1fs)',
+ broadcast_id,
+ len(recipient_ids),
+ _TG_BATCH_SIZE,
+ _TG_BATCH_DELAY,
+ )
+
+ sent_count, failed_count, cancelled_during_run = await self._send_batched(
+ broadcast_id,
+ recipient_ids,
+ config,
+ keyboard,
+ cancel_event,
+ )
if cancelled_during_run:
logger.info(
@@ -211,140 +221,125 @@ class BroadcastService:
logger.exception('Критическая ошибка при выполнении рассылки %s: %s', broadcast_id, exc)
await self._mark_failed(broadcast_id, sent_count, failed_count)
- async def _fetch_recipients(self, target: str):
+ async def _fetch_recipients(self, target: str) -> list[int]:
+ """Загружает получателей и возвращает список telegram_id (скаляры, не ORM-объекты)."""
async with AsyncSessionLocal() as session:
if target.startswith('custom_'):
criteria = target[len('custom_') :]
- return await get_custom_users(session, criteria)
- return await get_target_users(session, target)
+ users_orm = await get_custom_users(session, criteria)
+ else:
+ users_orm = await get_target_users(session, target)
- async def _run_standard_broadcast(
+ # Извлекаем telegram_id сразу, пока сессия жива.
+ # После выхода из блока ORM-объекты станут detached.
+ return [u.telegram_id for u in users_orm if u.telegram_id is not None]
+
+ async def _send_batched(
self,
broadcast_id: int,
- recipients: list,
+ recipient_ids: list[int],
config: BroadcastConfig,
keyboard: InlineKeyboardMarkup | None,
cancel_event: asyncio.Event,
) -> tuple[int, int, bool]:
- """Базовый режим рассылки для небольших списков."""
+ """
+ Единый метод рассылки для любого количества получателей.
+ Батчинг по _TG_BATCH_SIZE сообщений с _TG_BATCH_DELAY задержкой.
+ Прогресс обновляется каждые _PROGRESS_UPDATE_MESSAGES сообщений.
+ Глобальная пауза при FloodWait.
+ """
sent_count = 0
failed_count = 0
- # Ограничение на количество одновременных отправок
- semaphore = asyncio.Semaphore(20)
+ # Глобальная пауза при FloodWait — все корутины ждут
+ flood_wait_until: float = 0.0
+ last_progress_update: float = 0.0
+ last_progress_count: int = 0
+
+ async def send_single(telegram_id: int) -> bool:
+ nonlocal flood_wait_until
+
+ for attempt in range(_TG_MAX_RETRIES):
+ # Глобальная пауза при FloodWait
+ now = asyncio.get_event_loop().time()
+ if flood_wait_until > now:
+ await asyncio.sleep(flood_wait_until - now)
- async def send_single_message(user):
- """Отправляет одно сообщение с семафором ограничения"""
- async with semaphore:
if cancel_event.is_set():
return False
- telegram_id = getattr(user, 'telegram_id', None)
- if telegram_id is None:
- # Email-пользователи без telegram_id - пропускаем (не считаем ошибкой)
- return None
-
try:
await self._deliver_message(telegram_id, config, keyboard)
return True
- except Exception as exc:
- logger.error(
- 'Ошибка отправки рассылки %s пользователю %s: %s',
+
+ except TelegramRetryAfter as e:
+ wait_seconds = e.retry_after + 1
+ flood_wait_until = asyncio.get_event_loop().time() + wait_seconds
+ logger.warning(
+ 'FloodWait рассылки %s: Telegram просит %d сек (user=%d, попытка %d/%d)',
broadcast_id,
+ e.retry_after,
telegram_id,
- exc,
+ attempt + 1,
+ _TG_MAX_RETRIES,
)
+ await asyncio.sleep(wait_seconds)
+
+ except TelegramForbiddenError:
return False
- # Отправляем сообщения пакетами для эффективности
- batch_size = 100
- skipped_count = 0
- for i in range(0, len(recipients), batch_size):
+ except TelegramBadRequest:
+ return False
+
+ except Exception as exc:
+ logger.error(
+ 'Ошибка отправки рассылки %s пользователю %d (попытка %d/%d): %s',
+ broadcast_id,
+ telegram_id,
+ attempt + 1,
+ _TG_MAX_RETRIES,
+ exc,
+ )
+ if attempt < _TG_MAX_RETRIES - 1:
+ await asyncio.sleep(0.5 * (attempt + 1))
+
+ return False
+
+ for i in range(0, len(recipient_ids), _TG_BATCH_SIZE):
if cancel_event.is_set():
await self._mark_cancelled(broadcast_id, sent_count, failed_count)
return sent_count, failed_count, True
- batch = recipients[i : i + batch_size]
- tasks = [send_single_message(user) for user in batch]
- results = await asyncio.gather(*tasks, return_exceptions=True)
+ batch = recipient_ids[i : i + _TG_BATCH_SIZE]
+ results = await asyncio.gather(
+ *[send_single(tid) for tid in batch],
+ return_exceptions=True,
+ )
for result in results:
- if result is True:
- sent_count += 1
- elif result is None:
- # Email-пользователи - пропускаем без ошибки
- skipped_count += 1
- else:
- failed_count += 1
-
- # Небольшая задержка между пакетами для снижения нагрузки на API
- await asyncio.sleep(0.1)
-
- return sent_count, failed_count, False
-
- async def _run_resilient_broadcast(
- self,
- broadcast_id: int,
- recipients: list,
- config: BroadcastConfig,
- keyboard: InlineKeyboardMarkup | None,
- cancel_event: asyncio.Event,
- ) -> tuple[int, int, bool]:
- """Режим рассылки с периодическим обновлением статуса для больших списков."""
-
- sent_count = 0
- failed_count = 0
-
- # Ограничение на количество одновременных отправок
- semaphore = asyncio.Semaphore(15)
-
- async def send_single_message(user):
- async with semaphore:
- if cancel_event.is_set():
- return False
-
- telegram_id = getattr(user, 'telegram_id', None)
- if telegram_id is None:
- # Email-пользователи без telegram_id - пропускаем (не считаем ошибкой)
- return None
-
- try:
- await self._deliver_message(telegram_id, config, keyboard)
- return True
- except Exception as exc:
- logger.error(
- 'Ошибка отправки рассылки %s пользователю %s: %s',
- broadcast_id,
- telegram_id,
- exc,
- )
- return False
-
- batch_size = 100
- for i in range(0, len(recipients), batch_size):
- if cancel_event.is_set():
- await self._mark_cancelled(broadcast_id, sent_count, failed_count)
- return sent_count, failed_count, True
-
- batch = recipients[i : i + batch_size]
- tasks = [send_single_message(user) for user in batch]
- results = await asyncio.gather(*tasks, return_exceptions=True)
-
- for result in results:
- if result is True:
- sent_count += 1
- elif result is None:
- # Email-пользователи - пропускаем без ошибки
- pass
- else:
+ if isinstance(result, bool):
+ if result:
+ sent_count += 1
+ else:
+ failed_count += 1
+ elif isinstance(result, Exception):
failed_count += 1
+ logger.error('Необработанное исключение в рассылке %s: %s', broadcast_id, result)
+ # Обновляем прогресс в БД периодически
processed = sent_count + failed_count
- if processed % PROGRESS_UPDATE_STEP == 0:
+ now = asyncio.get_event_loop().time()
+ if (
+ processed - last_progress_count >= _PROGRESS_UPDATE_MESSAGES
+ or now - last_progress_update >= _PROGRESS_MIN_INTERVAL_SEC
+ ):
await self._update_progress(broadcast_id, sent_count, failed_count)
+ last_progress_count = processed
+ last_progress_update = now
- await asyncio.sleep(0.1)
+ # Задержка между батчами для rate limiting
+ await asyncio.sleep(_TG_BATCH_DELAY)
return sent_count, failed_count, False
@@ -359,37 +354,36 @@ class BroadcastService:
config: BroadcastConfig,
keyboard: InlineKeyboardMarkup | None,
) -> None:
+ """
+ Отправляет одно сообщение.
+
+ НЕ ловит исключения — TelegramRetryAfter, TelegramForbiddenError и др.
+ обрабатываются в вызывающем коде (_send_batched).
+ """
if not self._bot:
raise RuntimeError('Телеграм-бот не инициализирован')
if config.media and config.media.type in VALID_MEDIA_TYPES:
caption = config.media.caption or config.message_text
- if config.media.type == 'photo':
- await self._bot.send_photo(
- chat_id=telegram_id,
- photo=config.media.file_id,
- caption=caption,
- reply_markup=keyboard,
- )
- elif config.media.type == 'video':
- await self._bot.send_video(
- chat_id=telegram_id,
- video=config.media.file_id,
- caption=caption,
- reply_markup=keyboard,
- )
- elif config.media.type == 'document':
- await self._bot.send_document(
- chat_id=telegram_id,
- document=config.media.file_id,
- caption=caption,
- reply_markup=keyboard,
- )
+ media_methods = {
+ 'photo': ('photo', self._bot.send_photo),
+ 'video': ('video', self._bot.send_video),
+ 'document': ('document', self._bot.send_document),
+ }
+ kwarg_name, send_method = media_methods[config.media.type]
+ await send_method(
+ chat_id=telegram_id,
+ **{kwarg_name: config.media.file_id},
+ caption=caption,
+ parse_mode='HTML',
+ reply_markup=keyboard,
+ )
return
await self._bot.send_message(
chat_id=telegram_id,
text=config.message_text,
+ parse_mode='HTML',
reply_markup=keyboard,
)
@@ -617,8 +611,13 @@ class EmailBroadcastService:
logger.exception('Critical error in email broadcast %s: %s', broadcast_id, exc)
await self._mark_failed(broadcast_id, sent_count, failed_count)
- async def _fetch_email_recipients(self, target: str) -> list:
- """Fetch email recipients based on target filter."""
+ async def _fetch_email_recipients(self, target: str) -> list[_EmailRecipient]:
+ """
+ Загружает получателей email-рассылки.
+
+ Возвращает список _EmailRecipient (скалярные данные), а не ORM-объектов,
+ чтобы избежать detached state при долгих рассылках.
+ """
from sqlalchemy import select
from app.database.models import Subscription, SubscriptionStatus, User
@@ -632,18 +631,15 @@ class EmailBroadcastService:
]
if target == 'all_email':
- # All users with verified email
query = select(User).where(*base_conditions)
elif target == 'email_only':
- # Only email-registered users (no telegram)
query = select(User).where(
*base_conditions,
User.auth_type == 'email',
)
elif target == 'telegram_with_email':
- # Telegram users who also have email
query = select(User).where(
*base_conditions,
User.auth_type == 'telegram',
@@ -651,7 +647,6 @@ class EmailBroadcastService:
)
elif target == 'active_email':
- # Email users with active subscription
query = (
select(User)
.join(Subscription, User.id == Subscription.user_id)
@@ -662,7 +657,6 @@ class EmailBroadcastService:
)
elif target == 'expired_email':
- # Email users with expired subscription
query = (
select(User)
.join(Subscription, User.id == Subscription.user_id)
@@ -681,8 +675,8 @@ class EmailBroadcastService:
logger.warning('Unknown email target filter: %s', target)
return []
- # Load users in batches
- users: list = []
+ # Загружаем батчами и извлекаем скаляры сразу
+ recipients: list[_EmailRecipient] = []
offset = 0
batch_size = 1000
@@ -693,108 +687,116 @@ class EmailBroadcastService:
if not batch:
break
- users.extend(batch)
+ for user in batch:
+ email = user.email
+ if not email:
+ continue
+
+ # Формируем имя пользователя
+ user_name = user.username
+ if not user_name:
+ user_name = user.first_name or ''
+ if last_name := user.last_name:
+ user_name = f'{user_name} {last_name}'.strip()
+ if not user_name:
+ user_name = email.split('@')[0]
+
+ recipients.append(_EmailRecipient(email=email, user_name=user_name))
+
offset += batch_size
- return users
+ return recipients
async def _send_emails(
self,
broadcast_id: int,
- recipients: list,
+ recipients: list[_EmailRecipient],
config: EmailBroadcastConfig,
cancel_event: asyncio.Event,
) -> tuple[int, int, bool]:
- """Send emails with rate limiting."""
+ """
+ Отправляет email-рассылку с rate limiting.
+
+ Использует run_in_executor для синхронного SMTP, ограничивая
+ параллельность семафором EMAIL_RATE_LIMIT.
+ """
sent_count = 0
failed_count = 0
+ last_progress_count = 0
+ last_progress_time: float = 0.0
- # Semaphore for rate limiting (max EMAIL_RATE_LIMIT concurrent sends)
semaphore = asyncio.Semaphore(EMAIL_RATE_LIMIT)
- async def send_single_email(user) -> bool | None:
- """Send single email with rate limiting."""
+ async def send_single_email(recipient: _EmailRecipient) -> bool | None:
+ """Отправляет один email."""
async with semaphore:
if cancel_event.is_set():
return None
- email = getattr(user, 'email', None)
- if not email:
- return None
-
- # Render template with variables
- html_content = self._render_template(config.email_html_content, user)
- subject = self._render_template(config.email_subject, user)
+ html_content = self._render_template(config.email_html_content, recipient)
+ subject = self._render_template(config.email_subject, recipient)
try:
- # Run sync email send in executor to not block event loop
loop = asyncio.get_event_loop()
success = await loop.run_in_executor(
None,
self._email_service.send_email,
- email,
+ recipient.email,
subject,
html_content,
)
return success
except Exception as exc:
logger.error(
- 'Error sending email broadcast %s to %s: %s',
+ 'Ошибка отправки email рассылки %s на %s: %s',
broadcast_id,
- email,
+ recipient.email,
exc,
)
return False
- # Process in batches
for i in range(0, len(recipients), EMAIL_BATCH_SIZE):
if cancel_event.is_set():
await self._mark_cancelled(broadcast_id, sent_count, failed_count)
return sent_count, failed_count, True
batch = recipients[i : i + EMAIL_BATCH_SIZE]
- tasks = [send_single_email(user) for user in batch]
+ tasks = [send_single_email(r) for r in batch]
results = await asyncio.gather(*tasks, return_exceptions=True)
for result in results:
if result is True:
sent_count += 1
elif result is None:
- # Skipped (cancelled or no email)
- pass
+ pass # Cancelled or skipped
else:
failed_count += 1
- # Update progress periodically
+ # Обновляем прогресс периодически
processed = sent_count + failed_count
- if processed % PROGRESS_UPDATE_STEP == 0 or i + EMAIL_BATCH_SIZE >= len(recipients):
+ now = asyncio.get_event_loop().time()
+ if (
+ processed - last_progress_count >= _PROGRESS_UPDATE_MESSAGES
+ or now - last_progress_time >= _PROGRESS_MIN_INTERVAL_SEC
+ or i + EMAIL_BATCH_SIZE >= len(recipients)
+ ):
await self._update_progress(broadcast_id, sent_count, failed_count)
+ last_progress_count = processed
+ last_progress_time = now
- # Rate limiting delay between batches (ensure ~8 emails/sec)
+ # Rate limiting: ~8 emails/sec
await asyncio.sleep(EMAIL_BATCH_SIZE / EMAIL_RATE_LIMIT)
return sent_count, failed_count, False
- def _render_template(self, template: str, user) -> str:
- """Render template with user variables."""
+ @staticmethod
+ def _render_template(template: str, recipient: _EmailRecipient) -> str:
+ """Подставляет переменные в шаблон email."""
if not template:
return template
- # Get user name
- user_name = getattr(user, 'username', None)
- if not user_name:
- user_name = getattr(user, 'first_name', None) or ''
- if last_name := getattr(user, 'last_name', None):
- user_name = f'{user_name} {last_name}'.strip()
- if not user_name:
- user_name = getattr(user, 'email', '').split('@')[0] if getattr(user, 'email', None) else 'User'
-
- email = getattr(user, 'email', '') or ''
-
- # Replace template variables
- result = template.replace('{{user_name}}', user_name)
- result = result.replace('{{email}}', email)
-
+ result = template.replace('{{user_name}}', recipient.user_name)
+ result = result.replace('{{email}}', recipient.email)
return result
async def _mark_finished(