From d196801e2e60a7a47e0b498c8e488997aae84f0f Mon Sep 17 00:00:00 2001
From: Machka Pasla <161734431+machka-pasla@users.noreply.github.com>
Date: Fri, 27 Jun 2025 20:38:03 +0300
Subject: [PATCH] Add promo management and unified user lookup
---
bot/handlers/admin/common.py | 5 +
bot/handlers/admin/promo_codes.py | 146 +++++++++++++++++++++++-
bot/handlers/admin/user_management.py | 18 ++-
bot/keyboards/inline/admin_keyboards.py | 2 +
bot/services/panel_api_service.py | 22 ++++
bot/states/admin_states.py | 1 +
db/dal/promo_code_dal.py | 21 ++++
db/dal/user_dal.py | 16 +++
locales/en.json | 8 ++
locales/ru.json | 8 ++
10 files changed, 234 insertions(+), 13 deletions(-)
diff --git a/bot/handlers/admin/common.py b/bot/handlers/admin/common.py
index 9e92f11..e23df9d 100644
--- a/bot/handlers/admin/common.py
+++ b/bot/handlers/admin/common.py
@@ -24,6 +24,7 @@ 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,
):
@@ -34,6 +35,7 @@ async def 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(
@@ -73,6 +75,9 @@ async def admin_panel_actions_callback_handler(
elif action == "create_promo":
await admin_promo_handlers.create_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)
diff --git a/bot/handlers/admin/promo_codes.py b/bot/handlers/admin/promo_codes.py
index 96042ff..8080148 100644
--- a/bot/handlers/admin/promo_codes.py
+++ b/bot/handlers/admin/promo_codes.py
@@ -11,6 +11,7 @@ from db.dal import promo_code_dal
from bot.states.admin_states import AdminStates
from bot.keyboards.inline.admin_keyboards import get_back_to_admin_panel_keyboard, get_admin_panel_keyboard
+from aiogram.utils.keyboard import InlineKeyboardBuilder, InlineKeyboardButton
from bot.middlewares.i18n import JsonI18n
router = Router(name="admin_promo_codes_router")
@@ -194,8 +195,149 @@ async def view_promo_codes_handler(callback: types.CallbackQuery,
await callback.answer()
-@router.callback_query(F.data == "admin_action:main",
- AdminStates.waiting_for_promo_details)
+@router.callback_query(F.data == "admin_action:manage_promos")
+async def manage_promo_codes_handler(callback: types.CallbackQuery,
+ 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 displaying promo codes.", show_alert=True)
+ return
+ _ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs)
+
+ promo_models = await promo_code_dal.get_all_active_promo_codes(session,
+ limit=20,
+ offset=0)
+ if not promo_models:
+ await callback.message.edit_text(
+ _("admin_no_active_promos"),
+ reply_markup=get_back_to_admin_panel_keyboard(current_lang, i18n))
+ await callback.answer()
+ return
+
+ kb = InlineKeyboardBuilder()
+ for promo in promo_models:
+ kb.row(
+ InlineKeyboardButton(
+ text=promo.code,
+ callback_data=f"promo_edit:{promo.promo_code_id}"),
+ InlineKeyboardButton(
+ text=_("admin_promo_delete_button"),
+ callback_data=f"promo_delete:{promo.promo_code_id}"),
+ )
+ kb.row(
+ InlineKeyboardButton(text=_("back_to_admin_panel_button"),
+ callback_data="admin_action:main"))
+
+ await callback.message.edit_text(
+ _("admin_manage_promos_title"),
+ reply_markup=kb.as_markup())
+ await callback.answer()
+
+
+@router.callback_query(F.data.startswith("promo_edit:"))
+async def promo_edit_select_handler(callback: types.CallbackQuery, state: FSMContext,
+ i18n_data: dict, settings: Settings,
+ session: AsyncSession):
+ promo_id = int(callback.data.split(":")[1])
+ promo = await promo_code_dal.get_promo_code_by_id(session, promo_id)
+ current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE)
+ i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance")
+ if not i18n or not promo or not callback.message:
+ await callback.answer("Error", show_alert=True)
+ return
+ _ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs)
+
+ await state.update_data(edit_promo_id=promo_id)
+ await state.set_state(AdminStates.waiting_for_promo_edit_details)
+ await callback.message.edit_text(
+ _("admin_promo_edit_prompt", code=promo.code),
+ reply_markup=get_back_to_admin_panel_keyboard(current_lang, i18n),
+ parse_mode="HTML")
+ await callback.answer()
+
+
+@router.message(AdminStates.waiting_for_promo_edit_details, F.text)
+async def process_promo_edit_details(message: types.Message, state: FSMContext,
+ i18n_data: dict, settings: Settings,
+ session: AsyncSession):
+ data = await state.get_data()
+ promo_id = data.get("edit_promo_id")
+ current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE)
+ i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance")
+ if not i18n or not promo_id:
+ await message.answer("Error")
+ await state.clear()
+ return
+ _ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs)
+
+ parts = message.text.strip().split()
+ if not (3 <= len(parts) <= 4):
+ await message.answer(_("admin_promo_invalid_format"))
+ return
+ try:
+ code_str = parts[0].upper()
+ bonus = int(parts[1])
+ max_act = int(parts[2])
+ valid_until = None
+ if len(parts) == 4:
+ days = int(parts[3])
+ valid_until = datetime.now(timezone.utc) + timedelta(days=days)
+ except Exception:
+ await message.answer(_("admin_promo_invalid_format_general"))
+ return
+
+ update_data = {
+ "code": code_str,
+ "bonus_days": bonus,
+ "max_activations": max_act,
+ "valid_until": valid_until,
+ }
+ updated = await promo_code_dal.update_promo_code(session, promo_id, update_data)
+ if updated:
+ await session.commit()
+ await message.answer(
+ _("admin_promo_updated_success", code=code_str),
+ reply_markup=get_back_to_admin_panel_keyboard(current_lang, i18n),
+ parse_mode="HTML",
+ )
+ else:
+ await session.rollback()
+ await message.answer(_("admin_promo_not_found"))
+ await state.clear()
+
+
+@router.callback_query(F.data.startswith("promo_delete:"))
+async def promo_delete_handler(callback: types.CallbackQuery, i18n_data: dict,
+ settings: Settings, session: AsyncSession):
+ promo_id = int(callback.data.split(":")[1])
+ promo = await promo_code_dal.delete_promo_code(session, promo_id)
+ 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", show_alert=True)
+ return
+ _ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs)
+
+ if promo:
+ await session.commit()
+ await callback.message.edit_text(
+ _("admin_promo_deleted_success", code=promo.code),
+ reply_markup=get_back_to_admin_panel_keyboard(current_lang, i18n),
+ parse_mode="HTML")
+ else:
+ await session.rollback()
+ await callback.message.edit_text(
+ _("admin_promo_not_found"),
+ reply_markup=get_back_to_admin_panel_keyboard(current_lang, i18n))
+ await callback.answer()
+
+
+@router.callback_query(
+ F.data == "admin_action:main",
+ AdminStates.waiting_for_promo_details | AdminStates.waiting_for_promo_edit_details,
+)
async def cancel_promo_creation_state_to_menu(callback: types.CallbackQuery,
state: FSMContext,
settings: Settings,
diff --git a/bot/handlers/admin/user_management.py b/bot/handlers/admin/user_management.py
index 8e2b601..d98a3f2 100644
--- a/bot/handlers/admin/user_management.py
+++ b/bot/handlers/admin/user_management.py
@@ -28,20 +28,16 @@ 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]:
-
- user_model: Optional[User] = None
if input_text.isdigit():
try:
- user_model = await user_dal.get_user_by_id(session,
- int(input_text))
+ return await user_dal.get_user(session, user_id=int(input_text))
except ValueError:
- pass
- elif input_text.startswith("@") and USERNAME_REGEX.match(input_text[1:]):
- user_model = await user_dal.get_user_by_username(
- session, input_text[1:])
- elif USERNAME_REGEX.match(input_text):
- user_model = await user_dal.get_user_by_username(session, input_text)
- return user_model
+ 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,
diff --git a/bot/keyboards/inline/admin_keyboards.py b/bot/keyboards/inline/admin_keyboards.py
index 55fecf0..6dae708 100644
--- a/bot/keyboards/inline/admin_keyboards.py
+++ b/bot/keyboards/inline/admin_keyboards.py
@@ -18,6 +18,8 @@ def get_admin_panel_keyboard(i18n_instance, lang: str,
callback_data="admin_action:broadcast")
builder.button(text=_(key="admin_create_promo_button"),
callback_data="admin_action:create_promo")
+ builder.button(text=_(key="admin_manage_promos_button"),
+ callback_data="admin_action:manage_promos")
builder.button(text=_(key="admin_view_promos_button"),
callback_data="admin_action:view_promos")
builder.button(text=_(key="admin_ban_user_button"),
diff --git a/bot/services/panel_api_service.py b/bot/services/panel_api_service.py
index 8aab103..e5639ae 100644
--- a/bot/services/panel_api_service.py
+++ b/bot/services/panel_api_service.py
@@ -226,6 +226,28 @@ class PanelApiService:
return None
+ async def get_user(
+ self,
+ *,
+ uuid: Optional[str] = None,
+ telegram_id: Optional[int] = None,
+ username: Optional[str] = None,
+ email: Optional[str] = None,
+ log_response: bool = True,
+ ) -> Optional[Dict[str, Any]]:
+ if uuid:
+ return await self.get_user_by_uuid(uuid, log_response=log_response)
+
+ users = await self.get_users_by_filter(
+ telegram_id=telegram_id,
+ username=username,
+ email=email,
+ log_response=log_response,
+ )
+ if users:
+ return users[0]
+ return None
+
async def get_users_by_filter(
self,
telegram_id: Optional[int] = None,
diff --git a/bot/states/admin_states.py b/bot/states/admin_states.py
index 7ea9e00..854bc6c 100644
--- a/bot/states/admin_states.py
+++ b/bot/states/admin_states.py
@@ -6,6 +6,7 @@ class AdminStates(StatesGroup):
waiting_for_broadcast_message = State()
confirming_broadcast = State()
waiting_for_promo_details = State()
+ waiting_for_promo_edit_details = State()
waiting_for_user_id_to_ban = State()
waiting_for_user_id_to_unban = State()
diff --git a/db/dal/promo_code_dal.py b/db/dal/promo_code_dal.py
index 522d189..df3ea56 100644
--- a/db/dal/promo_code_dal.py
+++ b/db/dal/promo_code_dal.py
@@ -49,6 +49,27 @@ async def get_all_active_promo_codes(session: AsyncSession,
return result.scalars().all()
+async def update_promo_code(session: AsyncSession, promo_id: int,
+ update_data: Dict[str, Any]) -> Optional[PromoCode]:
+ promo = await get_promo_code_by_id(session, promo_id)
+ if not promo:
+ return None
+ for key, value in update_data.items():
+ setattr(promo, key, value)
+ await session.flush()
+ await session.refresh(promo)
+ return promo
+
+
+async def delete_promo_code(session: AsyncSession, promo_id: int) -> Optional[PromoCode]:
+ promo = await get_promo_code_by_id(session, promo_id)
+ if not promo:
+ return None
+ await session.delete(promo)
+ await session.flush()
+ return promo
+
+
async def increment_promo_code_usage(
session: AsyncSession, promo_code_id: int) -> Optional[PromoCode]:
promo = await get_promo_code_by_id(session, promo_code_id)
diff --git a/db/dal/user_dal.py b/db/dal/user_dal.py
index e6cb4be..70d3c57 100644
--- a/db/dal/user_dal.py
+++ b/db/dal/user_dal.py
@@ -30,6 +30,22 @@ 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
+
+
async def create_user(session: AsyncSession, user_data: Dict[str, Any]) -> User:
if "registration_date" not in user_data:
diff --git a/locales/en.json b/locales/en.json
index 359394f..f17bac5 100644
--- a/locales/en.json
+++ b/locales/en.json
@@ -90,6 +90,7 @@
"admin_stats_button": "📊 Statistics",
"admin_broadcast_button": "📢 Broadcast",
"admin_create_promo_button": "🎁 Create Promo",
+ "admin_manage_promos_button": "🛠 Manage Promos",
"admin_view_promos_button": "👀 Promo List",
"admin_ban_user_button": "🚫 Ban User",
"admin_unban_user_button": "✅ Unban User",
@@ -141,6 +142,13 @@
"admin_promo_list_item": "{code}: +{bonus}d, {current}/{max} used, valid until {valid_until}",
"admin_promo_valid_indefinitely": "indefinite",
"admin_promo_valid_until_display": "until {date}",
+ "admin_manage_promos_title": "Manage Promo Codes",
+ "admin_promo_edit_button": "✏️ Edit",
+ "admin_promo_delete_button": "🗑 Delete",
+ "admin_promo_edit_prompt": "Send new details for {code} in format: CODE BONUS_DAYS MAX_USES [VALIDITY_DAYS]",
+ "admin_promo_updated_success": "Promo {code} updated.",
+ "admin_promo_deleted_success": "Promo {code} deleted.",
+ "admin_promo_not_found": "Promo not found.",
"admin_ban_user_prompt": "Enter user ID or @username to ban:",
"admin_user_not_found_in_bot_db": "User {user_id} not found in bot database.",
diff --git a/locales/ru.json b/locales/ru.json
index 1250214..e08e427 100644
--- a/locales/ru.json
+++ b/locales/ru.json
@@ -90,6 +90,7 @@
"admin_stats_button": "📊 Статистика",
"admin_broadcast_button": "📢 Рассылка",
"admin_create_promo_button": "🎁 Создать промо",
+ "admin_manage_promos_button": "🛠 Управление промо",
"admin_view_promos_button": "👀 Список промо",
"admin_ban_user_button": "🚫 Забанить",
"admin_unban_user_button": "✅ Разбанить",
@@ -141,6 +142,13 @@
"admin_promo_list_item": "{code}: +{bonus}дн, {current}/{max} акт., до {valid_until}",
"admin_promo_valid_indefinitely": "бессрочно",
"admin_promo_valid_until_display": "до {date}",
+ "admin_manage_promos_title": "Управление промокодами",
+ "admin_promo_edit_button": "✏️ Изменить",
+ "admin_promo_delete_button": "🗑 Удалить",
+ "admin_promo_edit_prompt": "Отправьте новые данные для {code} в формате: КОД ДНИ_БОНУСА МАКС_АКТИВАЦИЙ [СРОК]",
+ "admin_promo_updated_success": "Промокод {code} обновлен.",
+ "admin_promo_deleted_success": "Промокод {code} удален.",
+ "admin_promo_not_found": "Промокод не найден.",
"admin_ban_user_prompt": "Введите ID или @username пользователя для блокировки:",
"admin_user_not_found_in_bot_db": "Пользователь {user_id} не найден в базе данных бота.",