machka-pasla
4f6822f8e4
- Refactored the promo code creation process into a step-by-step flow, improving user guidance and validation. - Introduced bulk promo code creation with a structured multi-step process, allowing admins to create multiple codes efficiently. - Added new states for managing the promo code creation process and bulk operations. - Implemented panel statistics retrieval to provide admins with insights into system performance and user activity. - Updated localization files to support new messages and prompts related to promo code management and statistics.
1354 lines
56 KiB
Python
1354 lines
56 KiB
Python
import logging
|
|
import random
|
|
import string
|
|
from aiogram import Router, F, types, Bot
|
|
from aiogram.filters import StateFilter
|
|
from aiogram.fsm.context import FSMContext
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Optional
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from config.settings import Settings
|
|
|
|
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")
|
|
|
|
|
|
async def create_promo_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 preparing promo creation.",
|
|
show_alert=True)
|
|
return
|
|
_ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs)
|
|
|
|
# Step 1: Ask for promo code
|
|
prompt_text = _(
|
|
"admin_promo_step1_code",
|
|
default="🎟 <b>Создание промокода</b>\n\n<b>Шаг 1 из 5:</b> Код промокода\n\nВведите код промокода (3-30 символов, только буквы и цифры):"
|
|
)
|
|
|
|
try:
|
|
await callback.message.edit_text(
|
|
prompt_text,
|
|
reply_markup=get_back_to_admin_panel_keyboard(current_lang, i18n),
|
|
parse_mode="HTML")
|
|
except Exception as e:
|
|
logging.warning(
|
|
f"Could not edit message for promo prompt: {e}. Sending new.")
|
|
await callback.message.answer(
|
|
prompt_text,
|
|
reply_markup=get_back_to_admin_panel_keyboard(current_lang, i18n),
|
|
parse_mode="HTML")
|
|
await callback.answer()
|
|
await state.set_state(AdminStates.waiting_for_promo_code)
|
|
|
|
|
|
# Step 1: Process promo code
|
|
@router.message(AdminStates.waiting_for_promo_code, F.text)
|
|
async def process_promo_code_handler(message: types.Message,
|
|
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:
|
|
await message.reply("Language service error.")
|
|
return
|
|
_ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs)
|
|
|
|
try:
|
|
code_str = message.text.strip().upper()
|
|
if not (3 <= len(code_str) <= 30 and code_str.isalnum()):
|
|
await message.answer(_(
|
|
"admin_promo_invalid_code_format",
|
|
default="❌ Код промокода должен содержать 3-30 символов (только буквы и цифры)"
|
|
))
|
|
return
|
|
|
|
# Check if code already exists
|
|
existing_promo = await promo_code_dal.get_promo_code_by_code(session, code_str)
|
|
if existing_promo:
|
|
await message.answer(_(
|
|
"admin_promo_code_already_exists",
|
|
default="❌ Промокод с таким кодом уже существует"
|
|
))
|
|
return
|
|
|
|
await state.update_data(promo_code=code_str)
|
|
|
|
# Step 2: Ask for bonus days
|
|
prompt_text = _(
|
|
"admin_promo_step2_bonus_days",
|
|
default="🎟 <b>Создание промокода</b>\n\n<b>Шаг 2 из 5:</b> Бонусные дни\n\nКод: <b>{code}</b>\n\nВведите количество бонусных дней (1-365):",
|
|
code=code_str
|
|
)
|
|
|
|
await message.answer(
|
|
prompt_text,
|
|
reply_markup=get_back_to_admin_panel_keyboard(current_lang, i18n),
|
|
parse_mode="HTML"
|
|
)
|
|
await state.set_state(AdminStates.waiting_for_promo_bonus_days)
|
|
|
|
except Exception as e:
|
|
logging.error(f"Error processing promo code: {e}")
|
|
await message.answer(_(
|
|
"admin_promo_processing_error",
|
|
default="❌ Ошибка обработки промокода"
|
|
))
|
|
|
|
# Step 2: Process bonus days
|
|
@router.message(AdminStates.waiting_for_promo_bonus_days, F.text)
|
|
async def process_promo_bonus_days_handler(message: types.Message,
|
|
state: FSMContext,
|
|
i18n_data: dict,
|
|
settings: Settings):
|
|
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)
|
|
|
|
try:
|
|
bonus_days = int(message.text.strip())
|
|
if bonus_days <= 0 or bonus_days > 365:
|
|
await message.answer(_(
|
|
"admin_promo_invalid_bonus_days",
|
|
default="❌ Бонусные дни должны быть от 1 до 365"
|
|
))
|
|
return
|
|
|
|
data = await state.get_data()
|
|
await state.update_data(bonus_days=bonus_days)
|
|
|
|
# Step 3: Ask for max activations
|
|
prompt_text = _(
|
|
"admin_promo_step3_max_activations",
|
|
default="🎟 <b>Создание промокода</b>\n\n<b>Шаг 3 из 5:</b> Максимальные активации\n\nКод: <b>{code}</b>\nБонусные дни: <b>{bonus_days}</b>\n\nВведите максимальное количество активаций (1-10000):",
|
|
code=data['promo_code'],
|
|
bonus_days=bonus_days
|
|
)
|
|
|
|
await message.answer(
|
|
prompt_text,
|
|
reply_markup=get_back_to_admin_panel_keyboard(current_lang, i18n),
|
|
parse_mode="HTML"
|
|
)
|
|
await state.set_state(AdminStates.waiting_for_promo_max_activations)
|
|
|
|
except ValueError:
|
|
await message.answer(_(
|
|
"admin_promo_invalid_bonus_days",
|
|
default="❌ Бонусные дни должны быть от 1 до 365"
|
|
))
|
|
|
|
# Step 3: Process max activations
|
|
@router.message(AdminStates.waiting_for_promo_max_activations, F.text)
|
|
async def process_promo_max_activations_handler(message: types.Message,
|
|
state: FSMContext,
|
|
i18n_data: dict,
|
|
settings: Settings):
|
|
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)
|
|
|
|
try:
|
|
max_activations = int(message.text.strip())
|
|
if max_activations <= 0 or max_activations > 10000:
|
|
await message.answer(_(
|
|
"admin_promo_invalid_max_activations",
|
|
default="❌ Максимальные активации должны быть от 1 до 10000"
|
|
))
|
|
return
|
|
|
|
data = await state.get_data()
|
|
await state.update_data(max_activations=max_activations)
|
|
|
|
# Step 4: Ask for validity days (with buttons)
|
|
keyboard = InlineKeyboardBuilder()
|
|
keyboard.row(
|
|
InlineKeyboardButton(
|
|
text=_("admin_promo_unlimited_validity", default="♾️ Неограниченно"),
|
|
callback_data="promo_unlimited_validity"
|
|
)
|
|
)
|
|
keyboard.row(
|
|
InlineKeyboardButton(
|
|
text=_("admin_promo_set_validity", default="⏰ Установить срок"),
|
|
callback_data="promo_set_validity"
|
|
)
|
|
)
|
|
keyboard.row(
|
|
InlineKeyboardButton(
|
|
text=_("admin_panel_back_button", default="⬅️ Назад"),
|
|
callback_data="admin_panel_back"
|
|
)
|
|
)
|
|
|
|
prompt_text = _(
|
|
"admin_promo_step4_validity",
|
|
default="🎟 <b>Создание промокода</b>\n\n<b>Шаг 4 из 5:</b> Срок действия\n\nКод: <b>{code}</b>\nБонусные дни: <b>{bonus_days}</b>\nМакс. активации: <b>{max_activations}</b>\n\nВыберите срок действия промокода:",
|
|
code=data['promo_code'],
|
|
bonus_days=data['bonus_days'],
|
|
max_activations=max_activations
|
|
)
|
|
|
|
await message.answer(
|
|
prompt_text,
|
|
reply_markup=keyboard.as_markup(),
|
|
parse_mode="HTML"
|
|
)
|
|
await state.set_state(AdminStates.waiting_for_promo_validity_days)
|
|
|
|
except ValueError:
|
|
await message.answer(_(
|
|
"admin_promo_invalid_max_activations",
|
|
default="❌ Максимальные активации должны быть от 1 до 10000"
|
|
))
|
|
|
|
# Handle validity choice buttons
|
|
@router.callback_query(F.data == "promo_unlimited_validity", StateFilter(AdminStates.waiting_for_promo_validity_days))
|
|
async def process_promo_unlimited_validity(callback: types.CallbackQuery,
|
|
state: FSMContext,
|
|
i18n_data: dict,
|
|
settings: Settings):
|
|
await state.update_data(validity_days=None)
|
|
await ask_for_promo_description(callback, state, i18n_data, settings)
|
|
|
|
@router.callback_query(F.data == "promo_set_validity", StateFilter(AdminStates.waiting_for_promo_validity_days))
|
|
async def process_promo_set_validity(callback: types.CallbackQuery,
|
|
state: FSMContext,
|
|
i18n_data: dict,
|
|
settings: Settings):
|
|
current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE)
|
|
i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance")
|
|
if not i18n:
|
|
await callback.answer("Language service error.", show_alert=True)
|
|
return
|
|
_ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs)
|
|
|
|
prompt_text = _(
|
|
"admin_promo_enter_validity_days",
|
|
default="⏰ Введите количество дней действия промокода (1-365):"
|
|
)
|
|
|
|
try:
|
|
await callback.message.edit_text(
|
|
prompt_text,
|
|
reply_markup=get_back_to_admin_panel_keyboard(current_lang, i18n)
|
|
)
|
|
except Exception as e:
|
|
await callback.message.answer(
|
|
prompt_text,
|
|
reply_markup=get_back_to_admin_panel_keyboard(current_lang, i18n)
|
|
)
|
|
await callback.answer()
|
|
|
|
# Step 4: Process validity days (when user chooses to set)
|
|
@router.message(AdminStates.waiting_for_promo_validity_days, F.text)
|
|
async def process_promo_validity_days_handler(message: types.Message,
|
|
state: FSMContext,
|
|
i18n_data: dict,
|
|
settings: Settings):
|
|
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)
|
|
|
|
try:
|
|
validity_days = int(message.text.strip())
|
|
if validity_days <= 0 or validity_days > 365:
|
|
await message.answer(_(
|
|
"admin_promo_invalid_validity_days",
|
|
default="❌ Дни действия должны быть от 1 до 365"
|
|
))
|
|
return
|
|
|
|
await state.update_data(validity_days=validity_days)
|
|
|
|
# Create a fake callback for consistency
|
|
fake_callback = types.CallbackQuery(
|
|
id="fake",
|
|
from_user=message.from_user,
|
|
chat_instance="fake",
|
|
message=message,
|
|
data="fake"
|
|
)
|
|
|
|
await ask_for_promo_description(fake_callback, state, i18n_data, settings)
|
|
|
|
except ValueError:
|
|
await message.answer(_(
|
|
"admin_promo_invalid_validity_days",
|
|
default="❌ Дни действия должны быть от 1 до 365"
|
|
))
|
|
|
|
async def ask_for_promo_description(callback_or_message,
|
|
state: FSMContext,
|
|
i18n_data: dict,
|
|
settings: Settings):
|
|
"""Ask for promo description (final step)"""
|
|
current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE)
|
|
i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance")
|
|
if not i18n:
|
|
if hasattr(callback_or_message, 'answer'):
|
|
await callback_or_message.answer("Language service error.", show_alert=True)
|
|
else:
|
|
await callback_or_message.reply("Language service error.")
|
|
return
|
|
_ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs)
|
|
|
|
data = await state.get_data()
|
|
validity_days = data.get('validity_days')
|
|
validity_text = "Неограниченно" if validity_days is None else f"{validity_days} дн."
|
|
|
|
# Step 5: Ask for description (optional)
|
|
keyboard = InlineKeyboardBuilder()
|
|
keyboard.row(
|
|
InlineKeyboardButton(
|
|
text=_("admin_promo_skip_description", default="⏭️ Пропустить"),
|
|
callback_data="promo_skip_description"
|
|
)
|
|
)
|
|
keyboard.row(
|
|
InlineKeyboardButton(
|
|
text=_("admin_panel_back_button", default="⬅️ Назад"),
|
|
callback_data="admin_panel_back"
|
|
)
|
|
)
|
|
|
|
prompt_text = _(
|
|
"admin_promo_step5_description",
|
|
default="🎟 <b>Создание промокода</b>\n\n<b>Шаг 5 из 5:</b> Описание (опционально)\n\nКод: <b>{code}</b>\nБонусные дни: <b>{bonus_days}</b>\nМакс. активации: <b>{max_activations}</b>\nДействует: <b>{validity}</b>\n\nВведите описание промокода или нажмите \"Пропустить\":",
|
|
code=data['promo_code'],
|
|
bonus_days=data['bonus_days'],
|
|
max_activations=data['max_activations'],
|
|
validity=validity_text
|
|
)
|
|
|
|
if hasattr(callback_or_message, 'message'):
|
|
try:
|
|
await callback_or_message.message.edit_text(
|
|
prompt_text,
|
|
reply_markup=keyboard.as_markup(),
|
|
parse_mode="HTML"
|
|
)
|
|
except:
|
|
await callback_or_message.message.answer(
|
|
prompt_text,
|
|
reply_markup=keyboard.as_markup(),
|
|
parse_mode="HTML"
|
|
)
|
|
if hasattr(callback_or_message, 'answer'):
|
|
await callback_or_message.answer()
|
|
else:
|
|
await callback_or_message.answer(
|
|
prompt_text,
|
|
reply_markup=keyboard.as_markup(),
|
|
parse_mode="HTML"
|
|
)
|
|
|
|
await state.set_state(AdminStates.waiting_for_promo_description)
|
|
|
|
# Handle skip description button
|
|
@router.callback_query(F.data == "promo_skip_description", StateFilter(AdminStates.waiting_for_promo_description))
|
|
async def process_promo_skip_description(callback: types.CallbackQuery,
|
|
state: FSMContext,
|
|
i18n_data: dict,
|
|
settings: Settings,
|
|
session: AsyncSession):
|
|
await state.update_data(description=None)
|
|
await create_promo_code_final(callback, state, i18n_data, settings, session)
|
|
|
|
# Step 5: Process description (optional)
|
|
@router.message(AdminStates.waiting_for_promo_description, F.text)
|
|
async def process_promo_description_handler(message: types.Message,
|
|
state: FSMContext,
|
|
i18n_data: dict,
|
|
settings: Settings,
|
|
session: AsyncSession):
|
|
description = message.text.strip()
|
|
if len(description) > 200:
|
|
current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE)
|
|
i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance")
|
|
if i18n:
|
|
_ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs)
|
|
await message.answer(_(
|
|
"admin_promo_description_too_long",
|
|
default="❌ Описание не должно превышать 200 символов"
|
|
))
|
|
else:
|
|
await message.answer("Description too long")
|
|
return
|
|
|
|
await state.update_data(description=description)
|
|
|
|
# Create a fake callback for consistency
|
|
fake_callback = types.CallbackQuery(
|
|
id="fake",
|
|
from_user=message.from_user,
|
|
chat_instance="fake",
|
|
message=message,
|
|
data="fake"
|
|
)
|
|
|
|
await create_promo_code_final(fake_callback, state, i18n_data, settings, session)
|
|
|
|
async def create_promo_code_final(callback_or_message,
|
|
state: FSMContext,
|
|
i18n_data: dict,
|
|
settings: Settings,
|
|
session: AsyncSession):
|
|
"""Final step to create the promo code"""
|
|
current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE)
|
|
i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance")
|
|
if not i18n:
|
|
if hasattr(callback_or_message, 'answer'):
|
|
await callback_or_message.answer("Language service error.", show_alert=True)
|
|
else:
|
|
await callback_or_message.reply("Language service error.")
|
|
return
|
|
_ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs)
|
|
|
|
data = await state.get_data()
|
|
promo_code = data['promo_code']
|
|
bonus_days = data['bonus_days']
|
|
max_activations = data['max_activations']
|
|
validity_days = data.get('validity_days')
|
|
description = data.get('description')
|
|
|
|
# Create validity date
|
|
valid_until_date: Optional[datetime] = None
|
|
valid_until_str_display = _("admin_promo_valid_indefinitely", default="Неограниченно")
|
|
|
|
if validity_days:
|
|
valid_until_date = datetime.now(timezone.utc) + timedelta(days=validity_days)
|
|
valid_until_str_display = valid_until_date.strftime('%Y-%m-%d')
|
|
|
|
# Show creating message
|
|
creating_text = _(
|
|
"admin_promo_creating",
|
|
default="⏳ Создаю промокод..."
|
|
)
|
|
|
|
if hasattr(callback_or_message, 'message'):
|
|
try:
|
|
await callback_or_message.message.edit_text(creating_text)
|
|
except:
|
|
await callback_or_message.message.answer(creating_text)
|
|
if hasattr(callback_or_message, 'answer'):
|
|
await callback_or_message.answer()
|
|
else:
|
|
await callback_or_message.answer(creating_text)
|
|
|
|
# Create promo code
|
|
try:
|
|
promo_data = {
|
|
"code": promo_code,
|
|
"bonus_days": bonus_days,
|
|
"max_activations": max_activations,
|
|
"current_activations": 0,
|
|
"is_active": True,
|
|
"valid_until": valid_until_date,
|
|
"description": description
|
|
}
|
|
|
|
await promo_code_dal.create_promo_code(session, promo_data)
|
|
await session.commit()
|
|
|
|
# Send success message
|
|
success_text = _(
|
|
"admin_promo_created_success",
|
|
default="✅ Промокод создан!\n\n🎟 Код: <b>{code}</b>\n🎁 Бонусные дни: <b>{days}</b>\n🔢 Максимальные активации: <b>{max_act}</b>\n⏰ Действует до: <b>{validity}</b>\n📝 Описание: {description}",
|
|
code=promo_code,
|
|
days=bonus_days,
|
|
max_act=max_activations,
|
|
validity=valid_until_str_display,
|
|
description=description or "Не указано"
|
|
)
|
|
|
|
if hasattr(callback_or_message, 'message'):
|
|
target_message = callback_or_message.message
|
|
else:
|
|
target_message = callback_or_message
|
|
|
|
await target_message.answer(
|
|
success_text,
|
|
reply_markup=get_back_to_admin_panel_keyboard(current_lang, i18n),
|
|
parse_mode="HTML"
|
|
)
|
|
|
|
except Exception as e:
|
|
logging.error(f"Error creating promo code: {e}")
|
|
await session.rollback()
|
|
error_text = _(
|
|
"admin_promo_creation_failed",
|
|
default="❌ Ошибка создания промокода: {error}",
|
|
error=str(e)
|
|
)
|
|
|
|
if hasattr(callback_or_message, 'message'):
|
|
target_message = callback_or_message.message
|
|
else:
|
|
target_message = callback_or_message
|
|
|
|
await target_message.answer(
|
|
error_text,
|
|
reply_markup=get_back_to_admin_panel_keyboard(current_lang, i18n)
|
|
)
|
|
|
|
await state.clear()
|
|
|
|
# Legacy handler removed - using step-by-step creation now
|
|
|
|
|
|
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:
|
|
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
|
|
|
|
response_text_parts = [f"<b>{_('admin_active_promos_list_header')}</b>\n"]
|
|
for promo in promo_models:
|
|
valid_until_display_text = _("admin_promo_valid_indefinitely")
|
|
if promo.valid_until:
|
|
|
|
valid_until_display_text = promo.valid_until.strftime('%Y-%m-%d')
|
|
|
|
response_text_parts.append(
|
|
_("admin_promo_list_item",
|
|
code=promo.code,
|
|
bonus=promo.bonus_days,
|
|
current=promo.current_activations,
|
|
max=promo.max_activations,
|
|
valid_until=valid_until_display_text))
|
|
|
|
final_text = "\n".join(response_text_parts)
|
|
try:
|
|
await callback.message.edit_text(
|
|
final_text,
|
|
reply_markup=get_back_to_admin_panel_keyboard(current_lang, i18n),
|
|
parse_mode="HTML")
|
|
except Exception as e:
|
|
logging.warning(
|
|
f"Failed to edit message for promo list: {e}. Sending new.")
|
|
if callback.message:
|
|
await callback.message.answer(
|
|
final_text,
|
|
reply_markup=get_back_to_admin_panel_keyboard(
|
|
current_lang, i18n),
|
|
parse_mode="HTML")
|
|
await callback.answer()
|
|
|
|
|
|
@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",
|
|
StateFilter(
|
|
AdminStates.waiting_for_promo_details,
|
|
AdminStates.waiting_for_promo_code,
|
|
AdminStates.waiting_for_promo_bonus_days,
|
|
AdminStates.waiting_for_promo_max_activations,
|
|
AdminStates.waiting_for_promo_validity_days,
|
|
AdminStates.waiting_for_promo_description,
|
|
AdminStates.waiting_for_promo_edit_details,
|
|
AdminStates.waiting_for_bulk_promo_details,
|
|
AdminStates.waiting_for_bulk_promo_quantity,
|
|
AdminStates.waiting_for_bulk_promo_bonus_days,
|
|
AdminStates.waiting_for_bulk_promo_max_activations,
|
|
AdminStates.waiting_for_bulk_promo_validity_days,
|
|
),
|
|
)
|
|
async def cancel_promo_creation_state_to_menu(callback: types.CallbackQuery,
|
|
state: FSMContext,
|
|
settings: Settings,
|
|
i18n_data: dict,
|
|
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 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()
|
|
|
|
|
|
async def create_bulk_promo_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 preparing bulk promo creation.",
|
|
show_alert=True)
|
|
return
|
|
_ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs)
|
|
|
|
# Step 1: Ask for quantity
|
|
prompt_text = _(
|
|
"admin_bulk_promo_step1_quantity",
|
|
default="📦 <b>Массовое создание промокодов</b>\n\n<b>Шаг 1 из 4:</b> Количество промокодов\n\nВведите количество промокодов для создания (1-1000):"
|
|
)
|
|
|
|
try:
|
|
await callback.message.edit_text(
|
|
prompt_text,
|
|
reply_markup=get_back_to_admin_panel_keyboard(current_lang, i18n),
|
|
parse_mode="HTML")
|
|
except Exception as e:
|
|
logging.warning(
|
|
f"Could not edit message for bulk promo prompt: {e}. Sending new.")
|
|
await callback.message.answer(
|
|
prompt_text,
|
|
reply_markup=get_back_to_admin_panel_keyboard(current_lang, i18n),
|
|
parse_mode="HTML")
|
|
await callback.answer()
|
|
await state.set_state(AdminStates.waiting_for_bulk_promo_quantity)
|
|
|
|
|
|
def generate_unique_promo_code(length: int = 8) -> str:
|
|
"""Generate a unique promotional code"""
|
|
characters = string.ascii_uppercase + string.digits
|
|
# Exclude confusing characters
|
|
characters = characters.replace('0', '').replace('O', '').replace('1', '').replace('I', '').replace('L', '')
|
|
return ''.join(random.choice(characters) for _ in range(length))
|
|
|
|
|
|
# Step 1: Process quantity
|
|
@router.message(AdminStates.waiting_for_bulk_promo_quantity, F.text)
|
|
async def process_bulk_promo_quantity_handler(message: types.Message,
|
|
state: FSMContext,
|
|
i18n_data: dict,
|
|
settings: Settings):
|
|
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)
|
|
|
|
try:
|
|
quantity = int(message.text.strip())
|
|
if quantity <= 0 or quantity > 1000:
|
|
await message.answer(_(
|
|
"admin_bulk_promo_invalid_quantity",
|
|
default="❌ Количество должно быть от 1 до 1000"
|
|
))
|
|
return
|
|
|
|
await state.update_data(quantity=quantity)
|
|
|
|
# Step 2: Ask for bonus days
|
|
prompt_text = _(
|
|
"admin_bulk_promo_step2_bonus_days",
|
|
default="📦 <b>Массовое создание промокодов</b>\n\n<b>Шаг 2 из 4:</b> Бонусные дни\n\nКоличество: <b>{quantity}</b>\n\nВведите количество бонусных дней (1-365):",
|
|
quantity=quantity
|
|
)
|
|
|
|
await message.answer(
|
|
prompt_text,
|
|
reply_markup=get_back_to_admin_panel_keyboard(current_lang, i18n),
|
|
parse_mode="HTML"
|
|
)
|
|
await state.set_state(AdminStates.waiting_for_bulk_promo_bonus_days)
|
|
|
|
except ValueError:
|
|
await message.answer(_(
|
|
"admin_bulk_promo_invalid_quantity",
|
|
default="❌ Количество должно быть от 1 до 1000"
|
|
))
|
|
|
|
# Step 2: Process bonus days
|
|
@router.message(AdminStates.waiting_for_bulk_promo_bonus_days, F.text)
|
|
async def process_bulk_promo_bonus_days_handler(message: types.Message,
|
|
state: FSMContext,
|
|
i18n_data: dict,
|
|
settings: Settings):
|
|
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)
|
|
|
|
try:
|
|
bonus_days = int(message.text.strip())
|
|
if bonus_days <= 0 or bonus_days > 365:
|
|
await message.answer(_(
|
|
"admin_bulk_promo_invalid_bonus_days",
|
|
default="❌ Бонусные дни должны быть от 1 до 365"
|
|
))
|
|
return
|
|
|
|
data = await state.get_data()
|
|
await state.update_data(bonus_days=bonus_days)
|
|
|
|
# Step 3: Ask for max activations
|
|
prompt_text = _(
|
|
"admin_bulk_promo_step3_max_activations",
|
|
default="📦 <b>Массовое создание промокодов</b>\n\n<b>Шаг 3 из 4:</b> Максимальные активации\n\nКоличество: <b>{quantity}</b>\nБонусные дни: <b>{bonus_days}</b>\n\nВведите максимальное количество активаций для каждого промокода (1-10000):",
|
|
quantity=data['quantity'],
|
|
bonus_days=bonus_days
|
|
)
|
|
|
|
await message.answer(
|
|
prompt_text,
|
|
reply_markup=get_back_to_admin_panel_keyboard(current_lang, i18n),
|
|
parse_mode="HTML"
|
|
)
|
|
await state.set_state(AdminStates.waiting_for_bulk_promo_max_activations)
|
|
|
|
except ValueError:
|
|
await message.answer(_(
|
|
"admin_bulk_promo_invalid_bonus_days",
|
|
default="❌ Бонусные дни должны быть от 1 до 365"
|
|
))
|
|
|
|
# Step 3: Process max activations
|
|
@router.message(AdminStates.waiting_for_bulk_promo_max_activations, F.text)
|
|
async def process_bulk_promo_max_activations_handler(message: types.Message,
|
|
state: FSMContext,
|
|
i18n_data: dict,
|
|
settings: Settings):
|
|
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)
|
|
|
|
try:
|
|
max_activations = int(message.text.strip())
|
|
if max_activations <= 0 or max_activations > 10000:
|
|
await message.answer(_(
|
|
"admin_bulk_promo_invalid_max_activations",
|
|
default="❌ Максимальные активации должны быть от 1 до 10000"
|
|
))
|
|
return
|
|
|
|
data = await state.get_data()
|
|
await state.update_data(max_activations=max_activations)
|
|
|
|
# Step 4: Ask for validity days (optional)
|
|
keyboard = InlineKeyboardBuilder()
|
|
keyboard.row(
|
|
InlineKeyboardButton(
|
|
text=_("admin_bulk_promo_unlimited_validity", default="♾️ Неограниченно"),
|
|
callback_data="bulk_promo_unlimited_validity"
|
|
)
|
|
)
|
|
keyboard.row(
|
|
InlineKeyboardButton(
|
|
text=_("admin_bulk_promo_set_validity", default="⏰ Установить срок"),
|
|
callback_data="bulk_promo_set_validity"
|
|
)
|
|
)
|
|
keyboard.row(
|
|
InlineKeyboardButton(
|
|
text=_("admin_panel_back_button", default="⬅️ Назад"),
|
|
callback_data="admin_panel_back"
|
|
)
|
|
)
|
|
|
|
prompt_text = _(
|
|
"admin_bulk_promo_step4_validity",
|
|
default="📦 <b>Массовое создание промокодов</b>\n\n<b>Шаг 4 из 4:</b> Срок действия\n\nКоличество: <b>{quantity}</b>\nБонусные дни: <b>{bonus_days}</b>\nМакс. активации: <b>{max_activations}</b>\n\nВыберите срок действия промокодов:",
|
|
quantity=data['quantity'],
|
|
bonus_days=data['bonus_days'],
|
|
max_activations=max_activations
|
|
)
|
|
|
|
await message.answer(
|
|
prompt_text,
|
|
reply_markup=keyboard.as_markup(),
|
|
parse_mode="HTML"
|
|
)
|
|
await state.set_state(AdminStates.waiting_for_bulk_promo_validity_days)
|
|
|
|
except ValueError:
|
|
await message.answer(_(
|
|
"admin_bulk_promo_invalid_max_activations",
|
|
default="❌ Максимальные активации должны быть от 1 до 10000"
|
|
))
|
|
|
|
# Handle validity choice buttons
|
|
@router.callback_query(F.data == "bulk_promo_unlimited_validity", StateFilter(AdminStates.waiting_for_bulk_promo_validity_days))
|
|
async def process_bulk_promo_unlimited_validity(callback: types.CallbackQuery,
|
|
state: FSMContext,
|
|
i18n_data: dict,
|
|
settings: Settings,
|
|
session: AsyncSession):
|
|
await state.update_data(validity_days=None)
|
|
await create_bulk_promo_codes_final(callback, state, i18n_data, settings, session)
|
|
|
|
@router.callback_query(F.data == "bulk_promo_set_validity", StateFilter(AdminStates.waiting_for_bulk_promo_validity_days))
|
|
async def process_bulk_promo_set_validity(callback: types.CallbackQuery,
|
|
state: FSMContext,
|
|
i18n_data: dict,
|
|
settings: Settings):
|
|
current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE)
|
|
i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance")
|
|
if not i18n:
|
|
await callback.answer("Language service error.", show_alert=True)
|
|
return
|
|
_ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs)
|
|
|
|
prompt_text = _(
|
|
"admin_bulk_promo_enter_validity_days",
|
|
default="⏰ Введите количество дней действия промокодов (1-365):"
|
|
)
|
|
|
|
try:
|
|
await callback.message.edit_text(
|
|
prompt_text,
|
|
reply_markup=get_back_to_admin_panel_keyboard(current_lang, i18n)
|
|
)
|
|
except Exception as e:
|
|
await callback.message.answer(
|
|
prompt_text,
|
|
reply_markup=get_back_to_admin_panel_keyboard(current_lang, i18n)
|
|
)
|
|
await callback.answer()
|
|
|
|
# Step 4: Process validity days (when user chooses to set)
|
|
@router.message(AdminStates.waiting_for_bulk_promo_validity_days, F.text)
|
|
async def process_bulk_promo_validity_days_handler(message: types.Message,
|
|
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:
|
|
await message.reply("Language service error.")
|
|
return
|
|
_ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs)
|
|
|
|
try:
|
|
validity_days = int(message.text.strip())
|
|
if validity_days <= 0 or validity_days > 365:
|
|
await message.answer(_(
|
|
"admin_bulk_promo_invalid_validity_days",
|
|
default="❌ Дни действия должны быть от 1 до 365"
|
|
))
|
|
return
|
|
|
|
await state.update_data(validity_days=validity_days)
|
|
|
|
# Create a fake callback for consistency
|
|
fake_callback = types.CallbackQuery(
|
|
id="fake",
|
|
from_user=message.from_user,
|
|
chat_instance="fake",
|
|
message=message,
|
|
data="fake"
|
|
)
|
|
|
|
await create_bulk_promo_codes_final(fake_callback, state, i18n_data, settings, session)
|
|
|
|
except ValueError:
|
|
await message.answer(_(
|
|
"admin_bulk_promo_invalid_validity_days",
|
|
default="❌ Дни действия должны быть от 1 до 365"
|
|
))
|
|
|
|
async def create_bulk_promo_codes_final(callback_or_message,
|
|
state: FSMContext,
|
|
i18n_data: dict,
|
|
settings: Settings,
|
|
session: AsyncSession):
|
|
"""Final step to create all promo codes"""
|
|
current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE)
|
|
i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance")
|
|
if not i18n:
|
|
if hasattr(callback_or_message, 'answer'):
|
|
await callback_or_message.answer("Language service error.", show_alert=True)
|
|
else:
|
|
await callback_or_message.reply("Language service error.")
|
|
return
|
|
_ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs)
|
|
|
|
data = await state.get_data()
|
|
quantity = data['quantity']
|
|
bonus_days = data['bonus_days']
|
|
max_activations = data['max_activations']
|
|
validity_days = data.get('validity_days')
|
|
|
|
# Create validity date
|
|
valid_until_date: Optional[datetime] = None
|
|
valid_until_str_display = _("admin_promo_valid_indefinitely", default="Неограниченно")
|
|
|
|
if validity_days:
|
|
valid_until_date = datetime.now(timezone.utc) + timedelta(days=validity_days)
|
|
valid_until_str_display = valid_until_date.strftime('%Y-%m-%d')
|
|
|
|
# Show creating message
|
|
creating_text = _(
|
|
"admin_bulk_promo_creating",
|
|
default="⏳ Создаю {quantity} промокодов...",
|
|
quantity=quantity
|
|
)
|
|
|
|
if hasattr(callback_or_message, 'message'):
|
|
try:
|
|
await callback_or_message.message.edit_text(creating_text)
|
|
except:
|
|
await callback_or_message.message.answer(creating_text)
|
|
if hasattr(callback_or_message, 'answer'):
|
|
await callback_or_message.answer()
|
|
else:
|
|
await callback_or_message.answer(creating_text)
|
|
|
|
# Generate and create promo codes
|
|
created_codes = []
|
|
try:
|
|
for i in range(quantity):
|
|
# Generate unique code
|
|
promo_code = generate_unique_promo_code()
|
|
|
|
# Ensure uniqueness
|
|
existing_promo = await promo_code_dal.get_promo_code_by_code(session, promo_code)
|
|
retries = 0
|
|
while existing_promo and retries < 10:
|
|
promo_code = generate_unique_promo_code()
|
|
existing_promo = await promo_code_dal.get_promo_code_by_code(session, promo_code)
|
|
retries += 1
|
|
|
|
if retries >= 10:
|
|
raise Exception("Не удалось создать уникальный промокод")
|
|
|
|
# Create promo code
|
|
promo_data = {
|
|
"code": promo_code,
|
|
"bonus_days": bonus_days,
|
|
"max_activations": max_activations,
|
|
"current_activations": 0,
|
|
"is_active": True,
|
|
"valid_until": valid_until_date,
|
|
"description": f"Bulk created #{i+1}/{quantity}"
|
|
}
|
|
|
|
await promo_code_dal.create_promo_code(session, promo_data)
|
|
created_codes.append(promo_code)
|
|
|
|
await session.commit()
|
|
|
|
# Send success message
|
|
codes_text = " ".join(created_codes)
|
|
success_text = _(
|
|
"admin_bulk_promo_created_success",
|
|
default="✅ Создано {count} промокодов на {days} дней!\n\nДействуют до: {validity}\nМакс. активации: {max_act}\n\nПромокоды:\n{codes}",
|
|
count=quantity,
|
|
days=bonus_days,
|
|
validity=valid_until_str_display,
|
|
max_act=max_activations,
|
|
codes=codes_text[:3500] + "..." if len(codes_text) > 3500 else codes_text
|
|
)
|
|
|
|
if hasattr(callback_or_message, 'message'):
|
|
target_message = callback_or_message.message
|
|
else:
|
|
target_message = callback_or_message
|
|
|
|
await target_message.answer(
|
|
success_text,
|
|
reply_markup=get_back_to_admin_panel_keyboard(current_lang, i18n),
|
|
parse_mode="HTML"
|
|
)
|
|
|
|
# Send codes as file if too many
|
|
if len(codes_text) > 3500:
|
|
codes_file_content = "\n".join(created_codes)
|
|
codes_file = types.BufferedInputFile(
|
|
codes_file_content.encode('utf-8'),
|
|
filename=f"bulk_promo_codes_{datetime.now().strftime('%Y%m%d_%H%M%S')}.txt"
|
|
)
|
|
|
|
await target_message.answer_document(
|
|
codes_file,
|
|
caption=_(
|
|
"admin_bulk_promo_codes_file",
|
|
default="📄 Все созданные промокоды в файле",
|
|
count=quantity
|
|
)
|
|
)
|
|
|
|
except Exception as e:
|
|
logging.error(f"Error creating bulk promo codes: {e}")
|
|
await session.rollback()
|
|
error_text = _(
|
|
"admin_bulk_promo_creation_failed",
|
|
default="❌ Ошибка создания промокодов: {error}",
|
|
error=str(e)
|
|
)
|
|
|
|
if hasattr(callback_or_message, 'message'):
|
|
target_message = callback_or_message.message
|
|
else:
|
|
target_message = callback_or_message
|
|
|
|
await target_message.answer(
|
|
error_text,
|
|
reply_markup=get_back_to_admin_panel_keyboard(current_lang, i18n)
|
|
)
|
|
|
|
await state.clear()
|
|
|
|
@router.message(AdminStates.waiting_for_bulk_promo_details, F.text)
|
|
async def process_bulk_promo_details_handler(message: types.Message,
|
|
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:
|
|
await message.reply("Language service error.")
|
|
return
|
|
_ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs)
|
|
|
|
if not message.text:
|
|
await message.answer(_("admin_promo_invalid_format"))
|
|
return
|
|
|
|
parts = message.text.strip().split()
|
|
if not (3 <= len(parts) <= 4):
|
|
await message.answer(_(
|
|
"admin_bulk_promo_invalid_format",
|
|
default="❌ Неверный формат. Используйте: <количество> <бонусные_дни> <максимальное_использование> [дни_действия]"
|
|
))
|
|
return
|
|
|
|
try:
|
|
count = int(parts[0])
|
|
bonus_days = int(parts[1])
|
|
max_activations = int(parts[2])
|
|
|
|
if count <= 0 or count > 1000:
|
|
raise ValueError(_("admin_bulk_promo_invalid_count", default="Количество должно быть от 1 до 1000"))
|
|
|
|
if bonus_days <= 0 or max_activations <= 0:
|
|
raise ValueError(_("admin_promo_invalid_bonus_or_activations"))
|
|
|
|
valid_until_date: Optional[datetime] = None
|
|
valid_until_str_display = _("admin_promo_valid_indefinitely")
|
|
|
|
if len(parts) == 4:
|
|
valid_days_from_now = int(parts[3])
|
|
if valid_days_from_now <= 0:
|
|
raise ValueError(_("admin_promo_invalid_validity_days"))
|
|
valid_until_date = datetime.now(
|
|
timezone.utc) + timedelta(days=valid_days_from_now)
|
|
valid_until_str_display = _(
|
|
"admin_promo_valid_until_display",
|
|
date=valid_until_date.strftime('%Y-%m-%d'))
|
|
|
|
except ValueError as e:
|
|
await message.answer(_(
|
|
"admin_bulk_promo_invalid_values",
|
|
default="❌ Неверные значения: {error}",
|
|
error=str(e)
|
|
))
|
|
return
|
|
except Exception as e_parse:
|
|
logging.error(f"Error parsing bulk promo details '{message.text}': {e_parse}")
|
|
await message.answer(_("admin_promo_invalid_format_general"))
|
|
return
|
|
|
|
admin_id = message.from_user.id if message.from_user else 0
|
|
|
|
# Generate unique codes and create promo codes
|
|
created_codes = []
|
|
failed_codes = []
|
|
|
|
await message.answer(_(
|
|
"admin_bulk_promo_creating",
|
|
default="🔄 Создаю {count} промокодов...",
|
|
count=count
|
|
))
|
|
|
|
for i in range(count):
|
|
try:
|
|
# Generate unique code
|
|
attempts = 0
|
|
while attempts < 10: # Max 10 attempts to generate unique code
|
|
code = generate_unique_promo_code()
|
|
|
|
# Check if code already exists
|
|
existing_promo = await promo_code_dal.get_promo_code_by_code(session, code)
|
|
if not existing_promo:
|
|
break
|
|
attempts += 1
|
|
|
|
if attempts >= 10:
|
|
failed_codes.append(f"Failed to generate unique code #{i+1}")
|
|
continue
|
|
|
|
promo_data_to_create = {
|
|
"code": code,
|
|
"bonus_days": bonus_days,
|
|
"max_activations": max_activations,
|
|
"created_by_admin_id": admin_id,
|
|
"valid_until": valid_until_date,
|
|
"is_active": True,
|
|
"current_activations": 0
|
|
}
|
|
|
|
created_promo = await promo_code_dal.create_promo_code(
|
|
session, promo_data_to_create)
|
|
|
|
if created_promo:
|
|
created_codes.append(created_promo.code)
|
|
else:
|
|
failed_codes.append(f"Failed to create code #{i+1}")
|
|
|
|
except Exception as e:
|
|
logging.error(f"Error creating bulk promo code #{i+1}: {e}")
|
|
failed_codes.append(f"Error creating code #{i+1}: {str(e)}")
|
|
|
|
try:
|
|
await session.commit()
|
|
|
|
# Prepare success message
|
|
success_text_parts = [
|
|
_(
|
|
"admin_bulk_promo_created_success",
|
|
default="✅ Массовое создание завершено!\n\n📦 Создано промокодов: {created_count}\n💎 Бонусных дней: {bonus_days}\n🔄 Макс. активаций каждого: {max_activations}\n⏰ Действительны до: {valid_until}",
|
|
created_count=len(created_codes),
|
|
bonus_days=bonus_days,
|
|
max_activations=max_activations,
|
|
valid_until=valid_until_str_display
|
|
)
|
|
]
|
|
|
|
if failed_codes:
|
|
success_text_parts.append(f"\n❌ Ошибок: {len(failed_codes)}")
|
|
|
|
if created_codes:
|
|
# Show first few codes as examples
|
|
codes_to_show = created_codes[:10] # Show first 10
|
|
success_text_parts.append(f"\n📝 Примеры созданных кодов:")
|
|
success_text_parts.append("\n".join([f"• {code}" for code in codes_to_show]))
|
|
|
|
if len(created_codes) > 10:
|
|
success_text_parts.append(f"... и еще {len(created_codes) - 10} кодов")
|
|
|
|
# Send codes as a file if many were created
|
|
if len(created_codes) > 20:
|
|
try:
|
|
codes_text = "\n".join(created_codes)
|
|
codes_file = types.BufferedInputFile(
|
|
codes_text.encode('utf-8'),
|
|
filename=f"promo_codes_{datetime.now().strftime('%Y%m%d_%H%M%S')}.txt"
|
|
)
|
|
await message.answer_document(
|
|
codes_file,
|
|
caption=_(
|
|
"admin_bulk_promo_codes_file",
|
|
default="📄 Файл со всеми созданными промокодами"
|
|
)
|
|
)
|
|
except Exception as e:
|
|
logging.error(f"Failed to send codes file: {e}")
|
|
|
|
final_text = "\n".join(success_text_parts)
|
|
await message.answer(
|
|
final_text,
|
|
reply_markup=get_back_to_admin_panel_keyboard(current_lang, i18n))
|
|
|
|
except Exception as e_db_commit:
|
|
await session.rollback()
|
|
logging.error(f"Failed to commit bulk promo codes creation: {e_db_commit}", exc_info=True)
|
|
await message.answer(_(
|
|
"admin_bulk_promo_creation_failed",
|
|
default="❌ Ошибка при сохранении промокодов в базу данных"
|
|
))
|
|
|
|
await state.clear()
|
|
|
|
|
|
|