diff --git a/bot/handlers/admin/promo/manage.py b/bot/handlers/admin/promo/manage.py
index 97a5620..9d7c4f5 100644
--- a/bot/handlers/admin/promo/manage.py
+++ b/bot/handlers/admin/promo/manage.py
@@ -1,4 +1,6 @@
import logging
+import csv
+import io
from aiogram import Router, F, types
from aiogram.filters import StateFilter
from aiogram.fsm.context import FSMContext
@@ -17,10 +19,45 @@ from bot.middlewares.i18n import JsonI18n
router = Router(name="promo_manage_router")
-async def view_promo_codes_handler(callback: types.CallbackQuery,
- i18n_data: dict, settings: Settings,
- session: AsyncSession):
- """View all active promo codes"""
+async def get_promo_detail_text_and_keyboard(promo_id: int, session: AsyncSession, i18n: JsonI18n, current_lang: str):
+ _ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs)
+ promo = await promo_code_dal.get_promo_code_by_id(session, promo_id)
+ 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)
i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance")
if not i18n or not callback.message:
@@ -28,53 +65,19 @@ async def view_promo_codes_handler(callback: types.CallbackQuery,
return
_ = 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)
-
- if not promo_models:
- text = f"{_('admin_active_promos_list_header')}\n\n{_('admin_no_active_promos')}"
- else:
- lines = [_("admin_active_promos_list_header"), ""]
-
- 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"🎟 {promo.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)
+ promo_models = await promo_code_dal.get_all_active_promo_codes(session, limit=20, offset=0)
+ text = f"{_('admin_active_promos_list_header')}\n\n{_('admin_no_active_promos')}" if not promo_models else "\n".join(
+ [_("admin_active_promos_list_header"), ""] + [
+ f"🎟 {p.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
+ ]
+ )
+
+ await callback.message.edit_text(text, reply_markup=get_back_to_admin_panel_keyboard(current_lang, i18n), parse_mode="HTML")
+ await callback.answer()
-async def promo_management_handler(callback: types.CallbackQuery,
- i18n_data: dict, settings: Settings,
- session: AsyncSession):
- """Main promo management interface"""
+async def promo_management_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:
@@ -82,301 +85,251 @@ async def promo_management_handler(callback: types.CallbackQuery,
return
_ = 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)
-
- 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"
- )
+ 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")
await callback.answer()
-
- except Exception as e:
- logging.error(f"Error in promo management: {e}")
- await callback.answer(_("error_occurred_try_again"), show_alert=True)
+ return
+
+ builder = InlineKeyboardBuilder()
+ 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:"))
-async def promo_detail_handler(callback: types.CallbackQuery,
- 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)
+async def promo_detail_handler(callback: types.CallbackQuery, i18n_data: dict, session: AsyncSession):
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)
return
- _ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs)
-
+
try:
promo_id = int(callback.data.split(":")[1])
- promo = await promo_code_dal.get_promo_code_by_id(session, promo_id)
-
- if not promo:
- await callback.answer(_("admin_promo_not_found"), show_alert=True)
- return
-
- 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)
- ])
-
- # 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)
+ text, keyboard = await get_promo_detail_text_and_keyboard(promo_id, session, i18n, current_lang)
+ if text:
+ await callback.message.edit_text(text, reply_markup=keyboard, parse_mode="HTML")
+ else:
+ await callback.answer(i18n.gettext(current_lang, "admin_promo_not_found"), show_alert=True)
+ except (ValueError, IndexError):
+ await callback.answer(i18n.gettext(current_lang, "admin_promo_not_found"), show_alert=True)
+ await callback.answer()
@router.callback_query(F.data.startswith("promo_toggle:"))
-async def promo_toggle_handler(callback: types.CallbackQuery, i18n_data: dict,
- settings: Settings, session: AsyncSession):
- """Toggle promo code active status"""
- current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE)
+async def promo_toggle_handler(callback: types.CallbackQuery, i18n_data: dict, session: AsyncSession):
i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance")
- if not i18n:
- await callback.answer("Language service error.", show_alert=True)
- return
+ current_lang = i18n_data.get("current_language")
+ if not i18n or not callback.message or not current_lang:
+ return await callback.answer("Language service error.", show_alert=True)
_ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs)
-
+
try:
promo_id = int(callback.data.split(":")[1])
promo = await promo_code_dal.get_promo_code_by_id(session, promo_id)
-
if not promo:
- await callback.answer(_("admin_promo_not_found"), show_alert=True)
- return
-
+ return await callback.answer(_("admin_promo_not_found"), show_alert=True)
+
new_status = not promo.is_active
- update_data = {"is_active": new_status}
-
- updated = await promo_code_dal.update_promo_code(session, promo_id, update_data)
- if updated:
+ if await promo_code_dal.update_promo_code(session, promo_id, {"is_active": new_status}):
await session.commit()
status_text = _("admin_promo_status_activated") if new_status else _("admin_promo_status_deactivated")
- await callback.answer(
- _("admin_promo_toggle_success", code=promo.code, status=status_text)
- )
- # Refresh the detail view
- callback.data = f"promo_detail:{promo_id}"
- await promo_detail_handler(callback, i18n_data, settings, session)
+ await callback.answer(_("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)
+ if text:
+ await callback.message.edit_text(text, reply_markup=keyboard, parse_mode="HTML")
else:
await callback.answer(_("error_occurred_try_again"), show_alert=True)
-
- except ValueError:
+ except (ValueError, IndexError):
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:"))
-async def promo_activations_handler(callback: types.CallbackQuery, i18n_data: dict,
- settings: Settings, session: AsyncSession):
- """Show promo code activations"""
- current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE)
+async def promo_activations_handler(callback: types.CallbackQuery, i18n_data: dict, settings: Settings, session: AsyncSession):
i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance")
- if not i18n or not callback.message:
- await callback.answer("Error processing request.", show_alert=True)
- return
+ 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)
+
+ 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)
try:
promo_id = int(callback.data.split(":")[1])
promo = await promo_code_dal.get_promo_code_by_id(session, promo_id)
-
if not promo:
- await callback.answer(_("admin_promo_not_found"), show_alert=True)
- return
-
+ return await callback.answer(_("admin_promo_not_found"), show_alert=True)
+
activations = await promo_code_dal.get_promo_activations_by_code_id(session, promo_id)
-
if not activations:
- text = _("admin_promo_no_activations", code=promo.code)
- else:
- text_lines = [
- _("admin_promo_activations_header", code=promo.code)
- ]
- for activation in activations[:20]:
- text_lines.append(
- _("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)
+ return await callback.answer(_("admin_promo_no_activations", code=promo.code), show_alert=True)
+
+ output = io.StringIO()
+ writer = csv.writer(output)
+ writer.writerow(["User ID", "Activation Date"])
+ for act in activations:
+ writer.writerow([act.user_id, act.activated_at.strftime("%Y-%m-%d %H:%M:%S")])
- builder = InlineKeyboardBuilder()
- builder.row(
- InlineKeyboardButton(
- text=_("admin_promo_back_to_detail_button"),
- callback_data=f"promo_detail:{promo_id}"
- )
- )
-
- 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:
+ output.seek(0)
+ file = types.BufferedInputFile(output.getvalue().encode('utf-8'), filename=f"promo_{promo.code}_activations.csv")
+ await callback.message.answer_document(file, caption=_("admin_promo_export_caption", code=promo.code))
+
+ except (ValueError, IndexError):
await callback.answer(_("admin_promo_not_found"), show_alert=True)
- except Exception as e:
- logging.error(f"Error viewing activations: {e}")
- await callback.answer(_("error_occurred_try_again"), show_alert=True)
+ await callback.answer()
@router.callback_query(F.data.startswith("promo_delete:"))
-async def promo_delete_handler(callback: types.CallbackQuery, i18n_data: dict,
- settings: Settings, session: AsyncSession):
- """Delete promo code"""
- current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE)
+async def promo_delete_handler(callback: types.CallbackQuery, i18n_data: dict, session: AsyncSession):
i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance")
- if not i18n:
- await callback.answer("Language service error.", show_alert=True)
- return
+ current_lang = i18n_data.get("current_language")
+ if not i18n or not callback.message or not current_lang:
+ return await callback.answer("Language service error.", show_alert=True)
_ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs)
try:
promo_id = int(callback.data.split(":")[1])
promo = await promo_code_dal.delete_promo_code(session, promo_id)
-
if promo:
await session.commit()
- await callback.answer(
- _("admin_promo_deleted_success", code=promo.code)
- )
- # Go back to management
- callback.data = "admin_action:promo_management"
- await promo_management_handler(callback, i18n_data, settings, session)
+ await callback.answer(_("admin_promo_deleted_success", code=promo.code), show_alert=True)
+ await promo_management_handler(callback, i18n_data, {}, session) # Settings not needed here
else:
await callback.answer(_("admin_promo_not_found"), show_alert=True)
-
- except ValueError:
+ except (ValueError, IndexError):
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
-async def manage_promo_codes_handler(callback: types.CallbackQuery,
- i18n_data: dict, settings: Settings,
- session: AsyncSession):
- """Redirect to new unified handler"""
+# --- Promo Edit Handlers ---
+@router.callback_query(F.data.startswith("promo_edit_select:"))
+async def promo_edit_select_handler(callback: types.CallbackQuery, 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)
+ 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)
diff --git a/db/dal/promo_code_dal.py b/db/dal/promo_code_dal.py
index 5d1f017..8d2f964 100644
--- a/db/dal/promo_code_dal.py
+++ b/db/dal/promo_code_dal.py
@@ -10,7 +10,6 @@ from db.models import PromoCode, PromoCodeActivation, User, Payment
async def create_promo_code(session: AsyncSession,
promo_data: Dict[str, Any]) -> PromoCode:
-
new_promo = PromoCode(**promo_data)
session.add(new_promo)
await session.flush()
@@ -65,24 +64,25 @@ async def get_all_promo_codes_with_details(session: AsyncSession, limit: int = 5
return result.scalars().all()
-async def get_promo_code_by_id(session: AsyncSession, promo_id: int) -> Optional[PromoCode]:
- """Get promo code by ID"""
- 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
+async def get_promo_activations_by_code_id(session: AsyncSession, promo_code_id: int, limit: Optional[int] = None, offset: int = 0) -> List[PromoCodeActivation]:
+ """Get activation history for a specific promo code with optional pagination."""
stmt = (select(PromoCodeActivation)
.where(PromoCodeActivation.promo_code_id == promo_code_id)
.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)
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,
update_data: Dict[str, Any]) -> Optional[PromoCode]:
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)
if not promo:
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.flush()
return promo
@@ -124,7 +129,6 @@ async def increment_promo_code_usage(
async def get_user_activation_for_promo(
session: AsyncSession, promo_code_id: int,
user_id: int) -> Optional[PromoCodeActivation]:
-
stmt = select(PromoCodeActivation).where(
PromoCodeActivation.promo_code_id == promo_code_id,
PromoCodeActivation.user_id == user_id).limit(1)
@@ -137,7 +141,6 @@ async def record_promo_activation(
promo_code_id: int,
user_id: int,
payment_id: Optional[int] = None) -> Optional[PromoCodeActivation]:
-
existing_activation = await get_user_activation_for_promo(
session, promo_code_id, user_id)
if existing_activation:
@@ -162,7 +165,6 @@ async def record_promo_activation(
logging.error(
f"Cannot record promo activation: Payment {payment_id} not found."
)
-
return None
activation_data = {
diff --git a/locales/en.json b/locales/en.json
index 6b8405c..738369e 100644
--- a/locales/en.json
+++ b/locales/en.json
@@ -161,9 +161,20 @@
"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_updated_success": "Promo {code} updated.",
+ "admin_promo_deleted_success": "Promo {code} deleted.",
"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_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 9f50ed9..9c75ebc 100644
--- a/locales/ru.json
+++ b/locales/ru.json
@@ -161,9 +161,20 @@
"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_updated_success": "Промокод {code} обновлен.",
+ "admin_promo_deleted_success": "Промокод {code} удален.",
"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_user_not_found_in_bot_db": "Пользователь {user_id} не найден в базе данных бота.",