Refactor bot initialization and service registration for improved modularity

- Integrated new service building functions to streamline bot initialization, enhancing maintainability and clarity. - Updated the dispatcher setup to include core services dynamically, reducing redundancy in service registration. - Added support for multiple broadcast targets in the admin handler, allowing for more flexible message distribution. - Enhanced localization for new broadcast target options, improving user experience in both English and Russian. - Implemented new database access methods to retrieve user IDs based on subscription status, optimizing broadcast logic.
2025-08-17 12:05:06 +03:00
parent a75a1f483c
commit 3d58f60a4d
9 changed files with 200 additions and 169 deletions
@@ -9,7 +9,7 @@ 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
+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
@@ -1,8 +1,8 @@
 from aiogram import Bot
 from sqlalchemy.orm import sessionmaker
-from typing import Tuple

 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
@@ -14,18 +14,24 @@ 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):
+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)
-    referral_service = ReferralService(settings, subscription_service, bot)
-    promo_code_service = PromoCodeService(settings, subscription_service, bot)
-    stars_service = StarsService(bot, settings, None, subscription_service, referral_service)
+    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,
-        None,
+        i18n,
        async_session_factory,
        subscription_service,
        referral_service,
@@ -33,13 +39,20 @@ def build_core_services(settings: Settings, bot: Bot, async_session_factory: ses
    tribute_service = TributeService(
        bot,
        settings,
-        None,
+        i18n,
        async_session_factory,
        panel_service,
        subscription_service,
        referral_service,
    )
-    panel_webhook_service = PanelWebhookService(bot, settings, None, async_session_factory)
+    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,
@@ -50,6 +63,7 @@ def build_core_services(settings: Settings, bot: Bot, async_session_factory: ses
        "cryptopay_service": cryptopay_service,
        "tribute_service": tribute_service,
        "panel_webhook_service": panel_webhook_service,
+        "yookassa_service": yookassa_service,
    }


@@ -19,6 +19,20 @@ async def build_and_start_web_app(
    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",
+    ):
+        if key in dp:  # type: ignore
+            app[key] = dp[key]  # type: ignore

    setup_application(app, dp, bot=bot)

@@ -31,6 +45,7 @@ async def build_and_start_web_app(
            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
@@ -45,6 +60,12 @@ async def build_and_start_web_app(
        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)
@@ -113,17 +113,58 @@ async def process_broadcast_message_handler(
    await state.update_data(
        broadcast_text=text,
        broadcast_entities=entities,
+        broadcast_target="all",
    )

    confirmation_prompt = _("admin_broadcast_confirm_prompt", message_preview=text)

    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()
+    text = user_fsm_data.get("broadcast_text", "")
+    _ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs)
+    confirmation_prompt = _(
+        "admin_broadcast_confirm_prompt", message_preview=text
+    )
+    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
 )
@@ -194,7 +235,13 @@ 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
@@ -260,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()
@@ -21,6 +21,9 @@ 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.routers import build_root_router

@@ -225,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()
@@ -248,82 +248,19 @@ 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)
-    )
-    # Keep profile data in DB fresh (username/first_name/last_name)
-    dp.update.outer_middleware(ProfileSyncMiddleware())
-    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)

    await register_all_routers(dp, settings_param)
@@ -353,88 +290,10 @@ async def run_bot(settings_param: Settings):
    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
-
-        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
-
-        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)
-            )
+            await build_and_start_web_app(dp, bot, settings_param, local_async_session_factory)

-        main_tasks.append(
-            asyncio.create_task(web_server_task(), name="AIOHTTPServerTask")
-        )
+        main_tasks.append(asyncio.create_task(web_server_task(), name="AIOHTTPServerTask"))

    if run_telegram_polling:
        logging.info("Starting bot in Telegram Polling mode...")
@@ -155,3 +155,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()
@@ -155,6 +155,9 @@

  "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?",
+  "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...",
@@ -155,6 +155,9 @@

  "admin_broadcast_enter_message": "Введите сообщение для рассылки (HTML поддерживается):",
  "admin_broadcast_confirm_prompt": "Вы собираетесь отправить следующее сообщение:\n\n{message_preview}\n\nПодтверждаете отправку?",
+  "broadcast_target_all_button": "👥 Все",
+  "broadcast_target_active_button": "✅ Активные",
+  "broadcast_target_inactive_button": "⌛ Неактивные",
  "confirm_broadcast_send_button": "✅ Отправить",
  "cancel_broadcast_button": "❌ Отмена",
  "admin_broadcast_sending_started": "Начинаю рассылку...",