250 lines
11 KiB
Python
250 lines
11 KiB
Python
"""
|
||
Хендлеры для управления кастомными эмодзи админом
|
||
Админ отправляет эмодзи боту, бот сохраняет emoji_id и использует его в сообщениях в чатах
|
||
"""
|
||
from src.utils.errors import public_error
|
||
import logging
|
||
from html import escape
|
||
from aiogram import Router, F
|
||
from aiogram.types import Message, CallbackQuery, InlineKeyboardMarkup, InlineKeyboardButton
|
||
from aiogram.filters import Command, StateFilter
|
||
from aiogram.fsm.context import FSMContext
|
||
from aiogram.fsm.state import State, StatesGroup
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from ..core.database import async_session_maker
|
||
from ..core.access import is_admin
|
||
from ..core.config import ADMIN_IDS
|
||
from ..core.emoji_mapping_service import EmojiMappingService
|
||
|
||
logger = logging.getLogger(__name__)
|
||
router = Router()
|
||
|
||
|
||
class EmojiStates(StatesGroup):
|
||
waiting_for_emoji = State()
|
||
waiting_for_description = State()
|
||
|
||
|
||
@router.message(Command("add_emoji"), StateFilter(None))
|
||
async def add_emoji_start(message: Message, state: FSMContext):
|
||
"""Начать процесс добавления нового эмодзи"""
|
||
if not await is_admin(message.from_user.id):
|
||
await message.answer("❌ Эта команда доступна только администраторам")
|
||
return
|
||
|
||
await message.answer(
|
||
"🎨 Отправьте один премиум-эмодзи из панели Telegram.\n\n"
|
||
"Бот сохранит его в каталоге для использования в шаблонах. "
|
||
"В чатах и рассылках оформление сохраняется прямо из отправленного сообщения.",
|
||
parse_mode="HTML"
|
||
)
|
||
await state.set_state(EmojiStates.waiting_for_emoji)
|
||
|
||
|
||
@router.message(EmojiStates.waiting_for_emoji, F.text)
|
||
async def receive_emoji(message: Message, state: FSMContext):
|
||
"""Получить эмодзи от админа и сохранить его emoji_id"""
|
||
entities = [entity for entity in message.entities or () if entity.type == "custom_emoji"]
|
||
if len(entities) != 1:
|
||
await message.answer("Отправьте один премиум-эмодзи из панели эмодзи Telegram. Обычный символ не содержит его ID.")
|
||
return
|
||
entity = entities[0]
|
||
emoji_text = entity.extract_from(message.text)
|
||
emoji_id = entity.custom_emoji_id
|
||
if message.text.strip() != emoji_text or len(emoji_text) > 10 or not emoji_id or not emoji_id.isascii() or not emoji_id.isdigit():
|
||
await message.answer("Отправьте только один премиум-эмодзи, без дополнительного текста.")
|
||
return
|
||
await state.update_data(emoji_text=emoji_text, emoji_id=emoji_id)
|
||
|
||
await message.answer(
|
||
f"✅ Получил эмодзи: <code>{escape(emoji_text or '')}</code>\n\n"
|
||
f"Теперь отправьте описание этого эмодзи (для чего его использовать?)\n"
|
||
f"Например: <code>Для лотереи</code>, <code>Для победителей</code> и т.д.",
|
||
parse_mode="HTML"
|
||
)
|
||
await state.set_state(EmojiStates.waiting_for_description)
|
||
|
||
|
||
@router.message(EmojiStates.waiting_for_description, F.text)
|
||
async def receive_emoji_description(message: Message, state: FSMContext):
|
||
"""Получить описание эмодзи и сохранить в БД"""
|
||
if not message.text:
|
||
await message.answer("❌ Пожалуйста, отправьте текстовое описание")
|
||
return
|
||
|
||
description = message.text.strip()
|
||
if not description or len(description) > 255:
|
||
await message.answer("Введите описание длиной от 1 до 255 символов.")
|
||
return
|
||
data = await state.get_data()
|
||
emoji_text = data.get("emoji_text")
|
||
emoji_id = data.get("emoji_id")
|
||
|
||
if not emoji_text or not emoji_id:
|
||
await state.clear()
|
||
await message.answer("Начните добавление заново: /add_emoji")
|
||
return
|
||
|
||
# Сохраняем в БД
|
||
async with async_session_maker() as session:
|
||
emoji_service = EmojiMappingService(session)
|
||
|
||
# Проверяем не существует ли уже такой эмодзи
|
||
existing = await emoji_service.get_emoji_by_text(emoji_text, message.from_user.id)
|
||
if existing:
|
||
await message.answer(
|
||
f"⚠️ Вы уже зарегистрировали этот эмодзи: {escape(emoji_text or '')}\n"
|
||
f"Описание: <code>{escape(existing.description or '')}</code>",
|
||
parse_mode="HTML"
|
||
)
|
||
await state.clear()
|
||
return
|
||
|
||
try:
|
||
emoji_mapping = await emoji_service.register_emoji(
|
||
emoji_text=emoji_text,
|
||
emoji_id=emoji_id,
|
||
admin_id=message.from_user.id,
|
||
description=description
|
||
)
|
||
|
||
await message.answer(
|
||
f"✅ <b>Эмодзи успешно зарегистрировано!</b>\n\n"
|
||
f"Эмодзи: <code>{escape(emoji_text or '')}</code>\n"
|
||
f"Описание: <code>{escape(description or '')}</code>\n"
|
||
f"ID: <code>{emoji_id[:50]}</code>...\n\n"
|
||
"Эмодзи сохранён в каталоге. В чатах и рассылках отправляйте его из панели Telegram.",
|
||
parse_mode="HTML"
|
||
)
|
||
except ValueError as error:
|
||
await message.answer(str(error), parse_mode=None)
|
||
except Exception as e:
|
||
logger.error(f"Error registering emoji: {e}")
|
||
await message.answer(
|
||
f"❌ Ошибка при сохранении эмодзи: {public_error(e)}",
|
||
parse_mode="HTML"
|
||
)
|
||
|
||
await state.clear()
|
||
|
||
|
||
@router.message(Command("my_emojis"))
|
||
async def list_my_emojis(message: Message):
|
||
"""Показать все эмодзи, добавленные этим админом"""
|
||
if not await is_admin(message.from_user.id):
|
||
await message.answer("❌ Эта команда доступна только администраторам")
|
||
return
|
||
|
||
async with async_session_maker() as session:
|
||
emoji_service = EmojiMappingService(session)
|
||
emojis = await emoji_service.get_all_emoji_by_admin(message.from_user.id)
|
||
|
||
if not emojis:
|
||
await message.answer(
|
||
"📭 Вы еще не добавили ни один эмодзи.\n\n"
|
||
"Используйте /add_emoji чтобы добавить новый эмодзи"
|
||
)
|
||
return
|
||
|
||
text = "🎨 <b>Ваши зарегистрированные эмодзи:</b>\n\n"
|
||
for emoji in emojis:
|
||
text += (
|
||
f"<code>{escape(emoji.emoji_text or '')}</code> — {escape(emoji.description or '')}\n"
|
||
f" ID: <code>{emoji.emoji_id[:30]}</code>...\n"
|
||
f" Добавлено: <code>{emoji.created_at.strftime('%d.%m.%Y %H:%M')}</code>\n\n"
|
||
)
|
||
|
||
await message.answer(text, parse_mode="HTML")
|
||
|
||
|
||
@router.message(Command("all_emojis"))
|
||
async def list_all_emojis(message: Message):
|
||
"""Показать все зарегистрированные эмодзи (для всех админов)"""
|
||
if not await is_admin(message.from_user.id):
|
||
await message.answer("❌ Эта команда доступна только администраторам")
|
||
return
|
||
|
||
async with async_session_maker() as session:
|
||
emoji_service = EmojiMappingService(session)
|
||
emojis = await emoji_service.get_all_emojis()
|
||
|
||
if not emojis:
|
||
await message.answer(
|
||
"📭 Нет зарегистрированных эмодзи в системе"
|
||
)
|
||
return
|
||
|
||
text = "🎨 <b>Все зарегистрированные эмодзи в системе:</b>\n\n"
|
||
for emoji in emojis:
|
||
text += (
|
||
f"<code>{escape(emoji.emoji_text or '')}</code> — {escape(emoji.description or '')}\n"
|
||
f" Админ: <code>{escape(emoji.admin.first_name or 'Администратор') if emoji.admin else 'Администратор'}</code> "
|
||
f"(ID: {emoji.admin_id})\n"
|
||
f" Добавлено: <code>{emoji.created_at.strftime('%d.%m.%Y %H:%M')}</code>\n\n"
|
||
)
|
||
|
||
await message.answer(text, parse_mode="HTML")
|
||
|
||
|
||
@router.message(Command("delete_emoji"))
|
||
async def delete_emoji_start(message: Message, state: FSMContext):
|
||
"""Удалить эмодзи"""
|
||
if not await is_admin(message.from_user.id):
|
||
await message.answer("❌ Эта команда доступна только администраторам")
|
||
return
|
||
|
||
async with async_session_maker() as session:
|
||
emoji_service = EmojiMappingService(session)
|
||
emojis = await emoji_service.get_all_emoji_by_admin(message.from_user.id)
|
||
|
||
if not emojis:
|
||
await message.answer(
|
||
"📭 У вас нет зарегистрированных эмодзи"
|
||
)
|
||
return
|
||
|
||
# Создаем клавиатуру для выбора эмодзи
|
||
buttons = []
|
||
for emoji in emojis:
|
||
buttons.append([
|
||
InlineKeyboardButton(
|
||
text=f"{emoji.emoji_text} ({emoji.description or ''})",
|
||
callback_data=f"delete_emoji_{emoji.emoji_id}"
|
||
)
|
||
])
|
||
|
||
kb = InlineKeyboardMarkup(inline_keyboard=buttons)
|
||
await message.answer(
|
||
"🗑️ Выберите эмодзи для удаления:",
|
||
reply_markup=kb
|
||
)
|
||
|
||
|
||
@router.callback_query(F.data.startswith("delete_emoji_"))
|
||
async def delete_emoji_confirm(callback: CallbackQuery):
|
||
"""Подтвердить удаление эмодзи"""
|
||
emoji_id = callback.data.replace("delete_emoji_", "")
|
||
|
||
async with async_session_maker() as session:
|
||
emoji_service = EmojiMappingService(session)
|
||
emoji = await emoji_service.get_emoji_by_id(emoji_id)
|
||
|
||
if not emoji:
|
||
await callback.answer("❌ Эмодзи не найден", show_alert=True)
|
||
return
|
||
|
||
if (not emoji.admin or emoji.admin.telegram_id != callback.from_user.id) and callback.from_user.id not in ADMIN_IDS:
|
||
await callback.answer("❌ Вы не можете удалить эмодзи другого админа", show_alert=True)
|
||
return
|
||
|
||
success = await emoji_service.delete_emoji(emoji_id)
|
||
if success:
|
||
await callback.answer(
|
||
f"✅ Эмодзи {emoji.emoji_text} удалено",
|
||
show_alert=True
|
||
)
|
||
await callback.message.delete()
|
||
else:
|
||
await callback.answer("❌ Ошибка при удалении эмодзи", show_alert=True)
|