diff --git a/app/cabinet/routes/admin_broadcasts.py b/app/cabinet/routes/admin_broadcasts.py index b5d67157..950c2690 100644 --- a/app/cabinet/routes/admin_broadcasts.py +++ b/app/cabinet/routes/admin_broadcasts.py @@ -4,7 +4,7 @@ import logging from datetime import datetime from fastapi import APIRouter, Depends, HTTPException, Query, status -from sqlalchemy import func, select +from sqlalchemy import distinct, func, select from sqlalchemy.ext.asyncio import AsyncSession from app.database.models import BroadcastHistory, Subscription, SubscriptionStatus, Tariff, User @@ -13,7 +13,9 @@ from app.keyboards.admin import BROADCAST_BUTTONS, DEFAULT_BROADCAST_BUTTONS from app.services.broadcast_service import ( BroadcastConfig, BroadcastMediaConfig, + EmailBroadcastConfig, broadcast_service, + email_broadcast_service, ) from ..dependencies import get_cabinet_db, get_current_admin_user @@ -28,6 +30,11 @@ from ..schemas.broadcasts import ( BroadcastPreviewResponse, BroadcastResponse, BroadcastTariffsResponse, + CombinedBroadcastCreateRequest, + EmailFilterItem, + EmailFiltersResponse, + EmailPreviewRequest, + EmailPreviewResponse, TariffFilter, TariffForBroadcast, ) @@ -87,6 +94,25 @@ CUSTOM_FILTER_GROUPS = { } +# ============ Email Filter Labels ============ + +EMAIL_FILTER_LABELS = { + 'all_email': 'Все с email', + 'email_only': 'Только email-регистрация', + 'telegram_with_email': 'Telegram с email', + 'active_email': 'С активной подпиской', + 'expired_email': 'С истекшей подпиской', +} + +EMAIL_FILTER_GROUPS = { + 'all_email': 'basic', + 'email_only': 'auth_type', + 'telegram_with_email': 'auth_type', + 'active_email': 'subscription', + 'expired_email': 'subscription', +} + + # ============ Helper Functions ============ @@ -113,9 +139,73 @@ def _serialize_broadcast(broadcast: BroadcastHistory) -> BroadcastResponse: created_at=broadcast.created_at, completed_at=broadcast.completed_at, progress_percent=progress, + channel=getattr(broadcast, 'channel', 'telegram') or 'telegram', + email_subject=getattr(broadcast, 'email_subject', None), + email_html_content=getattr(broadcast, 'email_html_content', None), ) +async def _get_email_filter_count(db: AsyncSession, target: str) -> int: + """Get count of email users matching the filter.""" + base_conditions = [ + User.email.isnot(None), + User.email_verified == True, + User.status == 'active', + ] + + if target == 'all_email': + query = select(func.count(User.id)).where(*base_conditions) + + elif target == 'email_only': + query = select(func.count(User.id)).where( + *base_conditions, + User.auth_type == 'email', + ) + + elif target == 'telegram_with_email': + query = select(func.count(User.id)).where( + *base_conditions, + User.auth_type == 'telegram', + User.telegram_id.isnot(None), + ) + + elif target == 'active_email': + query = ( + select(func.count(distinct(User.id))) + .join(Subscription, User.id == Subscription.user_id) + .where( + *base_conditions, + Subscription.status == SubscriptionStatus.ACTIVE.value, + ) + ) + + elif target == 'expired_email': + query = ( + select(func.count(distinct(User.id))) + .join(Subscription, User.id == Subscription.user_id) + .where( + *base_conditions, + Subscription.status.in_( + [ + SubscriptionStatus.EXPIRED.value, + SubscriptionStatus.DISABLED.value, + ] + ), + ) + ) + + else: + return 0 + + result = await db.execute(query) + return result.scalar() or 0 + + +def _validate_email_target(target: str) -> bool: + """Validate email target filter.""" + return target in EMAIL_FILTER_LABELS + + async def _get_tariff_user_counts(db: AsyncSession) -> dict: """Get count of active users per tariff.""" result = await db.execute( @@ -388,6 +478,194 @@ async def list_broadcasts( ) +# ============ Email Broadcast Endpoints ============ + + +@router.get('/email-filters', response_model=EmailFiltersResponse) +async def get_email_filters( + admin: User = Depends(get_current_admin_user), + db: AsyncSession = Depends(get_cabinet_db), +) -> EmailFiltersResponse: + """Get all available email filters with user counts.""" + filters = [] + total_with_email = 0 + + for key, label in EMAIL_FILTER_LABELS.items(): + try: + count = await _get_email_filter_count(db, key) + except Exception as e: + logger.warning(f'Failed to get count for email filter {key}: {e}') + count = 0 + + filters.append( + EmailFilterItem( + key=key, + label=label, + count=count, + group=EMAIL_FILTER_GROUPS.get(key), + ) + ) + + # Track total with email (all_email filter) + if key == 'all_email': + total_with_email = count + + return EmailFiltersResponse( + filters=filters, + total_with_email=total_with_email, + ) + + +@router.post('/email-preview', response_model=EmailPreviewResponse) +async def preview_email_broadcast( + request: EmailPreviewRequest, + admin: User = Depends(get_current_admin_user), + db: AsyncSession = Depends(get_cabinet_db), +) -> EmailPreviewResponse: + """Preview email broadcast recipients count.""" + if not _validate_email_target(request.target): + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=f'Invalid email target: {request.target}', + ) + + try: + count = await _get_email_filter_count(db, request.target) + except Exception as e: + logger.error(f'Failed to get email count for target {request.target}: {e}') + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail='Failed to count email recipients', + ) + + return EmailPreviewResponse(target=request.target, count=count) + + +@router.post('/send', response_model=BroadcastResponse, status_code=status.HTTP_201_CREATED) +async def create_combined_broadcast( + request: CombinedBroadcastCreateRequest, + admin: User = Depends(get_current_admin_user), + db: AsyncSession = Depends(get_cabinet_db), +) -> BroadcastResponse: + """Create and start a combined broadcast (telegram/email/both).""" + # Get tariff IDs for target validation + result = await db.execute(select(Tariff.id)) + tariff_ids = {row[0] for row in result.all()} + + admin_name = admin.username or f'Admin #{admin.id}' + + # Validate based on channel + if request.channel in ('telegram', 'both'): + # Validate telegram target + if not _validate_target(request.target, tariff_ids): + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=f'Invalid target: {request.target}', + ) + + # Validate telegram message + if not request.message_text or not request.message_text.strip(): + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail='Message text is required for Telegram broadcast', + ) + + # Validate buttons + if not _validate_buttons(request.selected_buttons): + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail='Invalid button key', + ) + + if request.channel in ('email', 'both'): + # For email channel, target must be email filter or we use telegram target for 'both' + if request.channel == 'email' and not _validate_email_target(request.target): + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=f'Invalid email target: {request.target}', + ) + + # Validate email fields + if not request.email_subject or not request.email_subject.strip(): + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail='Email subject is required for email broadcast', + ) + + if not request.email_html_content or not request.email_html_content.strip(): + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail='Email HTML content is required for email broadcast', + ) + + media_payload = request.media + + # Create broadcast record + broadcast = BroadcastHistory( + target_type=request.target, + message_text=request.message_text.strip() if request.message_text else None, + has_media=media_payload is not None, + media_type=media_payload.type if media_payload else None, + media_file_id=media_payload.file_id if media_payload else None, + media_caption=media_payload.caption if media_payload else None, + total_count=0, + sent_count=0, + failed_count=0, + status='queued', + admin_id=admin.id, + admin_name=admin_name, + channel=request.channel, + email_subject=request.email_subject.strip() if request.email_subject else None, + email_html_content=request.email_html_content.strip() if request.email_html_content else None, + ) + db.add(broadcast) + await db.commit() + await db.refresh(broadcast) + + # Start broadcasts based on channel + if request.channel in ('telegram', 'both'): + # Prepare media config + media_config = None + if media_payload: + media_config = BroadcastMediaConfig( + type=media_payload.type, + file_id=media_payload.file_id, + caption=media_payload.caption or request.message_text, + ) + + # Create telegram broadcast config + telegram_config = BroadcastConfig( + target=request.target, + message_text=request.message_text.strip(), + selected_buttons=request.selected_buttons, + media=media_config, + initiator_name=admin_name, + ) + + await broadcast_service.start_broadcast(broadcast.id, telegram_config) + + if request.channel in ('email', 'both'): + # For 'both' channel, we use 'all_email' as default email target + # since telegram target won't match email filters + email_target = request.target if request.channel == 'email' else 'all_email' + + # Create email broadcast config + email_config = EmailBroadcastConfig( + target=email_target, + email_subject=request.email_subject.strip(), + email_html_content=request.email_html_content.strip(), + initiator_name=admin_name, + ) + + await email_broadcast_service.start_broadcast(broadcast.id, email_config) + + await db.refresh(broadcast) + + logger.info(f"Admin {admin.id} created {request.channel} broadcast {broadcast.id} for target '{request.target}'") + + return _serialize_broadcast(broadcast) + + @router.get('/{broadcast_id}', response_model=BroadcastResponse) async def get_broadcast( broadcast_id: int, @@ -410,7 +688,7 @@ async def stop_broadcast( admin: User = Depends(get_current_admin_user), db: AsyncSession = Depends(get_cabinet_db), ) -> BroadcastResponse: - """Stop a running broadcast.""" + """Stop a running broadcast (telegram or email).""" broadcast = await db.get(BroadcastHistory, broadcast_id) if not broadcast: raise HTTPException( @@ -424,7 +702,15 @@ async def stop_broadcast( detail='Broadcast is not running', ) - is_running = await broadcast_service.request_stop(broadcast_id) + # Try to stop both telegram and email broadcasts (one or both may be running) + channel = getattr(broadcast, 'channel', 'telegram') or 'telegram' + + is_running = False + if channel in ('telegram', 'both'): + is_running = await broadcast_service.request_stop(broadcast_id) or is_running + + if channel in ('email', 'both'): + is_running = await email_broadcast_service.request_stop(broadcast_id) or is_running if is_running: broadcast.status = 'cancelling' diff --git a/app/cabinet/schemas/broadcasts.py b/app/cabinet/schemas/broadcasts.py index b0675538..94f72bbc 100644 --- a/app/cabinet/schemas/broadcasts.py +++ b/app/cabinet/schemas/broadcasts.py @@ -1,10 +1,16 @@ """Pydantic schemas for cabinet broadcasts.""" from datetime import datetime +from typing import Literal from pydantic import BaseModel, Field +# ============ Channel Types ============ + +BroadcastChannel = Literal['telegram', 'email', 'both'] + + # ============ Filters ============ @@ -100,7 +106,7 @@ class BroadcastResponse(BaseModel): id: int target_type: str - message_text: str + message_text: str | None = None has_media: bool media_type: str | None = None media_file_id: str | None = None @@ -115,6 +121,11 @@ class BroadcastResponse(BaseModel): completed_at: datetime | None = None progress_percent: float = 0.0 + # Email/channel fields + channel: str = 'telegram' # telegram|email|both + email_subject: str | None = None + email_html_content: str | None = None + class Config: from_attributes = True @@ -142,3 +153,57 @@ class BroadcastPreviewResponse(BaseModel): target: str count: int + + +# ============ Email Filters ============ + + +class EmailFilterItem(BaseModel): + """Single email filter with count.""" + + key: str + label: str + count: int + group: str | None = None + + +class EmailFiltersResponse(BaseModel): + """Response with all email filters and their counts.""" + + filters: list[EmailFilterItem] + total_with_email: int + + +# ============ Combined Broadcast ============ + + +class CombinedBroadcastCreateRequest(BaseModel): + """Request to create a combined (telegram/email/both) broadcast.""" + + channel: BroadcastChannel + target: str + + # Telegram-specific fields + message_text: str | None = Field(default=None, max_length=4000) + selected_buttons: list[str] = Field(default_factory=lambda: ['home']) + media: BroadcastMediaRequest | None = None + + # Email-specific fields + email_subject: str | None = Field(default=None, max_length=255) + email_html_content: str | None = Field(default=None, max_length=100000) + + +# ============ Email Preview ============ + + +class EmailPreviewRequest(BaseModel): + """Request to preview email broadcast recipients.""" + + target: str + + +class EmailPreviewResponse(BaseModel): + """Preview response for email broadcast.""" + + target: str + count: int diff --git a/app/database/models.py b/app/database/models.py index e8019445..3f32f98f 100644 --- a/app/database/models.py +++ b/app/database/models.py @@ -1866,7 +1866,7 @@ class BroadcastHistory(Base): id = Column(Integer, primary_key=True, index=True) target_type = Column(String(100), nullable=False) - message_text = Column(Text, nullable=False) + message_text = Column(Text, nullable=True) # Nullable for email-only broadcasts has_media = Column(Boolean, default=False) media_type = Column(String(20), nullable=True) media_file_id = Column(String(255), nullable=True) @@ -1879,6 +1879,12 @@ class BroadcastHistory(Base): admin_name = Column(String(255)) created_at = Column(DateTime(timezone=True), server_default=func.now()) completed_at = Column(DateTime(timezone=True), nullable=True) + + # Email broadcast fields + channel = Column(String(20), default='telegram', nullable=False) # telegram|email|both + email_subject = Column(String(255), nullable=True) + email_html_content = Column(Text, nullable=True) + admin = relationship('User', back_populates='broadcasts') diff --git a/app/database/universal_migration.py b/app/database/universal_migration.py index ab239062..a6c72f2e 100644 --- a/app/database/universal_migration.py +++ b/app/database/universal_migration.py @@ -3357,6 +3357,38 @@ async def add_media_fields_to_broadcast_history(): return False +async def add_email_fields_to_broadcast_history(): + """Добавление полей для email-рассылки в broadcast_history.""" + logger.info('=== ДОБАВЛЕНИЕ ПОЛЕЙ EMAIL В BROADCAST_HISTORY ===') + + email_fields = { + 'channel': "VARCHAR(20) DEFAULT 'telegram'", + 'email_subject': 'VARCHAR(255)', + 'email_html_content': 'TEXT', + } + + try: + async with engine.begin() as conn: + for field_name, field_type in email_fields.items(): + field_exists = await check_column_exists('broadcast_history', field_name) + + if not field_exists: + logger.info(f'Добавление поля {field_name} в таблицу broadcast_history') + + alter_sql = f'ALTER TABLE broadcast_history ADD COLUMN {field_name} {field_type}' + await conn.execute(text(alter_sql)) + logger.info(f'✅ Поле {field_name} успешно добавлено') + else: + logger.info(f'Поле {field_name} уже существует в broadcast_history') + + logger.info('✅ Все поля email в broadcast_history готовы') + return True + + except Exception as e: + logger.error(f'Ошибка при добавлении полей email в broadcast_history: {e}') + return False + + async def add_ticket_reply_block_columns(): try: col_perm_exists = await check_column_exists('tickets', 'user_reply_block_permanent') @@ -6701,6 +6733,13 @@ async def run_universal_migration(): else: logger.warning('⚠️ Проблемы с добавлением медиа полей') + logger.info('=== ДОБАВЛЕНИЕ EMAIL ПОЛЕЙ В BROADCAST_HISTORY ===') + email_fields_added = await add_email_fields_to_broadcast_history() + if email_fields_added: + logger.info('✅ Email поля в broadcast_history готовы') + else: + logger.warning('⚠️ Проблемы с добавлением email полей') + logger.info('=== ДОБАВЛЕНИЕ ПОЛЕЙ БЛОКИРОВКИ В TICKETS ===') tickets_block_cols_added = await add_ticket_reply_block_columns() if tickets_block_cols_added: @@ -7055,6 +7094,7 @@ async def check_migration_status(): 'pinned_messages_start_mode_column': False, 'users_last_pinned_column': False, 'broadcast_history_media_fields': False, + 'broadcast_history_email_fields': False, 'subscription_duplicates': False, 'subscription_conversions_table': False, 'subscription_events_table': False, @@ -7202,6 +7242,13 @@ async def check_migration_status(): ) status['broadcast_history_media_fields'] = media_fields_exist + email_fields_exist = ( + await check_column_exists('broadcast_history', 'channel') + and await check_column_exists('broadcast_history', 'email_subject') + and await check_column_exists('broadcast_history', 'email_html_content') + ) + status['broadcast_history_email_fields'] = email_fields_exist + pinned_media_columns_exist = ( status['pinned_messages_table'] and await check_column_exists('pinned_messages', 'media_type') @@ -7254,6 +7301,7 @@ async def check_migration_status(): 'pinned_messages_start_mode_column': 'Режим отправки закрепа при /start', 'users_last_pinned_column': 'Колонка last_pinned_message_id у пользователей', 'broadcast_history_media_fields': 'Медиа поля в broadcast_history', + 'broadcast_history_email_fields': 'Email поля в broadcast_history', 'subscription_conversions_table': 'Таблица конверсий подписок', 'subscription_events_table': 'Таблица событий подписок', 'subscription_duplicates': 'Отсутствие дубликатов подписок', diff --git a/app/services/broadcast_service.py b/app/services/broadcast_service.py index 36353645..f3ab9da3 100644 --- a/app/services/broadcast_service.py +++ b/app/services/broadcast_service.py @@ -4,6 +4,7 @@ import asyncio import logging from dataclasses import dataclass from datetime import datetime +from typing import TYPE_CHECKING from aiogram import Bot from aiogram.types import InlineKeyboardMarkup @@ -18,6 +19,10 @@ from app.handlers.admin.messages import ( ) +if TYPE_CHECKING: + from app.cabinet.services.email_service import EmailService + + logger = logging.getLogger(__name__) @@ -25,6 +30,10 @@ VALID_MEDIA_TYPES = {'photo', 'video', 'document'} LARGE_BROADCAST_THRESHOLD = 20_000 PROGRESS_UPDATE_STEP = 5_000 +# Email broadcast rate limiting: max 8 emails per second +EMAIL_RATE_LIMIT = 8 +EMAIL_BATCH_SIZE = 50 + @dataclass(slots=True) class BroadcastMediaConfig: @@ -42,6 +51,16 @@ class BroadcastConfig: initiator_name: str | None = None +@dataclass +class EmailBroadcastConfig: + """Configuration for email broadcast.""" + + target: str + email_subject: str + email_html_content: str + initiator_name: str | None = None + + @dataclass(slots=True) class _BroadcastTask: task: asyncio.Task @@ -473,3 +492,396 @@ class BroadcastService: broadcast_service = BroadcastService() + + +class EmailBroadcastService: + """Handles email broadcast execution triggered from the admin web API.""" + + def __init__(self) -> None: + self._email_service: EmailService | None = None + self._tasks: dict[int, _BroadcastTask] = {} + self._lock = asyncio.Lock() + + def set_email_service(self, email_service: EmailService) -> None: + """Set email service instance.""" + self._email_service = email_service + + def is_running(self, broadcast_id: int) -> bool: + """Check if broadcast is currently running.""" + task_entry = self._tasks.get(broadcast_id) + return bool(task_entry and not task_entry.task.done()) + + async def start_broadcast(self, broadcast_id: int, config: EmailBroadcastConfig) -> None: + """Start email broadcast in background.""" + if self._email_service is None: + logger.error('Cannot start email broadcast %s: email service not initialized', broadcast_id) + await self._mark_failed(broadcast_id) + return + + if not self._email_service.is_configured(): + logger.error('Cannot start email broadcast %s: SMTP not configured', broadcast_id) + await self._mark_failed(broadcast_id) + return + + cancel_event = asyncio.Event() + + async with self._lock: + if broadcast_id in self._tasks and not self._tasks[broadcast_id].task.done(): + logger.warning('Email broadcast %s is already running', broadcast_id) + return + + task = asyncio.create_task( + self._run_broadcast(broadcast_id, config, cancel_event), + name=f'email-broadcast-{broadcast_id}', + ) + self._tasks[broadcast_id] = _BroadcastTask(task=task, cancel_event=cancel_event) + task.add_done_callback(lambda _: self._tasks.pop(broadcast_id, None)) + + async def request_stop(self, broadcast_id: int) -> bool: + """Request to stop a running broadcast.""" + async with self._lock: + task_entry = self._tasks.get(broadcast_id) + if not task_entry: + return False + + task_entry.cancel_event.set() + return True + + async def _run_broadcast( + self, + broadcast_id: int, + config: EmailBroadcastConfig, + cancel_event: asyncio.Event, + ) -> None: + """Execute email broadcast.""" + sent_count = 0 + failed_count = 0 + + try: + if cancel_event.is_set(): + await self._mark_cancelled(broadcast_id, sent_count, failed_count) + return + + # Update status to in_progress + async with AsyncSessionLocal() as session: + broadcast = await session.get(BroadcastHistory, broadcast_id) + if not broadcast: + logger.error('Broadcast record %s not found', broadcast_id) + return + + broadcast.status = 'in_progress' + broadcast.sent_count = 0 + broadcast.failed_count = 0 + await session.commit() + + # Fetch email recipients + recipients = await self._fetch_email_recipients(config.target) + + # Update total count + async with AsyncSessionLocal() as session: + broadcast = await session.get(BroadcastHistory, broadcast_id) + if not broadcast: + logger.error('Broadcast record %s deleted before start', broadcast_id) + return + + broadcast.total_count = len(recipients) + await session.commit() + + if cancel_event.is_set(): + await self._mark_cancelled(broadcast_id, sent_count, failed_count) + return + + if not recipients: + logger.info('Email broadcast %s: no recipients found', broadcast_id) + await self._mark_finished(broadcast_id, sent_count, failed_count, cancelled=False) + return + + # Send emails with rate limiting + sent_count, failed_count, was_cancelled = await self._send_emails( + broadcast_id, + recipients, + config, + cancel_event, + ) + + if was_cancelled: + logger.info('Email broadcast %s was cancelled during execution', broadcast_id) + return + + await self._mark_finished(broadcast_id, sent_count, failed_count, cancelled=False) + + except asyncio.CancelledError: + await self._mark_cancelled(broadcast_id, sent_count, failed_count) + raise + except Exception as exc: + 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.""" + from sqlalchemy import select + + from app.database.models import Subscription, SubscriptionStatus, User + + async with AsyncSessionLocal() as session: + # Base query: verified email users with active status + base_conditions = [ + User.email.isnot(None), + User.email_verified == True, + User.status == 'active', + ] + + 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', + User.telegram_id.isnot(None), + ) + + elif target == 'active_email': + # Email users with active subscription + query = ( + select(User) + .join(Subscription, User.id == Subscription.user_id) + .where( + *base_conditions, + Subscription.status == SubscriptionStatus.ACTIVE.value, + ) + ) + + elif target == 'expired_email': + # Email users with expired subscription + query = ( + select(User) + .join(Subscription, User.id == Subscription.user_id) + .where( + *base_conditions, + Subscription.status.in_( + [ + SubscriptionStatus.EXPIRED.value, + SubscriptionStatus.DISABLED.value, + ] + ), + ) + ) + + else: + logger.warning('Unknown email target filter: %s', target) + return [] + + # Load users in batches + users: list = [] + offset = 0 + batch_size = 1000 + + while True: + result = await session.execute(query.offset(offset).limit(batch_size)) + batch = result.scalars().all() + + if not batch: + break + + users.extend(batch) + offset += batch_size + + return users + + async def _send_emails( + self, + broadcast_id: int, + recipients: list, + config: EmailBroadcastConfig, + cancel_event: asyncio.Event, + ) -> tuple[int, int, bool]: + """Send emails with rate limiting.""" + sent_count = 0 + failed_count = 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 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) + + 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, + subject, + html_content, + ) + return success + except Exception as exc: + logger.error( + 'Error sending email broadcast %s to %s: %s', + broadcast_id, + 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] + 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 + 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): + await self._update_progress(broadcast_id, sent_count, failed_count) + + # Rate limiting delay between batches (ensure ~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.""" + 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) + + return result + + async def _mark_finished( + self, + broadcast_id: int, + sent_count: int, + failed_count: int, + *, + cancelled: bool, + ) -> None: + """Mark broadcast as finished.""" + status = 'cancelled' if cancelled else ('completed' if failed_count == 0 else 'partial') + await self._safe_status_update(broadcast_id, sent_count, failed_count, status=status) + + async def _mark_cancelled( + self, + broadcast_id: int, + sent_count: int, + failed_count: int, + ) -> None: + """Mark broadcast as cancelled.""" + await self._mark_finished(broadcast_id, sent_count, failed_count, cancelled=True) + + async def _mark_failed( + self, + broadcast_id: int, + sent_count: int = 0, + failed_count: int = 0, + ) -> None: + """Mark broadcast as failed.""" + await self._safe_status_update(broadcast_id, sent_count, failed_count, status='failed') + + async def _update_progress( + self, + broadcast_id: int, + sent_count: int, + failed_count: int, + ) -> None: + """Update broadcast progress.""" + await self._safe_status_update( + broadcast_id, + sent_count, + failed_count, + status='in_progress', + update_completed_at=False, + ) + + async def _safe_status_update( + self, + broadcast_id: int, + sent_count: int, + failed_count: int, + *, + status: str, + update_completed_at: bool = True, + ) -> None: + """Safely update broadcast status with retry.""" + attempts = 0 + + while attempts < 2: + try: + async with AsyncSessionLocal() as session: + broadcast = await session.get(BroadcastHistory, broadcast_id) + if not broadcast: + return + + broadcast.sent_count = sent_count + broadcast.failed_count = failed_count + broadcast.status = status + + if update_completed_at: + broadcast.completed_at = datetime.utcnow() + + await session.commit() + return + except InterfaceError as exc: + attempts += 1 + logger.warning( + 'Connection issue updating email broadcast %s: %s. Retry %s/2', + broadcast_id, + exc, + attempts, + ) + await asyncio.sleep(0.2) + except SQLAlchemyError: + logger.exception('Failed to update email broadcast status %s', broadcast_id) + return + + +email_broadcast_service = EmailBroadcastService() diff --git a/main.py b/main.py index 158b8a40..0f606ab3 100644 --- a/main.py +++ b/main.py @@ -287,6 +287,12 @@ async def main(): traffic_monitoring_scheduler.set_bot(bot) daily_subscription_service.set_bot(bot) + # Initialize email broadcast service + from app.cabinet.services.email_service import email_service + from app.services.broadcast_service import email_broadcast_service + + email_broadcast_service.set_email_service(email_service) + from app.services.admin_notification_service import AdminNotificationService async with timeline.stage(