from typing import Any
from aiogram import F, Router
from aiogram.enums import ParseMode
from aiogram.fsm.context import FSMContext
from aiogram.fsm.state import State, StatesGroup
from aiogram.types import (
CallbackQuery,
InlineQuery,
InlineQueryResultArticle,
InputTextMessageContent,
Message,
)
from aiogram.utils.keyboard import InlineKeyboardBuilder
from config import INLINE_MODE, USERNAME_BOT
from database import create_coupon, delete_coupon, get_all_coupons
from filters.admin import IsAdminFilter
from handlers.buttons import BACK
from handlers.utils import format_days
from logger import logger
from ..panel.keyboard import AdminPanelCallback, build_admin_back_kb
from .keyboard import (
AdminCouponDeleteCallback,
build_coupons_kb,
build_coupons_list_kb,
format_coupons_list,
)
router = Router()
class AdminCouponsState(StatesGroup):
waiting_for_coupon_type = State()
waiting_for_balance_data = State()
waiting_for_days_data = State()
@router.callback_query(
AdminPanelCallback.filter(F.action == "coupons"),
IsAdminFilter(),
)
async def handle_coupons(callback_query: CallbackQuery):
await callback_query.message.edit_text(
text="π ΠΠ΅Π½Ρ ΡΠΏΡΠ°Π²Π»Π΅Π½ΠΈΡ ΠΊΡΠΏΠΎΠ½Π°ΠΌΠΈ:", reply_markup=build_coupons_kb()
)
@router.callback_query(
AdminPanelCallback.filter(F.action == "coupons_create"),
IsAdminFilter(),
)
async def handle_coupons_create(callback_query: CallbackQuery, state: FSMContext):
text = "π« ΠΡΠ±Π΅ΡΠΈΡΠ΅ ΡΠΈΠΏ ΠΊΡΠΏΠΎΠ½Π°:"
kb = InlineKeyboardBuilder()
kb.button(text="π° ΠΠ°Π»Π°Π½Ρ", callback_data="coupon_type_balance")
kb.button(text="β³ ΠΡΠ΅ΠΌΡ", callback_data="coupon_type_days")
kb.button(text=BACK, callback_data=AdminPanelCallback(action="coupons").pack())
kb.adjust(1)
await callback_query.message.edit_text(text=text, reply_markup=kb.as_markup())
await state.set_state(AdminCouponsState.waiting_for_coupon_type)
@router.callback_query(F.data == "coupon_type_balance")
async def handle_balance_coupon_selection(
callback_query: CallbackQuery, state: FSMContext
):
text = (
"π« ΠΠ²Π΅Π΄ΠΈΡΠ΅ Π΄Π°Π½Π½ΡΠ΅ Π΄Π»Ρ ΡΠΎΠ·Π΄Π°Π½ΠΈΡ ΠΊΡΠΏΠΎΠ½Π° Π² ΡΠΎΡΠΌΠ°ΡΠ΅:\n\n"
"π ΠΊΠΎΠ΄ π° ΡΡΠΌΠΌΠ° π’ Π»ΠΈΠΌΠΈΡ\n\n"
"ΠΡΠΈΠΌΠ΅Ρ: 'COUPON1 50 5' π\n\n"
)
kb = InlineKeyboardBuilder()
kb.button(text=BACK, callback_data=AdminPanelCallback(action="coupons").pack())
await callback_query.message.edit_text(text=text, reply_markup=kb.as_markup())
await state.set_state(AdminCouponsState.waiting_for_balance_data)
@router.callback_query(F.data == "coupon_type_days")
async def handle_days_coupon_selection(
callback_query: CallbackQuery, state: FSMContext
):
text = (
"π« ΠΠ²Π΅Π΄ΠΈΡΠ΅ Π΄Π°Π½Π½ΡΠ΅ Π΄Π»Ρ ΡΠΎΠ·Π΄Π°Π½ΠΈΡ ΠΊΡΠΏΠΎΠ½Π° Π² ΡΠΎΡΠΌΠ°ΡΠ΅:\n\n"
"π ΠΊΠΎΠ΄ β³ Π΄Π½ΠΈ π’ Π»ΠΈΠΌΠΈΡ\n\n"
"ΠΡΠΈΠΌΠ΅Ρ: 'DAYS10 10 50' π\n\n"
)
kb = InlineKeyboardBuilder()
kb.button(text=BACK, callback_data=AdminPanelCallback(action="coupons").pack())
await callback_query.message.edit_text(text=text, reply_markup=kb.as_markup())
await state.set_state(AdminCouponsState.waiting_for_days_data)
@router.message(AdminCouponsState.waiting_for_balance_data, IsAdminFilter())
async def handle_balance_coupon_input(
message: Message, state: FSMContext, session: Any
):
text = message.text.strip()
parts = text.split()
kb = InlineKeyboardBuilder()
kb.button(text=BACK, callback_data=AdminPanelCallback(action="coupons").pack())
if len(parts) != 3:
text = (
"β ΠΠ΅ΠΊΠΎΡΡΠ΅ΠΊΡΠ½ΡΠΉ ΡΠΎΡΠΌΠ°Ρ! π ΠΠΎΠΆΠ°Π»ΡΠΉΡΡΠ°, Π²Π²Π΅Π΄ΠΈΡΠ΅ Π΄Π°Π½Π½ΡΠ΅ Π² ΡΠΎΡΠΌΠ°ΡΠ΅:\n"
"π·οΈ ΠΊΠΎΠ΄ π° ΡΡΠΌΠΌΠ° π’ Π»ΠΈΠΌΠΈΡ\n"
"ΠΡΠΈΠΌΠ΅Ρ: 'COUPON1 50 5' π"
)
await message.answer(text=text, reply_markup=kb.as_markup())
return
try:
coupon_code = parts[0]
coupon_amount = int(parts[1])
usage_limit = int(parts[2])
if coupon_amount <= 0:
raise ValueError("Π‘ΡΠΌΠΌΠ° Π΄ΠΎΠ»ΠΆΠ½Π° Π±ΡΡΡ Π±ΠΎΠ»ΡΡΠ΅ 0")
except ValueError:
text = "β οΈ ΠΡΠΎΠ²Π΅ΡΡΡΠ΅ ΠΏΡΠ°Π²ΠΈΠ»ΡΠ½ΠΎΡΡΡ Π²Π²Π΅Π΄Π΅Π½Π½ΡΡ
Π΄Π°Π½Π½ΡΡ
!\nπ± Π‘ΡΠΌΠΌΠ° Π΄ΠΎΠ»ΠΆΠ½Π° Π±ΡΡΡ ΡΠΈΡΠ»ΠΎΠΌ, Π° Π»ΠΈΠΌΠΈΡ β ΡΠ΅Π»ΡΠΌ ΡΠΈΡΠ»ΠΎΠΌ."
await message.answer(text=text, reply_markup=kb.as_markup())
return
try:
await create_coupon(session, coupon_code, coupon_amount, usage_limit, days=None)
coupon_link = f"https://t.me/{USERNAME_BOT}?start=coupons_{coupon_code}"
text = (
f"β
ΠΡΠΏΠΎΠ½ Ρ ΠΊΠΎΠ΄ΠΎΠΌ {coupon_code} ΡΡΠΏΠ΅ΡΠ½ΠΎ ΡΠΎΠ·Π΄Π°Π½!\n"
f"π° Π‘ΡΠΌΠΌΠ°: {coupon_amount} ΡΡΠ±Π»Π΅ΠΉ\n"
f"π’ ΠΠΈΠΌΠΈΡ ΠΈΡΠΏΠΎΠ»ΡΠ·ΠΎΠ²Π°Π½ΠΈΡ: {usage_limit} ΡΠ°Π·\n"
f"π Π‘ΡΡΠ»ΠΊΠ°: {coupon_link}\n"
)
kb = InlineKeyboardBuilder()
if INLINE_MODE:
kb.button(text="π€ ΠΠΎΠ΄Π΅Π»ΠΈΡΡΡΡ", switch_inline_query=f"coupon_{coupon_code}")
kb.button(text=BACK, callback_data=AdminPanelCallback(action="coupons").pack())
kb.adjust(1)
await message.answer(text=text, reply_markup=kb.as_markup())
await state.clear()
except Exception as e:
logger.error(f"ΠΡΠΈΠ±ΠΊΠ° ΠΏΡΠΈ ΡΠΎΠ·Π΄Π°Π½ΠΈΠΈ ΠΊΡΠΏΠΎΠ½Π°: {e}")
await message.answer(
"β ΠΡΠΎΠΈΠ·ΠΎΡΠ»Π° ΠΎΡΠΈΠ±ΠΊΠ° ΠΏΡΠΈ ΡΠΎΠ·Π΄Π°Π½ΠΈΠΈ ΠΊΡΠΏΠΎΠ½Π°.", reply_markup=kb.as_markup()
)
@router.message(AdminCouponsState.waiting_for_days_data, IsAdminFilter())
async def handle_days_coupon_input(message: Message, state: FSMContext, session: Any):
text = message.text.strip()
parts = text.split()
kb = InlineKeyboardBuilder()
kb.button(text=BACK, callback_data=AdminPanelCallback(action="coupons").pack())
if len(parts) != 3:
text = (
"β ΠΠ΅ΠΊΠΎΡΡΠ΅ΠΊΡΠ½ΡΠΉ ΡΠΎΡΠΌΠ°Ρ! π ΠΠΎΠΆΠ°Π»ΡΠΉΡΡΠ°, Π²Π²Π΅Π΄ΠΈΡΠ΅ Π΄Π°Π½Π½ΡΠ΅ Π² ΡΠΎΡΠΌΠ°ΡΠ΅:\n"
"π·οΈ ΠΊΠΎΠ΄ β³ Π΄Π½ΠΈ π’ Π»ΠΈΠΌΠΈΡ\n"
"ΠΡΠΈΠΌΠ΅Ρ: 'DAYS10 10 50' π"
)
await message.answer(text=text, reply_markup=kb.as_markup())
return
try:
coupon_code = parts[0]
days = int(parts[1])
usage_limit = int(parts[2])
if days <= 0:
raise ValueError("ΠΠΎΠ»ΠΈΡΠ΅ΡΡΠ²ΠΎ Π΄Π½Π΅ΠΉ Π΄ΠΎΠ»ΠΆΠ½ΠΎ Π±ΡΡΡ Π±ΠΎΠ»ΡΡΠ΅ 0")
except ValueError:
text = "β οΈ ΠΡΠΎΠ²Π΅ΡΡΡΠ΅ ΠΏΡΠ°Π²ΠΈΠ»ΡΠ½ΠΎΡΡΡ Π²Π²Π΅Π΄Π΅Π½Π½ΡΡ
Π΄Π°Π½Π½ΡΡ
!\nπ± ΠΠ½ΠΈ Π΄ΠΎΠ»ΠΆΠ½Ρ Π±ΡΡΡ ΡΠΈΡΠ»ΠΎΠΌ, Π° Π»ΠΈΠΌΠΈΡ β ΡΠ΅Π»ΡΠΌ ΡΠΈΡΠ»ΠΎΠΌ."
await message.answer(text=text, reply_markup=kb.as_markup())
return
try:
await create_coupon(session, coupon_code, 0, usage_limit, days=days)
coupon_link = f"https://t.me/{USERNAME_BOT}?start=coupons_{coupon_code}"
text = (
f"β
ΠΡΠΏΠΎΠ½ Ρ ΠΊΠΎΠ΄ΠΎΠΌ {coupon_code} ΡΡΠΏΠ΅ΡΠ½ΠΎ ΡΠΎΠ·Π΄Π°Π½!\n"
f"β³ {format_days(days)}\n"
f"π’ ΠΠΈΠΌΠΈΡ ΠΈΡΠΏΠΎΠ»ΡΠ·ΠΎΠ²Π°Π½ΠΈΡ: {usage_limit} ΡΠ°Π·\n"
f"π Π‘ΡΡΠ»ΠΊΠ°: {coupon_link}\n"
)
kb = InlineKeyboardBuilder()
if INLINE_MODE:
kb.button(text="π€ ΠΠΎΠ΄Π΅Π»ΠΈΡΡΡΡ", switch_inline_query=f"coupon_{coupon_code}")
kb.button(text=BACK, callback_data=AdminPanelCallback(action="coupons").pack())
kb.adjust(1)
await message.answer(text=text, reply_markup=kb.as_markup())
await state.clear()
except Exception as e:
logger.error(f"ΠΡΠΈΠ±ΠΊΠ° ΠΏΡΠΈ ΡΠΎΠ·Π΄Π°Π½ΠΈΠΈ ΠΊΡΠΏΠΎΠ½Π°: {e}")
await message.answer(
"β ΠΡΠΎΠΈΠ·ΠΎΡΠ»Π° ΠΎΡΠΈΠ±ΠΊΠ° ΠΏΡΠΈ ΡΠΎΠ·Π΄Π°Π½ΠΈΠΈ ΠΊΡΠΏΠΎΠ½Π°.", reply_markup=kb.as_markup()
)
@router.callback_query(
AdminPanelCallback.filter(F.action == "coupons_list"),
IsAdminFilter(),
)
async def handle_coupons_list(callback_query: CallbackQuery, session: Any):
try:
data = AdminPanelCallback.unpack(callback_query.data)
page = data.page if data.page is not None else 1
await update_coupons_list(callback_query.message, session, page)
except Exception as e:
logger.error(f"ΠΡΠΈΠ±ΠΊΠ° ΠΏΡΠΈ ΠΏΠΎΠ»ΡΡΠ΅Π½ΠΈΠΈ ΡΠΏΠΈΡΠΊΠ° ΠΊΡΠΏΠΎΠ½ΠΎΠ²: {e}")
await callback_query.message.edit_text(
"ΠΡΠΎΠΈΠ·ΠΎΡΠ»Π° ΠΎΡΠΈΠ±ΠΊΠ° ΠΏΡΠΈ ΠΏΠΎΠ»ΡΡΠ΅Π½ΠΈΠΈ ΡΠΏΠΈΡΠΊΠ° ΠΊΡΠΏΠΎΠ½ΠΎΠ²."
)
@router.callback_query(
AdminCouponDeleteCallback.filter(F.confirm.is_(None)), IsAdminFilter()
)
async def handle_coupon_delete(
callback_query: CallbackQuery,
callback_data: AdminCouponDeleteCallback,
session: Any,
):
coupon_code = callback_data.coupon_code
kb = InlineKeyboardBuilder()
kb.button(
text="β
ΠΠ°, ΡΠ΄Π°Π»ΠΈΡΡ",
callback_data=AdminCouponDeleteCallback(
coupon_code=coupon_code, confirm=True
).pack(),
)
kb.button(
text="β ΠΠ΅Ρ, ΠΎΡΠΌΠ΅Π½ΠΈΡΡ",
callback_data=AdminCouponDeleteCallback(
coupon_code=coupon_code, confirm=False
).pack(),
)
kb.adjust(1)
await callback_query.message.edit_text(
f"ΠΡ ΡΠ²Π΅ΡΠ΅Π½Ρ, ΡΡΠΎ Ρ
ΠΎΡΠΈΡΠ΅ ΡΠ΄Π°Π»ΠΈΡΡ ΠΊΡΠΏΠΎΠ½ {coupon_code}?",
reply_markup=kb.as_markup(),
)
@router.callback_query(
AdminCouponDeleteCallback.filter(F.confirm.is_not(None)), IsAdminFilter()
)
async def confirm_coupon_delete(
callback_query: CallbackQuery,
callback_data: AdminCouponDeleteCallback,
session: Any,
):
coupon_code = callback_data.coupon_code
confirm = callback_data.confirm
if confirm:
try:
result = await delete_coupon(session, coupon_code)
if not result:
await callback_query.message.edit_text(
f"β ΠΡΠΏΠΎΠ½ Ρ ΠΊΠΎΠ΄ΠΎΠΌ {coupon_code} Π½Π΅ Π½Π°ΠΉΠ΄Π΅Π½.",
reply_markup=build_admin_back_kb("coupons"),
)
return
except Exception as e:
logger.error(f"ΠΡΠΈΠ±ΠΊΠ° ΠΏΡΠΈ ΡΠ΄Π°Π»Π΅Π½ΠΈΠΈ ΠΊΡΠΏΠΎΠ½Π°: {e}")
await callback_query.message.edit_text(
"ΠΡΠΎΠΈΠ·ΠΎΡΠ»Π° ΠΎΡΠΈΠ±ΠΊΠ° ΠΏΡΠΈ ΡΠ΄Π°Π»Π΅Π½ΠΈΠΈ ΠΊΡΠΏΠΎΠ½Π°.",
reply_markup=build_admin_back_kb("coupons"),
)
return
await update_coupons_list(callback_query.message, session)
async def update_coupons_list(message, session: Any, page: int = 1):
per_page = 10
result = await get_all_coupons(session, page, per_page)
coupons = result["coupons"]
if not coupons:
await message.edit_text(
text="β ΠΠ° Π΄Π°Π½Π½ΡΠΉ ΠΌΠΎΠΌΠ΅Π½Ρ Π½Π΅Ρ Π΄ΠΎΡΡΡΠΏΠ½ΡΡ
ΠΊΡΠΏΠΎΠ½ΠΎΠ²!",
reply_markup=build_admin_back_kb("coupons"),
)
return
kb = build_coupons_list_kb(coupons, result["current_page"], result["pages"])
text = format_coupons_list(coupons, USERNAME_BOT)
await message.edit_text(text=text, reply_markup=kb)
@router.inline_query(F.query.startswith("coupon_"))
async def inline_coupon_handler(inline_query: InlineQuery, session: Any):
if not INLINE_MODE:
return
coupon_code = inline_query.query.split("coupon_")[1]
coupon_link = f"https://t.me/{USERNAME_BOT}?start=coupons_{coupon_code}"
coupons = await get_all_coupons(session, page=1, per_page=10)
coupon = next((c for c in coupons["coupons"] if c["code"] == coupon_code), None)
if not coupon:
await inline_query.answer(
results=[],
switch_pm_text="ΠΡΠΏΠΎΠ½ Π½Π΅ Π½Π°ΠΉΠ΄Π΅Π½",
switch_pm_parameter="coupons",
cache_time=1,
)
return
title = f"ΠΡΠΏΠΎΠ½ {coupon['code']}"
description = (
f"ΠΠΎΠ»ΡΡΠΈ {coupon['amount']} ΡΡΠ±Π»Π΅ΠΉ!"
if coupon["amount"] > 0
else f"ΠΡΠΎΠ΄Π»ΠΈ ΠΏΠΎΠ΄ΠΏΠΈΡΠΊΡ Π½Π° {format_days(coupon['days'])}!"
)
message_text = (
f"π« ΠΡΠΏΠΎΠ½: {coupon['code']}\n"
f"{'π° ΠΠΎΠ½ΡΡ: ' + str(coupon['amount']) + ' ΡΡΠ±Π»Π΅ΠΉ' if coupon['amount'] > 0 else 'β³ ΠΡΠΎΠ΄Π»Π΅Π½ΠΈΠ΅: ' + format_days(coupon['days'])}\n"
f"π ΠΠ°ΠΆΠΌΠΈ, ΡΡΠΎΠ±Ρ Π°ΠΊΡΠΈΠ²ΠΈΡΠΎΠ²Π°ΡΡ!"
)
builder = InlineKeyboardBuilder()
builder.button(text="ΠΠΊΡΠΈΠ²ΠΈΡΠΎΠ²Π°ΡΡ ΠΊΡΠΏΠΎΠ½", url=coupon_link)
result = InlineQueryResultArticle(
id=coupon_code,
title=title,
description=description,
input_message_content=InputTextMessageContent(
message_text=message_text, parse_mode=ParseMode.HTML
),
reply_markup=builder.as_markup(),
)
await inline_query.answer(results=[result], cache_time=86400, is_personal=True)