Прочитал я намедни, такую новость 8-летний ребёнок без опыта программирования создал чат-бота за 45 минут, ролик стал вирусным. Ясно, что кликбейт, ЛПП и прочие нехорошие слова… Но я решил проверить на что же способен ChatGPT. Признаюсь, я его использую для написания простых скриптов на 300-500 строчек кода. Сегодня я решил проверить правда или миф сия кликбейтная новость… и она оказалась правдой (частично).
Я до этого не писал ботов для Телеграм. Да, я знаю Python, знаю какие библиотеки для это используются, так как сидя в чате про Python с соевыми омега-обрыганами и школьниками, постоянно вижу вопросы про aiogram.
Началось все с такого задания:
Напиши мне на Python чат-бота. Алгоритм работы такой. Сторонний пользователь пишет сообщение и оно отправляется в чат к владельцу бота. Если владелец бота отвечает на конкретное сообщение то бот его пересылает лицу которое его отправило, если просто отвечает то сообщение отправляется последнему написавшему
На что он мне родил такой код:
import telebot
from collections import defaultdict
# Токен вашего бота
API_TOKEN = 'YOUR_API_TOKEN_HERE'
bot = telebot.TeleBot(API_TOKEN)
# Хранилище сообщений (user_id -> last_message)
user_last_message = {}
# Хранилище последних сообщений каждого пользователя, отправленных владельцу
message_store = defaultdict(list)
# ID владельца бота
OWNER_ID = 'YOUR_OWNER_ID_HERE'
@bot.message_handler(func=lambda message: True)
def handle_message(message):
user_id = message.from_user.id
if user_id == OWNER_ID:
# Ответ от владельца
if message.reply_to_message:
# Если владелец отвечает на конкретное сообщение
original_message_id = message.reply_to_message.message_id
for uid, msg_list in message_store.items():
for msg in msg_list:
if msg.message_id == original_message_id:
bot.send_message(uid, f"Ответ на ваше сообщение: {message.text}")
return
else:
# Если владелец отвечает без привязки
if user_last_message:
last_user_id = max(user_last_message, key=user_last_message.get)
bot.send_message(last_user_id, f"Ответ от владельца: {message.text}")
else:
# Сообщение от пользователя
bot.send_message(OWNER_ID, f"Сообщение от {message.from_user.username} (id: {user_id}): {message.text}")
user_last_message[user_id] = message.message_id
message_store[user_id].append(message)
bot.polling()
Устроил ли он меня? - Нет. Я люблю асинхронщину:
Используй какую нибудь асинхронную библиотек, храни данные в sqlite. Владельцу добавь кнопки типа заблокировать, очистить диалог (есть ли у бота возможность очищать диалог?). Напиши Dockerfile и docker-compose для запуска.
Был высран такой код:
import asyncio
import aiosqlite
from aiogram import Bot, Dispatcher, types
from aiogram.types import InlineKeyboardMarkup, InlineKeyboardButton
from aiogram.utils.keyboard import InlineKeyboardBuilder
from aiogram.filters import Text, Command
from aiogram.fsm.storage.memory import MemoryStorage
# Токен вашего бота
API_TOKEN = 'YOUR_API_TOKEN_HERE'
# ID владельца бота
OWNER_ID = 'YOUR_OWNER_ID_HERE'
# Инициализация бота и диспетчера
bot = Bot(token=API_TOKEN)
dp = Dispatcher(storage=MemoryStorage())
# Подключение к базе данных
async def init_db():
async with aiosqlite.connect('bot.db') as db:
await db.execute('''CREATE TABLE IF NOT EXISTS users
(user_id INTEGER PRIMARY KEY, blocked INTEGER DEFAULT 0)''')
await db.execute('''CREATE TABLE IF NOT EXISTS messages
(user_id INTEGER, message_id INTEGER, content TEXT)''')
await db.commit()
# Кнопки управления
def owner_keyboard(user_id):
builder = InlineKeyboardBuilder()
builder.add(
InlineKeyboardButton(text="Заблокировать", callback_data=f"block_{user_id}"),
InlineKeyboardButton(text="Очистить диалог", callback_data=f"clear_{user_id}")
)
return builder.as_markup()
# Обработчик сообщений от пользователя
@dp.message(Command(commands=['start']))
async def start(message: types.Message):
await message.answer("Привет! Отправь мне сообщение, и я перешлю его владельцу.")
@dp.message()
async def forward_to_owner(message: types.Message):
user_id = message.from_user.id
async with aiosqlite.connect('bot.db') as db:
async with db.execute("SELECT blocked FROM users WHERE user_id = ?", (user_id,)) as cursor:
row = await cursor.fetchone()
if row and row[0] == 1:
await message.answer("Вы заблокированы.")
return
await db.execute("INSERT INTO messages (user_id, message_id, content) VALUES (?, ?, ?)",
(user_id, message.message_id, message.text))
await db.commit()
# Пересылка сообщения владельцу
keyboard = owner_keyboard(user_id)
await bot.send_message(OWNER_ID, f"Сообщение от {message.from_user.username} (id: {user_id}): {message.text}", reply_markup=keyboard)
# Обработчик сообщений от владельца
@dp.message(lambda message: message.from_user.id == int(OWNER_ID))
async def handle_owner_reply(message: types.Message):
if message.reply_to_message:
# Если ответ на конкретное сообщение
original_message = message.reply_to_message.text.split("(id: ")[-1].rstrip('):')
target_user_id = int(original_message)
await bot.send_message(target_user_id, f"Ответ от владельца: {message.text}")
else:
async with aiosqlite.connect('bot.db') as db:
async with db.execute("SELECT user_id FROM messages ORDER BY rowid DESC LIMIT 1") as cursor:
row = await cursor.fetchone()
if row:
last_user_id = row[0]
await bot.send_message(last_user_id, f"Ответ от владельца: {message.text}")
# Обработчик кнопок
@dp.callback_query(Text(startswith="block_"))
async def block_user(callback: types.CallbackQuery):
user_id = int(callback.data.split("_")[1])
async with aiosqlite.connect('bot.db') as db:
await db.execute("UPDATE users SET blocked = 1 WHERE user_id = ?", (user_id,))
await db.commit()
await callback.message.answer(f"Пользователь {user_id} заблокирован.")
@dp.callback_query(Text(startswith="clear_"))
async def clear_dialog(callback: types.CallbackQuery):
user_id = int(callback.data.split("_")[1])
async with aiosqlite.connect('bot.db') as db:
await db.execute("DELETE FROM messages WHERE user_id = ?", (user_id,))
await db.commit()
await callback.message.answer(f"Диалог с пользователем {user_id} очищен.")
async def main():
await init_db()
await dp.start_polling(bot)
if __name__ == "__main__":
asyncio.run(main())
В целом как основа я подошел. Композ и докерфайл были правильными. Докерфайл я попросил переписать для использования alpine.
На это у меня ушло 6 запросов. После чего я переписал творчество бота, добавил логгирование, argparse и тп. Я скинул боту код с исправлениями:
<код>
Перепиши его. В общем когда боту пишет пользователь, то сообщение должно пересылаться полностью со всеми файлами и вложениями, а не только текст. Когда владелец просто пишет в чат с ботом, то ответ приходить должен последнему написавшему стороннему пользователю, если владелец отвечает (Reply) на конкретно сообщение, то ответ пересылается ботом в чат с этим пользователем.
То что он родил меня опять не устроило:
Убери это. await bot.send_message(args.owner_id, forward_text). Айди отправившего сообщения в базе. Когда мы отвечаем на сообщение, то нужно смотреть в базе кто его отправитель. Те нужно создать еще одну таблицу.
Но бот меня не понял:
Эта таблица должна выглядеть так:
CREATE TABLE IF NOT EXISTS messages (
message_id INTEGER PRIMARY KEY,
sender_id INTEGER,
)
Так же нужно добавить таблицу user_info где хранить записи типа user_id, fullname, username. Эти данные каждый раз обновляются когда пишет пользователь. Из этой же таблицы берутся данные, когда нажимаешь на кнопку Кто отправитель?
После еще пары перепиываний кода, я окончательно забил на интеллект этот и переписал уже все сам…
Из грубых ошибок в коде отмечу это:
def owner_keyboard(message_id: int) -> InlineKeyboardMarkup:
"""Создает клавиатуру с кнопками управления для владельца."""
keyboard = InlineKeyboardMarkup()
keyboard.add(
InlineKeyboardButton(
text="👁️ Whois",
callback_data=f"view_user_{message_id}",
),
InlineKeyboardButton(
text="🚫 Ban", callback_data=f"block_{message_id}"
),
)
return keyboard
Это какой-то неправильный счинтаксис из-за которого бот, написанный ChatGPT падал. Я это фрагмент переписал так:
def owner_keyboard(user_id: int) -> InlineKeyboardMarkup:
"""Создает клавиатуру с кнопками управления для владельца."""
return InlineKeyboardMarkup(
inline_keyboard=[
[
InlineKeyboardButton(
text="👁️ Кто это?",
callback_data=f"whois_{user_id}",
),
InlineKeyboardButton(
text="🚫 Бан",
callback_data=f"block_{user_id}",
),
],
]
)
Так же мне пришлось добавить обработку ошибок:
# https://docs.aiogram.dev/en/latest/dispatcher/errors.html
@dp.error()
async def error_handler(event: ErrorEvent):
logger.error("Error caused by %s", event.exception, exc_info=True)
Сделать закрытие курсоров после использования:
# Такое
cursor = await db_connection.execute(
"SELECT sender_id FROM messages ORDER BY ROWID DESC LIMIT 1"
)
result = await cursor.fetchone()
return result[0] if result else None
# Заменять на это
async with connection.execute(
"SELECT sender_id FROM message_senders WHERE message_id = ?",
(message_id,),
) as cursor:
result = await cursor.fetchone()
return result[0] if result else None
# Что аналогично
cursor = await execute()
result = await cursor.fetchone()
try:
return result[0] if result else None
finally:
await cursor.close()
Я переписал создание таблиц и часть запросов.
Меня порадовало то, что ChatGPT может генерировать довольно таки грамотные запросы:
await connection.execute(
"""
INSERT INTO user_info (user_id, full_name, username)
VALUES (?, ?, ?)
ON CONFLICT(user_id) DO UPDATE SET
full_name=excluded.full_name,
username=excluded.username,
updated_at=CURRENT_TIMESTAMP
""",
(user_id, full_name, username),
)
И вот что вышло:
https://github.com/s3rgeym/feedback-tgbot/tree/main
Протестировать можно тут:
https://t.me/feedback_s3rgeym_bot
