Merge pull request #2469 from BEDOLAGA-DEV/dev

Dev

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'

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

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')
 
 

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': 'Отсутствие дубликатов подписок',

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()

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(