546 lines
23 KiB
Python
546 lines
23 KiB
Python
import logging
|
|
import re
|
|
import math
|
|
from aiogram import Router, F, types, Bot
|
|
from aiogram.fsm.context import FSMContext
|
|
from typing import Optional, Tuple, List, Any
|
|
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from datetime import datetime
|
|
|
|
from config.settings import Settings
|
|
|
|
from db.dal import user_dal, subscription_dal
|
|
from db.models import User, Subscription
|
|
|
|
from bot.services.panel_api_service import PanelApiService
|
|
|
|
from bot.states.admin_states import AdminStates
|
|
from bot.keyboards.inline.admin_keyboards import (
|
|
get_back_to_admin_panel_keyboard, get_user_card_keyboard,
|
|
get_banned_users_keyboard, get_confirmation_keyboard,
|
|
get_admin_panel_keyboard)
|
|
from bot.middlewares.i18n import JsonI18n
|
|
|
|
router = Router(name="admin_user_management_router")
|
|
USERNAME_REGEX = re.compile(r"^[a-zA-Z0-9_]{5,32}$")
|
|
|
|
|
|
async def _get_user_model_by_input(session: AsyncSession,
|
|
input_text: str) -> Optional[User]:
|
|
if input_text.isdigit():
|
|
try:
|
|
return await user_dal.get_user(session, user_id=int(input_text))
|
|
except ValueError:
|
|
return None
|
|
if input_text.startswith("@") and USERNAME_REGEX.match(input_text[1:]):
|
|
return await user_dal.get_user(session, username=input_text[1:])
|
|
if USERNAME_REGEX.match(input_text):
|
|
return await user_dal.get_user(session, username=input_text)
|
|
return None
|
|
|
|
|
|
async def ban_user_prompt_handler(callback: types.CallbackQuery,
|
|
state: FSMContext, i18n_data: dict,
|
|
settings: Settings, session: AsyncSession):
|
|
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 ban prompt.", show_alert=True)
|
|
return
|
|
_ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs)
|
|
|
|
prompt_text = _("admin_ban_user_prompt")
|
|
try:
|
|
await callback.message.edit_text(
|
|
prompt_text,
|
|
reply_markup=get_back_to_admin_panel_keyboard(current_lang, i18n))
|
|
except Exception as e:
|
|
logging.warning(
|
|
f"Edit failed for ban_user_prompt: {e}. Sending new message.")
|
|
await callback.message.answer(
|
|
prompt_text,
|
|
reply_markup=get_back_to_admin_panel_keyboard(current_lang, i18n))
|
|
|
|
await callback.answer()
|
|
await state.set_state(AdminStates.waiting_for_user_id_to_ban)
|
|
|
|
|
|
@router.message(AdminStates.waiting_for_user_id_to_ban, F.text)
|
|
async def process_user_input_to_ban_handler(message: types.Message,
|
|
state: FSMContext, i18n_data: dict,
|
|
settings: Settings,
|
|
panel_service: PanelApiService,
|
|
session: AsyncSession):
|
|
current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE)
|
|
i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance")
|
|
if not i18n:
|
|
await message.reply("Language service error.")
|
|
return
|
|
_ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs)
|
|
|
|
input_text = message.text.strip() if message.text else ""
|
|
user_to_ban_model = await _get_user_model_by_input(session, input_text)
|
|
|
|
if not user_to_ban_model:
|
|
await message.answer(
|
|
_("admin_user_not_found_in_bot_db", user_id=input_text))
|
|
await state.clear()
|
|
return
|
|
|
|
user_id_to_ban = user_to_ban_model.user_id
|
|
user_display_for_msg = user_to_ban_model.username or str(user_id_to_ban)
|
|
|
|
if message.from_user and (user_id_to_ban == message.from_user.id
|
|
or user_id_to_ban in settings.ADMIN_IDS):
|
|
await message.answer(_("admin_cannot_ban_self_or_admin"))
|
|
await state.clear()
|
|
return
|
|
|
|
if user_to_ban_model.is_banned:
|
|
await message.answer(
|
|
_("admin_user_already_banned",
|
|
user_id_or_username=user_display_for_msg))
|
|
await state.clear()
|
|
return
|
|
|
|
ban_success_local = await user_dal.set_user_ban_status(
|
|
session, user_id_to_ban, True)
|
|
|
|
if ban_success_local:
|
|
|
|
panel_ban_message_part = ""
|
|
if user_to_ban_model.panel_user_uuid:
|
|
panel_ban_api_success = await panel_service.update_user_status_on_panel(
|
|
user_to_ban_model.panel_user_uuid, enable=False)
|
|
if panel_ban_api_success:
|
|
panel_ban_message_part = _("admin_panel_ban_success_part")
|
|
else:
|
|
panel_ban_message_part = _("admin_panel_ban_fail_part")
|
|
|
|
await session.commit()
|
|
await message.answer(_("admin_user_banned_success_combined",
|
|
user_id_or_username=user_display_for_msg,
|
|
panel_status_part=panel_ban_message_part),
|
|
reply_markup=get_back_to_admin_panel_keyboard(
|
|
current_lang, i18n))
|
|
else:
|
|
await session.rollback()
|
|
await message.answer(_("admin_user_ban_failed_local_db_error"))
|
|
|
|
await state.clear()
|
|
|
|
|
|
async def unban_user_prompt_handler(callback: types.CallbackQuery,
|
|
state: FSMContext, i18n_data: dict,
|
|
settings: Settings, session: AsyncSession):
|
|
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 unban prompt.",
|
|
show_alert=True)
|
|
return
|
|
_ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs)
|
|
|
|
prompt_text = _("admin_unban_user_prompt")
|
|
try:
|
|
await callback.message.edit_text(
|
|
prompt_text,
|
|
reply_markup=get_back_to_admin_panel_keyboard(current_lang, i18n))
|
|
except Exception as e:
|
|
logging.warning(
|
|
f"Edit failed for unban_user_prompt: {e}. Sending new message.")
|
|
await callback.message.answer(
|
|
prompt_text,
|
|
reply_markup=get_back_to_admin_panel_keyboard(current_lang, i18n))
|
|
|
|
await callback.answer()
|
|
await state.set_state(AdminStates.waiting_for_user_id_to_unban)
|
|
|
|
|
|
@router.message(AdminStates.waiting_for_user_id_to_unban, F.text)
|
|
async def process_user_input_to_unban_handler(message: types.Message,
|
|
state: FSMContext,
|
|
i18n_data: dict,
|
|
settings: Settings,
|
|
panel_service: PanelApiService,
|
|
session: AsyncSession):
|
|
current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE)
|
|
i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance")
|
|
if not i18n:
|
|
await message.reply("Language service error.")
|
|
return
|
|
_ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs)
|
|
|
|
input_text = message.text.strip() if message.text else ""
|
|
user_to_unban_model = await _get_user_model_by_input(session, input_text)
|
|
|
|
if not user_to_unban_model:
|
|
await message.answer(
|
|
_("admin_user_not_found_in_bot_db", user_id=input_text))
|
|
await state.clear()
|
|
return
|
|
|
|
user_id_to_unban = user_to_unban_model.user_id
|
|
user_display_for_msg = user_to_unban_model.username or str(
|
|
user_id_to_unban)
|
|
|
|
if not user_to_unban_model.is_banned:
|
|
await message.answer(
|
|
_("admin_user_not_banned",
|
|
user_id_or_username=user_display_for_msg))
|
|
await state.clear()
|
|
return
|
|
|
|
unban_success_local = await user_dal.set_user_ban_status(
|
|
session, user_id_to_unban, False)
|
|
|
|
if unban_success_local:
|
|
panel_unban_message_part = ""
|
|
if user_to_unban_model.panel_user_uuid:
|
|
panel_unban_api_success = await panel_service.update_user_status_on_panel(
|
|
user_to_unban_model.panel_user_uuid, enable=True)
|
|
if panel_unban_api_success:
|
|
panel_unban_message_part = _("admin_panel_unban_success_part")
|
|
else:
|
|
panel_unban_message_part = _("admin_panel_unban_fail_part")
|
|
|
|
await session.commit()
|
|
await message.answer(_("admin_user_unbanned_success_combined",
|
|
user_id_or_username=user_display_for_msg,
|
|
panel_status_part=panel_unban_message_part),
|
|
reply_markup=get_back_to_admin_panel_keyboard(
|
|
current_lang, i18n))
|
|
else:
|
|
await session.rollback()
|
|
await message.answer(_("admin_user_unban_failed_local_db_error"))
|
|
|
|
await state.clear()
|
|
|
|
|
|
async def view_banned_users_handler(callback: types.CallbackQuery,
|
|
state: FSMContext, i18n_data: dict,
|
|
settings: Settings, session: AsyncSession):
|
|
await state.clear()
|
|
current_page_idx = 0
|
|
if ":" in callback.data and callback.data.count(":") == 2:
|
|
try:
|
|
current_page_idx = int(callback.data.split(":")[-1])
|
|
except ValueError:
|
|
current_page_idx = 0
|
|
|
|
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 displaying banned users.",
|
|
show_alert=True)
|
|
return
|
|
_ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs)
|
|
|
|
page_size = settings.LOGS_PAGE_SIZE
|
|
offset = current_page_idx * page_size
|
|
|
|
banned_user_models, total_banned_count = await user_dal.get_banned_users_paginated(
|
|
session, limit=page_size, offset=offset)
|
|
|
|
if total_banned_count == 0:
|
|
await callback.message.edit_text(
|
|
_("admin_no_banned_users"),
|
|
reply_markup=get_back_to_admin_panel_keyboard(current_lang, i18n))
|
|
else:
|
|
total_pages = math.ceil(total_banned_count /
|
|
page_size) if page_size > 0 else 1
|
|
await callback.message.edit_text(
|
|
text=_("admin_banned_list_title",
|
|
current_page=current_page_idx + 1,
|
|
total_pages=max(1, total_pages)),
|
|
reply_markup=get_banned_users_keyboard(banned_user_models,
|
|
current_page_idx,
|
|
total_banned_count, i18n,
|
|
current_lang, settings))
|
|
await callback.answer()
|
|
|
|
|
|
async def _show_user_card_actual(target_message: types.Message,
|
|
user_id_to_show: int,
|
|
banned_list_page_to_return: int,
|
|
i18n_data: dict, settings: Settings,
|
|
panel_service: PanelApiService,
|
|
session: AsyncSession):
|
|
current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE)
|
|
i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance")
|
|
if not i18n: return
|
|
|
|
_ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs)
|
|
|
|
user_model = await user_dal.get_user_by_id(session, user_id_to_show)
|
|
if not user_model:
|
|
await target_message.edit_text(
|
|
_("admin_user_not_found_in_bot_db", user_id=user_id_to_show),
|
|
reply_markup=get_back_to_admin_panel_keyboard(current_lang, i18n))
|
|
return
|
|
|
|
user_display_name = user_model.first_name or (f"@{user_model.username}"
|
|
if user_model.username else
|
|
f"ID: {user_id_to_show}")
|
|
|
|
sub_end_date_str = await subscription_dal.get_user_active_subscription_end_date_str(
|
|
session, user_id_to_show) or _("user_card_sub_na")
|
|
|
|
reg_date_display = user_model.registration_date.strftime(
|
|
'%Y-%m-%d %H:%M') if user_model.registration_date else "N/A"
|
|
|
|
card_text = _("user_card_info",
|
|
user_id=user_model.user_id,
|
|
username=user_model.username or "N/A",
|
|
first_name=user_model.first_name or "",
|
|
last_name=user_model.last_name or "",
|
|
language_code=user_model.language_code or "N/A",
|
|
panel_user_uuid=user_model.panel_user_uuid or "N/A",
|
|
ban_status=_("user_card_banned")
|
|
if user_model.is_banned else _("user_card_active"),
|
|
reg_date=reg_date_display,
|
|
sub_end_date=sub_end_date_str)
|
|
await target_message.edit_text(
|
|
text=
|
|
f"{_('admin_user_card_title', user_display=user_display_name)}\n\n{card_text}",
|
|
reply_markup=get_user_card_keyboard(user_id_to_show,
|
|
bool(user_model.is_banned), i18n,
|
|
current_lang,
|
|
banned_list_page_to_return),
|
|
parse_mode="HTML")
|
|
|
|
|
|
@router.callback_query(F.data.startswith("admin_user_card:"))
|
|
async def show_user_card_handler(callback: types.CallbackQuery,
|
|
state: FSMContext,
|
|
i18n_data: dict,
|
|
settings: Settings,
|
|
panel_service: PanelApiService,
|
|
session: AsyncSession,
|
|
force_user_id: Optional[int] = None,
|
|
force_page: Optional[int] = None):
|
|
await state.clear()
|
|
|
|
user_id_to_show = 0
|
|
banned_list_page_to_return = 0
|
|
|
|
if force_user_id is not None and force_page is not None:
|
|
user_id_to_show = force_user_id
|
|
banned_list_page_to_return = force_page
|
|
else:
|
|
try:
|
|
parts = callback.data.split(":")
|
|
user_id_to_show = int(parts[1])
|
|
banned_list_page_to_return = int(parts[2]) if len(parts) > 2 else 0
|
|
except (IndexError, ValueError):
|
|
await callback.answer("Invalid user card data.", show_alert=True)
|
|
return
|
|
|
|
i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance")
|
|
if not i18n or not callback.message:
|
|
await callback.answer("Error displaying user card.", show_alert=True)
|
|
return
|
|
|
|
await _show_user_card_actual(
|
|
target_message=callback.message,
|
|
user_id_to_show=user_id_to_show,
|
|
banned_list_page_to_return=banned_list_page_to_return,
|
|
i18n_data=i18n_data,
|
|
settings=settings,
|
|
panel_service=panel_service,
|
|
session=session)
|
|
if force_user_id is None:
|
|
await callback.answer()
|
|
|
|
|
|
async def _confirm_action_handler(callback: types.CallbackQuery,
|
|
i18n_data: dict, settings: Settings,
|
|
session: AsyncSession, action_type: str):
|
|
try:
|
|
|
|
_, user_id_str, page_str = callback.data.split(":")
|
|
user_id = int(user_id_str)
|
|
banned_list_page = int(page_str)
|
|
except (ValueError, IndexError):
|
|
await callback.answer("Invalid confirmation data.", show_alert=True)
|
|
return
|
|
|
|
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 confirmation.",
|
|
show_alert=True)
|
|
return
|
|
_ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs)
|
|
|
|
user_model = await user_dal.get_user_by_id(session, user_id)
|
|
user_display = (user_model.first_name or
|
|
(f"@{user_model.username}" if user_model.username else
|
|
f"ID {user_id}")) if user_model else f"ID {user_id}"
|
|
|
|
prompt_key = f"admin_confirm_{action_type}_prompt"
|
|
yes_callback = f"admin_{action_type}_do:{user_id}:{banned_list_page}"
|
|
no_callback = f"admin_user_card:{user_id}:{banned_list_page}"
|
|
|
|
await callback.message.edit_text(
|
|
text=_("admin_confirm_action_title",
|
|
action_text=_(f"{action_type}_verb_l")) + "\n\n" +
|
|
_(prompt_key, user_display=user_display, user_id=user_id),
|
|
reply_markup=get_confirmation_keyboard(yes_callback_data=yes_callback,
|
|
no_callback_data=no_callback,
|
|
i18n_instance=i18n,
|
|
lang=current_lang),
|
|
parse_mode="HTML")
|
|
await callback.answer()
|
|
|
|
|
|
@router.callback_query(F.data.startswith("admin_ban_confirm:"))
|
|
async def confirm_ban_handler(callback: types.CallbackQuery, i18n_data: dict,
|
|
settings: Settings, session: AsyncSession):
|
|
await _confirm_action_handler(callback, i18n_data, settings, session,
|
|
"ban")
|
|
|
|
|
|
@router.callback_query(F.data.startswith("admin_unban_confirm:"))
|
|
async def confirm_unban_handler(callback: types.CallbackQuery, i18n_data: dict,
|
|
settings: Settings, session: AsyncSession):
|
|
await _confirm_action_handler(callback, i18n_data, settings, session,
|
|
"unban")
|
|
|
|
|
|
async def _do_ban_unban_action_handler(callback: types.CallbackQuery,
|
|
i18n_data: dict, settings: Settings,
|
|
panel_service: PanelApiService,
|
|
session: AsyncSession,
|
|
state: FSMContext, action_type: str):
|
|
try:
|
|
|
|
_, user_id_str, page_str = callback.data.split(":")
|
|
user_id_target = int(user_id_str)
|
|
banned_list_page = int(page_str)
|
|
except (ValueError, IndexError):
|
|
await callback.answer("Invalid action data.", show_alert=True)
|
|
return
|
|
|
|
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 action.", show_alert=True)
|
|
return
|
|
_ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs)
|
|
|
|
user_model = await user_dal.get_user_by_id(session, user_id_target)
|
|
if not user_model:
|
|
await callback.answer(_("admin_user_not_found_in_bot_db",
|
|
user_id=user_id_target),
|
|
show_alert=True)
|
|
return
|
|
|
|
user_display_name = user_model.first_name or (f"@{user_model.username}"
|
|
if user_model.username else
|
|
f"ID {user_id_target}")
|
|
|
|
is_banning = action_type == "ban"
|
|
|
|
if is_banning:
|
|
if user_model.is_banned:
|
|
await callback.answer(_("admin_user_already_banned",
|
|
user_id_or_username=user_display_name),
|
|
show_alert=True)
|
|
return
|
|
if user_id_target == callback.from_user.id or user_id_target in settings.ADMIN_IDS:
|
|
await callback.answer(_("admin_cannot_ban_self_or_admin"),
|
|
show_alert=True)
|
|
return
|
|
else:
|
|
if not user_model.is_banned:
|
|
await callback.answer(_("admin_user_not_banned",
|
|
user_id_or_username=user_display_name),
|
|
show_alert=True)
|
|
return
|
|
|
|
action_success_local = await user_dal.set_user_ban_status(
|
|
session, user_id_target, is_banning)
|
|
|
|
if action_success_local:
|
|
panel_action_message = ""
|
|
if user_model.panel_user_uuid:
|
|
panel_api_success = await panel_service.update_user_status_on_panel(
|
|
user_model.panel_user_uuid, enable=not is_banning)
|
|
if not panel_api_success:
|
|
panel_action_message = _("admin_panel_status_update_fail_part")
|
|
logging.warning(
|
|
f"Panel status update failed for {action_type} of user {user_id_target} (panel: {user_model.panel_user_uuid})"
|
|
)
|
|
|
|
await session.commit()
|
|
alert_message_key = f"admin_user_{action_type}ned_from_card_alert"
|
|
await callback.answer(_(alert_message_key,
|
|
user_display=user_display_name,
|
|
user_id=user_id_target) + " " +
|
|
panel_action_message,
|
|
show_alert=False)
|
|
else:
|
|
await session.rollback()
|
|
await callback.answer(_(f"admin_user_{action_type}_failed_db_error"),
|
|
show_alert=True)
|
|
return
|
|
|
|
await show_user_card_handler(callback,
|
|
state,
|
|
i18n_data,
|
|
settings,
|
|
panel_service,
|
|
session,
|
|
force_user_id=user_id_target,
|
|
force_page=banned_list_page)
|
|
|
|
|
|
@router.callback_query(F.data.startswith("admin_ban_do:"))
|
|
async def do_ban_user_handler(callback: types.CallbackQuery, i18n_data: dict,
|
|
settings: Settings,
|
|
panel_service: PanelApiService,
|
|
session: AsyncSession, state: FSMContext):
|
|
await _do_ban_unban_action_handler(callback, i18n_data, settings,
|
|
panel_service, session, state, "ban")
|
|
|
|
|
|
@router.callback_query(F.data.startswith("admin_unban_do:"))
|
|
async def do_unban_user_handler(callback: types.CallbackQuery, i18n_data: dict,
|
|
settings: Settings,
|
|
panel_service: PanelApiService,
|
|
session: AsyncSession, state: FSMContext):
|
|
await _do_ban_unban_action_handler(callback, i18n_data, settings,
|
|
panel_service, session, state, "unban")
|
|
|
|
|
|
@router.callback_query(F.data == "admin_action:main",
|
|
AdminStates.waiting_for_user_id_to_ban)
|
|
@router.callback_query(F.data == "admin_action:main",
|
|
AdminStates.waiting_for_user_id_to_unban)
|
|
async def cancel_user_management_input_state(
|
|
callback: types.CallbackQuery,
|
|
state: FSMContext,
|
|
settings: Settings,
|
|
i18n_data: dict,
|
|
bot: Bot,
|
|
):
|
|
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 cancelling.", show_alert=True)
|
|
return
|
|
_ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs)
|
|
|
|
try:
|
|
await callback.message.edit_text(_("admin_action_cancelled_default"),
|
|
reply_markup=get_admin_panel_keyboard(
|
|
i18n, current_lang, settings))
|
|
except Exception:
|
|
await callback.message.answer(_("admin_action_cancelled_default"),
|
|
reply_markup=get_admin_panel_keyboard(
|
|
i18n, current_lang, settings))
|
|
await callback.answer(_("admin_action_cancelled_default_alert"))
|
|
await state.clear()
|