austnv/remnawave-minishop
1
0
Fork 0
Code Issues 1 Pull Requests Activity

Add promo management and unified user lookup

Browse Source
This commit is contained in:
Machka Pasla
2025-06-27 20:38:03 +03:00
parent c12958e9a2
commit d196801e2e
10 changed files with 234 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bot/handlers/admin/common.py
+5
View File
@@ -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)
bot/handlers/admin/promo_codes.py
+144 -2
View File
@@ -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,
bot/handlers/admin/user_management.py
+7 -11
View File
@@ -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,
bot/keyboards/inline/admin_keyboards.py
+2
View File
@@ -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"),
bot/services/panel_api_service.py
+22
View File
@@ -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,
bot/states/admin_states.py
+1
View File
@@ -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()