machka-pasla
ebca408227
- Introduced a new action for managing promo codes within the admin panel, enhancing admin capabilities. - Removed the panel statistics retrieval code from the statistics handler to streamline functionality and reduce complexity. - Updated localization files to reflect changes related to promo management.
194 lines
8.2 KiB
Python
194 lines
8.2 KiB
Python
import logging
|
|
from aiogram import Router, F, types, Bot
|
|
from aiogram.filters import Command
|
|
from aiogram.fsm.context import FSMContext
|
|
from typing import Optional
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from config.settings import Settings
|
|
from bot.keyboards.inline.admin_keyboards import (
|
|
get_admin_panel_keyboard, get_stats_monitoring_keyboard,
|
|
get_user_management_keyboard, get_ban_management_keyboard,
|
|
get_promo_marketing_keyboard, get_system_functions_keyboard
|
|
)
|
|
from bot.middlewares.i18n import JsonI18n
|
|
from bot.services.panel_api_service import PanelApiService
|
|
from bot.services.subscription_service import SubscriptionService
|
|
|
|
from . import broadcast as admin_broadcast_handlers
|
|
from . import promo_codes as admin_promo_handlers
|
|
from . import user_management as admin_user_mgmnt_handlers
|
|
from . import statistics as admin_stats_handlers
|
|
from . import sync_admin as admin_sync_handlers
|
|
from . import logs_admin as admin_logs_handlers
|
|
|
|
router = Router(name="admin_common_router")
|
|
|
|
|
|
@router.message(Command("admin"))
|
|
async def admin_panel_command_handler(
|
|
message: types.Message,
|
|
state: FSMContext,
|
|
settings: Settings,
|
|
i18n_data: dict,
|
|
):
|
|
current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE)
|
|
i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance")
|
|
if not i18n:
|
|
logging.error("i18n missing in admin_panel_command_handler")
|
|
await message.answer("Language service error.")
|
|
return
|
|
|
|
await state.clear()
|
|
_ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs)
|
|
await message.answer(_(key="admin_panel_title"),
|
|
reply_markup=get_admin_panel_keyboard(
|
|
i18n, current_lang, settings))
|
|
|
|
|
|
@router.callback_query(F.data.startswith("admin_action:"))
|
|
async def admin_panel_actions_callback_handler(
|
|
callback: types.CallbackQuery, state: FSMContext, settings: Settings,
|
|
i18n_data: dict, bot: Bot, panel_service: PanelApiService,
|
|
subscription_service: SubscriptionService, session: AsyncSession):
|
|
action_parts = callback.data.split(":")
|
|
action = action_parts[1]
|
|
|
|
current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE)
|
|
i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance")
|
|
if not i18n:
|
|
logging.error("i18n missing in admin_panel_actions_callback_handler")
|
|
await callback.answer("Language error.", show_alert=True)
|
|
return
|
|
_ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs)
|
|
|
|
if not callback.message:
|
|
logging.error(
|
|
f"CallbackQuery {callback.id} from {callback.from_user.id} has no message for admin_action {action}"
|
|
)
|
|
await callback.answer("Error processing action: message context lost.",
|
|
show_alert=True)
|
|
return
|
|
|
|
if action == "stats":
|
|
await admin_stats_handlers.show_statistics_handler(
|
|
callback, i18n_data, settings, session)
|
|
elif action == "broadcast":
|
|
await admin_broadcast_handlers.broadcast_message_prompt_handler(
|
|
callback, state, i18n_data, settings, session)
|
|
elif action == "create_promo":
|
|
await admin_promo_handlers.create_promo_prompt_handler(
|
|
callback, state, i18n_data, settings, session)
|
|
elif action == "create_bulk_promo":
|
|
await admin_promo_handlers.create_bulk_promo_prompt_handler(
|
|
callback, state, i18n_data, settings, session)
|
|
elif action == "manage_promos":
|
|
await admin_promo_handlers.manage_promo_codes_handler(
|
|
callback, i18n_data, settings, session)
|
|
elif action == "view_promos":
|
|
await admin_promo_handlers.view_promo_codes_handler(
|
|
callback, i18n_data, settings, session)
|
|
elif action == "ban_user_prompt":
|
|
await admin_user_mgmnt_handlers.ban_user_prompt_handler(
|
|
callback, state, i18n_data, settings, session)
|
|
elif action == "unban_user_prompt":
|
|
await admin_user_mgmnt_handlers.unban_user_prompt_handler(
|
|
callback, state, i18n_data, settings, session)
|
|
elif action == "users_management":
|
|
from . import user_management as admin_user_management_handlers
|
|
await admin_user_management_handlers.user_management_menu_handler(
|
|
callback, state, i18n_data, settings, session)
|
|
elif action == "view_banned":
|
|
|
|
await admin_user_mgmnt_handlers.view_banned_users_handler(
|
|
callback, state, i18n_data, settings, session)
|
|
elif action == "view_logs_menu":
|
|
await admin_logs_handlers.display_logs_menu(callback, i18n_data,
|
|
settings, session)
|
|
elif action == "promo_management":
|
|
from . import promo_codes as admin_promo_handlers
|
|
await admin_promo_handlers.promo_management_handler(
|
|
callback, i18n_data, settings, session)
|
|
elif action == "sync_panel":
|
|
|
|
await admin_sync_handlers.sync_command_handler(
|
|
message_event=callback,
|
|
bot=bot,
|
|
settings=settings,
|
|
i18n_data=i18n_data,
|
|
panel_service=panel_service,
|
|
session=session)
|
|
await callback.answer(_("admin_sync_initiated_from_panel"))
|
|
elif action == "main":
|
|
try:
|
|
await callback.message.edit_text(
|
|
_(key="admin_panel_title"),
|
|
reply_markup=get_admin_panel_keyboard(i18n, current_lang,
|
|
settings))
|
|
except Exception:
|
|
await callback.message.answer(
|
|
_(key="admin_panel_title"),
|
|
reply_markup=get_admin_panel_keyboard(i18n, current_lang,
|
|
settings))
|
|
await callback.answer()
|
|
else:
|
|
logging.warning(
|
|
f"Unknown admin_action received: {action} from callback {callback.data}"
|
|
)
|
|
await callback.answer(_("admin_unknown_action"), show_alert=True)
|
|
|
|
|
|
@router.callback_query(F.data.startswith("admin_section:"))
|
|
async def admin_section_handler(callback: types.CallbackQuery, state: FSMContext,
|
|
settings: Settings, i18n_data: dict, session: AsyncSession):
|
|
section = callback.data.split(":")[1]
|
|
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 error.", show_alert=True)
|
|
return
|
|
_ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs)
|
|
|
|
if not callback.message:
|
|
await callback.answer("Error: message context lost.", show_alert=True)
|
|
return
|
|
|
|
try:
|
|
if section == "stats_monitoring":
|
|
await callback.message.edit_text(
|
|
_("admin_stats_and_monitoring_section"),
|
|
reply_markup=get_stats_monitoring_keyboard(i18n, current_lang)
|
|
)
|
|
elif section == "user_management":
|
|
await callback.message.edit_text(
|
|
_("admin_user_management_section"),
|
|
reply_markup=get_user_management_keyboard(i18n, current_lang)
|
|
)
|
|
elif section == "ban_management":
|
|
await callback.message.edit_text(
|
|
_("admin_ban_management_section"),
|
|
reply_markup=get_ban_management_keyboard(i18n, current_lang)
|
|
)
|
|
elif section == "promo_marketing":
|
|
await callback.message.edit_text(
|
|
_("admin_promo_marketing_section"),
|
|
reply_markup=get_promo_marketing_keyboard(i18n, current_lang)
|
|
)
|
|
elif section == "system_functions":
|
|
await callback.message.edit_text(
|
|
_("admin_system_functions_section"),
|
|
reply_markup=get_system_functions_keyboard(i18n, current_lang)
|
|
)
|
|
else:
|
|
await callback.answer(_("admin_unknown_action"), show_alert=True)
|
|
return
|
|
|
|
await callback.answer()
|
|
except Exception as e:
|
|
logging.error(f"Error handling admin section {section}: {e}")
|
|
await callback.message.answer(
|
|
_("error_occurred_try_again"),
|
|
reply_markup=get_admin_panel_keyboard(i18n, current_lang, settings)
|
|
)
|
|
await callback.answer()
|