bot/bot.py - master - burevol/ChatGPTBot

bot/bot.py
83 строки3 KB
Alexander Maximov
Реструктуризация кода
08 дек 2024, 18:03
08 дек 2024, 18:036ce2ab6
import logging

from aiogram import Bot, Dispatcher, html
from aiogram.filters import CommandStart
from aiogram.types import Message
from aiogram.client.default import DefaultBotProperties
from aiogram.enums import ParseMode

from config import Config
from gpt import GPT

dp = Dispatcher()
config = Config()

gpt = GPT(config.OPENAI_API_KEY)

bot_name = ''


@dp.message(CommandStart())
async def command_start_handler(message: Message) -> None:
    """
    This handler receives messages with `/start` command
    """
    # Most event objects have aliases for API methods that can be called in events' context
    # For example if you want to answer to incoming message you can use `message.answer(...)` alias
    # and the target chat will be passed to :ref:`aiogram.methods.send_message.SendMessage`
    # method automatically or call API method directly via
    # Bot instance: `bot.send_message(chat_id=message.chat.id, ...)`
    await message.answer(f"Hello, {html.bold(message.from_user.full_name)}!")


@dp.message()
async def message_handler(message: Message) -> None:
    """
    Handler will forward receive a message back to the sender

    By default, message handler will handle all message types (like a text, photo, sticker etc.)
    """
    global bot_name
    logging.debug(message)

    if message.chat.type == "private":
        if message.from_user.id not in config.ACCESS_LIST:
            logging.warning(f"User {message.from_user.id} is not allowed to use bot")
            await message.answer("У пользователя нет прав на использование бота.")
            return
        elif message.text is None:
            return
        else:
            try:
                response = await gpt.chat(message.text)
                await message.answer(response)
            except TypeError:
                # But not all the types is supported to be copied so need to handle it
                await message.answer("Неподдерживаемый тип сообщения")

    elif message.chat.type == "group" or message.chat.type == "supergroup":
        if message.chat.id not in config.GROUP_LIST:
            logging.warning(f"Group {message.chat.id} is not allowed to use bot")
            await message.answer("У группы нет прав на использование бота.")
            return
        if message.text is None or not message.text.startswith(bot_name):
            return
        else:
            try:
                response = await gpt.chat(message.text[len(bot_name):])
                await message.answer(response)
            except TypeError:
                # But not all the types is supported to be copied so need to handle it
                await message.answer("Неподдерживаемый тип сообщения")


async def start_bot() -> None:
    """
    This function will be called when bot will be ready to process incoming messages
    """
    global bot_name
    bot = Bot(token=config.TOKEN, default=DefaultBotProperties(parse_mode=ParseMode.MARKDOWN))
    me = await bot.get_me()
    bot_name = '@' + me.username
    logging.info(f"Bot {bot_name} started")
    await dp.start_polling(bot)