Модуль установки аватара (В разработке!)
This commit is contained in:
223
bot/handlers/commands/settings/set_avatar.py
Normal file
223
bot/handlers/commands/settings/set_avatar.py
Normal file
@@ -0,0 +1,223 @@
|
||||
"""
|
||||
Модуль смены аватарки бота.
|
||||
Совместим с aiogram 3.22.0 и Bot API 9.4+
|
||||
|
||||
Использует:
|
||||
- SetMyProfilePhoto
|
||||
- InputProfilePhotoStatic
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from typing import Union
|
||||
|
||||
from aiogram import Router, Bot, F
|
||||
from aiogram.filters import Command
|
||||
from aiogram.fsm.context import FSMContext
|
||||
from aiogram.fsm.state import StatesGroup, State
|
||||
from aiogram.types import (
|
||||
Message,
|
||||
CallbackQuery,
|
||||
FSInputFile,
|
||||
InputProfilePhotoStatic,
|
||||
)
|
||||
from aiogram.methods.set_my_profile_photo import SetMyProfilePhoto
|
||||
from aiogram.exceptions import TelegramAPIError, TelegramRetryAfter
|
||||
from aiogram.utils.i18n import gettext as _
|
||||
|
||||
from bot.filters import IsSuperAdmin
|
||||
from bot.templates import msg
|
||||
from bot.utils import format_retry_time, status_clear
|
||||
from bot.core.bots import BotInfo
|
||||
from bot.handlers.commands.settings.settings_cmd import settings_keyboard
|
||||
from configs import COMMANDS
|
||||
from middleware.loggers import logger
|
||||
|
||||
__all__ = ("router",)
|
||||
|
||||
CMD: str = "set_avatar".lower()
|
||||
router: Router = Router(name=f"{CMD}_router")
|
||||
|
||||
|
||||
# ================= FSM =================
|
||||
|
||||
class SetBotAvatarForm(StatesGroup):
|
||||
"""
|
||||
FSM состояния для смены аватарки.
|
||||
"""
|
||||
waiting_for_photo: State = State()
|
||||
|
||||
|
||||
# ================= CORE =================
|
||||
|
||||
async def handle_set_avatar(
|
||||
update: Union[Message, CallbackQuery],
|
||||
state: FSMContext,
|
||||
bot: Bot
|
||||
) -> None:
|
||||
"""
|
||||
Устанавливает новую аватарку бота.
|
||||
|
||||
Args:
|
||||
update: Message или CallbackQuery
|
||||
state: FSM контекст
|
||||
bot: Экземпляр бота
|
||||
"""
|
||||
|
||||
message: Message = update.message if isinstance(update, CallbackQuery) else update
|
||||
|
||||
if not message.photo:
|
||||
return
|
||||
|
||||
largest_photo = message.photo[-1]
|
||||
|
||||
# Получаем файл от Telegram
|
||||
tg_file = await bot.get_file(largest_photo.file_id)
|
||||
|
||||
temp_path: str = f"/tmp/{largest_photo.file_unique_id}.jpg"
|
||||
|
||||
await bot.download_file(tg_file.file_path, destination=temp_path)
|
||||
|
||||
try:
|
||||
method = SetMyProfilePhoto(
|
||||
photo=InputProfilePhotoStatic(
|
||||
photo=FSInputFile(temp_path)
|
||||
)
|
||||
)
|
||||
|
||||
result: bool = await bot(method)
|
||||
|
||||
if result:
|
||||
logger.info("Аватарка бота успешно обновлена", log_type="BOT_SETUP")
|
||||
|
||||
await state.clear()
|
||||
|
||||
await msg(
|
||||
update=update,
|
||||
text=_("✅ Аватарка бота успешно обновлена."),
|
||||
markup=settings_keyboard(),
|
||||
state=state
|
||||
)
|
||||
|
||||
except TelegramRetryAfter as e:
|
||||
retry_text: str = format_retry_time(e.retry_after)
|
||||
|
||||
logger.warning(
|
||||
f"Rate limit при смене аватарки. Повтор через {retry_text}",
|
||||
log_type="BOT_SETUP"
|
||||
)
|
||||
|
||||
await msg(
|
||||
update=update,
|
||||
text=_(
|
||||
"⚠️ Слишком частая смена аватарки.\n"
|
||||
"Попробуйте снова через: <b>{retry}</b>"
|
||||
).format(retry=retry_text),
|
||||
markup=settings_keyboard(),
|
||||
state=state
|
||||
)
|
||||
|
||||
except TelegramAPIError as e:
|
||||
logger.error(
|
||||
f"Ошибка Telegram API при смене аватарки: {e}",
|
||||
log_type="BOT_SETUP"
|
||||
)
|
||||
|
||||
await msg(
|
||||
update=update,
|
||||
text=_(
|
||||
"❌ Ошибка Telegram API:\n"
|
||||
"<pre>{error}</pre>"
|
||||
).format(error=str(e)),
|
||||
markup=settings_keyboard(),
|
||||
state=state
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
f"Непредвиденная ошибка при смене аватарки: {e}",
|
||||
log_type="BOT_SETUP"
|
||||
)
|
||||
|
||||
await msg(
|
||||
update=update,
|
||||
text=_(
|
||||
"❌ Непредвиденная ошибка:\n"
|
||||
"<pre>{error}</pre>"
|
||||
).format(error=str(e)),
|
||||
markup=settings_keyboard(),
|
||||
state=state
|
||||
)
|
||||
|
||||
finally:
|
||||
if os.path.exists(temp_path):
|
||||
os.remove(temp_path)
|
||||
|
||||
|
||||
# ================= COMMAND =================
|
||||
|
||||
@router.callback_query(F.data.lower() == CMD, IsSuperAdmin())
|
||||
@router.message(
|
||||
Command(*COMMANDS.get(CMD, [CMD]), prefix=BotInfo.prefix, ignore_case=True),
|
||||
IsSuperAdmin()
|
||||
)
|
||||
async def set_avatar_cmd(
|
||||
message: Message | CallbackQuery,
|
||||
state: FSMContext,
|
||||
bot: Bot
|
||||
) -> None:
|
||||
"""
|
||||
Команда /set_avatar
|
||||
|
||||
Поддерживает:
|
||||
- Фото вместе с командой
|
||||
- Ответ на фото
|
||||
- FSM режим
|
||||
"""
|
||||
|
||||
await status_clear(update=message, state=state)
|
||||
|
||||
msg_obj: Message = message.message if isinstance(message, CallbackQuery) else message
|
||||
|
||||
if msg_obj.photo:
|
||||
await handle_set_avatar(message, state, bot)
|
||||
return
|
||||
|
||||
if msg_obj.reply_to_message and msg_obj.reply_to_message.photo:
|
||||
await handle_set_avatar(msg_obj.reply_to_message, state, bot)
|
||||
return
|
||||
|
||||
await msg(
|
||||
update=message,
|
||||
text=_(
|
||||
"🖼 <b>Смена аватарки бота</b>\n\n"
|
||||
"Отправьте фотографию."
|
||||
),
|
||||
markup=settings_keyboard(),
|
||||
state=state
|
||||
)
|
||||
|
||||
await state.set_state(SetBotAvatarForm.waiting_for_photo)
|
||||
|
||||
|
||||
# ================= FSM =================
|
||||
|
||||
@router.message(SetBotAvatarForm.waiting_for_photo, IsSuperAdmin(), F.photo)
|
||||
async def process_avatar_photo(
|
||||
message: Message,
|
||||
state: FSMContext,
|
||||
bot: Bot
|
||||
) -> None:
|
||||
"""
|
||||
Обработка фото через FSM.
|
||||
"""
|
||||
await handle_set_avatar(message, state, bot)
|
||||
|
||||
|
||||
@router.message(SetBotAvatarForm.waiting_for_photo, IsSuperAdmin())
|
||||
async def invalid_input(message: Message) -> None:
|
||||
"""
|
||||
Обработка некорректного ввода.
|
||||
"""
|
||||
await message.answer(_("❌ Пожалуйста, отправьте фотографию."))
|
||||
Reference in New Issue
Block a user