224 lines
6.0 KiB
Python
224 lines
6.0 KiB
Python
"""
|
||
Модуль смены аватарки бота.
|
||
Совместим с 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(_("❌ Пожалуйста, отправьте фотографию."))
|