Enhance promo code management in admin panel with detailed views and export functionality

- Introduced a new method to retrieve detailed promo code information, improving the clarity of promo management.
- Added functionality to export promo code activations to CSV, enhancing data accessibility for admins.
- Streamlined the handling of promo code edits and improved user prompts for better interaction.
- Updated localization files to reflect new messages and ensure consistency across the admin panel.
This commit is contained in:
machka-pasla
2025-08-06 01:21:47 +03:00
parent 67f31d62e4
commit 4675141974
4 changed files with 288 additions and 311 deletions
+245 -292
View File
@@ -1,4 +1,6 @@
import logging import logging
import csv
import io
from aiogram import Router, F, types from aiogram import Router, F, types
from aiogram.filters import StateFilter from aiogram.filters import StateFilter
from aiogram.fsm.context import FSMContext from aiogram.fsm.context import FSMContext
@@ -17,10 +19,45 @@ from bot.middlewares.i18n import JsonI18n
router = Router(name="promo_manage_router") router = Router(name="promo_manage_router")
async def view_promo_codes_handler(callback: types.CallbackQuery, async def get_promo_detail_text_and_keyboard(promo_id: int, session: AsyncSession, i18n: JsonI18n, current_lang: str):
i18n_data: dict, settings: Settings, _ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs)
session: AsyncSession): promo = await promo_code_dal.get_promo_code_by_id(session, promo_id)
"""View all active promo codes""" if not promo:
return None, None
status = _("admin_promo_status_active") if promo.is_active else _("admin_promo_status_inactive")
if promo.valid_until and promo.valid_until < datetime.now(timezone.utc):
status = _("admin_promo_status_expired")
elif promo.current_activations >= promo.max_activations:
status = _("admin_promo_status_used_up")
validity = _("admin_promo_valid_indefinitely")
if promo.valid_until:
validity = promo.valid_until.strftime("%d.%m.%Y %H:%M")
created = promo.created_at.strftime("%d.%m.%Y %H:%M") if promo.created_at else "N/A"
text = "\n".join([
_("admin_promo_card_title", code=promo.code),
_("admin_promo_card_bonus_days", days=promo.bonus_days),
_("admin_promo_card_activations", current=promo.current_activations, max=promo.max_activations),
_("admin_promo_card_validity", validity=validity),
_("admin_promo_card_status", status=status),
_("admin_promo_card_created", created=created),
_("admin_promo_card_created_by", creator=promo.created_by_admin_id)
])
builder = InlineKeyboardBuilder()
builder.row(InlineKeyboardButton(text=_("admin_promo_edit_button"), callback_data=f"promo_edit_select:{promo_id}"))
builder.row(InlineKeyboardButton(text=_("admin_promo_toggle_status_button"), callback_data=f"promo_toggle:{promo_id}"))
builder.row(InlineKeyboardButton(text=_("admin_promo_view_activations_button"), callback_data=f"promo_activations:{promo_id}:0"))
builder.row(InlineKeyboardButton(text=_("admin_promo_delete_button"), callback_data=f"promo_delete:{promo_id}"))
builder.row(InlineKeyboardButton(text=_("admin_promo_back_to_list_button"), callback_data="admin_action:promo_management"))
return text, builder.as_markup()
async def view_promo_codes_handler(callback: types.CallbackQuery, i18n_data: dict, settings: Settings, session: AsyncSession):
current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE) current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE)
i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance") i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance")
if not i18n or not callback.message: if not i18n or not callback.message:
@@ -28,53 +65,19 @@ async def view_promo_codes_handler(callback: types.CallbackQuery,
return return
_ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs) _ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs)
try: promo_models = await promo_code_dal.get_all_active_promo_codes(session, limit=20, offset=0)
promo_models = await promo_code_dal.get_all_active_promo_codes(session, text = f"{_('admin_active_promos_list_header')}\n\n{_('admin_no_active_promos')}" if not promo_models else "\n".join(
limit=20, [_("admin_active_promos_list_header"), ""] + [
offset=0) f"🎟 <code>{p.code}</code> | 🎁 {p.bonus_days}д | 📊 {p.current_activations}/{p.max_activations} | ⏰ {p.valid_until.strftime('%d.%m.%Y') if p.valid_until else _('admin_promo_valid_indefinitely')}"
for p in promo_models
if not promo_models: ]
text = f"{_('admin_active_promos_list_header')}\n\n{_('admin_no_active_promos')}" )
else:
lines = [_("admin_active_promos_list_header"), ""] await callback.message.edit_text(text, reply_markup=get_back_to_admin_panel_keyboard(current_lang, i18n), parse_mode="HTML")
await callback.answer()
for promo in promo_models:
validity_str = _("admin_promo_valid_indefinitely")
if promo.valid_until:
validity_str = promo.valid_until.strftime("%d.%m.%Y")
lines.append(
f"🎟 <code>{promo.code}</code> | "
f"🎁 {promo.bonus_days}д | "
f"📊 {promo.current_activations}/{promo.max_activations} | "
f"{validity_str}"
)
text = "\n".join(lines)
try:
await callback.message.edit_text(
text,
reply_markup=get_back_to_admin_panel_keyboard(current_lang, i18n),
parse_mode="HTML"
)
except Exception:
await callback.message.answer(
text,
reply_markup=get_back_to_admin_panel_keyboard(current_lang, i18n),
parse_mode="HTML"
)
await callback.answer()
except Exception as e:
logging.error(f"Error viewing promo codes: {e}")
await callback.answer(_("error_occurred_try_again"), show_alert=True)
async def promo_management_handler(callback: types.CallbackQuery, async def promo_management_handler(callback: types.CallbackQuery, i18n_data: dict, settings: Settings, session: AsyncSession):
i18n_data: dict, settings: Settings,
session: AsyncSession):
"""Main promo management interface"""
current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE) current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE)
i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance") i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance")
if not i18n or not callback.message: if not i18n or not callback.message:
@@ -82,301 +85,251 @@ async def promo_management_handler(callback: types.CallbackQuery,
return return
_ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs) _ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs)
try: promo_models = await promo_code_dal.get_all_promo_codes_with_details(session, limit=50, offset=0)
promo_models = await promo_code_dal.get_all_promo_codes_with_details(session, limit=50, offset=0) if not promo_models:
await callback.message.edit_text(_("admin_promo_management_empty"), reply_markup=get_back_to_admin_panel_keyboard(current_lang, i18n), parse_mode="HTML")
if not promo_models:
text = _("admin_promo_management_empty")
await callback.message.edit_text(
text,
reply_markup=get_back_to_admin_panel_keyboard(current_lang, i18n),
parse_mode="HTML"
)
await callback.answer()
return
builder = InlineKeyboardBuilder()
text = _("admin_promo_management_title")
for promo in promo_models:
builder.row(
InlineKeyboardButton(
text=f"📝 {promo.code}",
callback_data=f"promo_detail:{promo.promo_code_id}"
)
)
builder.row(
InlineKeyboardButton(
text=_("back_to_admin_panel_button"),
callback_data="admin_action:main"
)
)
try:
await callback.message.edit_text(
text,
reply_markup=builder.as_markup(),
parse_mode="HTML"
)
except Exception:
await callback.message.answer(
text,
reply_markup=builder.as_markup(),
parse_mode="HTML"
)
await callback.answer() await callback.answer()
return
except Exception as e:
logging.error(f"Error in promo management: {e}") builder = InlineKeyboardBuilder()
await callback.answer(_("error_occurred_try_again"), show_alert=True) for promo in promo_models:
builder.row(InlineKeyboardButton(text=f"📝 {promo.code}", callback_data=f"promo_detail:{promo.promo_code_id}"))
builder.row(InlineKeyboardButton(text=_("back_to_admin_panel_button"), callback_data="admin_action:main"))
await callback.message.edit_text(_("admin_promo_management_title"), reply_markup=builder.as_markup(), parse_mode="HTML")
await callback.answer()
@router.callback_query(F.data.startswith("promo_detail:")) @router.callback_query(F.data.startswith("promo_detail:"))
async def promo_detail_handler(callback: types.CallbackQuery, async def promo_detail_handler(callback: types.CallbackQuery, i18n_data: dict, session: AsyncSession):
i18n_data: dict, settings: Settings,
session: AsyncSession):
"""Show detailed promo code information with management options"""
current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE)
i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance") i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance")
if not i18n or not callback.message: current_lang = i18n_data.get("current_language")
if not i18n or not callback.message or not current_lang:
await callback.answer("Error processing request.", show_alert=True) await callback.answer("Error processing request.", show_alert=True)
return return
_ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs)
try: try:
promo_id = int(callback.data.split(":")[1]) promo_id = int(callback.data.split(":")[1])
promo = await promo_code_dal.get_promo_code_by_id(session, promo_id) text, keyboard = await get_promo_detail_text_and_keyboard(promo_id, session, i18n, current_lang)
if text:
if not promo: await callback.message.edit_text(text, reply_markup=keyboard, parse_mode="HTML")
await callback.answer(_("admin_promo_not_found"), show_alert=True) else:
return await callback.answer(i18n.gettext(current_lang, "admin_promo_not_found"), show_alert=True)
except (ValueError, IndexError):
status = _("admin_promo_status_active") if promo.is_active else _("admin_promo_status_inactive") await callback.answer(i18n.gettext(current_lang, "admin_promo_not_found"), show_alert=True)
if promo.valid_until and promo.valid_until < datetime.now(timezone.utc): await callback.answer()
status = _("admin_promo_status_expired")
elif promo.current_activations >= promo.max_activations:
status = _("admin_promo_status_used_up")
validity = _("admin_promo_valid_indefinitely")
if promo.valid_until:
validity = promo.valid_until.strftime("%d.%m.%Y %H:%M")
created = promo.created_at.strftime("%d.%m.%Y %H:%M") if promo.created_at else "N/A"
text = "\n".join([
_("admin_promo_card_title", code=promo.code),
_("admin_promo_card_bonus_days", days=promo.bonus_days),
_("admin_promo_card_activations", current=promo.current_activations, max=promo.max_activations),
_("admin_promo_card_validity", validity=validity),
_("admin_promo_card_status", status=status),
_("admin_promo_card_created", created=created),
_("admin_promo_card_created_by", creator=promo.created_by_admin_id)
])
# Create management buttons
builder = InlineKeyboardBuilder()
builder.row(
InlineKeyboardButton(
text=_("admin_promo_edit_button"),
callback_data=f"promo_edit:{promo_id}"
)
)
builder.row(
InlineKeyboardButton(
text=_("admin_promo_toggle_status_button"),
callback_data=f"promo_toggle:{promo_id}"
)
)
builder.row(
InlineKeyboardButton(
text=_("admin_promo_view_activations_button"),
callback_data=f"promo_activations:{promo_id}"
)
)
builder.row(
InlineKeyboardButton(
text=_("admin_promo_delete_button"),
callback_data=f"promo_delete:{promo_id}"
)
)
builder.row(
InlineKeyboardButton(
text=_("admin_promo_back_to_list_button"),
callback_data="admin_action:promo_management"
)
)
try:
await callback.message.edit_text(
text,
reply_markup=builder.as_markup(),
parse_mode="HTML"
)
except Exception:
await callback.message.answer(
text,
reply_markup=builder.as_markup(),
parse_mode="HTML"
)
await callback.answer()
except ValueError:
await callback.answer(_("admin_promo_not_found"), show_alert=True)
except Exception as e:
logging.error(f"Error in promo detail: {e}")
await callback.answer(_("error_occurred_try_again"), show_alert=True)
@router.callback_query(F.data.startswith("promo_toggle:")) @router.callback_query(F.data.startswith("promo_toggle:"))
async def promo_toggle_handler(callback: types.CallbackQuery, i18n_data: dict, async def promo_toggle_handler(callback: types.CallbackQuery, i18n_data: dict, session: AsyncSession):
settings: Settings, session: AsyncSession):
"""Toggle promo code active status"""
current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE)
i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance") i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance")
if not i18n: current_lang = i18n_data.get("current_language")
await callback.answer("Language service error.", show_alert=True) if not i18n or not callback.message or not current_lang:
return return await callback.answer("Language service error.", show_alert=True)
_ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs) _ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs)
try: try:
promo_id = int(callback.data.split(":")[1]) promo_id = int(callback.data.split(":")[1])
promo = await promo_code_dal.get_promo_code_by_id(session, promo_id) promo = await promo_code_dal.get_promo_code_by_id(session, promo_id)
if not promo: if not promo:
await callback.answer(_("admin_promo_not_found"), show_alert=True) return await callback.answer(_("admin_promo_not_found"), show_alert=True)
return
new_status = not promo.is_active new_status = not promo.is_active
update_data = {"is_active": new_status} if await promo_code_dal.update_promo_code(session, promo_id, {"is_active": new_status}):
updated = await promo_code_dal.update_promo_code(session, promo_id, update_data)
if updated:
await session.commit() await session.commit()
status_text = _("admin_promo_status_activated") if new_status else _("admin_promo_status_deactivated") status_text = _("admin_promo_status_activated") if new_status else _("admin_promo_status_deactivated")
await callback.answer( await callback.answer(_("admin_promo_toggle_success", code=promo.code, status=status_text))
_("admin_promo_toggle_success", code=promo.code, status=status_text)
) text, keyboard = await get_promo_detail_text_and_keyboard(promo_id, session, i18n, current_lang)
# Refresh the detail view if text:
callback.data = f"promo_detail:{promo_id}" await callback.message.edit_text(text, reply_markup=keyboard, parse_mode="HTML")
await promo_detail_handler(callback, i18n_data, settings, session)
else: else:
await callback.answer(_("error_occurred_try_again"), show_alert=True) await callback.answer(_("error_occurred_try_again"), show_alert=True)
except (ValueError, IndexError):
except ValueError:
await callback.answer(_("admin_promo_not_found"), show_alert=True) await callback.answer(_("admin_promo_not_found"), show_alert=True)
except Exception as e:
logging.error(f"Error toggling promo: {e}")
await callback.answer(_("error_occurred_try_again"), show_alert=True)
@router.callback_query(F.data.startswith("promo_activations:")) @router.callback_query(F.data.startswith("promo_activations:"))
async def promo_activations_handler(callback: types.CallbackQuery, i18n_data: dict, async def promo_activations_handler(callback: types.CallbackQuery, i18n_data: dict, settings: Settings, session: AsyncSession):
settings: Settings, session: AsyncSession):
"""Show promo code activations"""
current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE)
i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance") i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance")
if not i18n or not callback.message: current_lang = i18n_data.get("current_language")
await callback.answer("Error processing request.", show_alert=True) if not i18n or not callback.message or not current_lang:
return return await callback.answer("Error processing request.", show_alert=True)
_ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs)
try:
parts = callback.data.split(":")
promo_id = int(parts[1])
page = int(parts[2])
page_size = settings.LOGS_PAGE_SIZE
promo = await promo_code_dal.get_promo_code_by_id(session, promo_id)
if not promo:
return await callback.answer(_("admin_promo_not_found"), show_alert=True)
total_activations = await promo_code_dal.count_promo_activations_by_code_id(session, promo_id)
activations = await promo_code_dal.get_promo_activations_by_code_id(session, promo_id, limit=page_size, offset=page * page_size)
builder = InlineKeyboardBuilder()
if not activations:
text = _("admin_promo_no_activations", code=promo.code)
else:
text = _("admin_promo_activations_header", code=promo.code) + "\n\n"
text += "\n".join([_("admin_promo_activation_item", user_id=a.user_id, date=a.activated_at.strftime("%d.%m.%Y %H:%M")) for a in activations])
nav_buttons = []
if page > 0:
nav_buttons.append(InlineKeyboardButton(text="⬅️", callback_data=f"promo_activations:{promo_id}:{page-1}"))
if (page + 1) * page_size < total_activations:
nav_buttons.append(InlineKeyboardButton(text="➡️", callback_data=f"promo_activations:{promo_id}:{page+1}"))
if nav_buttons:
builder.row(*nav_buttons)
builder.row(InlineKeyboardButton(text=_("admin_promo_export_csv_button"), callback_data=f"promo_export:{promo_id}"))
builder.row(InlineKeyboardButton(text=_("admin_promo_back_to_detail_button"), callback_data=f"promo_detail:{promo_id}"))
await callback.message.edit_text(text, reply_markup=builder.as_markup(), parse_mode="HTML")
except (ValueError, IndexError):
await callback.answer(_("admin_promo_not_found"), show_alert=True)
await callback.answer()
@router.callback_query(F.data.startswith("promo_export:"))
async def promo_export_activations_handler(callback: types.CallbackQuery, i18n_data: dict, session: AsyncSession):
i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance")
current_lang = i18n_data.get("current_language")
if not i18n or not callback.message or not current_lang:
return await callback.answer("Error processing request.", show_alert=True)
_ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs) _ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs)
try: try:
promo_id = int(callback.data.split(":")[1]) promo_id = int(callback.data.split(":")[1])
promo = await promo_code_dal.get_promo_code_by_id(session, promo_id) promo = await promo_code_dal.get_promo_code_by_id(session, promo_id)
if not promo: if not promo:
await callback.answer(_("admin_promo_not_found"), show_alert=True) return await callback.answer(_("admin_promo_not_found"), show_alert=True)
return
activations = await promo_code_dal.get_promo_activations_by_code_id(session, promo_id) activations = await promo_code_dal.get_promo_activations_by_code_id(session, promo_id)
if not activations: if not activations:
text = _("admin_promo_no_activations", code=promo.code) return await callback.answer(_("admin_promo_no_activations", code=promo.code), show_alert=True)
else:
text_lines = [ output = io.StringIO()
_("admin_promo_activations_header", code=promo.code) writer = csv.writer(output)
] writer.writerow(["User ID", "Activation Date"])
for activation in activations[:20]: for act in activations:
text_lines.append( writer.writerow([act.user_id, act.activated_at.strftime("%Y-%m-%d %H:%M:%S")])
_("admin_promo_activation_item",
user_id=activation.user_id,
date=activation.activated_at.strftime("%d.%m.%Y %H:%M"))
)
if len(activations) > 20:
text_lines.append(f"\n... (еще {len(activations) - 20})")
text = "\n".join(text_lines)
builder = InlineKeyboardBuilder() output.seek(0)
builder.row( file = types.BufferedInputFile(output.getvalue().encode('utf-8'), filename=f"promo_{promo.code}_activations.csv")
InlineKeyboardButton( await callback.message.answer_document(file, caption=_("admin_promo_export_caption", code=promo.code))
text=_("admin_promo_back_to_detail_button"),
callback_data=f"promo_detail:{promo_id}" except (ValueError, IndexError):
)
)
try:
await callback.message.edit_text(
text,
reply_markup=builder.as_markup(),
parse_mode="HTML"
)
except Exception:
await callback.message.answer(
text,
reply_markup=builder.as_markup(),
parse_mode="HTML"
)
await callback.answer()
except ValueError:
await callback.answer(_("admin_promo_not_found"), show_alert=True) await callback.answer(_("admin_promo_not_found"), show_alert=True)
except Exception as e: await callback.answer()
logging.error(f"Error viewing activations: {e}")
await callback.answer(_("error_occurred_try_again"), show_alert=True)
@router.callback_query(F.data.startswith("promo_delete:")) @router.callback_query(F.data.startswith("promo_delete:"))
async def promo_delete_handler(callback: types.CallbackQuery, i18n_data: dict, async def promo_delete_handler(callback: types.CallbackQuery, i18n_data: dict, session: AsyncSession):
settings: Settings, session: AsyncSession):
"""Delete promo code"""
current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE)
i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance") i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance")
if not i18n: current_lang = i18n_data.get("current_language")
await callback.answer("Language service error.", show_alert=True) if not i18n or not callback.message or not current_lang:
return return await callback.answer("Language service error.", show_alert=True)
_ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs) _ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs)
try: try:
promo_id = int(callback.data.split(":")[1]) promo_id = int(callback.data.split(":")[1])
promo = await promo_code_dal.delete_promo_code(session, promo_id) promo = await promo_code_dal.delete_promo_code(session, promo_id)
if promo: if promo:
await session.commit() await session.commit()
await callback.answer( await callback.answer(_("admin_promo_deleted_success", code=promo.code), show_alert=True)
_("admin_promo_deleted_success", code=promo.code) await promo_management_handler(callback, i18n_data, {}, session) # Settings not needed here
)
# Go back to management
callback.data = "admin_action:promo_management"
await promo_management_handler(callback, i18n_data, settings, session)
else: else:
await callback.answer(_("admin_promo_not_found"), show_alert=True) await callback.answer(_("admin_promo_not_found"), show_alert=True)
except (ValueError, IndexError):
except ValueError:
await callback.answer(_("admin_promo_not_found"), show_alert=True) await callback.answer(_("admin_promo_not_found"), show_alert=True)
except Exception as e:
logging.error(f"Error deleting promo: {e}")
await callback.answer(_("error_occurred_try_again"), show_alert=True)
# Legacy handlers that redirect to new system # --- Promo Edit Handlers ---
async def manage_promo_codes_handler(callback: types.CallbackQuery, @router.callback_query(F.data.startswith("promo_edit_select:"))
i18n_data: dict, settings: Settings, async def promo_edit_select_handler(callback: types.CallbackQuery, i18n_data: dict):
session: AsyncSession): i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance")
"""Redirect to new unified handler""" current_lang = i18n_data.get("current_language")
if not i18n or not callback.message or not current_lang:
return
_ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs)
promo_id = int(callback.data.split(":")[1])
builder = InlineKeyboardBuilder()
builder.row(InlineKeyboardButton(text=_("admin_promo_edit_bonus_days"), callback_data=f"promo_edit_field:bonus_days:{promo_id}"))
builder.row(InlineKeyboardButton(text=_("admin_promo_edit_max_activations"), callback_data=f"promo_edit_field:max_activations:{promo_id}"))
builder.row(InlineKeyboardButton(text=_("admin_promo_edit_validity"), callback_data=f"promo_edit_field:valid_until:{promo_id}"))
builder.row(InlineKeyboardButton(text=_("admin_promo_back_to_detail_button"), callback_data=f"promo_detail:{promo_id}"))
await callback.message.edit_text(_("admin_promo_edit_select_field"), reply_markup=builder.as_markup())
await callback.answer()
@router.callback_query(F.data.startswith("promo_edit_field:"))
async def promo_edit_field_handler(callback: types.CallbackQuery, state: FSMContext, i18n_data: dict):
i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance")
current_lang = i18n_data.get("current_language")
if not i18n or not callback.message or not current_lang: return
_ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs)
_, field, promo_id_str = callback.data.split(":")
await state.update_data(promo_id=int(promo_id_str), field_to_edit=field)
prompts = {
"bonus_days": "admin_promo_prompt_bonus_days",
"max_activations": "admin_promo_prompt_max_activations",
"valid_until": "admin_promo_prompt_validity_days"
}
await state.set_state(AdminStates.waiting_for_promo_edit_details)
await callback.message.edit_text(_(prompts.get(field, "error_occurred_try_again")))
await callback.answer()
@router.message(StateFilter(AdminStates.waiting_for_promo_edit_details))
async def process_promo_edit_details(message: types.Message, state: FSMContext, session: AsyncSession, i18n_data: dict):
i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance")
current_lang = i18n_data.get("current_language")
if not i18n or not message or not current_lang: return
_ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs)
data = await state.get_data()
promo_id = data.get("promo_id")
field = data.get("field_to_edit")
try:
value = message.text
update_data = {}
if field == "bonus_days":
update_data["bonus_days"] = int(value)
elif field == "max_activations":
update_data["max_activations"] = int(value)
elif field == "valid_until":
if value.lower() in ['0', 'вечно', 'бессрочно', 'indefinite']:
update_data["valid_until"] = None
else:
days = int(value)
update_data["valid_until"] = datetime.now(timezone.utc) + timedelta(days=days)
if await promo_code_dal.update_promo_code(session, promo_id, update_data):
await session.commit()
await message.answer(_("admin_promo_edit_success"))
# Reset state and show updated details
await state.clear()
text, keyboard = await get_promo_detail_text_and_keyboard(promo_id, session, i18n, current_lang)
if text:
await message.answer(text, reply_markup=keyboard, parse_mode="HTML")
else:
await message.answer(_("error_occurred_try_again"))
await state.clear()
except (ValueError, TypeError):
await message.answer(_("admin_promo_invalid_input"))
# Don't clear state, let them try again
async def manage_promo_codes_handler(callback: types.CallbackQuery, i18n_data: dict, settings: Settings, session: AsyncSession):
await promo_management_handler(callback, i18n_data, settings, session) await promo_management_handler(callback, i18n_data, settings, session)
+17 -15
View File
@@ -10,7 +10,6 @@ from db.models import PromoCode, PromoCodeActivation, User, Payment
async def create_promo_code(session: AsyncSession, async def create_promo_code(session: AsyncSession,
promo_data: Dict[str, Any]) -> PromoCode: promo_data: Dict[str, Any]) -> PromoCode:
new_promo = PromoCode(**promo_data) new_promo = PromoCode(**promo_data)
session.add(new_promo) session.add(new_promo)
await session.flush() await session.flush()
@@ -65,24 +64,25 @@ async def get_all_promo_codes_with_details(session: AsyncSession, limit: int = 5
return result.scalars().all() return result.scalars().all()
async def get_promo_code_by_id(session: AsyncSession, promo_id: int) -> Optional[PromoCode]: async def get_promo_activations_by_code_id(session: AsyncSession, promo_code_id: int, limit: Optional[int] = None, offset: int = 0) -> List[PromoCodeActivation]:
"""Get promo code by ID""" """Get activation history for a specific promo code with optional pagination."""
stmt = select(PromoCode).where(PromoCode.promo_code_id == promo_id)
result = await session.execute(stmt)
return result.scalar_one_or_none()
async def get_promo_activations_by_code_id(session: AsyncSession, promo_code_id: int) -> List:
"""Get activation history for a specific promo code"""
from db.models import PromoCodeActivation
stmt = (select(PromoCodeActivation) stmt = (select(PromoCodeActivation)
.where(PromoCodeActivation.promo_code_id == promo_code_id) .where(PromoCodeActivation.promo_code_id == promo_code_id)
.order_by(PromoCodeActivation.activated_at.desc()) .order_by(PromoCodeActivation.activated_at.desc())
.limit(50)) # Limit to last 50 activations .offset(offset))
if limit is not None:
stmt = stmt.limit(limit)
result = await session.execute(stmt) result = await session.execute(stmt)
return result.scalars().all() return result.scalars().all()
async def count_promo_activations_by_code_id(session: AsyncSession, promo_code_id: int) -> int:
"""Count total activations for a specific promo code."""
stmt = select(func.count()).select_from(PromoCodeActivation).where(PromoCodeActivation.promo_code_id == promo_code_id)
result = await session.execute(stmt)
return result.scalar_one()
async def update_promo_code(session: AsyncSession, promo_id: int, async def update_promo_code(session: AsyncSession, promo_id: int,
update_data: Dict[str, Any]) -> Optional[PromoCode]: update_data: Dict[str, Any]) -> Optional[PromoCode]:
promo = await get_promo_code_by_id(session, promo_id) promo = await get_promo_code_by_id(session, promo_id)
@@ -99,6 +99,11 @@ async def delete_promo_code(session: AsyncSession, promo_id: int) -> Optional[Pr
promo = await get_promo_code_by_id(session, promo_id) promo = await get_promo_code_by_id(session, promo_id)
if not promo: if not promo:
return None return None
# First, delete related activations due to foreign key constraint
activations = await get_promo_activations_by_code_id(session, promo_id)
for activation in activations:
await session.delete(activation)
await session.delete(promo) await session.delete(promo)
await session.flush() await session.flush()
return promo return promo
@@ -124,7 +129,6 @@ async def increment_promo_code_usage(
async def get_user_activation_for_promo( async def get_user_activation_for_promo(
session: AsyncSession, promo_code_id: int, session: AsyncSession, promo_code_id: int,
user_id: int) -> Optional[PromoCodeActivation]: user_id: int) -> Optional[PromoCodeActivation]:
stmt = select(PromoCodeActivation).where( stmt = select(PromoCodeActivation).where(
PromoCodeActivation.promo_code_id == promo_code_id, PromoCodeActivation.promo_code_id == promo_code_id,
PromoCodeActivation.user_id == user_id).limit(1) PromoCodeActivation.user_id == user_id).limit(1)
@@ -137,7 +141,6 @@ async def record_promo_activation(
promo_code_id: int, promo_code_id: int,
user_id: int, user_id: int,
payment_id: Optional[int] = None) -> Optional[PromoCodeActivation]: payment_id: Optional[int] = None) -> Optional[PromoCodeActivation]:
existing_activation = await get_user_activation_for_promo( existing_activation = await get_user_activation_for_promo(
session, promo_code_id, user_id) session, promo_code_id, user_id)
if existing_activation: if existing_activation:
@@ -162,7 +165,6 @@ async def record_promo_activation(
logging.error( logging.error(
f"Cannot record promo activation: Payment {payment_id} not found." f"Cannot record promo activation: Payment {payment_id} not found."
) )
return None return None
activation_data = { activation_data = {
+13 -2
View File
@@ -161,9 +161,20 @@
"admin_promo_edit_button": "✏️ Edit", "admin_promo_edit_button": "✏️ Edit",
"admin_promo_delete_button": "🗑 Delete", "admin_promo_delete_button": "🗑 Delete",
"admin_promo_edit_prompt": "Send new details for <code>{code}</code> in format: CODE BONUS_DAYS MAX_USES [VALIDITY_DAYS]", "admin_promo_edit_prompt": "Send new details for <code>{code}</code> in format: CODE BONUS_DAYS MAX_USES [VALIDITY_DAYS]",
"admin_promo_updated_success": "Promo <code>{code}</code> updated.", "admin_promo_updated_success": "Promo {code} updated.",
"admin_promo_deleted_success": "Promo <code>{code}</code> deleted.", "admin_promo_deleted_success": "Promo {code} deleted.",
"admin_promo_not_found": "Promo not found.", "admin_promo_not_found": "Promo not found.",
"admin_promo_export_csv_button": "📄 Export to CSV",
"admin_promo_export_caption": "📄 Activations for promo code {code}",
"admin_promo_edit_select_field": "Select a field to edit:",
"admin_promo_prompt_bonus_days": "Enter the new number of bonus days:",
"admin_promo_prompt_max_activations": "Enter the new maximum number of activations:",
"admin_promo_prompt_validity_days": "Enter the new validity period in days (0 for indefinite):",
"admin_promo_edit_success": "Promo code updated successfully.",
"admin_promo_invalid_input": "Invalid input, please try again.",
"admin_promo_edit_bonus_days": "🎁 Bonus Days",
"admin_promo_edit_max_activations": "🔢 Max Activations",
"admin_promo_edit_validity": "⏰ Validity",
"admin_ban_user_prompt": "Enter user ID or @username to ban:", "admin_ban_user_prompt": "Enter user ID or @username to ban:",
"admin_user_not_found_in_bot_db": "User <code>{user_id}</code> not found in bot database.", "admin_user_not_found_in_bot_db": "User <code>{user_id}</code> not found in bot database.",
+13 -2
View File
@@ -161,9 +161,20 @@
"admin_promo_edit_button": "✏️ Изменить", "admin_promo_edit_button": "✏️ Изменить",
"admin_promo_delete_button": "🗑 Удалить", "admin_promo_delete_button": "🗑 Удалить",
"admin_promo_edit_prompt": "Отправьте новые данные для <code>{code}</code> в формате: КОД ДНИ_БОНУСА МАКС_АКТИВАЦИЙ [СРОК]", "admin_promo_edit_prompt": "Отправьте новые данные для <code>{code}</code> в формате: КОД ДНИ_БОНУСА МАКС_АКТИВАЦИЙ [СРОК]",
"admin_promo_updated_success": "Промокод <code>{code}</code> обновлен.", "admin_promo_updated_success": "Промокод {code} обновлен.",
"admin_promo_deleted_success": "Промокод <code>{code}</code> удален.", "admin_promo_deleted_success": "Промокод {code} удален.",
"admin_promo_not_found": "Промокод не найден.", "admin_promo_not_found": "Промокод не найден.",
"admin_promo_export_csv_button": "📄 Экспорт в CSV",
"admin_promo_export_caption": "📄 Активации промокода {code}",
"admin_promo_edit_select_field": "Выберите поле для редактирования:",
"admin_promo_prompt_bonus_days": "Введите новое количество бонусных дней:",
"admin_promo_prompt_max_activations": "Введите новое максимальное количество активаций:",
"admin_promo_prompt_validity_days": "Введите новый срок действия в днях (0 для бессрочного):",
"admin_promo_edit_success": "Промокод успешно обновлен.",
"admin_promo_invalid_input": "Неверный ввод, попробуйте еще раз.",
"admin_promo_edit_bonus_days": "🎁 Бонусные дни",
"admin_promo_edit_max_activations": "🔢 Макс. активации",
"admin_promo_edit_validity": "⏰ Срок действия",
"admin_ban_user_prompt": "Введите ID или @username пользователя для блокировки:", "admin_ban_user_prompt": "Введите ID или @username пользователя для блокировки:",
"admin_user_not_found_in_bot_db": "Пользователь <code>{user_id}</code> не найден в базе данных бота.", "admin_user_not_found_in_bot_db": "Пользователь <code>{user_id}</code> не найден в базе данных бота.",