Merge pull request #78 from machka-pasla/dev

Implement user creation and synchronization enhancements in admin syn…

.env.example

@@ -76,6 +76,8 @@ SUBSCRIPTION_NOTIFY_ON_EXPIRE=True
 SUBSCRIPTION_NOTIFY_AFTER_EXPIRE=True
 SUBSCRIPTION_NOTIFY_DAYS_BEFORE=3
 
+
+REFERRAL_ONE_BONUS_PER_REFEREE=False
 # Referral Bonus Days
 REFERRAL_BONUS_DAYS_1_MONTH=3
 REFERRAL_BONUS_DAYS_3_MONTHS=7

README.md

@@ -28,7 +28,6 @@
 -   **aiohttp:** Для запуска веб-сервера (вебхуки).
 -   **SQLAlchemy 2.x & asyncpg:** Асинхронная работа с базой данных PostgreSQL.
 -   **YooKassa, aiocryptopay:** SDK для интеграции с платежными системами.
--   **APScheduler:** Для выполнения отложенных задач (например, уведомления об окончании подписки).
 -   **Pydantic:** Для управления настройками из `.env` файла.
 -   **Docker & Docker Compose:** Для контейнеризации и развертывания.
 

bot/app/controllers/__init__.py

@@ -0,0 +1,2 @@
+
+

bot/app/controllers/dispatcher_controller.py

@@ -0,0 +1,38 @@
+import logging
+from typing import Dict
+
+from aiogram import Bot, Dispatcher
+from aiogram.enums import ParseMode
+from aiogram.client.default import DefaultBotProperties
+from aiogram.fsm.storage.memory import MemoryStorage
+from sqlalchemy.orm import sessionmaker
+
+from config.settings import Settings
+from bot.middlewares.db_session import DBSessionMiddleware
+from bot.middlewares.i18n import I18nMiddleware, get_i18n_instance, JsonI18n
+from bot.middlewares.ban_check_middleware import BanCheckMiddleware
+from bot.middlewares.action_logger_middleware import ActionLoggerMiddleware
+from bot.middlewares.profile_sync import ProfileSyncMiddleware
+
+
+def build_dispatcher(settings: Settings, async_session_factory: sessionmaker) -> tuple[Dispatcher, Bot, Dict]:
+    storage = MemoryStorage()
+    default_props = DefaultBotProperties(parse_mode=ParseMode.HTML)
+    bot = Bot(token=settings.BOT_TOKEN, default=default_props)
+
+    dp = Dispatcher(storage=storage, settings=settings, bot_instance=bot)
+
+    i18n_instance = get_i18n_instance(path="locales", default=settings.DEFAULT_LANGUAGE)
+
+    dp["i18n_instance"] = i18n_instance
+    dp["async_session_factory"] = async_session_factory
+
+    dp.update.outer_middleware(DBSessionMiddleware(async_session_factory))
+    dp.update.outer_middleware(I18nMiddleware(i18n=i18n_instance, settings=settings))
+    dp.update.outer_middleware(ProfileSyncMiddleware())
+    dp.update.outer_middleware(BanCheckMiddleware(settings=settings, i18n_instance=i18n_instance))
+    dp.update.outer_middleware(ActionLoggerMiddleware(settings=settings))
+
+    return dp, bot, {"i18n_instance": i18n_instance}
+
+

bot/app/factories/__init__.py

@@ -0,0 +1,2 @@
+
+

bot/app/factories/build_services.py

@@ -0,0 +1,69 @@
+from aiogram import Bot
+from sqlalchemy.orm import sessionmaker
+
+from config.settings import Settings
+from bot.middlewares.i18n import JsonI18n
+from bot.services.yookassa_service import YooKassaService
+from bot.services.panel_api_service import PanelApiService
+from bot.services.subscription_service import SubscriptionService
+from bot.services.referral_service import ReferralService
+from bot.services.promo_code_service import PromoCodeService
+from bot.services.stars_service import StarsService
+from bot.services.tribute_service import TributeService
+from bot.services.crypto_pay_service import CryptoPayService
+from bot.services.panel_webhook_service import PanelWebhookService
+
+
+def build_core_services(
+    settings: Settings,
+    bot: Bot,
+    async_session_factory: sessionmaker,
+    i18n: JsonI18n,
+    bot_username_for_default_return: str,
+):
+    panel_service = PanelApiService(settings)
+    subscription_service = SubscriptionService(settings, panel_service, bot, i18n)
+    referral_service = ReferralService(settings, subscription_service, bot, i18n)
+    promo_code_service = PromoCodeService(settings, subscription_service, bot, i18n)
+    stars_service = StarsService(bot, settings, i18n, subscription_service, referral_service)
+    cryptopay_service = CryptoPayService(
+        settings.CRYPTOPAY_TOKEN,
+        settings.CRYPTOPAY_NETWORK,
+        bot,
+        settings,
+        i18n,
+        async_session_factory,
+        subscription_service,
+        referral_service,
+    )
+    tribute_service = TributeService(
+        bot,
+        settings,
+        i18n,
+        async_session_factory,
+        panel_service,
+        subscription_service,
+        referral_service,
+    )
+    panel_webhook_service = PanelWebhookService(bot, settings, i18n, async_session_factory)
+    yookassa_service = YooKassaService(
+        shop_id=settings.YOOKASSA_SHOP_ID,
+        secret_key=settings.YOOKASSA_SECRET_KEY,
+        configured_return_url=settings.YOOKASSA_RETURN_URL,
+        bot_username_for_default_return=bot_username_for_default_return,
+        settings_obj=settings,
+    )
+
+    return {
+        "panel_service": panel_service,
+        "subscription_service": subscription_service,
+        "referral_service": referral_service,
+        "promo_code_service": promo_code_service,
+        "stars_service": stars_service,
+        "cryptopay_service": cryptopay_service,
+        "tribute_service": tribute_service,
+        "panel_webhook_service": panel_webhook_service,
+        "yookassa_service": yookassa_service,
+    }
+
+

bot/app/web/__init__.py

@@ -0,0 +1,2 @@
+
+

bot/app/web/web_server.py

@@ -0,0 +1,91 @@
+import asyncio
+import logging
+from aiohttp import web
+from aiogram import Bot, Dispatcher
+from aiogram.webhook.aiohttp_server import SimpleRequestHandler, setup_application
+from sqlalchemy.orm import sessionmaker
+
+from config.settings import Settings
+
+
+async def build_and_start_web_app(
+    dp: Dispatcher,
+    bot: Bot,
+    settings: Settings,
+    async_session_factory: sessionmaker,
+):
+    app = web.Application()
+    app["bot"] = bot
+    app["dp"] = dp
+    app["settings"] = settings
+    app["async_session_factory"] = async_session_factory
+    # Inject shared instances used by webhook handlers
+    app["i18n"] = dp.get("i18n_instance")
+    for key in (
+        "yookassa_service",
+        "subscription_service",
+        "referral_service",
+        "panel_service",
+        "stars_service",
+        "cryptopay_service",
+        "tribute_service",
+        "panel_webhook_service",
+    ):
+        # Access dispatcher workflow_data directly to avoid sequence protocol issues
+        if hasattr(dp, "workflow_data") and key in dp.workflow_data:  # type: ignore
+            app[key] = dp.workflow_data[key]  # type: ignore
+
+    setup_application(app, dp, bot=bot)
+
+    telegram_uses_webhook_mode = bool(settings.WEBHOOK_BASE_URL)
+
+    if telegram_uses_webhook_mode:
+        telegram_webhook_path = f"/{settings.BOT_TOKEN}"
+        app.router.add_post(telegram_webhook_path, SimpleRequestHandler(dispatcher=dp, bot=bot))
+        logging.info(
+            f"Telegram webhook route configured at: [POST] {telegram_webhook_path} (relative to base URL)"
+        )
+
+    from bot.handlers.user.payment import yookassa_webhook_route
+    from bot.services.tribute_service import tribute_webhook_route
+    from bot.services.crypto_pay_service import cryptopay_webhook_route
+    from bot.services.panel_webhook_service import panel_webhook_route
+
+    tribute_path = settings.tribute_webhook_path
+    if tribute_path.startswith("/"):
+        app.router.add_post(tribute_path, tribute_webhook_route)
+        logging.info(f"Tribute webhook route configured at: [POST] {tribute_path}")
+
+    cp_path = settings.cryptopay_webhook_path
+    if cp_path.startswith("/"):
+        app.router.add_post(cp_path, cryptopay_webhook_route)
+        logging.info(f"CryptoPay webhook route configured at: [POST] {cp_path}")
+
+    # YooKassa webhook (register only when base URL present and path configured)
+    yk_path = settings.yookassa_webhook_path
+    if settings.WEBHOOK_BASE_URL and yk_path and yk_path.startswith("/"):
+        app.router.add_post(yk_path, yookassa_webhook_route)
+        logging.info(f"YooKassa webhook route configured at: [POST] {yk_path}")
+
+    panel_path = settings.panel_webhook_path
+    if panel_path.startswith("/"):
+        app.router.add_post(panel_path, panel_webhook_route)
+        logging.info(f"Panel webhook route configured at: [POST] {panel_path}")
+
+    web_app_runner = web.AppRunner(app)
+    await web_app_runner.setup()
+    site = web.TCPSite(
+        web_app_runner,
+        host=settings.WEB_SERVER_HOST,
+        port=settings.WEB_SERVER_PORT,
+    )
+
+    await site.start()
+    logging.info(
+        f"AIOHTTP server started on http://{settings.WEB_SERVER_HOST}:{settings.WEB_SERVER_PORT}"
+    )
+
+    # Run until cancelled
+    await asyncio.Event().wait()
+
+

bot/handlers/admin/__init__.py

@@ -7,6 +7,7 @@ from . import user_management
 from . import statistics
 from . import sync_admin
 from . import logs_admin
+from . import payments
 
 admin_router_aggregate = Router(name="admin_features_router")
 
@@ -17,5 +18,6 @@ admin_router_aggregate.include_router(user_management.router)
 admin_router_aggregate.include_router(statistics.router)
 admin_router_aggregate.include_router(sync_admin.router)
 admin_router_aggregate.include_router(logs_admin.router)
+admin_router_aggregate.include_router(payments.router)
 
 __all__ = ("admin_router_aggregate", )

bot/handlers/admin/broadcast.py

@@ -1,7 +1,7 @@
 import logging
 import asyncio
 from aiogram import Router, F, types, Bot
-from aiogram.exceptions import TelegramRetryAfter
+from aiogram.exceptions import TelegramRetryAfter, TelegramBadRequest
 
 from aiogram.fsm.context import FSMContext
 from typing import Optional
@@ -19,6 +19,7 @@ from bot.keyboards.inline.admin_keyboards import (
 )
 from bot.middlewares.i18n import JsonI18n
 from bot.utils.message_queue import get_queue_manager
+from bot.utils import get_message_content, send_message_by_type, send_message_via_queue, MessageContent
 
 router = Router(name="admin_broadcast_router")
 
@@ -58,13 +59,14 @@ async def broadcast_message_prompt_handler(
     await state.set_state(AdminStates.waiting_for_broadcast_message)
 
 
-@router.message(AdminStates.waiting_for_broadcast_message, F.text)
+@router.message(AdminStates.waiting_for_broadcast_message)
 async def process_broadcast_message_handler(
     message: types.Message,
     state: FSMContext,
     i18n_data: dict,
     settings: Settings,
     session: AsyncSession,
+    bot: Bot,
 ):
     current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE)
     i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance")
@@ -75,24 +77,93 @@ async def process_broadcast_message_handler(
 
     _ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs)
 
-    # Сохраняем в state исходный текст и entities
-    text = message.text or message.caption or ""
+    # Определяем тип содержимого и сохраняем данные в state
     entities = message.entities or message.caption_entities or []
+    content = get_message_content(message)
 
+    # Если нет ни текста, ни медиа — ошибка
+    if not content.text and not content.file_id:
+        await message.answer(_("admin_broadcast_error_no_message"))
+        return
+
+    # Сохраняем данные для рассылки
     await state.update_data(
-        broadcast_text=text,
+        broadcast_text=content.text,
         broadcast_entities=entities,
+        broadcast_content_type=content.content_type,
+        broadcast_file_id=content.file_id,
+        broadcast_target="all",
     )
 
-    confirmation_prompt = _("admin_broadcast_confirm_prompt", message_preview=text)
+    # Отправляем превью-копию того, что будет разослано
+    try:
+        await send_message_by_type(
+            bot, 
+            chat_id=message.chat.id, 
+            content=content,
+            parse_mode="HTML",
+            disable_web_page_preview=True,
+            disable_notification=True,
+        )
+    except TelegramBadRequest as e:
+        await message.answer(
+            _(
+                "admin_broadcast_invalid_html",
+                default="❌ Некорректный HTML в сообщении. Пожалуйста, отправьте корректный HTML (поддерживаются теги Telegram) или уберите теги.\nОшибка: {error}",
+                error=str(e),
+            )
+        )
+        return
+
+    # Показываем короткое подтверждение без дублирования текста — сообщение выше служит превью
+    confirmation_prompt = _("admin_broadcast_confirm_prompt_short")
 
     await message.answer(
         confirmation_prompt,
-        reply_markup=get_broadcast_confirmation_keyboard(current_lang, i18n),
+        reply_markup=get_broadcast_confirmation_keyboard(current_lang, i18n, target="all"),
     )
     await state.set_state(AdminStates.confirming_broadcast)
 
 
+@router.callback_query(
+    F.data.startswith("broadcast_target:"),
+    AdminStates.confirming_broadcast,
+)
+async def change_broadcast_target_handler(
+    callback: types.CallbackQuery,
+    state: FSMContext,
+    i18n_data: dict,
+    settings: Settings,
+):
+    current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE)
+    i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance")
+    if not i18n or not callback.message:
+        await callback.answer("Error updating selection.", show_alert=True)
+        return
+
+    new_target = callback.data.split(":")[1]
+    if new_target not in {"all", "active", "inactive"}:
+        await callback.answer("Unknown target.", show_alert=True)
+        return
+
+    await state.update_data(broadcast_target=new_target)
+    user_fsm_data = await state.get_data()
+    _ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs)
+    confirmation_prompt = _(
+        "admin_broadcast_confirm_prompt_short"
+    )
+    try:
+        await callback.message.edit_text(
+            confirmation_prompt,
+            reply_markup=get_broadcast_confirmation_keyboard(
+                current_lang, i18n, target=new_target
+            ),
+        )
+    except Exception:
+        pass
+    await callback.answer()
+
+
 @router.callback_query(
     F.data == "admin_action:main", AdminStates.waiting_for_broadcast_message
 )
@@ -149,10 +220,15 @@ async def confirm_broadcast_callback_handler(
     user_fsm_data = await state.get_data()
 
     if action == "send":
-        text = user_fsm_data.get("broadcast_text")
+        # Создаем объект контента из сохраненных данных
+        content = MessageContent(
+            content_type=user_fsm_data.get("broadcast_content_type", "text"),
+            file_id=user_fsm_data.get("broadcast_file_id"),
+            text=user_fsm_data.get("broadcast_text")
+        )
         entities = user_fsm_data.get("broadcast_entities", [])
-
-        if not text:
+        
+        if not content.text and content.content_type == "text":
             await callback.message.edit_text(_("admin_broadcast_error_no_message"))
             await state.clear()
             await callback.answer(
@@ -163,13 +239,19 @@ async def confirm_broadcast_callback_handler(
         await callback.message.edit_text(_("admin_broadcast_sending_started"), reply_markup=None)
         await callback.answer()
 
-        user_ids = await user_dal.get_all_active_user_ids_for_broadcast(session)
+        target = user_fsm_data.get("broadcast_target", "all")
+        if target == "active":
+            user_ids = await user_dal.get_user_ids_with_active_subscription(session)
+        elif target == "inactive":
+            user_ids = await user_dal.get_user_ids_without_active_subscription(session)
+        else:
+            user_ids = await user_dal.get_all_active_user_ids_for_broadcast(session)
 
         sent_count = 0
         failed_count = 0
         admin_user = callback.from_user
         logging.info(
-            f"Admin {admin_user.id} broadcasting '{text[:50]}...' to {len(user_ids)} users."
+            f"Admin {admin_user.id} broadcasting '{(content.text or '')[:50]}...' to {len(user_ids)} users."
         )
 
         # Get message queue manager
@@ -181,10 +263,12 @@ async def confirm_broadcast_callback_handler(
         # Queue all messages for sending
         for uid in user_ids:
             try:
-                await queue_manager.send_message(
-                    chat_id=uid,
-                    text=text,
-                    entities=entities,
+                await send_message_via_queue(
+                    queue_manager, 
+                    uid, 
+                    content,
+                    parse_mode="HTML",
+                    disable_web_page_preview=True,
                 )
                 sent_count += 1
                 
@@ -196,7 +280,7 @@ async def confirm_broadcast_callback_handler(
                         "telegram_username": admin_user.username,
                         "telegram_first_name": admin_user.first_name,
                         "event_type": "admin_broadcast_queued",
-                        "content": f"To user {uid}: {text[:70]}...",
+                        "content": f"To user {uid}: [{content.content_type}] {(content.text or '')[:70]}...",
                         "is_admin_event": True,
                         "target_user_id": uid,
                     },

bot/handlers/admin/common.py

@@ -123,6 +123,10 @@ async def admin_panel_actions_callback_handler(
         await callback.answer(_("admin_sync_initiated_from_panel"))
     elif action == "queue_status":
         await show_queue_status_handler(callback, i18n_data)
+    elif action == "view_payments":
+        from . import payments as admin_payments_handlers
+        await admin_payments_handlers.view_payments_handler(
+            callback, i18n_data, settings, session)
     elif action == "main":
         try:
             await callback.message.edit_text(

bot/handlers/admin/payments.py

@@ -0,0 +1,245 @@
+import logging
+import csv
+import io
+from aiogram import Router, F, types
+from aiogram.filters import StateFilter
+from aiogram.fsm.context import FSMContext
+from datetime import datetime, timedelta, timezone
+from typing import Optional, List
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from config.settings import Settings
+from db.dal import payment_dal
+from db.models import Payment
+from bot.keyboards.inline.admin_keyboards import get_back_to_admin_panel_keyboard
+from aiogram.utils.keyboard import InlineKeyboardBuilder, InlineKeyboardButton
+from bot.middlewares.i18n import JsonI18n
+
+router = Router(name="admin_payments_router")
+
+
+async def get_payments_with_pagination(session: AsyncSession, page: int = 0, 
+                                     page_size: int = 10) -> tuple[List[Payment], int]:
+    """Get payments with pagination and total count."""
+    offset = page * page_size
+    
+    # Get total count
+    total_count = await payment_dal.get_payments_count(session)
+    
+    # Get payments for current page
+    payments = await payment_dal.get_recent_payment_logs_with_user(
+        session, limit=page_size, offset=offset
+    )
+    
+    return payments, total_count
+
+
+def format_payment_text(payment: Payment, i18n: JsonI18n, lang: str) -> str:
+    """Format single payment info as text."""
+    _ = lambda key, **kwargs: i18n.gettext(lang, key, **kwargs)
+    
+    status_emoji = "✅" if payment.status == 'succeeded' else (
+        "⏳" if payment.status in ['pending', 'pending_yookassa'] else "❌"
+    )
+    
+    user_info = f"User {payment.user_id}"
+    if payment.user and payment.user.username:
+        user_info += f" (@{payment.user.username})"
+    elif payment.user and payment.user.first_name:
+        user_info += f" ({payment.user.first_name})"
+    
+    payment_date = payment.created_at.strftime('%Y-%m-%d %H:%M') if payment.created_at else "N/A"
+    
+    provider_text = {
+        'yookassa': 'YooKassa',
+        'tribute': 'Tribute', 
+        'telegram_stars': 'Telegram Stars',
+        'cryptopay': 'CryptoPay'
+    }.get(payment.provider, payment.provider or 'Unknown')
+    
+    return (
+        f"{status_emoji} <b>{payment.amount} {payment.currency}</b>\n"
+        f"👤 {user_info}\n"
+        f"💳 {provider_text}\n"
+        f"📅 {payment_date}\n"
+        f"📋 {payment.status}\n"
+        f"📝 {payment.description or 'N/A'}"
+    )
+
+
+async def view_payments_handler(callback: types.CallbackQuery, i18n_data: dict, 
+                              settings: Settings, session: AsyncSession, page: int = 0):
+    """Display paginated list of all payments."""
+    current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE)
+    i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance")
+    if not i18n or not callback.message:
+        await callback.answer("Error processing request.", show_alert=True)
+        return
+    _ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs)
+
+    page_size = 5  # Показываем по 5 платежей на странице
+    payments, total_count = await get_payments_with_pagination(session, page, page_size)
+    total_pages = (total_count + page_size - 1) // page_size if total_count > 0 else 1
+
+    if not payments and page == 0:
+        await callback.message.edit_text(
+            _("admin_no_payments_found", default="Платежи не найдены."),
+            reply_markup=get_back_to_admin_panel_keyboard(current_lang, i18n),
+            parse_mode="HTML"
+        )
+        await callback.answer()
+        return
+
+    # Format payments text
+    text_parts = [_("admin_payments_header", default="💰 <b>Все платежи</b>")]
+    text_parts.append(f"📊 Показано {len(payments)} из {total_count} платежей (стр. {page + 1}/{total_pages})\n")
+    
+    for i, payment in enumerate(payments, 1):
+        text_parts.append(f"<b>{page * page_size + i}.</b> {format_payment_text(payment, i18n, current_lang)}")
+        text_parts.append("")  # Empty line between payments
+
+    # Build keyboard with pagination and export
+    builder = InlineKeyboardBuilder()
+    
+    # Pagination buttons
+    nav_buttons = []
+    if page > 0:
+        nav_buttons.append(InlineKeyboardButton(text="⬅️", callback_data=f"payments_page:{page-1}"))
+    
+    nav_buttons.append(InlineKeyboardButton(text=f"{page + 1}/{total_pages}", callback_data="noop"))
+    
+    if page < total_pages - 1:
+        nav_buttons.append(InlineKeyboardButton(text="➡️", callback_data=f"payments_page:{page+1}"))
+    
+    if nav_buttons:
+        builder.row(*nav_buttons)
+    
+    # Export and refresh buttons
+    builder.row(
+        InlineKeyboardButton(
+            text=_("admin_export_payments_csv", default="📊 Экспорт CSV"), 
+            callback_data="payments_export_csv"
+        ),
+        InlineKeyboardButton(
+            text=_("admin_refresh_payments", default="🔄 Обновить"), 
+            callback_data=f"payments_page:{page}"
+        )
+    )
+    
+    # Back button
+    builder.row(InlineKeyboardButton(
+        text=_("back_to_admin_panel_button"), 
+        callback_data="admin_section:stats_monitoring"
+    ))
+
+    await callback.message.edit_text(
+        "\n".join(text_parts),
+        reply_markup=builder.as_markup(),
+        parse_mode="HTML"
+    )
+    await callback.answer()
+
+
+@router.callback_query(F.data.startswith("payments_page:"))
+async def payments_pagination_handler(callback: types.CallbackQuery, i18n_data: dict, 
+                                    settings: Settings, session: AsyncSession):
+    """Handle pagination for payments list."""
+    try:
+        page = int(callback.data.split(":")[1])
+        await view_payments_handler(callback, i18n_data, settings, session, page)
+    except (ValueError, IndexError):
+        await callback.answer("Error processing pagination.", show_alert=True)
+
+
+@router.callback_query(F.data == "payments_export_csv")
+async def export_payments_csv_handler(callback: types.CallbackQuery, i18n_data: dict, 
+                                    settings: Settings, session: AsyncSession):
+    """Export all successful payments to CSV file."""
+    current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE)
+    i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance")
+    if not i18n:
+        await callback.answer("Language service error.", show_alert=True)
+        return
+    _ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs)
+
+    try:
+        # Get all successful payments
+        all_payments = await payment_dal.get_all_succeeded_payments_with_user(session)
+        
+        if not all_payments:
+            await callback.answer(
+                _("admin_no_payments_to_export", default="Нет платежей для экспорта."),
+                show_alert=True
+            )
+            return
+
+        # Create CSV in memory
+        output = io.StringIO()
+        writer = csv.writer(output)
+        
+        # Write header
+        writer.writerow([
+            _("admin_csv_payment_id", default="ID"),
+            _("admin_csv_user_id", default="User ID"),
+            _("admin_csv_username", default="Username"),
+            _("admin_csv_first_name", default="First Name"),
+            _("admin_csv_amount", default="Amount"),
+            _("admin_csv_currency", default="Currency"),
+            _("admin_csv_provider", default="Provider"),
+            _("admin_csv_status", default="Status"),
+            _("admin_csv_description", default="Description"),
+            _("admin_csv_months", default="Months"),
+            _("admin_csv_created_at", default="Created At"),
+            _("admin_csv_provider_payment_id", default="Provider Payment ID")
+        ])
+        
+        # Write payment data
+        for payment in all_payments:
+            writer.writerow([
+                payment.payment_id,
+                payment.user_id,
+                payment.user.username if payment.user and payment.user.username else "",
+                payment.user.first_name if payment.user and payment.user.first_name else "",
+                payment.amount,
+                payment.currency,
+                payment.provider or "",
+                payment.status,
+                payment.description or "",
+                payment.subscription_duration_months or "",
+                payment.created_at.strftime('%Y-%m-%d %H:%M:%S') if payment.created_at else "",
+                payment.provider_payment_id or ""
+            ])
+        
+        # Prepare file
+        csv_content = output.getvalue().encode('utf-8-sig')  # UTF-8 with BOM for Excel
+        output.close()
+        
+        # Generate filename with current date
+        current_time = datetime.now().strftime('%Y-%m-%d_%H-%M')
+        filename = f"payments_export_{current_time}.csv"
+        
+        # Send file
+        from aiogram.types import BufferedInputFile
+        file = BufferedInputFile(csv_content, filename=filename)
+        
+        await callback.message.reply_document(
+            document=file,
+            caption=_("admin_payments_export_success", 
+                     default="📊 Экспорт платежей завершен!\nВсего записей: {count}",
+                     count=len(all_payments))
+        )
+        
+        await callback.answer(
+            _("admin_export_sent", default="Файл отправлен!"),
+            show_alert=False
+        )
+        
+    except Exception as e:
+        logging.error(f"Failed to export payments CSV: {e}", exc_info=True)
+        await callback.answer(f"❌ Ошибка экспорта: {str(e)}", show_alert=True)
+
+
+@router.callback_query(F.data == "noop")
+async def noop_handler(callback: types.CallbackQuery):
+    """Handle no-op callback (for pagination display)."""
+    await callback.answer()

bot/handlers/admin/promo/manage.py

@@ -332,7 +332,7 @@ async def promo_export_all_handler(callback: types.CallbackQuery, i18n_data: dic
 
 
 @router.callback_query(F.data.startswith("promo_delete:"))
-async def promo_delete_handler(callback: types.CallbackQuery, i18n_data: dict, session: AsyncSession):
+async def promo_delete_handler(callback: types.CallbackQuery, i18n_data: dict, settings: Settings, session: AsyncSession):
     i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance")
     current_lang = i18n_data.get("current_language")
     if not i18n or not callback.message or not current_lang:
@@ -345,7 +345,7 @@ async def promo_delete_handler(callback: types.CallbackQuery, i18n_data: dict, s
         if promo:
             await session.commit()
             await callback.answer(_("admin_promo_deleted_success", code=promo.code), show_alert=True)
-            await promo_management_handler(callback, i18n_data, get_settings(), session, 0)
+            await promo_management_handler(callback, i18n_data, settings, session, 0)
         else:
             await callback.answer(_("admin_promo_not_found"), show_alert=True)
     except (ValueError, IndexError):
@@ -354,7 +354,7 @@ async def promo_delete_handler(callback: types.CallbackQuery, i18n_data: dict, s
 
 # --- Promo Edit Handlers ---
 @router.callback_query(F.data.startswith("promo_edit_select:"))
-async def promo_edit_select_handler(callback: types.CallbackQuery, i18n_data: dict):
+async def promo_edit_select_handler(callback: types.CallbackQuery, i18n_data: dict, session: AsyncSession):
     i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance")
     current_lang = i18n_data.get("current_language")
     if not i18n or not callback.message or not current_lang:
@@ -373,13 +373,13 @@ async def promo_edit_select_handler(callback: types.CallbackQuery, i18n_data: di
 
 
 @router.callback_query(F.data.startswith("promo_edit_field:"))
-async def promo_edit_field_handler(callback: types.CallbackQuery, state: FSMContext, i18n_data: dict):
+async def promo_edit_field_handler(callback: types.CallbackQuery, state: FSMContext, i18n_data: dict, session: AsyncSession):
     i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance")
     current_lang = i18n_data.get("current_language")
     if not i18n or not callback.message or not current_lang: return
     _ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs)
     
-    _, field, promo_id_str = callback.data.split(":")
+    action, field, promo_id_str = callback.data.split(":")
     await state.update_data(promo_id=int(promo_id_str), field_to_edit=field)
     
     prompts = {

bot/handlers/admin/sync_admin.py

@@ -7,7 +7,7 @@ from datetime import datetime, timezone
 
 from config.settings import Settings
 from bot.services.panel_api_service import PanelApiService
-from bot.services.notification_service import notify_admin_panel_sync
+from bot.services.notification_service import NotificationService
 
 from db.dal import user_dal, subscription_dal, panel_sync_dal
 
@@ -31,6 +31,7 @@ async def perform_sync(panel_service: PanelApiService, session: AsyncSession,
     # Additional counters for detailed logging
     users_without_telegram_id = 0
     users_not_found_in_db = 0
+    users_created = 0
     users_uuid_updated = 0
     subscriptions_created = 0
     subscriptions_updated = 0
@@ -93,10 +94,33 @@ async def perform_sync(panel_service: PanelApiService, session: AsyncSession,
                 if not existing_user:
                     users_not_found_in_db += 1
                     if telegram_id_from_panel:
-                        logging.debug(f"Panel user with telegramId {telegram_id_from_panel} and UUID {panel_uuid} not found in local DB")
+                        # Create new user if they have telegram_id
+                        try:
+                            user_data = {
+                                "user_id": telegram_id_from_panel,
+                                "username": None,  # Username will be updated when user interacts with bot
+                                "first_name": None,  # Panel doesn't provide this info
+                                "last_name": None,   # Panel doesn't provide this info
+                                "language_code": "ru",  # Default language
+                                "panel_user_uuid": panel_uuid,
+                                "is_banned": False,
+                                "referred_by_id": None
+                            }
+                            
+                            new_user, was_created = await user_dal.create_user(session, user_data)
+                            if was_created:
+                                users_created += 1
+                                logging.info(f"Created new user {telegram_id_from_panel} from panel sync with UUID {panel_uuid}")
+                            
+                            existing_user = new_user
+                            
+                        except Exception as e_create:
+                            sync_errors.append(f"Error creating user {telegram_id_from_panel}: {str(e_create)}")
+                            logging.error(f"Error creating user {telegram_id_from_panel}: {e_create}")
+                            continue
                     else:
-                        logging.debug(f"Panel user with UUID {panel_uuid} (no telegramId) not found in local DB")
-                    continue
+                        logging.debug(f"Panel user with UUID {panel_uuid} (no telegramId) not found in local DB - skipping")
+                        continue
 
                 # User found in local DB
                 users_found_in_db += 1
@@ -122,48 +146,88 @@ async def perform_sync(panel_service: PanelApiService, session: AsyncSession,
                             panel_expire_at_iso.replace("Z", "+00:00")
                         )
                         
-                        # Update or create subscription
-                        active_sub = await subscription_dal.get_active_subscription_by_user_id(
-                            session, actual_user_id, panel_uuid
+                        # Prefer syncing by concrete subscription UUID (shortUuid/subscriptionUuid)
+                        subscription_uuid_from_panel = (
+                            panel_user_dict.get("subscriptionUuid")
+                            or panel_user_dict.get("shortUuid")
                         )
-                        
-                        if active_sub:
-                            # Check if subscription needs update
-                            if (active_sub.end_date != panel_expire_at or 
-                                active_sub.status_from_panel != panel_status or
-                                active_sub.is_active != (panel_status == "ACTIVE")):
-                                
-                                await subscription_dal.update_subscription_end_date(
-                                    session, active_sub.subscription_id, panel_expire_at
+
+                        if subscription_uuid_from_panel:
+                            # Try to find subscription by its panel_subscription_uuid first (idempotent)
+                            existing_sub_by_uuid = (
+                                await subscription_dal.get_subscription_by_panel_subscription_uuid(
+                                    session, subscription_uuid_from_panel
+                                )
+                            )
+
+                            if existing_sub_by_uuid:
+                                # Atomic update of all relevant fields
+                                await subscription_dal.update_subscription(
+                                    session,
+                                    existing_sub_by_uuid.subscription_id,
+                                    {
+                                        "user_id": actual_user_id,
+                                        "panel_user_uuid": panel_uuid,
+                                        "end_date": panel_expire_at,
+                                        "is_active": panel_status == "ACTIVE",
+                                        "status_from_panel": panel_status,
+                                    },
                                 )
-                                # Update status fields
-                                active_sub.status_from_panel = panel_status
-                                active_sub.is_active = (panel_status == "ACTIVE")
                                 subscriptions_synced_count += 1
                                 subscriptions_updated += 1
                                 user_was_updated = True
-                                logging.info(f"Updated subscription for user {actual_user_id}: expires {panel_expire_at}, status {panel_status}")
+                                logging.info(
+                                    f"Synced existing subscription {existing_sub_by_uuid.subscription_id} for user {actual_user_id}: expires {panel_expire_at}, status {panel_status}"
+                                )
+                            else:
+                                # Create a new subscription only when we have a concrete subscription UUID
+                                sub_payload = {
+                                    "user_id": actual_user_id,
+                                    "panel_user_uuid": panel_uuid,
+                                    "panel_subscription_uuid": subscription_uuid_from_panel,
+                                    # Do not guess precise start_date from panel; keep nullable
+                                    "start_date": None,
+                                    "end_date": panel_expire_at,
+                                    "duration_months": None,
+                                    "is_active": panel_status == "ACTIVE",
+                                    "status_from_panel": panel_status,
+                                    "traffic_limit_bytes": settings.user_traffic_limit_bytes,
+                                }
+                                created_sub = await subscription_dal.upsert_subscription(
+                                    session, sub_payload
+                                )
+                                subscriptions_synced_count += 1
+                                subscriptions_created += 1
+                                user_was_updated = True
+                                logging.info(
+                                    f"Created subscription {created_sub.subscription_id} for user {actual_user_id} by panel_sub_uuid {subscription_uuid_from_panel}"
+                                )
                         else:
-                            # Create new subscription record
-                            subscription_uuid_to_use = panel_subscription_uuid or panel_uuid
-                            
-                            logging.info(f"Creating new subscription for user {actual_user_id} with UUID {subscription_uuid_to_use}")
-                            
-                            sub_payload = {
-                                "user_id": actual_user_id,
-                                "panel_user_uuid": panel_uuid,
-                                "panel_subscription_uuid": subscription_uuid_to_use,
-                                "start_date": datetime.now(timezone.utc),
-                                "end_date": panel_expire_at,
-                                "duration_months": 1,  # Default
-                                "is_active": panel_status == "ACTIVE",
-                                "status_from_panel": panel_status,
-                                "traffic_limit_bytes": settings.user_traffic_limit_bytes,
-                            }
-                            await subscription_dal.upsert_subscription(session, sub_payload)
-                            subscriptions_synced_count += 1
-                            subscriptions_created += 1
-                            user_was_updated = True
+                            # No subscription UUID from panel: only update an already active subscription for this user/panel UUID
+                            active_sub = await subscription_dal.get_active_subscription_by_user_id(
+                                session, actual_user_id, panel_uuid
+                            )
+                            if active_sub:
+                                await subscription_dal.update_subscription(
+                                    session,
+                                    active_sub.subscription_id,
+                                    {
+                                        "end_date": panel_expire_at,
+                                        "is_active": panel_status == "ACTIVE",
+                                        "status_from_panel": panel_status,
+                                    },
+                                )
+                                subscriptions_synced_count += 1
+                                subscriptions_updated += 1
+                                user_was_updated = True
+                                logging.info(
+                                    f"Updated active subscription {active_sub.subscription_id} for user {actual_user_id}: expires {panel_expire_at}, status {panel_status}"
+                                )
+                            else:
+                                # Without a concrete subscription UUID we avoid creating new records to keep sync idempotent
+                                logging.debug(
+                                    f"No subscriptionUuid for panel user {panel_uuid}; skipped creation for user {actual_user_id}"
+                                )
                             
                     except Exception as e:
                         sync_errors.append(f"Error syncing subscription for user {actual_user_id}: {str(e)}")
@@ -178,20 +242,27 @@ async def perform_sync(panel_service: PanelApiService, session: AsyncSession,
 
         # Update sync status
         status = "completed_with_errors" if sync_errors else "completed"
-        details = (f"📊 Статистика синхронизации:\n"
-                  f"🔍 Проверено записей панели: {panel_records_checked}\n"
-                  f"👥 Найдено пользователей в БД: {users_found_in_db}\n"
-                  f"🔄 Пользователей обновлено: {users_updated}\n"
-                  f"📋 Подписок синхронизировано: {subscriptions_synced_count}\n"
-                  f"   ├── Создано новых: {subscriptions_created}\n"
-                  f"   └── Обновлено существующих: {subscriptions_updated}")
-        
+        # Build additional stats
+        default_lang = settings.DEFAULT_LANGUAGE
+        additional_stats = ""
         if users_without_telegram_id > 0:
-            details += f"\n⚠️ Записей без telegramId: {users_without_telegram_id}"
+            additional_stats += i18n_instance.gettext(default_lang, "admin_sync_no_telegram_id", count=users_without_telegram_id)
         if users_not_found_in_db > 0:
-            details += f"\n❌ Не найдено в БД: {users_not_found_in_db}"
+            additional_stats += i18n_instance.gettext(default_lang, "admin_sync_not_found_in_db", count=users_not_found_in_db)
         if sync_errors:
-            details += f"\n🚫 Ошибок: {len(sync_errors)}"
+            additional_stats += i18n_instance.gettext(default_lang, "admin_sync_errors", count=len(sync_errors))
+
+        # Build full details using localization
+        details = i18n_instance.gettext(default_lang, "admin_sync_details", 
+            panel_records_checked=panel_records_checked,
+            users_found_in_db=users_found_in_db,
+            users_created=users_created,
+            users_updated=users_updated,
+            subscriptions_synced_count=subscriptions_synced_count,
+            subscriptions_created=subscriptions_created,
+            subscriptions_updated=subscriptions_updated,
+            additional_stats=additional_stats
+        )
 
         await panel_sync_dal.update_panel_sync_status(
             session, status, details, panel_records_checked, subscriptions_synced_count
@@ -204,6 +275,7 @@ async def perform_sync(panel_service: PanelApiService, session: AsyncSession,
         logging.info(f"  Users without telegramId: {users_without_telegram_id}")
         logging.info(f"  Users not found in local DB: {users_not_found_in_db}")
         logging.info(f"  Users found in local DB: {users_found_in_db}")
+        logging.info(f"  Users created: {users_created}")
         logging.info(f"  Users with UUID updated: {users_uuid_updated}")
         logging.info(f"  Users updated overall: {users_updated}")
         logging.info(f"  Subscriptions total synced: {subscriptions_synced_count}")
@@ -216,6 +288,7 @@ async def perform_sync(panel_service: PanelApiService, session: AsyncSession,
             "details": details,
             "users_processed": panel_records_checked,
             "users_synced": users_found_in_db,
+            "users_created": users_created,
             "subs_synced": subscriptions_synced_count,
             "errors": sync_errors
         }
@@ -265,7 +338,7 @@ async def sync_command_handler(
         return
 
     if isinstance(message_event, types.Message):
-        await message_event.answer(_("sync_started"))
+        await message_event.answer(_("sync_started_simple"))
 
     logging.info(f"Admin ({message_event.from_user.id}) triggered panel sync.")
 
@@ -277,32 +350,19 @@ async def sync_command_handler(
         details = sync_result.get("details", "No details available")
         errors = sync_result.get("errors", [])
         
+        # Simple confirmation message to admin
         if status == "failed":
-            await bot.send_message(target_chat_id, _("sync_failed", details=details))
+            await bot.send_message(target_chat_id, _("sync_failed_simple"))
         elif status == "completed_with_errors":
-            error_preview = "; ".join(errors[:3])  # Show first 3 errors
-            final_message = _(
-                "sync_completed_with_errors_details",
-                total_checked=sync_result.get("users_processed", 0),
-                users_synced=sync_result.get("users_synced", 0),
-                subs_synced=sync_result.get("subs_synced", 0),
-                errors_count=len(errors),
-                error_details_preview=error_preview
-            )
-            await bot.send_message(target_chat_id, final_message)
+            await bot.send_message(target_chat_id, _("sync_errors_simple", errors_count=len(errors)))
         else:
-            final_message = _(
-                "sync_completed_details",
-                total_checked=sync_result.get("users_processed", 0),
-                users_synced=sync_result.get("users_synced", 0),
-                subs_synced=sync_result.get("subs_synced", 0)
-            )
-            await bot.send_message(target_chat_id, _("sync_completed", status="Success", details=final_message))
+            await bot.send_message(target_chat_id, _("sync_success_simple"))
         
         # Send notification to log channel with proper thread handling
         try:
-            await notify_admin_panel_sync(
-                bot, settings, i18n, status, details,
+            notification_service = NotificationService(bot, settings, i18n)
+            await notification_service.notify_panel_sync(
+                status, details,
                 sync_result.get("users_processed", 0),
                 sync_result.get("subs_synced", 0)
             )
@@ -311,12 +371,13 @@ async def sync_command_handler(
             
     except Exception as e_sync_global:
         logging.error(f"Global error during /sync command: {e_sync_global}", exc_info=True)
-        await bot.send_message(target_chat_id, _("sync_failed", details=str(e_sync_global)))
+        await bot.send_message(target_chat_id, _("sync_critical_error"))
         
         # Send notification to log channel about failure
         try:
-            await notify_admin_panel_sync(
-                bot, settings, i18n, "failed", str(e_sync_global), 0, 0
+            notification_service = NotificationService(bot, settings, i18n)
+            await notification_service.notify_panel_sync(
+                "failed", str(e_sync_global), 0, 0
             )
         except Exception as e_notification:
             logging.error(f"Failed to send sync failure notification: {e_notification}")

bot/handlers/admin/user_management.py

@@ -1,6 +1,7 @@
 import logging
 import re
 from aiogram import Router, F, types, Bot
+from aiogram.exceptions import TelegramBadRequest
 from aiogram.fsm.context import FSMContext
 from aiogram.utils.markdown import hcode, hbold
 from typing import Optional, Dict, Any
@@ -15,6 +16,7 @@ from bot.keyboards.inline.admin_keyboards import get_back_to_admin_panel_keyboar
 from bot.services.subscription_service import SubscriptionService
 from bot.services.panel_api_service import PanelApiService
 from bot.middlewares.i18n import JsonI18n
+from bot.utils import get_message_content, send_direct_message
 from aiogram.utils.keyboard import InlineKeyboardBuilder, InlineKeyboardButton
 
 router = Router(name="admin_user_management_router")
@@ -578,7 +580,7 @@ async def process_subscription_days_handler(message: types.Message, state: FSMCo
     await state.clear()
 
 
-@router.message(AdminStates.waiting_for_direct_message_to_user, F.text)
+@router.message(AdminStates.waiting_for_direct_message_to_user)
 async def process_direct_message_handler(message: types.Message, state: FSMContext,
                                        settings: Settings, i18n_data: dict,
                                        bot: Bot, session: AsyncSession):
@@ -597,8 +599,9 @@ async def process_direct_message_handler(message: types.Message, state: FSMConte
         await state.clear()
         return
 
-    message_text = message.text.strip()
-    if len(message_text) > 4000:
+    # Determine content similar to broadcast
+    text = (message.text or message.caption or "").strip()
+    if len(text) > 4000:
         await message.answer(_(
             "admin_user_message_too_long",
             default="❌ Сообщение слишком длинное (максимум 4000 символов)"
@@ -613,15 +616,40 @@ async def process_direct_message_handler(message: types.Message, state: FSMConte
             await state.clear()
             return
 
-        # Prepare message with admin signature
+        # Prepare admin signature and get content
         admin_signature = _(
             "admin_direct_message_signature",
             default="\n\n---\n💬 Сообщение от администратора"
         )
-        full_message = message_text + admin_signature
+        
+        content = get_message_content(message)
 
-        # Send message to user
-        await bot.send_message(target_user_id, full_message)
+        if not content.text and not content.file_id:
+            await message.answer(_(
+                "admin_direct_empty_message",
+                default="❌ Пустое сообщение. Отправьте текст или медиа."
+            ))
+            return
+
+        caption_with_signature = (content.text + admin_signature) if content.text else None
+
+        # Send to target user using our fancy match/case function
+        try:
+            await send_direct_message(
+                bot,
+                target_user_id, 
+                content,
+                extra_text=admin_signature,
+                parse_mode="HTML",
+                disable_web_page_preview=True,
+            )
+        except TelegramBadRequest as e:
+            await message.answer(_(
+                "admin_broadcast_invalid_html",
+                default="❌ Некорректный HTML в сообщении. Пожалуйста, отправьте корректный HTML (поддерживаются теги Telegram) или уберите теги.\nОшибка: {error}",
+                error=str(e),
+            ))
+            return
         
         # Confirm to admin
         await message.answer(_(

bot/handlers/user/payment.py

@@ -123,7 +123,12 @@ async def process_successful_payment(session: AsyncSession, bot: Bot,
             "applied_promo_bonus_days", 0)
 
         referral_bonus_info = await referral_service.apply_referral_bonuses_for_payment(
-            session, user_id, subscription_months)
+            session,
+            user_id,
+            subscription_months,
+            current_payment_db_id=payment_db_id,
+            skip_if_active_before_payment=False,
+        )
         applied_referee_bonus_days_from_referral: Optional[int] = None
         if referral_bonus_info and referral_bonus_info.get(
                 "referee_new_end_date"):

bot/handlers/user/start.py

@@ -165,24 +165,25 @@ async def start_command_handler(message: types.Message,
             "registration_date": datetime.now(timezone.utc)
         }
         try:
-            db_user = await user_dal.create_user(session, user_data_to_create)
+            db_user, created = await user_dal.create_user(session, user_data_to_create)
 
-            logging.info(
-                f"New user {user_id} added to session. Referred by: {referred_by_user_id or 'N/A'}."
-            )
-            
-            # Send notification about new user registration
-            try:
-                from bot.services.notification_service import NotificationService
-                notification_service = NotificationService(message.bot, settings, i18n)
-                await notification_service.notify_new_user_registration(
-                    user_id=user_id,
-                    username=user.username,
-                    first_name=user.first_name,
-                    referred_by_id=referred_by_user_id
+            if created:
+                logging.info(
+                    f"New user {user_id} added to session. Referred by: {referred_by_user_id or 'N/A'}."
                 )
-            except Exception as e:
-                logging.error(f"Failed to send new user notification: {e}")
+
+                # Send notification about new user registration
+                try:
+                    from bot.services.notification_service import NotificationService
+                    notification_service = NotificationService(message.bot, settings, i18n)
+                    await notification_service.notify_new_user_registration(
+                        user_id=user_id,
+                        username=user.username,
+                        first_name=user.first_name,
+                        referred_by_id=referred_by_user_id
+                    )
+                except Exception as e:
+                    logging.error(f"Failed to send new user notification: {e}")
         except Exception as e_create:
 
             logging.error(
@@ -194,8 +195,15 @@ async def start_command_handler(message: types.Message,
         update_payload = {}
         if db_user.language_code != current_lang:
             update_payload["language_code"] = current_lang
+        # Set referral only if not already set AND user is not currently active.
+        # This allows previously subscribed but currently inactive users to be attributed.
         if referred_by_user_id and db_user.referred_by_id is None:
-            update_payload["referred_by_id"] = referred_by_user_id
+            try:
+                is_active_now = await subscription_service.has_active_subscription(session, user_id)
+            except Exception:
+                is_active_now = False
+            if not is_active_now:
+                update_payload["referred_by_id"] = referred_by_user_id
         if user.username != db_user.username:
             update_payload["username"] = user.username
         if user.first_name != db_user.first_name:

bot/handlers/user/trial_handler.py

@@ -7,7 +7,7 @@ from datetime import datetime
 from config.settings import Settings
 from bot.services.subscription_service import SubscriptionService
 from bot.services.panel_api_service import PanelApiService
-from bot.services.notification_service import notify_admin_new_trial
+from bot.services.notification_service import NotificationService
 from bot.keyboards.inline.user_keyboards import (
     get_trial_confirmation_keyboard,
     get_main_menu_inline_keyboard,
@@ -97,13 +97,8 @@ async def request_trial_confirmation_handler(
         )
         
         # Send notification to admin about new trial
-        await notify_admin_new_trial(
-            callback.bot,
-            settings,
-            i18n,
-            user_id,
-            end_date_obj,
-        )
+        notification_service = NotificationService(callback.bot, settings, i18n)
+        await notification_service.notify_trial_activation(user_id, end_date_obj)
     else:
         message_key_from_service = (
             activation_result.get("message_key", "trial_activation_failed")
@@ -264,13 +259,8 @@ async def confirm_activate_trial_handler(
             )
 
     if activation_result and activation_result.get("activated") and end_date_obj:
-        await notify_admin_new_trial(
-            callback.bot,
-            settings,
-            i18n,
-            user_id,
-            end_date_obj,
-        )
+        notification_service = NotificationService(callback.bot, settings, i18n)
+        await notification_service.notify_trial_activation(user_id, end_date_obj)
 
 
 @router.callback_query(F.data == "main_action:cancel_trial")

bot/keyboards/inline/admin_keyboards.py

@@ -39,12 +39,14 @@ def get_stats_monitoring_keyboard(i18n_instance, lang: str) -> InlineKeyboardMar
     
     builder.button(text=_(key="admin_stats_button"),
                    callback_data="admin_action:stats")
+    builder.button(text=_(key="admin_view_payments_button", default="💰 Платежи"),
+                   callback_data="admin_action:view_payments")
     builder.button(text=_(key="admin_view_logs_menu_button"),
                    callback_data="admin_action:view_logs_menu")
     
     builder.button(text=_(key="back_to_admin_panel_button"),
                    callback_data="admin_action:main")
-    builder.adjust(2, 1)
+    builder.adjust(2, 1, 1)
     return builder.as_markup()
 
 
@@ -258,12 +260,47 @@ def get_confirmation_keyboard(yes_callback_data: str, no_callback_data: str,
 
 
 def get_broadcast_confirmation_keyboard(lang: str,
-                                        i18n_instance) -> InlineKeyboardMarkup:
+                                        i18n_instance,
+                                        target: str = "all") -> InlineKeyboardMarkup:
     _ = lambda key, **kwargs: i18n_instance.gettext(lang, key, **kwargs)
     builder = InlineKeyboardBuilder()
-    builder.button(text=_(key="confirm_broadcast_send_button"),
+
+    # Row: target selection (all / active / inactive)
+    target_all_label = _(
+        key="broadcast_target_all_button",
+        default="👥 Все"
+    )
+    target_active_label = _(
+        key="broadcast_target_active_button",
+        default="✅ Активные"
+    )
+    target_inactive_label = _(
+        key="broadcast_target_inactive_button",
+        default="⌛ Неактивные"
+    )
+
+    # Highlight current selection with a prefix
+    def mark_selected(label: str, is_selected: bool) -> str:
+        return ("• " + label) if is_selected else label
+
+    builder.button(
+        text=mark_selected(target_all_label, target == "all"),
+        callback_data="broadcast_target:all",
+    )
+    builder.button(
+        text=mark_selected(target_active_label, target == "active"),
+        callback_data="broadcast_target:active",
+    )
+    builder.button(
+        text=mark_selected(target_inactive_label, target == "inactive"),
+        callback_data="broadcast_target:inactive",
+    )
+    builder.adjust(3)
+
+    # Row: confirmation
+    builder.button(text=_(key="confirm_broadcast_send_button", default="🚀 Отправить"),
                    callback_data="broadcast_final_action:send")
-    builder.button(text=_(key="cancel_broadcast_button"),
+    builder.button(text=_(key="cancel_broadcast_button", default="❌ Отмена"),
                    callback_data="broadcast_final_action:cancel")
     builder.adjust(2)
     return builder.as_markup()

bot/main_bot.py

@@ -1,17 +1,10 @@
 import logging
 import asyncio
-from typing import Callable, Dict, Any, Awaitable, Optional
+from typing import Dict, Any, Optional
 
-from aiogram import Bot, Dispatcher, BaseMiddleware, Router, F
-from aiogram.types import (
-    Update,
-    MenuButtonDefault,
-    MenuButtonWebApp,
-    WebAppInfo,
-    BotCommand,
-)
+from aiogram import Bot, Dispatcher
+from aiogram.types import (MenuButtonDefault, MenuButtonWebApp, WebAppInfo, BotCommand)
 from aiogram.enums import ParseMode
-from aiogram.filters import CommandStart, Command
 from aiogram.client.default import DefaultBotProperties
 from aiogram.webhook.aiohttp_server import SimpleRequestHandler, setup_application
 from aiogram.fsm.storage.memory import MemoryStorage
@@ -24,13 +17,15 @@ from config.settings import Settings
 from db.database_setup import init_db_connection
 
 from bot.middlewares.i18n import I18nMiddleware, get_i18n_instance, JsonI18n
+from bot.middlewares.db_session import DBSessionMiddleware
 from bot.middlewares.ban_check_middleware import BanCheckMiddleware
 from bot.middlewares.action_logger_middleware import ActionLoggerMiddleware
+from bot.middlewares.profile_sync import ProfileSyncMiddleware
+from bot.app.controllers.dispatcher_controller import build_dispatcher
+from bot.app.factories.build_services import build_core_services
+from bot.app.web.web_server import build_and_start_web_app
 
-from bot.handlers.user import user_router_aggregate
-from bot.handlers.admin import admin_router_aggregate
-from bot.handlers import inline_mode
-from bot.filters.admin_filter import AdminFilter
+from bot.routers import build_root_router
 
 from bot.services.yookassa_service import YooKassaService
 from bot.services.panel_api_service import PanelApiService
@@ -46,54 +41,8 @@ from bot.handlers.admin.sync_admin import perform_sync
 from bot.utils.message_queue import init_queue_manager
 
 
-class DBSessionMiddleware(BaseMiddleware):
-
-    def __init__(self, async_session_factory: sessionmaker):
-        super().__init__()
-        self.async_session_factory = async_session_factory
-
-    async def __call__(
-        self,
-        handler: Callable[[Update, Dict[str, Any]], Awaitable[Any]],
-        event: Update,
-        data: Dict[str, Any],
-    ) -> Any:
-        if self.async_session_factory is None:
-            logging.critical("DBSessionMiddleware: async_session_factory is None!")
-            raise RuntimeError(
-                "async_session_factory not provided to DBSessionMiddleware"
-            )
-
-        async with self.async_session_factory() as session:
-            data["session"] = session
-            try:
-                result = await handler(event, data)
-
-                await session.commit()
-                return result
-            except Exception:
-                await session.rollback()
-                logging.error(
-                    "DBSessionMiddleware: Exception caused rollback.", exc_info=True
-                )
-                raise
-
-
 async def register_all_routers(dp: Dispatcher, settings: Settings):
-    dp.include_router(user_router_aggregate)
-    
-    # Add inline mode router (available for all users)
-    dp.include_router(inline_mode.router)
-
-    admin_main_router = Router(name="admin_main_filtered_router")
-    admin_filter_instance = AdminFilter(admin_ids=settings.ADMIN_IDS)
-
-    admin_main_router.message.filter(admin_filter_instance)
-    admin_main_router.callback_query.filter(admin_filter_instance)
-
-    admin_main_router.include_router(admin_router_aggregate)
-
-    dp.include_router(admin_main_router)
+    dp.include_router(build_root_router(settings))
     logging.info("All application routers registered.")
 
 
@@ -158,10 +107,10 @@ async def on_startup_configured(dispatcher: Dispatcher):
                 "STARTUP: Skipped setting Telegram webhook due to security or configuration error."
             )
     else:
-        logging.info(
-            "STARTUP: WEBHOOK_BASE_URL not set in environment. Running in polling mode and clearing any existing webhook."
+        logging.error(
+            "STARTUP: WEBHOOK_BASE_URL not set in environment. Webhook mode is required. Exiting."
         )
-        await bot.delete_webhook(drop_pending_updates=True)
+        raise SystemExit("WEBHOOK_BASE_URL is required. Polling mode is disabled.")
 
     if settings.SUBSCRIPTION_MINI_APP_URL:
         try:
@@ -279,19 +228,16 @@ async def on_shutdown_configured(dispatcher: Dispatcher):
 
 
 async def run_bot(settings_param: Settings):
-    storage = MemoryStorage()
-    default_props = DefaultBotProperties(parse_mode=ParseMode.HTML)
-    bot = Bot(token=settings_param.BOT_TOKEN, default=default_props)
-
     local_async_session_factory = init_db_connection(settings_param)
     if local_async_session_factory is None:
         logging.critical(
             "Failed to initialize database connection and session factory. Exiting."
         )
         return
+    dp, bot, extra = build_dispatcher(settings_param, local_async_session_factory)
+    i18n_instance = extra["i18n_instance"]
 
-    dp = Dispatcher(storage=storage, settings=settings_param, bot_instance=bot)
-
+    # Get bot username for YooKassa default return URL if needed
     actual_bot_username = "your_bot_username"
     try:
         bot_info = await bot.get_me()
@@ -302,211 +248,53 @@ async def run_bot(settings_param: Settings):
             f"Failed to get bot info (e.g., for YooKassa default URL): {e}. Using fallback: {actual_bot_username}"
         )
 
-    i18n_instance = get_i18n_instance(
-        path="locales", default=settings_param.DEFAULT_LANGUAGE
-    )
-
-    yookassa_service = YooKassaService(
-        shop_id=settings_param.YOOKASSA_SHOP_ID,
-        secret_key=settings_param.YOOKASSA_SECRET_KEY,
-        configured_return_url=settings_param.YOOKASSA_RETURN_URL,
-        bot_username_for_default_return=actual_bot_username,
-        settings_obj=settings_param,
-    )
-    panel_service = PanelApiService(settings_param)
-
-    subscription_service = SubscriptionService(
-        settings_param, panel_service, bot, i18n_instance
-    )
-    referral_service = ReferralService(
-        settings_param, subscription_service, bot, i18n_instance
-    )
-    promo_code_service = PromoCodeService(
-        settings_param, subscription_service, bot, i18n_instance
-    )
-    stars_service = StarsService(
-        bot, settings_param, i18n_instance, subscription_service, referral_service
-    )
-    cryptopay_service = CryptoPayService(
-        settings_param.CRYPTOPAY_TOKEN,
-        settings_param.CRYPTOPAY_NETWORK,
-        bot,
+    services = build_core_services(
         settings_param,
-        i18n_instance,
-        local_async_session_factory,
-        subscription_service,
-        referral_service,
-    )
-    tribute_service = TributeService(
         bot,
-        settings_param,
-        i18n_instance,
         local_async_session_factory,
-        panel_service,
-        subscription_service,
-        referral_service,
-    )
-    panel_webhook_service = PanelWebhookService(
-        bot,
-        settings_param,
         i18n_instance,
-        local_async_session_factory,
+        actual_bot_username,
     )
-
-    dp["i18n_instance"] = i18n_instance
-    dp["yookassa_service"] = yookassa_service
-    dp["panel_service"] = panel_service
-    dp["subscription_service"] = subscription_service
-    dp["referral_service"] = referral_service
-    dp["promo_code_service"] = promo_code_service
-    dp["stars_service"] = stars_service
-    dp["cryptopay_service"] = cryptopay_service
-    dp["tribute_service"] = tribute_service
-    dp["panel_webhook_service"] = panel_webhook_service
+    for key, service in services.items():
+        dp[key] = service
+    dp["panel_service"] = services["panel_service"]
     dp["async_session_factory"] = local_async_session_factory
 
-    dp.update.outer_middleware(DBSessionMiddleware(local_async_session_factory))
-    dp.update.outer_middleware(
-        I18nMiddleware(i18n=i18n_instance, settings=settings_param)
-    )
-    dp.update.outer_middleware(
-        BanCheckMiddleware(settings=settings_param, i18n_instance=i18n_instance)
-    )
-    dp.update.outer_middleware(ActionLoggerMiddleware(settings=settings_param))
-
-    dp.startup.register(on_startup_configured)
-    # Register shutdown callback directly so Dispatcher instance is provided
-    dp.shutdown.register(on_shutdown_configured)
+    # Wrap startup/shutdown handlers to satisfy aiogram event signature (no args passed)
+    async def _on_startup_wrapper():
+        await on_startup_configured(dp)
+    async def _on_shutdown_wrapper():
+        await on_shutdown_configured(dp)
+    dp.startup.register(_on_startup_wrapper)
+    dp.shutdown.register(_on_shutdown_wrapper)
 
     await register_all_routers(dp, settings_param)
 
     tg_webhook_base = settings_param.WEBHOOK_BASE_URL
-    yk_webhook_base = settings_param.WEBHOOK_BASE_URL
 
-    should_run_aiohttp_server = bool(tg_webhook_base) or (
-        bool(yk_webhook_base) and bool(settings_param.yookassa_webhook_path)
-    )
-
-    telegram_uses_webhook_mode = bool(tg_webhook_base)
-    run_telegram_polling = not telegram_uses_webhook_mode
+    # Webhook mode is now required - exit if not configured
+    if not tg_webhook_base:
+        logging.error("WEBHOOK_BASE_URL is required. Polling mode is disabled. Exiting.")
+        await dp.emit_shutdown()
+        raise SystemExit("WEBHOOK_BASE_URL is required. Polling mode is disabled.")
 
     logging.info(f"--- Bot Run Mode Decision ---")
-    logging.info(
-        f"Configured WEBHOOK_BASE_URL: '{tg_webhook_base}' -> Telegram Webhook Mode: {telegram_uses_webhook_mode}"
-    )
-    logging.info(
-        f"YooKassa webhook path: '{settings_param.yookassa_webhook_path}'"
-    )
-    logging.info(f"Decision: Run AIOHTTP server: {should_run_aiohttp_server}")
-    logging.info(f"Decision: Run Telegram Polling: {run_telegram_polling}")
+    logging.info(f"Configured WEBHOOK_BASE_URL: '{tg_webhook_base}' -> Webhook Mode: ENABLED")
+    logging.info(f"YooKassa webhook path: '{settings_param.yookassa_webhook_path}'")
+    logging.info(f"Decision: Run AIOHTTP server: ENABLED (required for webhooks)")
     logging.info(f"--- End Bot Run Mode Decision ---")
 
     web_app_runner = None
     main_tasks = []
 
-    if should_run_aiohttp_server:
-        app = web.Application()
-        app["bot"] = bot
-        app["dp"] = dp
-        app["settings"] = settings_param
-        app["i18n"] = i18n_instance
-        app["async_session_factory"] = local_async_session_factory
+    # Only run AIOHTTP server for webhook mode
+    async def web_server_task():
+        await build_and_start_web_app(dp, bot, settings_param, local_async_session_factory)
 
-        app["yookassa_service"] = yookassa_service
-        app["subscription_service"] = subscription_service
-        app["referral_service"] = referral_service
-        app["panel_service"] = panel_service
-        app["stars_service"] = stars_service
-        app["cryptopay_service"] = cryptopay_service
-        app["tribute_service"] = tribute_service
-        app["panel_webhook_service"] = panel_webhook_service
+    main_tasks.append(asyncio.create_task(web_server_task(), name="AIOHTTPServerTask"))
 
-        setup_application(app, dp, bot=bot)
-
-        if telegram_uses_webhook_mode:
-            telegram_webhook_path = f"/{settings_param.BOT_TOKEN}"
-            if not telegram_webhook_path.startswith("/"):
-                telegram_webhook_path = "/" + telegram_webhook_path
-            app.router.add_post(
-                telegram_webhook_path, SimpleRequestHandler(dispatcher=dp, bot=bot)
-            )
-            logging.info(
-                f"Telegram webhook route configured at: [POST] {telegram_webhook_path} (relative to base URL)"
-            )
-
-        if yk_webhook_base and settings_param.yookassa_webhook_path:
-            yk_path = settings_param.yookassa_webhook_path
-            if not yk_path or not isinstance(yk_path, str):
-                logging.error(
-                    f"YooKassa webhook path is invalid or not configured in settings: {yk_path}. Skipping YooKassa webhook setup."
-                )
-            elif not yk_path.startswith("/"):
-                logging.error(
-                    f"CRITICAL: YooKassa webhook path '{yk_path}' from settings does not start with '/'. Correct settings.py or .env. Skipping YooKassa webhook."
-                )
-            else:
-                app.router.add_post(
-                    yk_path, user_payment_webhook_module.yookassa_webhook_route
-                )
-                logging.info(f"YooKassa webhook route configured at: [POST] {yk_path}")
-
-        tribute_path = settings_param.tribute_webhook_path
-        if tribute_path.startswith("/"):
-            app.router.add_post(tribute_path, tribute_webhook_route)
-            logging.info(f"Tribute webhook route configured at: [POST] {tribute_path}")
-
-        cp_path = settings_param.cryptopay_webhook_path
-        if cp_path.startswith("/"):
-            app.router.add_post(cp_path, cryptopay_webhook_route)
-            logging.info(f"CryptoPay webhook route configured at: [POST] {cp_path}")
-
-        panel_path = settings_param.panel_webhook_path
-        if panel_path.startswith("/"):
-            app.router.add_post(panel_path, panel_webhook_route)
-            logging.info(f"Panel webhook route configured at: [POST] {panel_path}")
-
-        web_app_runner = web.AppRunner(app)
-        await web_app_runner.setup()
-        site = web.TCPSite(
-            web_app_runner,
-            host=settings_param.WEB_SERVER_HOST,
-            port=settings_param.WEB_SERVER_PORT,
-        )
-
-        async def web_server_task():
-            await site.start()
-            logging.info(
-                f"AIOHTTP server started on http://{settings_param.WEB_SERVER_HOST}:{settings_param.WEB_SERVER_PORT}"
-            )
-            (
-                await asyncio.Event().wait()
-                if not run_telegram_polling
-                else await asyncio.sleep(31536000)
-            )
-
-        main_tasks.append(
-            asyncio.create_task(web_server_task(), name="AIOHTTPServerTask")
-        )
-
-    if run_telegram_polling:
-        logging.info("Starting bot in Telegram Polling mode...")
-        main_tasks.append(
-            asyncio.create_task(
-                dp.start_polling(bot, allowed_updates=dp.resolve_used_update_types()),
-                name="TelegramPollingTask",
-            )
-        )
-
-    if not main_tasks:
-        logging.error(
-            "Bot is not configured for any run mode (neither Webhook nor Polling). Exiting."
-        )
-        await dp.emit_shutdown()
-        return
-
-    logging.info(
-        f"Starting bot with main tasks: {[task.get_name() for task in main_tasks]}"
-    )
+    logging.info("Starting bot in Webhook mode with AIOHTTP server...")
+    logging.info(f"Starting bot with main tasks: {[task.get_name() for task in main_tasks]}")
 
     try:
         await asyncio.gather(*main_tasks)

bot/middlewares/db_session.py

@@ -0,0 +1,40 @@
+import logging
+from typing import Callable, Dict, Any, Awaitable
+
+from aiogram import BaseMiddleware
+from aiogram.types import Update
+from sqlalchemy.orm import sessionmaker
+
+
+class DBSessionMiddleware(BaseMiddleware):
+
+    def __init__(self, async_session_factory: sessionmaker):
+        super().__init__()
+        self.async_session_factory = async_session_factory
+
+    async def __call__(
+        self,
+        handler: Callable[[Update, Dict[str, Any]], Awaitable[Any]],
+        event: Update,
+        data: Dict[str, Any],
+    ) -> Any:
+        if self.async_session_factory is None:
+            logging.critical("DBSessionMiddleware: async_session_factory is None!")
+            raise RuntimeError(
+                "async_session_factory not provided to DBSessionMiddleware"
+            )
+
+        async with self.async_session_factory() as session:
+            data["session"] = session
+            try:
+                result = await handler(event, data)
+
+                await session.commit()
+                return result
+            except Exception:
+                await session.rollback()
+                logging.error(
+                    "DBSessionMiddleware: Exception caused rollback.", exc_info=True
+                )
+                raise
+

bot/middlewares/profile_sync.py

@@ -0,0 +1,47 @@
+import logging
+from typing import Callable, Dict, Any, Awaitable, Optional
+
+from aiogram import BaseMiddleware
+from aiogram.types import Update, User as TgUser
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from db.dal import user_dal
+
+
+class ProfileSyncMiddleware(BaseMiddleware):
+
+    async def __call__(
+        self,
+        handler: Callable[[Update, Dict[str, Any]], Awaitable[Any]],
+        event: Update,
+        data: Dict[str, Any],
+    ) -> Any:
+        session: AsyncSession = data.get("session")
+        tg_user: Optional[TgUser] = data.get("event_from_user")
+
+        if session and tg_user:
+            try:
+                db_user = await user_dal.get_user_by_id(session, tg_user.id)
+                if db_user:
+                    update_payload: Dict[str, Any] = {}
+                    if db_user.username != tg_user.username:
+                        update_payload["username"] = tg_user.username
+                    if db_user.first_name != tg_user.first_name:
+                        update_payload["first_name"] = tg_user.first_name
+                    if db_user.last_name != tg_user.last_name:
+                        update_payload["last_name"] = tg_user.last_name
+
+                    if update_payload:
+                        await user_dal.update_user(session, tg_user.id, update_payload)
+                        logging.info(
+                            f"ProfileSyncMiddleware: Updated user {tg_user.id} profile fields: {list(update_payload.keys())}"
+                        )
+            except Exception as e:
+                logging.error(
+                    f"ProfileSyncMiddleware: Failed to sync profile for user {getattr(tg_user, 'id', 'N/A')}: {e}",
+                    exc_info=True,
+                )
+
+        return await handler(event, data)
+
+

bot/routers.py

@@ -0,0 +1,30 @@
+from aiogram import Router, F
+
+from bot.handlers.user import user_router_aggregate
+from bot.handlers import inline_mode
+from bot.handlers.admin import admin_router_aggregate
+from bot.filters.admin_filter import AdminFilter
+from config.settings import Settings
+
+
+def build_root_router(settings: Settings) -> Router:
+    root = Router(name="root")
+
+    # Allow all updates only in private chats (messages, callback queries, etc.)
+    root.message.filter(F.chat.type == "private")
+    root.callback_query.filter(F.message.chat.type == "private")
+
+    # Public routers
+    root.include_router(user_router_aggregate)
+    root.include_router(inline_mode.router)
+
+    # Admin routers behind filter
+    admin_main_router = Router(name="admin_main_filtered_router")
+    admin_filter_instance = AdminFilter(admin_ids=settings.ADMIN_IDS)
+    admin_main_router.message.filter(admin_filter_instance)
+    admin_main_router.callback_query.filter(admin_filter_instance)
+    admin_main_router.include_router(admin_router_aggregate)
+    root.include_router(admin_main_router)
+
+    return root
+

bot/services/crypto_pay_service.py

@@ -142,7 +142,11 @@ class CryptoPayService:
                     provider="cryptopay",
                 )
                 referral_bonus = await referral_service.apply_referral_bonuses_for_payment(
-                    session, user_id, months
+                    session,
+                    user_id,
+                    months,
+                    current_payment_db_id=payment_db_id,
+                    skip_if_active_before_payment=False,
                 )
                 await session.commit()
             except Exception as e:

bot/services/notification_service.py

@@ -295,51 +295,4 @@ class NotificationService:
         if to_admins:
             await self._send_to_admins(message)
 
-
-# Legacy functions for backward compatibility
-async def notify_admins(bot: Bot, settings: Settings, i18n: JsonI18n,
-                        message_key: str, parse_mode: str | None = None,
-                        **kwargs) -> None:
-    if not settings.ADMIN_IDS:
-        return
-    admin_lang = settings.DEFAULT_LANGUAGE
-    msg = i18n.gettext(admin_lang, message_key, **kwargs)
-    for admin_id in settings.ADMIN_IDS:
-        try:
-            await bot.send_message(admin_id, msg, parse_mode=parse_mode)
-        except Exception as e:
-            logging.error(f"Failed to send admin notification to {admin_id}: {e}")
-
-
-async def notify_admin_new_trial(bot: Bot, settings: Settings, i18n: JsonI18n,
-                                 user_id: int, end_date: datetime) -> None:
-    """Send notification to admins about new trial activation (legacy)"""
-    notification_service = NotificationService(bot, settings, i18n)
-    await notification_service.notify_trial_activation(user_id, end_date)
-
-
-
-
-
-async def notify_admin_promo_activation(bot: Bot, settings: Settings,
-                                        i18n: JsonI18n, user_id: int,
-                                        code: str,
-                                        bonus_days: int) -> None:
-    await notify_admins(
-        bot,
-        settings,
-        i18n,
-        "admin_promo_activation_notification",
-        user_id=user_id,
-        code=code,
-        bonus_days=bonus_days,
-    )
-
-
-async def notify_admin_panel_sync(bot: Bot, settings: Settings,
-                                  i18n: JsonI18n, status: str,
-                                  details: str, users_processed: int,
-                                  subs_synced: int) -> None:
-    """Send notification to admins about panel sync (legacy)"""
-    notification_service = NotificationService(bot, settings, i18n)
-    await notification_service.notify_panel_sync(status, details, users_processed, subs_synced)
+# Removed legacy helper functions that duplicated NotificationService API

bot/services/panel_webhook_service.py

@@ -11,6 +11,7 @@ from config.settings import Settings
 from bot.middlewares.i18n import JsonI18n
 from bot.keyboards.inline.user_keyboards import get_subscribe_only_markup
 from db.dal import user_dal
+from bot.utils.date_utils import add_months
 
 EVENT_MAP = {
     "user.expires_in_72_hours": (3, "subscription_72h_notification"),
@@ -42,12 +43,16 @@ class PanelWebhookService:
             logging.error(f"Failed to send notification to {user_id}: {e}")
 
     async def _handle_expired_subscription(self, session, user_id: int, user_payload: dict, 
-                                         lang: str, markup, first_name: str):
-        """Handle expired subscription - auto-renew tribute users if no cancellation was received"""
+                                         lang: str, markup, first_name: str) -> bool:
+        """Handle expired subscription - auto-renew tribute users if no cancellation was received.
+
+        Returns True if an auto-renewal was performed (and renewal message sent), False otherwise.
+        """
         from db.dal import subscription_dal, payment_dal
-        from datetime import datetime, timezone, timedelta
+        from datetime import datetime, timezone
         
         try:
+            auto_renewed = False
             # Check if user has tribute subscriptions that weren't cancelled
             user_subs = await subscription_dal.get_active_subscriptions_for_user(session, user_id)
             
@@ -64,8 +69,8 @@ class PanelWebhookService:
                     # This user has tribute payments, auto-renew for the same duration
                     logging.info(f"Auto-renewing tribute subscription for user {user_id} for {last_tribute_duration} months")
                     
-                    # Extend subscription by the last payment duration
-                    new_end_date = datetime.now(timezone.utc) + timedelta(days=last_tribute_duration * 30)
+                    # Extend subscription by the last payment duration (calendar months)
+                    new_end_date = add_months(datetime.now(timezone.utc), last_tribute_duration)
                     
                     await subscription_dal.update_subscription(
                         session,
@@ -96,14 +101,17 @@ class PanelWebhookService:
                             reply_markup=markup,
                             parse_mode="HTML"
                         )
+                        auto_renewed = True
                     except Exception as e:
                         logging.error(f"Failed to send auto-renewal notification to user {user_id}: {e}")
                         
             await session.commit()
+            return auto_renewed
             
         except Exception as e:
             logging.error(f"Error handling expired subscription for user {user_id}: {e}")
             await session.rollback()
+            return False
 
     async def handle_event(self, event_name: str, user_payload: dict):
         telegram_id = user_payload.get("telegramId")
@@ -133,18 +141,20 @@ class PanelWebhookService:
                     user_name=first_name,
                     end_date=user_payload.get("expireAt", "")[:10],
                 )
-        elif event_name == "user.expired" and self.settings.SUBSCRIPTION_NOTIFY_ON_EXPIRE:
-            # Check if this is a tribute user that should be auto-renewed
-            await self._handle_expired_subscription(session, user_id, user_payload, lang, markup, first_name)
+        elif event_name == "user.expired":
+            # Check if this is a tribute user that should be auto-renewed (regardless of notification settings)
+            auto_renewed = await self._handle_expired_subscription(session, user_id, user_payload, lang, markup, first_name)
             
-            await self._send_message(
-                user_id,
-                lang,
-                "subscription_expired_notification",
-                reply_markup=markup,
-                user_name=first_name,
-                end_date=user_payload.get("expireAt", "")[:10],
-            )
+            # If auto-renewed via Tribute, suppress expiration notification. Otherwise, send it if enabled.
+            if not auto_renewed and self.settings.SUBSCRIPTION_NOTIFY_ON_EXPIRE:
+                await self._send_message(
+                    user_id,
+                    lang,
+                    "subscription_expired_notification",
+                    reply_markup=markup,
+                    user_name=first_name,
+                    end_date=user_payload.get("expireAt", "")[:10],
+                )
         elif event_name == "user.expired_24_hours_ago" and self.settings.SUBSCRIPTION_NOTIFY_AFTER_EXPIRE:
             await self._send_message(
                 user_id,

bot/services/referral_service.py

@@ -7,6 +7,7 @@ from datetime import datetime, timezone, timedelta
 
 from config.settings import Settings
 from db.dal import user_dal
+from db.dal import payment_dal
 from db.models import User
 from db.dal import subscription_dal
 from bot.middlewares.i18n import JsonI18n
@@ -24,8 +25,12 @@ class ReferralService:
         self.i18n = i18n
 
     async def apply_referral_bonuses_for_payment(
-            self, session: AsyncSession, referee_user_id: int,
-            purchased_subscription_months: int) -> Dict[str, Any]:
+            self,
+            session: AsyncSession,
+            referee_user_id: int,
+            purchased_subscription_months: int,
+            current_payment_db_id: Optional[int] = None,
+            skip_if_active_before_payment: bool = True) -> Dict[str, Any]:
 
         referee_final_end_date: Optional[datetime] = None
         referee_bonus_applied_days: Optional[int] = None
@@ -43,6 +48,39 @@ class ReferralService:
                     "referee_new_end_date": None
                 }
 
+            # If configured to apply referral bonuses only once per invited user,
+            # check if the referee already has succeeded payments.
+            # Use getattr with a safe default (True) to avoid AttributeError if
+            # running with an older settings schema.
+            if getattr(self.settings, "REFERRAL_ONE_BONUS_PER_REFEREE", True):
+                try:
+                    succeeded_count = await payment_dal.count_user_succeeded_payments(
+                        session, referee_user_id, exclude_payment_id=current_payment_db_id
+                    )
+                    if succeeded_count and succeeded_count > 0:
+                        logging.info(
+                            f"Referral bonuses skipped for user {referee_user_id}: already has {succeeded_count} succeeded payments.")
+                        return {
+                            "referee_bonus_applied_days": None,
+                            "referee_new_end_date": None
+                        }
+                except Exception as e_cnt:
+                    logging.error(f"Failed counting succeeded payments for user {referee_user_id}: {e_cnt}")
+
+            # Additionally, do not award referral bonuses if the user was active at payment time
+            # (has an active subscription now). This avoids giving bonuses to already active users.
+            if skip_if_active_before_payment:
+                try:
+                    if await self.subscription_service.has_active_subscription(session, referee_user_id):
+                        logging.info(
+                            f"Referral bonuses skipped for user {referee_user_id}: user currently has an active subscription.")
+                        return {
+                            "referee_bonus_applied_days": None,
+                            "referee_new_end_date": None
+                        }
+                except Exception as e_sub:
+                    logging.error(f"Failed to check active subscription for {referee_user_id}: {e_sub}")
+
             inviter_user_id = referee_user_model.referred_by_id
             inviter_user_model = await user_dal.get_user_by_id(
                 session, inviter_user_id)

bot/services/stars_service.py

@@ -96,7 +96,12 @@ class StarsService:
             return
 
         referral_bonus = await self.referral_service.apply_referral_bonuses_for_payment(
-            session, message.from_user.id, months)
+            session,
+            message.from_user.id,
+            months,
+            current_payment_db_id=payment_db_id,
+            skip_if_active_before_payment=False,
+        )
         await session.commit()
 
         applied_days = referral_bonus.get("referee_bonus_applied_days") if referral_bonus else None

bot/services/subscription_service.py

@@ -6,6 +6,7 @@ from aiogram import Bot
 from bot.middlewares.i18n import JsonI18n
 
 from db.dal import user_dal, subscription_dal, promo_code_dal, payment_dal
+from bot.utils.date_utils import add_months
 from db.models import User, Subscription
 
 from config.settings import Settings
@@ -34,12 +35,25 @@ class SubscriptionService:
             else self.settings.DEFAULT_LANGUAGE
         )
 
-    async def has_had_any_subscription(
-        self, session: AsyncSession, user_id: int
-    ) -> bool:
-
+    async def has_had_any_subscription(self, session: AsyncSession, user_id: int) -> bool:
         return await subscription_dal.has_any_subscription_for_user(session, user_id)
 
+    async def has_active_subscription(self, session: AsyncSession, user_id: int) -> bool:
+        """Return True if user currently has an active subscription (end_date in future)."""
+        try:
+            user_record = await user_dal.get_user_by_id(session, user_id)
+            if not user_record or not user_record.panel_user_uuid:
+                return False
+            active_sub = await subscription_dal.get_active_subscription_by_user_id(
+                session, user_id, user_record.panel_user_uuid
+            )
+            if not active_sub or not active_sub.end_date:
+                return False
+            from datetime import datetime, timezone
+            return active_sub.is_active and active_sub.end_date > datetime.now(timezone.utc)
+        except Exception:
+            return False
+
     async def _notify_admin_panel_user_creation_failed(self, user_id: int):
         if not self.bot or not self.i18n or not self.settings.ADMIN_IDS:
             return
@@ -219,23 +233,10 @@ class SubscriptionService:
                     "panel_user_uuid": actual_panel_uuid_from_api
                 }
 
-                if (
-                    actual_panel_username_from_api
-                    and actual_panel_username_from_api
-                    != panel_username_on_panel_standard
-                    and (
-                        db_user.username is None
-                        or db_user.username != actual_panel_username_from_api
-                    )
-                ):
-                    update_data_for_local_user["username"] = (
-                        actual_panel_username_from_api
-                    )
-
+                # Do not overwrite Telegram username with panel username.
+                # Only update the local linkage to panel UUID here.
                 await user_dal.update_user(session, user_id, update_data_for_local_user)
                 db_user.panel_user_uuid = actual_panel_uuid_from_api
-                if "username" in update_data_for_local_user:
-                    db_user.username = update_data_for_local_user["username"]
                 panel_user_created_or_linked_now = True
                 current_local_panel_uuid = actual_panel_uuid_from_api
         else:
@@ -348,19 +349,12 @@ class SubscriptionService:
                 "message_key": "trial_activation_failed_db",
             }
 
-        panel_update_payload: Dict[str, Any] = {
-            "uuid": panel_user_uuid,
-            "expireAt": end_date.isoformat(timespec="milliseconds").replace(
-                "+00:00", "Z"
-            ),
-            "status": "ACTIVE",
-            "trafficLimitBytes": self.settings.trial_traffic_limit_bytes,
-            "trafficLimitStrategy": self.settings.USER_TRAFFIC_STRATEGY,
-        }
-        if self.settings.parsed_user_squad_uuids:
-            panel_update_payload["activeInternalSquads"] = (
-                self.settings.parsed_user_squad_uuids
-            )
+        panel_update_payload = self._build_panel_update_payload(
+            panel_user_uuid=panel_user_uuid,
+            expire_at=end_date,
+            status="ACTIVE",
+            traffic_limit_bytes=self.settings.trial_traffic_limit_bytes,
+        )
 
         updated_panel_user = await self.panel_service.update_user_details_on_panel(
             panel_user_uuid, panel_update_payload
@@ -431,7 +425,9 @@ class SubscriptionService:
         ):
             start_date = current_active_sub.end_date
 
-        duration_days_total = months * 30
+        # base duration by months
+        end_after_months = add_months(start_date, months)
+        duration_days_total = (end_after_months - start_date).days
         applied_promo_bonus_days = 0
 
         if promo_code_id_from_payment:
@@ -495,19 +491,12 @@ class SubscriptionService:
             )
             return None
 
-        panel_update_payload = {
-            "uuid": panel_user_uuid,
-            "expireAt": final_end_date.isoformat(timespec="milliseconds").replace(
-                "+00:00", "Z"
-            ),
-            "status": "ACTIVE",
-            "trafficLimitBytes": self.settings.user_traffic_limit_bytes,
-            "trafficLimitStrategy": self.settings.USER_TRAFFIC_STRATEGY,
-        }
-        if self.settings.parsed_user_squad_uuids:
-            panel_update_payload["activeInternalSquads"] = (
-                self.settings.parsed_user_squad_uuids
-            )
+        panel_update_payload = self._build_panel_update_payload(
+            panel_user_uuid=panel_user_uuid,
+            expire_at=final_end_date,
+            status="ACTIVE",
+            traffic_limit_bytes=self.settings.user_traffic_limit_bytes,
+        )
 
         updated_panel_user = await self.panel_service.update_user_details_on_panel(
             panel_user_uuid, panel_update_payload
@@ -598,17 +587,13 @@ class SubscriptionService:
 
         if updated_sub_model:
             # Prepare panel update payload
-            panel_update_payload = {
-                "expireAt": new_end_date_obj.isoformat(
-                    timespec="milliseconds"
-                ).replace("+00:00", "Z")
-            }
-            
-            # For promo code activations, remove traffic limit
-            if "promo code" in reason.lower():
-                panel_update_payload["trafficLimitBytes"] = self.settings.user_traffic_limit_bytes
-                panel_update_payload["trafficLimitStrategy"] = self.settings.USER_TRAFFIC_STRATEGY
-                logging.info(f"Updating traffic limit for user {user_id} to {self.settings.user_traffic_limit_bytes} bytes due to promo code activation")
+            panel_update_payload = self._build_panel_update_payload(
+                expire_at=new_end_date_obj,
+                traffic_limit_bytes=(
+                    self.settings.user_traffic_limit_bytes if "promo code" in reason.lower() else None
+                ),
+                include_uuid=False,
+            )
             
             panel_update_success = (
                 await self.panel_service.update_user_details_on_panel(
@@ -775,3 +760,27 @@ class SubscriptionService:
             logging.warning(
                 f"Could not find subscription for user {user_id} ending at {subscription_end_date.isoformat()} to update notification time."
             )
+
+    # Helpers
+    def _build_panel_update_payload(
+        self,
+        *,
+        panel_user_uuid: Optional[str] = None,
+        expire_at: Optional[datetime] = None,
+        status: Optional[str] = None,
+        traffic_limit_bytes: Optional[int] = None,
+        include_uuid: bool = True,
+    ) -> Dict[str, Any]:
+        payload: Dict[str, Any] = {}
+        if include_uuid and panel_user_uuid:
+            payload["uuid"] = panel_user_uuid
+        if expire_at is not None:
+            payload["expireAt"] = expire_at.isoformat(timespec="milliseconds").replace("+00:00", "Z")
+        if status is not None:
+            payload["status"] = status
+        if traffic_limit_bytes is not None:
+            payload["trafficLimitBytes"] = traffic_limit_bytes
+            payload["trafficLimitStrategy"] = self.settings.USER_TRAFFIC_STRATEGY
+        if self.settings.parsed_user_squad_uuids:
+            payload["activeInternalSquads"] = self.settings.parsed_user_squad_uuids
+        return payload

bot/services/tribute_service.py

@@ -39,11 +39,16 @@ def convert_period_to_months(period: Optional[str]) -> int:
 
 
 class TributeService:
-    def __init__(self, bot: Bot, settings: Settings, i18n: JsonI18n,
-                 async_session_factory: sessionmaker,
-                 panel_service: PanelApiService,
-                 subscription_service: SubscriptionService,
-                 referral_service: ReferralService):
+    def __init__(
+        self,
+        bot: Bot,
+        settings: Settings,
+        i18n: JsonI18n,
+        async_session_factory: sessionmaker,
+        panel_service: PanelApiService,
+        subscription_service: SubscriptionService,
+        referral_service: ReferralService,
+    ):
         self.bot = bot
         self.settings = settings
         self.i18n = i18n
@@ -52,8 +57,7 @@ class TributeService:
         self.subscription_service = subscription_service
         self.referral_service = referral_service
 
-    async def handle_webhook(self, raw_body: bytes,
-                             signature_header: Optional[str]) -> web.Response:
+    async def handle_webhook(self, raw_body: bytes, signature_header: Optional[str]) -> web.Response:
         settings = self.settings
         bot = self.bot
         i18n = self.i18n
@@ -61,78 +65,107 @@ class TributeService:
         subscription_service = self.subscription_service
         referral_service = self.referral_service
 
+        def ok(data: Optional[dict] = None) -> web.Response:
+            payload = {"status": "ok"}
+            if data:
+                payload.update(data)
+            return web.json_response(payload, status=200)
+
+        def ignored(reason: str) -> web.Response:
+            return web.json_response({"status": "ignored", "reason": reason}, status=200)
+
+        def bad_request(reason: str) -> web.Response:
+            return web.json_response({"status": "error", "reason": reason}, status=400)
+
         if settings.TRIBUTE_API_KEY:
             if not signature_header:
-                return web.Response(status=403, text="no_signature")
+                return web.json_response({"status": "error", "reason": "no_signature"}, status=403)
             expected_sig = hmac.new(settings.TRIBUTE_API_KEY.encode(), raw_body,
                                     hashlib.sha256).hexdigest()
             if not hmac.compare_digest(expected_sig, signature_header):
-                return web.Response(status=403, text="invalid_signature")
+                return web.json_response({"status": "error", "reason": "invalid_signature"}, status=403)
 
         try:
             payload = json.loads(raw_body.decode())
         except Exception:
-            return web.Response(status=400, text="bad_request")
+            return bad_request("invalid_json")
 
         logging.info(
             "Tribute webhook data: %s",
             json.dumps(payload, ensure_ascii=False),
         )
 
-        event_name = payload.get('name')
-        data = payload.get('payload', {})
-        user_id = data.get('telegram_user_id')
-        price_val = (
-            data.get('amount')
-            or data.get('amount_paid')
-            or data.get('price')
-        )
+        # Tribute webhook spec: only two events are sent
+        # name: new_subscription | cancelled_subscription
+        event_name = payload.get("name")
+        data = payload.get("payload", {})
 
-        if not user_id or price_val is None:
-            return web.Response(status=200, text="ok_missing_fields")
+        # Mandatory routing fields
+        user_id = data.get("telegram_user_id")
+        if not user_id:
+            # Permanent format issue — acknowledge to avoid retries
+            return ignored("missing_telegram_user_id")
 
-        period_val = data.get('period')
+        period_val = data.get("period")
         months = convert_period_to_months(period_val)
-        price_rub = price_val / 100
+
+        # Tribute sends amount in minor units (kopecks/cents). Convert to major units before persisting.
+        amount_value = data.get("amount") or data.get("price")
+        currency = (data.get("currency") or settings.DEFAULT_CURRENCY_SYMBOL or "RUB").upper()
+        if amount_value is not None:
+            try:
+                amount_minor_units = float(amount_value)
+            except (TypeError, ValueError):
+                amount_minor_units = 0.0
+            amount_float = round(amount_minor_units / 100.0, 2)
+        else:
+            amount_float = 0.0
 
         async with async_session_factory() as session:
-            if event_name == 'new_subscription':
-                provider_payment_id = str(data.get('subscription_id'))
-                existing_payment = await payment_dal.get_payment_by_provider_payment_id(
-                    session, provider_payment_id)
-                if existing_payment:
-                    logging.info(
-                        "Duplicate Tribute payment webhook ignored for provider_payment_id %s",
-                        provider_payment_id,
-                    )
-                    payment_record = existing_payment
+            if event_name == "new_subscription":
+                # Use a unique, idempotent provider payment id per webhook event
+                # Prefer explicit event/payment identifiers if present; otherwise fall back to payload hash suffix
+                candidate_event_id = (
+                    str(data.get("event_id") or data.get("payment_id") or data.get("purchase_id") or data.get("invoice_id") or "")
+                )
+                if candidate_event_id:
+                    provider_payment_id = candidate_event_id
                 else:
-                    payment_record = await payment_dal.create_payment_record(
-                        session,
-                        {
-                            'user_id': user_id,
-                            'amount': float(price_rub),
-                            'currency': 'RUB',
-                            'status': 'succeeded',
-                            'description': 'Tribute subscription',
-                            'subscription_duration_months': months,
-                            'provider_payment_id': provider_payment_id,
-                            'provider': 'tribute',
-                        },
-                    )
+                    # Combine subscription_id (if any) with a stable hash of the raw payload to ensure uniqueness per event
+                    sub_id_part = str(data.get("subscription_id") or "sub")
+                    payload_hash = hashlib.sha256(raw_body).hexdigest()[:16]
+                    provider_payment_id = f"{sub_id_part}:{payload_hash}"
+
+                # Idempotent ensure payment
+                payment_record = await payment_dal.ensure_payment_with_provider_id(
+                    session,
+                    user_id=int(user_id),
+                    amount=amount_float,
+                    currency=currency,
+                    months=months,
+                    description="Tribute subscription",
+                    provider="tribute",
+                    provider_payment_id=provider_payment_id,
+                )
+
                 activation_details = await subscription_service.activate_subscription(
                     session,
-                    user_id,
+                    int(user_id),
                     months,
-                    float(price_rub),
+                    float(amount_float),
                     payment_record.payment_id,
-                    provider='tribute',
+                    provider="tribute",
                 )
                 referral_bonus = await referral_service.apply_referral_bonuses_for_payment(
-                    session, user_id, months)
+                    session,
+                    int(user_id),
+                    months,
+                    current_payment_db_id=payment_record.payment_id,
+                    skip_if_active_before_payment=False,
+                )
                 await session.commit()
 
-                db_user = await user_dal.get_user_by_id(session, user_id)
+                db_user = await user_dal.get_user_by_id(session, int(user_id))
                 lang = db_user.language_code if db_user and db_user.language_code else settings.DEFAULT_LANGUAGE
                 _ = lambda k, **kw: i18n.gettext(lang, k, **kw)
 
@@ -177,7 +210,7 @@ class TributeService:
 
                     try:
                         await bot.send_message(
-                            user_id,
+                            int(user_id),
                             success_msg,
                             reply_markup=markup,
                             parse_mode="HTML",
@@ -190,25 +223,24 @@ class TributeService:
                 # Send notification about payment
                 try:
                     notification_service = NotificationService(bot, settings, i18n)
-                    user = await user_dal.get_user_by_id(session, user_id)
+                    user = await user_dal.get_user_by_id(session, int(user_id))
                     await notification_service.notify_payment_received(
-                        user_id=user_id,
-                        amount=float(price_rub),
-                        currency="RUB",
+                        user_id=int(user_id),
+                        amount=float(amount_float),
+                        currency=currency,
                         months=months,
                         payment_provider="tribute",
                         username=user.username if user else None
                     )
                 except Exception as e:
                     logging.error(f"Failed to send tribute payment notification: {e}")
-                    
-            elif event_name == 'subscription_cancelled':
-                # Handle tribute subscription cancellation
-                await self._handle_tribute_cancellation(session, user_id, bot, i18n)
+            elif event_name == "cancelled_subscription":
+                await self._handle_tribute_cancellation(session, int(user_id), bot, i18n)
                 
             else:
                 await session.commit()
-        return web.Response(status=200, text="ok")
+        # Acknowledge to Tribute that webhook was received and processed/accepted
+        return ok({"event": event_name or "unknown"})
 
     async def _handle_tribute_cancellation(self, session, user_id: int, bot: Bot, i18n: JsonI18n):
         """Handle tribute subscription cancellation - set subscription to 1 day grace period"""
@@ -218,22 +250,7 @@ class TributeService:
         
         try:
             # Set all user's subscriptions to expire in 1 day (grace period)
-            grace_end_date = datetime.now(timezone.utc) + timedelta(days=1)
-            
-            # Get all active subscriptions for the user
-            user_subs = await subscription_dal.get_active_subscriptions_for_user(session, user_id)
-            
-            for sub in user_subs:
-                await subscription_dal.update_subscription(
-                    session, 
-                    sub.subscription_id, 
-                    {
-                        'end_date': grace_end_date,
-                        'status_from_panel': 'CANCELLED',
-                        'skip_notifications': True  # Skip future notifications for cancelled subs
-                    }
-                )
-            
+            await subscription_dal.set_user_subscriptions_cancelled_with_grace(session, user_id, grace_days=1)
             await session.commit()
             
             # Send notification about cancellation if enabled
@@ -256,7 +273,7 @@ class TributeService:
                 
                 try:
                     await bot.send_message(
-                        user_id,
+                        int(user_id),
                         cancellation_msg,
                         reply_markup=markup,
                         parse_mode="HTML"

bot/utils/__init__.py

@@ -1 +1,263 @@
-# Bot utilities package
+# Bot utilities package
+
+from dataclasses import dataclass
+from typing import Optional, Dict, Any
+from aiogram import types
+
+
+@dataclass
+class MessageContent:
+    """Класс для хранения информации о контенте сообщения"""
+    content_type: str
+    file_id: Optional[str] = None
+    text: Optional[str] = None
+
+
+# Словари поддерживаемых параметров для каждого типа сообщения
+SUPPORTED_PARAMS = {
+    "text": {"parse_mode", "entities", "disable_web_page_preview", "disable_notification", "protect_content", "reply_markup", "reply_to_message_id", "allow_sending_without_reply", "message_thread_id"},
+    "photo": {"caption", "parse_mode", "caption_entities", "disable_notification", "protect_content", "reply_markup", "reply_to_message_id", "allow_sending_without_reply", "message_thread_id", "has_spoiler"},
+    "video": {"duration", "width", "height", "thumbnail", "caption", "parse_mode", "caption_entities", "supports_streaming", "disable_notification", "protect_content", "reply_markup", "reply_to_message_id", "allow_sending_without_reply", "message_thread_id", "has_spoiler"},
+    "animation": {"duration", "width", "height", "thumbnail", "caption", "parse_mode", "caption_entities", "disable_notification", "protect_content", "reply_markup", "reply_to_message_id", "allow_sending_without_reply", "message_thread_id", "has_spoiler"},
+    "document": {"thumbnail", "caption", "parse_mode", "caption_entities", "disable_content_type_detection", "disable_notification", "protect_content", "reply_markup", "reply_to_message_id", "allow_sending_without_reply", "message_thread_id"},
+    "audio": {"caption", "parse_mode", "caption_entities", "duration", "performer", "title", "thumbnail", "disable_notification", "protect_content", "reply_markup", "reply_to_message_id", "allow_sending_without_reply", "message_thread_id"},
+    "voice": {"caption", "parse_mode", "caption_entities", "duration", "disable_notification", "protect_content", "reply_markup", "reply_to_message_id", "allow_sending_without_reply", "message_thread_id"},
+    "sticker": {"disable_notification", "protect_content", "reply_markup", "reply_to_message_id", "allow_sending_without_reply", "message_thread_id"},
+    "video_note": {"duration", "length", "thumbnail", "disable_notification", "protect_content", "reply_markup", "reply_to_message_id", "allow_sending_without_reply", "message_thread_id"},
+}
+
+
+def filter_kwargs(content_type: str, kwargs: Dict[str, Any]) -> Dict[str, Any]:
+    """Фильтрует kwargs, оставляя только поддерживаемые параметры для данного типа сообщения"""
+    supported = SUPPORTED_PARAMS.get(content_type, set())
+    return {k: v for k, v in kwargs.items() if k in supported}
+
+
+def get_message_content(message: types.Message) -> MessageContent:
+    """
+    Определяет тип контента сообщения и возвращает его данные.
+    Использует match/case вместо длинных if-elif цепочек.
+    """
+    text = (message.text or message.caption or "").strip()
+    
+    # Проверяем наличие медиа-контента
+    media_content = None
+    if message.photo:
+        media_content = ("photo", message.photo[-1].file_id)
+    elif message.video:
+        media_content = ("video", message.video.file_id)
+    elif message.animation:
+        media_content = ("animation", message.animation.file_id)
+    elif message.document:
+        media_content = ("document", message.document.file_id)
+    elif message.audio:
+        media_content = ("audio", message.audio.file_id)
+    elif message.voice:
+        media_content = ("voice", message.voice.file_id)
+    elif message.sticker:
+        media_content = ("sticker", message.sticker.file_id)
+    elif message.video_note:
+        media_content = ("video_note", message.video_note.file_id)
+    
+    # Используем match/case для определения типа контента
+    match media_content:
+        case (content_type, file_id):
+            return MessageContent(content_type=content_type, file_id=file_id, text=text)
+        case None:
+            return MessageContent(content_type="text", text=text)
+        case _:
+            return MessageContent(content_type="text", text=text)
+
+
+async def send_message_by_type(bot, chat_id: int, content: MessageContent, **kwargs) -> None:
+    """
+    Отправляет сообщение указанного типа.
+    Использует match/case вместо длинных if-elif цепочек.
+    Автоматически фильтрует неподдерживаемые параметры.
+    """
+    # Фильтруем kwargs для данного типа сообщения
+    filtered_kwargs = filter_kwargs(content.content_type, kwargs)
+    
+    match content.content_type:
+        case "text":
+            await bot.send_message(
+                chat_id=chat_id,
+                text=content.text,
+                **filtered_kwargs
+            )
+        case "photo":
+            await bot.send_photo(
+                chat_id=chat_id,
+                photo=content.file_id,
+                caption=content.text or None,
+                **filtered_kwargs
+            )
+        case "video":
+            await bot.send_video(
+                chat_id=chat_id,
+                video=content.file_id,
+                caption=content.text or None,
+                **filtered_kwargs
+            )
+        case "animation":
+            await bot.send_animation(
+                chat_id=chat_id,
+                animation=content.file_id,
+                caption=content.text or None,
+                **filtered_kwargs
+            )
+        case "document":
+            await bot.send_document(
+                chat_id=chat_id,
+                document=content.file_id,
+                caption=content.text or None,
+                **filtered_kwargs
+            )
+        case "audio":
+            await bot.send_audio(
+                chat_id=chat_id,
+                audio=content.file_id,
+                caption=content.text or None,
+                **filtered_kwargs
+            )
+        case "voice":
+            await bot.send_voice(
+                chat_id=chat_id,
+                voice=content.file_id,
+                caption=content.text or None,
+                **filtered_kwargs
+            )
+        case "sticker":
+            await bot.send_sticker(
+                chat_id=chat_id,
+                sticker=content.file_id,
+                **filtered_kwargs
+            )
+        case "video_note":
+            await bot.send_video_note(
+                chat_id=chat_id,
+                video_note=content.file_id,
+                **filtered_kwargs
+            )
+        case _:
+            # Fallback для неизвестных типов - отправляем как текст
+            text_kwargs = filter_kwargs("text", kwargs)
+            await bot.send_message(
+                chat_id=chat_id,
+                text=content.text or "Unknown content type",
+                **text_kwargs
+            )
+
+
+async def send_message_via_queue(queue_manager, uid: int, content: MessageContent, **kwargs) -> None:
+    """
+    Отправляет сообщение через очередь в зависимости от типа контента.
+    Использует match/case вместо длинных if-elif цепочек.
+    Автоматически фильтрует неподдерживаемые параметры.
+    """
+    # Фильтруем kwargs для данного типа сообщения
+    filtered_kwargs = filter_kwargs(content.content_type, kwargs)
+    
+    match content.content_type:
+        case "text":
+            await queue_manager.send_message(
+                chat_id=uid, text=content.text, **filtered_kwargs
+            )
+        case "photo":
+            await queue_manager.send_photo(
+                chat_id=uid, photo=content.file_id, caption=content.text or None, **filtered_kwargs
+            )
+        case "video":
+            await queue_manager.send_video(
+                chat_id=uid, video=content.file_id, caption=content.text or None, **filtered_kwargs
+            )
+        case "animation":
+            await queue_manager.send_animation(
+                chat_id=uid, animation=content.file_id, caption=content.text or None, **filtered_kwargs
+            )
+        case "document":
+            await queue_manager.send_document(
+                chat_id=uid, document=content.file_id, caption=content.text or None, **filtered_kwargs
+            )
+        case "audio":
+            await queue_manager.send_audio(
+                chat_id=uid, audio=content.file_id, caption=content.text or None, **filtered_kwargs
+            )
+        case "voice":
+            await queue_manager.send_voice(
+                chat_id=uid, voice=content.file_id, caption=content.text or None, **filtered_kwargs
+            )
+        case "sticker":
+            await queue_manager.send_sticker(
+                chat_id=uid, sticker=content.file_id, **filtered_kwargs
+            )
+        case "video_note":
+            await queue_manager.send_video_note(
+                chat_id=uid, video_note=content.file_id, **filtered_kwargs
+            )
+        case _:
+            # Fallback для неизвестных типов - отправляем как текст
+            text_kwargs = filter_kwargs("text", kwargs)
+            await queue_manager.send_message(
+                chat_id=uid, text=content.text or "Unknown content type", **text_kwargs
+            )
+
+
+async def send_direct_message(bot, chat_id: int, content: MessageContent, extra_text: str = "", **kwargs) -> None:
+    """
+    Отправляет прямое сообщение с дополнительной обработкой для sticker и video_note.
+    Для этих типов медиа отправляется отдельное текстовое сообщение, т.к. они не поддерживают caption.
+    Автоматически фильтрует неподдерживаемые параметры.
+    """
+    match content.content_type:
+        case "sticker":
+            # Отправляем стикер с отфильтрованными параметрами
+            sticker_kwargs = filter_kwargs("sticker", kwargs)
+            await bot.send_sticker(
+                chat_id=chat_id,
+                sticker=content.file_id,
+                **sticker_kwargs
+            )
+            # Если есть текст с подписью, отправляем отдельно
+            if content.text or extra_text:
+                text_to_send = (content.text + extra_text) if content.text else extra_text
+                text_kwargs = filter_kwargs("text", kwargs)
+                await bot.send_message(
+                    chat_id,
+                    text_to_send,
+                    **text_kwargs
+                )
+        case "video_note":
+            # Отправляем видео-заметку с отфильтрованными параметрами
+            video_note_kwargs = filter_kwargs("video_note", kwargs)
+            await bot.send_video_note(
+                chat_id=chat_id,
+                video_note=content.file_id,
+                **video_note_kwargs
+            )
+            # Если есть текст с подписью, отправляем отдельно
+            if content.text or extra_text:
+                text_to_send = (content.text + extra_text) if content.text else extra_text
+                text_kwargs = filter_kwargs("text", kwargs)
+                await bot.send_message(
+                    chat_id,
+                    text_to_send,
+                    **text_kwargs
+                )
+        case "text":
+            # Для текста объединяем с extra_text
+            final_text = (content.text + extra_text) if content.text else extra_text
+            text_kwargs = filter_kwargs("text", kwargs)
+            await bot.send_message(
+                chat_id=chat_id,
+                text=final_text,
+                **text_kwargs
+            )
+        case _:
+            # Для остальных типов медиа используем caption
+            final_caption = (content.text + extra_text) if content.text else None
+            await send_message_by_type(
+                bot, chat_id, 
+                MessageContent(content.content_type, content.file_id, final_caption),
+                **kwargs
+            )

bot/utils/date_utils.py

@@ -0,0 +1,27 @@
+from datetime import datetime, timedelta
+
+
+def add_months(base_dt: datetime, months_to_add: int) -> datetime:
+    """Add calendar months to a datetime, clamping the day to the month's length.
+
+    Preserves tzinfo from base_dt.
+    """
+    year = base_dt.year
+    month = base_dt.month + months_to_add
+    day = base_dt.day
+
+    # Normalize year and month
+    year += (month - 1) // 12
+    month = ((month - 1) % 12) + 1
+
+    # Determine last day of target month by rolling to next month's first day and subtracting 1 day
+    if month == 12:
+        next_month_first = datetime(year + 1, 1, 1, tzinfo=base_dt.tzinfo)
+    else:
+        next_month_first = datetime(year, month + 1, 1, tzinfo=base_dt.tzinfo)
+    last_day = (next_month_first - timedelta(days=1)).day
+
+    clamped_day = min(day, last_day)
+    return base_dt.replace(year=year, month=month, day=clamped_day)
+
+

bot/utils/message_queue.py

@@ -151,6 +151,76 @@ class MessageQueueManager:
         )
         await queue.add_message(message)
     
+    async def send_photo(self, chat_id: int, **kwargs) -> None:
+        """Queue a send_photo call"""
+        queue = self.group_queue if self._is_group_chat(chat_id) else self.user_queue
+        message = QueuedMessage(
+            chat_id=chat_id,
+            method_name='send_photo',
+            kwargs=kwargs
+        )
+        await queue.add_message(message)
+
+    async def send_video(self, chat_id: int, **kwargs) -> None:
+        """Queue a send_video call"""
+        queue = self.group_queue if self._is_group_chat(chat_id) else self.user_queue
+        message = QueuedMessage(
+            chat_id=chat_id,
+            method_name='send_video',
+            kwargs=kwargs
+        )
+        await queue.add_message(message)
+
+    async def send_animation(self, chat_id: int, **kwargs) -> None:
+        """Queue a send_animation (GIF) call"""
+        queue = self.group_queue if self._is_group_chat(chat_id) else self.user_queue
+        message = QueuedMessage(
+            chat_id=chat_id,
+            method_name='send_animation',
+            kwargs=kwargs
+        )
+        await queue.add_message(message)
+
+    async def send_audio(self, chat_id: int, **kwargs) -> None:
+        """Queue a send_audio call"""
+        queue = self.group_queue if self._is_group_chat(chat_id) else self.user_queue
+        message = QueuedMessage(
+            chat_id=chat_id,
+            method_name='send_audio',
+            kwargs=kwargs
+        )
+        await queue.add_message(message)
+
+    async def send_voice(self, chat_id: int, **kwargs) -> None:
+        """Queue a send_voice call"""
+        queue = self.group_queue if self._is_group_chat(chat_id) else self.user_queue
+        message = QueuedMessage(
+            chat_id=chat_id,
+            method_name='send_voice',
+            kwargs=kwargs
+        )
+        await queue.add_message(message)
+
+    async def send_sticker(self, chat_id: int, **kwargs) -> None:
+        """Queue a send_sticker call"""
+        queue = self.group_queue if self._is_group_chat(chat_id) else self.user_queue
+        message = QueuedMessage(
+            chat_id=chat_id,
+            method_name='send_sticker',
+            kwargs=kwargs
+        )
+        await queue.add_message(message)
+
+    async def send_video_note(self, chat_id: int, **kwargs) -> None:
+        """Queue a send_video_note call"""
+        queue = self.group_queue if self._is_group_chat(chat_id) else self.user_queue
+        message = QueuedMessage(
+            chat_id=chat_id,
+            method_name='send_video_note',
+            kwargs=kwargs
+        )
+        await queue.add_message(message)
+    
     async def answer_callback_query(self, callback_query_id: str, **kwargs) -> None:
         """Send callback query answer immediately (not rate limited)"""
         await self.bot.answer_callback_query(callback_query_id, **kwargs)

config/settings.py

@@ -93,6 +93,12 @@ class Settings(BaseSettings):
     REFERRAL_BONUS_DAYS_REFEREE_12_MONTHS: Optional[int] = Field(
         default=15, alias="REFEREE_BONUS_DAYS_12_MONTHS")
 
+    # Referral program configuration
+    REFERRAL_ONE_BONUS_PER_REFEREE: bool = Field(
+        default=True,
+        description="When true, referral bonuses (for inviter and referee) are applied only once per invited user – on their first successful payment."
+    )
+
     PANEL_API_URL: Optional[str] = None
     PANEL_API_KEY: Optional[str] = None
     USER_TRAFFIC_LIMIT_GB: Optional[float] = Field(default=0.0)

db/dal/payment_dal.py

@@ -2,7 +2,7 @@ import logging
 from typing import Optional, List, Dict, Any
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.future import select
-from sqlalchemy import update, func
+from sqlalchemy import update, func, and_
 from sqlalchemy.orm import selectinload
 
 from db.models import Payment, User
@@ -47,6 +47,38 @@ async def get_payment_by_provider_payment_id(
     return result.scalar_one_or_none()
 
 
+async def ensure_payment_with_provider_id(
+        session: AsyncSession,
+        *,
+        user_id: int,
+        amount: float,
+        currency: str,
+        months: int,
+        description: str,
+        provider: str,
+        provider_payment_id: str) -> Payment:
+    """Idempotently create a payment record for a provider event.
+
+    If a payment with the same provider_payment_id already exists, returns it.
+    Otherwise creates a new succeeded payment with provided data.
+    """
+    existing = await get_payment_by_provider_payment_id(session, provider_payment_id)
+    if existing:
+        return existing
+
+    payment_payload: Dict[str, Any] = {
+        "user_id": user_id,
+        "amount": float(amount),
+        "currency": currency,
+        "status": "succeeded",
+        "description": description,
+        "subscription_duration_months": months,
+        "provider_payment_id": provider_payment_id,
+        "provider": provider,
+    }
+    return await create_payment_record(session, payment_payload)
+
+
 async def get_payment_by_db_id(session: AsyncSession,
                                payment_db_id: int) -> Optional[Payment]:
 
@@ -82,12 +114,47 @@ async def update_payment_status_by_db_id(
 async def get_recent_payment_logs_with_user(session: AsyncSession,
                                             limit: int = 20,
                                             offset: int = 0) -> List[Payment]:
-    stmt = (select(Payment).options(selectinload(Payment.user)).order_by(
-        Payment.created_at.desc()).limit(limit).offset(offset))
+    stmt = (select(Payment).options(selectinload(Payment.user))
+            .where(Payment.status == 'succeeded')
+            .order_by(Payment.created_at.desc())
+            .limit(limit).offset(offset))
     result = await session.execute(stmt)
     return result.scalars().all()
 
 
+async def get_payments_count(session: AsyncSession) -> int:
+    """Get total count of successful payments."""
+    stmt = select(func.count(Payment.payment_id)).where(Payment.status == 'succeeded')
+    result = await session.execute(stmt)
+    return result.scalar() or 0
+
+
+async def get_all_succeeded_payments_with_user(session: AsyncSession) -> List[Payment]:
+    """Get all successful payments with user data for export."""
+    stmt = (select(Payment).options(selectinload(Payment.user))
+            .where(Payment.status == 'succeeded')
+            .order_by(Payment.created_at.desc()))
+    result = await session.execute(stmt)
+    return result.scalars().all()
+
+
+async def count_user_succeeded_payments(
+    session: AsyncSession, user_id: int, exclude_payment_id: Optional[int] = None
+) -> int:
+    """Count succeeded payments for a specific user.
+
+    If exclude_payment_id is provided, that specific payment will be excluded
+    from the count. Useful to check "prior" payments while processing the
+    current payment in the same transaction.
+    """
+    conditions = [Payment.user_id == user_id, Payment.status == 'succeeded']
+    if exclude_payment_id is not None:
+        conditions.append(Payment.payment_id != exclude_payment_id)
+    stmt = select(func.count(Payment.payment_id)).where(and_(*conditions))
+    result = await session.execute(stmt)
+    return result.scalar() or 0
+
+
 async def update_provider_payment_and_status(
         session: AsyncSession, payment_db_id: int,
         provider_payment_id: str, new_status: str) -> Optional[Payment]:

db/dal/subscription_dal.py

@@ -53,6 +53,29 @@ async def update_subscription(
     return sub
 
 
+async def set_user_subscriptions_cancelled_with_grace(
+        session: AsyncSession, user_id: int, grace_days: int = 1) -> int:
+    """Mark all active user subscriptions as cancelled with a short grace period.
+
+    Sets end_date to now + grace_days, status_from_panel to 'CANCELLED', and
+    skip future notifications to reduce noise after cancellation.
+    Returns number of updated rows.
+    """
+    from datetime import datetime, timezone, timedelta
+    grace_end = datetime.now(timezone.utc) + timedelta(days=grace_days)
+    stmt = (
+        update(Subscription)
+        .where(Subscription.user_id == user_id, Subscription.is_active == True)
+        .values(
+            end_date=grace_end,
+            status_from_panel="CANCELLED",
+            skip_notifications=True,
+        )
+    )
+    result = await session.execute(stmt)
+    return result.rowcount or 0
+
+
 async def upsert_subscription(session: AsyncSession,
                               sub_payload: Dict[str, Any]) -> Subscription:
     panel_sub_uuid = sub_payload.get("panel_subscription_uuid")

db/dal/user_dal.py

@@ -4,7 +4,8 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.future import select
 from sqlalchemy.orm import selectinload
 from sqlalchemy import update, delete, func, and_
-from datetime import datetime
+from datetime import datetime, timezone
+from sqlalchemy.dialects.postgresql import insert as pg_insert
 
 from ..models import User, Subscription
 
@@ -30,35 +31,44 @@ async def get_user_by_panel_uuid(
     return result.scalar_one_or_none()
 
 
-async def get_user(
-    session: AsyncSession,
-    *,
-    user_id: Optional[int] = None,
-    username: Optional[str] = None,
-    panel_uuid: Optional[str] = None,
-) -> Optional[User]:
-    if user_id is not None:
-        return await get_user_by_id(session, user_id)
-    if username is not None:
-        return await get_user_by_username(session, username)
-    if panel_uuid is not None:
-        return await get_user_by_panel_uuid(session, panel_uuid)
-    return None
+## Removed unused generic get_user helper to keep DAL explicit and simple
 
 
-async def create_user(session: AsyncSession, user_data: Dict[str, Any]) -> User:
+async def create_user(session: AsyncSession, user_data: Dict[str, Any]) -> Tuple[User, bool]:
+    """Create a user if not exists in a race-safe way.
+
+    Returns a tuple of (user, created_flag).
+    """
 
     if "registration_date" not in user_data:
-        user_data["registration_date"] = datetime.now()
+        user_data["registration_date"] = datetime.now(timezone.utc)
 
-    new_user = User(**user_data)
-    session.add(new_user)
-    await session.flush()
-    await session.refresh(new_user)
-    logging.info(
-        f"New user {new_user.user_id} created in DAL. Referred by: {new_user.referred_by_id or 'N/A'}."
+    # Use PostgreSQL upsert to avoid IntegrityError on concurrent inserts
+    stmt = (
+        pg_insert(User)
+        .values(**user_data)
+        .on_conflict_do_nothing(index_elements=[User.user_id])
+        .returning(User.user_id)
     )
-    return new_user
+
+    result = await session.execute(stmt)
+    inserted_row = result.first()
+    created = inserted_row is not None
+
+    # Fetch the user (inserted just now or pre-existing)
+    user_id: int = user_data["user_id"]
+    user = await get_user_by_id(session, user_id)
+
+    if created and user is not None:
+        logging.info(
+            f"New user {user.user_id} created in DAL. Referred by: {user.referred_by_id or 'N/A'}."
+        )
+    elif user is not None:
+        logging.info(
+            f"User {user.user_id} already exists in DAL. Proceeding without creation."
+        )
+
+    return user, created
 
 
 async def update_user(
@@ -106,9 +116,10 @@ async def get_all_users_with_panel_uuid(session: AsyncSession) -> List[User]:
 
 async def get_enhanced_user_statistics(session: AsyncSession) -> Dict[str, Any]:
     """Get comprehensive user statistics including active users, trial users, etc."""
-    from datetime import datetime, timedelta
+    from datetime import datetime, timezone
     
-    now = datetime.utcnow()
+    # Use timezone-aware UTC to avoid naive/aware comparison issues in SQL queries
+    now = datetime.now(timezone.utc)
     today_start = now.replace(hour=0, minute=0, second=0, microsecond=0)
     
     # Total users
@@ -119,13 +130,11 @@ async def get_enhanced_user_statistics(session: AsyncSession) -> Dict[str, Any]:
     banned_users_stmt = select(func.count(User.user_id)).where(User.is_banned == True)
     banned_users = (await session.execute(banned_users_stmt)).scalar() or 0
     
-    # Active users today (users with login activity - for now using registration as proxy)
-    active_today_stmt = select(func.count(User.user_id)).where(
-        User.registration_date >= today_start
-    )
+    # Active users today (proxy: registered today)
+    active_today_stmt = select(func.count(User.user_id)).where(User.registration_date >= today_start)
     active_today = (await session.execute(active_today_stmt)).scalar() or 0
     
-    # Users with active paid subscriptions
+    # Users with active paid subscriptions (non-trial providers only)
     paid_subs_stmt = (
         select(func.count(func.distinct(Subscription.user_id)))
         .join(User, Subscription.user_id == User.user_id)
@@ -169,3 +178,52 @@ async def get_enhanced_user_statistics(session: AsyncSession) -> Dict[str, Any]:
         "inactive_users": max(0, inactive_users),
         "referral_users": referral_users
     }
+
+
+async def get_user_ids_with_active_subscription(session: AsyncSession) -> List[int]:
+    """Return non-banned user IDs who have an active subscription (paid or trial)."""
+    from datetime import datetime, timezone
+    now = datetime.now(timezone.utc)
+
+    stmt = (
+        select(func.distinct(Subscription.user_id))
+        .join(User, Subscription.user_id == User.user_id)
+        .where(
+            and_(
+                User.is_banned == False,
+                Subscription.is_active == True,
+                Subscription.end_date > now,
+            )
+        )
+    )
+    result = await session.execute(stmt)
+    return result.scalars().all()
+
+
+async def get_user_ids_without_active_subscription(session: AsyncSession) -> List[int]:
+    """Return non-banned user IDs who do NOT have any active subscription."""
+    from datetime import datetime, timezone
+    now = datetime.now(timezone.utc)
+
+    # Subquery for users with active subscription
+    active_subs_subq = (
+        select(Subscription.user_id)
+        .where(
+            and_(
+                Subscription.is_active == True,
+                Subscription.end_date > now,
+            )
+        )
+    ).scalar_subquery()
+
+    stmt = (
+        select(User.user_id)
+        .where(
+            and_(
+                User.is_banned == False,
+                ~User.user_id.in_(active_subs_subq),
+            )
+        )
+    )
+    result = await session.execute(stmt)
+    return result.scalars().all()

locales/en.json

@@ -16,6 +16,7 @@
   "choose_language": "Choose language / Выберите язык:",
   "language_set_alert": "Language changed!",
   "error_occurred_try_again": "An error occurred, please try again.",
+  "error_try_again": "Please try again.",
   "error_displaying_menu": "Error displaying menu.",
   "main_menu_unknown_action": "Unknown action.",
 
@@ -122,6 +123,28 @@
   "admin_stats_recent_payments_header": "Recent Payments:",
   "admin_stats_payment_item": "{status_emoji} {amount} {currency} from {user_info} ({p_status}) [{p_date}]",
   "admin_stats_no_payments_found": "No payments found yet.",
+  
+  "admin_view_payments_button": "💰 Payments",
+  "admin_payments_header": "💰 <b>All Payments</b>",
+  "admin_no_payments_found": "No payments found.",
+  "admin_export_payments_csv": "📊 Export CSV",
+  "admin_refresh_payments": "🔄 Refresh",
+  "admin_no_payments_to_export": "No payments to export.",
+  "admin_payments_export_success": "📊 Payments export completed!\nTotal records: {count}",
+  "admin_export_sent": "File sent!",
+  
+  "admin_csv_payment_id": "ID",
+  "admin_csv_user_id": "User ID",
+  "admin_csv_username": "Username",
+  "admin_csv_first_name": "First Name",
+  "admin_csv_amount": "Amount",
+  "admin_csv_currency": "Currency",
+  "admin_csv_provider": "Provider",
+  "admin_csv_status": "Status",
+  "admin_csv_description": "Description",
+  "admin_csv_months": "Months",
+  "admin_csv_created_at": "Created At",
+  "admin_csv_provider_payment_id": "Provider Payment ID",
   "admin_stats_last_sync_header": "Last Panel Sync:",
   "admin_stats_sync_time": "Time",
   "admin_stats_sync_status": "Status",
@@ -132,6 +155,10 @@
 
   "admin_broadcast_enter_message": "Enter the broadcast message (HTML supported):",
   "admin_broadcast_confirm_prompt": "You are about to send the following message:\n\n{message_preview}\n\nConfirm sending?",
+  "admin_broadcast_confirm_prompt_short": "The message above will be sent. Confirm?",
+  "broadcast_target_all_button": "👥 All",
+  "broadcast_target_active_button": "✅ Active",
+  "broadcast_target_inactive_button": "⌛ Inactive",
   "confirm_broadcast_send_button": "✅ Send",
   "cancel_broadcast_button": "❌ Cancel",
   "admin_broadcast_sending_started": "Starting broadcast...",
@@ -142,6 +169,8 @@
   "admin_broadcast_cancelled_alert": "Broadcast cancelled!",
   "admin_broadcast_cancelled_nav_back": "Broadcast cancelled. You are returned to the admin panel.",
 
+  "admin_broadcast_invalid_html": "❌ Invalid HTML in message. Please send valid HTML (Telegram-supported tags) or remove tags.",
+
   "admin_promo_create_prompt": "Enter promo details in the format: CODE BONUS_DAYS MAX_USES [VALIDITY_DAYS]\nExample: <code>{example_format}</code>\n(Validity is optional; default is indefinite)",
   "admin_promo_invalid_format": "Invalid format. Please use: CODE BONUS_DAYS MAX_USES [VALIDITY_DAYS]",
   "admin_promo_invalid_code_format": "Code must be 3–30 alphanumeric characters.",
@@ -228,11 +257,11 @@
   "admin_log_user_not_found": "User \"{input}\" not found in bot database.",
   "admin_user_logs_title": "Logs for {user_display} (page {current_page}/{total_pages}):",
 
-  "sync_started": "🔄 Starting data sync with panel...",
-  "sync_failed": "❌ Sync with panel failed. Details: {details}",
-  "sync_completed": "✅ Sync with panel completed. Status: {status}. Details: {details}",
-  "sync_completed_details": "Checked: {total_checked} entries.\nUsers synced/updated: {users_synced}.\nSubscriptions synced/updated: {subs_synced}.",
-  "sync_completed_with_errors_details": "Checked: {total_checked} entries.\nUsers synced/updated: {users_synced}.\nSubscriptions synced/updated: {subs_synced}.\nErrors: {errors_count}.\n\nFirst errors:\n{error_details_preview}",
+  "sync_started_simple": "🔄 Starting synchronization...",
+  "sync_success_simple": "✅ Synchronization completed successfully",
+  "sync_failed_simple": "❌ Synchronization failed",
+  "sync_errors_simple": "⚠️ Synchronization completed with errors ({errors_count} errors)",
+  "sync_critical_error": "❌ Critical synchronization error",
   "no_errors_placeholder": "none",
   "admin_sync_initiated_from_panel": "Sync initiated...",
   "admin_panel_user_creation_failed": "❌ Failed to create panel user for TG ID {user_id}. Panel unreachable?",
@@ -489,5 +518,10 @@
   "admin_financial_week_label": "This week",
   "admin_financial_month_label": "This month",
   "admin_financial_all_time_label": "All time",
-  "admin_financial_payments_label": "payments"
+  "admin_financial_payments_label": "payments",
+
+  "admin_sync_details": "📊 Synchronization Statistics:\n🔍 Panel records checked: {panel_records_checked}\n👥 Users found in DB: {users_found_in_db}\n✨ New users created: {users_created}\n🔄 Users updated: {users_updated}\n📋 Subscriptions synced: {subscriptions_synced_count}\n   ├── Created new: {subscriptions_created}\n   └── Updated existing: {subscriptions_updated}{additional_stats}",
+  "admin_sync_no_telegram_id": "\n⚠️ Records without telegramId: {count}",
+  "admin_sync_not_found_in_db": "\n❌ Not found in DB: {count}",
+  "admin_sync_errors": "\n🚫 Errors: {count}"
 }

locales/ru.json

@@ -16,6 +16,7 @@
   "choose_language": "Выберите язык / Select language:",
   "language_set_alert": "Язык изменен!",
   "error_occurred_try_again": "Произошла ошибка, попробуйте снова.",
+  "error_try_again": "Попробуйте еще раз.",
   "error_displaying_menu": "Ошибка отображения меню.",
   "main_menu_unknown_action": "Неизвестное действие.",
 
@@ -122,6 +123,28 @@
   "admin_stats_recent_payments_header": "Последние платежи:",
   "admin_stats_payment_item": "{status_emoji} {amount} {currency} от {user_info} ({p_status}) [{p_date}]",
   "admin_stats_no_payments_found": "Платежей пока нет.",
+  
+  "admin_view_payments_button": "💰 Платежи",
+  "admin_payments_header": "💰 <b>Все платежи</b>",
+  "admin_no_payments_found": "Платежи не найдены.",
+  "admin_export_payments_csv": "📊 Экспорт CSV",
+  "admin_refresh_payments": "🔄 Обновить",
+  "admin_no_payments_to_export": "Нет платежей для экспорта.",
+  "admin_payments_export_success": "📊 Экспорт платежей завершен!\nВсего записей: {count}",
+  "admin_export_sent": "Файл отправлен!",
+  
+  "admin_csv_payment_id": "ID",
+  "admin_csv_user_id": "User ID",
+  "admin_csv_username": "Логин",
+  "admin_csv_first_name": "Имя",
+  "admin_csv_amount": "Сумма",
+  "admin_csv_currency": "Валюта",
+  "admin_csv_provider": "Платежная система",
+  "admin_csv_status": "Статус",
+  "admin_csv_description": "Описание",
+  "admin_csv_months": "Месяцев",
+  "admin_csv_created_at": "Дата создания",
+  "admin_csv_provider_payment_id": "ID платежа в системе",
   "admin_stats_last_sync_header": "Последняя синхронизация с панелью:",
   "admin_stats_sync_time": "Время",
   "admin_stats_sync_status": "Статус",
@@ -132,6 +155,10 @@
 
   "admin_broadcast_enter_message": "Введите сообщение для рассылки (HTML поддерживается):",
   "admin_broadcast_confirm_prompt": "Вы собираетесь отправить следующее сообщение:\n\n{message_preview}\n\nПодтверждаете отправку?",
+  "admin_broadcast_confirm_prompt_short": "Сообщение выше будет отправлено. Подтвердить отправку?",
+  "broadcast_target_all_button": "👥 Все",
+  "broadcast_target_active_button": "✅ Активные",
+  "broadcast_target_inactive_button": "⌛ Неактивные",
   "confirm_broadcast_send_button": "✅ Отправить",
   "cancel_broadcast_button": "❌ Отмена",
   "admin_broadcast_sending_started": "Начинаю рассылку...",
@@ -238,13 +265,14 @@
   "admin_log_user_not_found": "Пользователь по запросу \"{input}\" не найден в базе данных бота.",
   "admin_user_logs_title": "Логи пользователя {user_display} (стр. {current_page}/{total_pages}):",
   
-  "sync_started": "🔄 Начинаю синхронизацию данных с панелью...",
-  "sync_failed": "❌ Ошибка синхронизации с панелью. Детали: {details}",
-  "sync_completed": "✅ Синхронизация с панелью завершена. Статус: {status}. Детали: {details}",
-  "sync_completed_details": "Проверено: {total_checked} записей.\nПользователей синхронизировано/обновлено: {users_synced}.\nПодписок синхронизировано/обновлено: {subs_synced}.",
-  "sync_completed_with_errors_details": "Проверено: {total_checked} записей.\nПользователей синхронизировано/обновлено: {users_synced}.\nПодписок синхронизировано/обновлено: {subs_synced}.\nОшибок: {errors_count}.\n\nПервые ошибки:\n{error_details_preview}",
+  "sync_started_simple": "🔄 Начинаю синхронизацию...",
+  "sync_success_simple": "✅ Синхронизация успешно завершена",
+  "sync_failed_simple": "❌ Синхронизация завершилась с ошибкой",
+  "sync_errors_simple": "⚠️ Синхронизация завершена с ошибками ({errors_count} ошибок)",
+  "sync_critical_error": "❌ Критическая ошибка синхронизации",
   "no_errors_placeholder": "нет",
   "admin_sync_initiated_from_panel": "Синхронизация запущена...",
+  "admin_broadcast_invalid_html": "❌ Некорректный HTML в сообщении. Пожалуйста, отправьте корректный HTML (поддерживаются теги Telegram) или уберите теги.",
   "admin_panel_user_creation_failed": "❌ Не удалось создать пользователя на панели для TG ID {user_id}. Панель недоступна?",
   "error_displaying_logs_too_long": "Ошибка: логи слишком длинные для отображения одним сообщением. Попробуйте найти логи по конкретному пользователю.",
   "error_displaying_statistics": "Ошибка отображения статистики.",
@@ -498,5 +526,10 @@
   "admin_financial_week_label": "За неделю",
   "admin_financial_month_label": "За месяц",
   "admin_financial_all_time_label": "За все время",
-  "admin_financial_payments_label": "платежей"
+  "admin_financial_payments_label": "платежей",
+
+  "admin_sync_details": "📊 Статистика синхронизации:\n🔍 Проверено записей панели: {panel_records_checked}\n👥 Найдено пользователей в БД: {users_found_in_db}\n✨ Создано новых пользователей: {users_created}\n🔄 Пользователей обновлено: {users_updated}\n📋 Подписок синхронизировано: {subscriptions_synced_count}\n   ├── Создано новых: {subscriptions_created}\n   └── Обновлено существующих: {subscriptions_updated}{additional_stats}",
+  "admin_sync_no_telegram_id": "\n⚠️ Записей без telegramId: {count}",
+  "admin_sync_not_found_in_db": "\n❌ Не найдено в БД: {count}",
+  "admin_sync_errors": "\n🚫 Ошибок: {count}"
 }