Практическое руководство по созданию ботов с помощью Python

Расширенные примеры и сценарии

Пример 1: Бот с регистрацией, логированием и рассылкой

Этот бот сохраняет пользователей в SQLite, логирует все входящие сообщения и умеет отправлять сообщения всем зарегистрированным пользователям по команде администратора.

import asyncio
import aiosqlite
from aiogram import Bot, Dispatcher, types
from aiogram.filters import Command
from aiogram.types import Message
import logging

logging.basicConfig(level=logging.INFO)

BOT_TOKEN = 'ВАШ_ТОКЕН'
ADMIN_ID = 123456789  # Ваш Telegram ID

bot = Bot(token=BOT_TOKEN)
dp = Dispatcher()

DB = "bot.db"

async def init_db():
    async with aiosqlite.connect(DB) as db:
        await db.execute('''CREATE TABLE IF NOT EXISTS users (
            id INTEGER PRIMARY KEY,
            username TEXT,
            first_name TEXT,
            registered_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )''')
        await db.commit()

async def add_user(user_id: int, username: str, first_name: str):
    async with aiosqlite.connect(DB) as db:
        await db.execute('INSERT OR IGNORE INTO users (id, username, first_name) VALUES (?, ?, ?)',
                         (user_id, username, first_name))
        await db.commit()

async def get_all_users():
    async with aiosqlite.connect(DB) as db:
        cursor = await db.execute('SELECT id FROM users')
        rows = await cursor.fetchall()
        return [row[0] for row in rows]

@dp.message(Command("start"))
async def cmd_start(message: Message):
    await add_user(message.from_user.id, message.from_user.username, message.from_user.first_name)
    await message.answer("Вы зарегистрированы! Используйте /help для списка команд.")

@dp.message(Command("broadcast"))
async def cmd_broadcast(message: Message):
    if message.from_user.id != ADMIN_ID:
        await message.answer("Нет прав.")
        return
    # Отправляем сообщение всем пользователям
    text = "Внимание! Бот обновлён."
    users = await get_all_users()
    success = 0
    for uid in users:
        try:
            await bot.send_message(uid, text)
            success += 1
        except Exception as e:
            logging.error(f"Не удалось отправить {uid}: {e}")
    await message.answer(f"Рассылка завершена. Отправлено {success} из {len(users)}.")

@dp.message()
async def log_message(message: Message):
    # Логируем все сообщения
    logging.info(f"{message.from_user.id}: {message.text}")

async def main():
    await init_db()
    await dp.start_polling(bot)

if __name__ == '__main__':
    asyncio.run(main())

Результат работы (пример логов в консоли):

INFO:root:123456789: Привет
INFO:root:987654321: /start
INFO:root:123456789: /broadcast

Такой бот подходит для оповещений, новостных рассылок.

Пример 2: Бот с планировщиком задач (apscheduler)

Позволяет выполнять периодические действия, например, публикацию прогноза погоды каждое утро.

from apscheduler.schedulers.asyncio import AsyncIOScheduler
from aiogram import Bot
import asyncio

BOT_TOKEN = '...'
CHAT_ID = -1001234567890  # ID группы

bot = Bot(token=BOT_TOKEN)
scheduler = AsyncIOScheduler()

async def morning_message():
    await bot.send_message(CHAT_ID, "Доброе утро! Сегодня хороший день.")

async def main():
    scheduler.add_job(morning_message, 'cron', hour=7, minute=0)
    scheduler.start()
    # keep running
    while True:
        await asyncio.sleep(1)

if __name__ == '__main__':
    asyncio.run(main())

Результат: каждое утро в 7:00 бот отправляет сообщение в указанный чат.

Пример 3: Использование Middleware для проверки подписки на канал

Middleware позволяет выполнять код до или после каждого хэндлера. Проверим, подписан ли пользователь на канал.

from aiogram import BaseMiddleware
from aiogram.types import Message
from typing import Callable, Dict, Any, Awaitable

class SubscriptionMiddleware(BaseMiddleware):
    async def __call__(
        self,
        handler: Callable[[Message, Dict[str, Any]], Awaitable[Any]],
        event: Message,
        data: Dict[str, Any]
    ) -> Any:
        chat_member = await event.bot.get_chat_member(chat_id='@my_channel', user_id=event.from_user.id)
        if chat_member.status in ('member', 'administrator', 'creator'):
            return await handler(event, data)
        else:
            await event.answer("Подпишитесь на канал @my_channel, чтобы пользоваться ботом.")

# При подключении:
dp.message.middleware(SubscriptionMiddleware())

Такой подход универсален: можно проверять доступ по ролям, блокировать спам.

Пример 4: Обработка изображений и файлов

Бот, который принимает фото, сохраняет его на диск и отправляет обратно с наложенным текстом.

from aiogram import types
from aiogram.filters import Command
from PIL import Image, ImageDraw, ImageFont
import io

@dp.message(Command("photo"))
async def cmd_photo(message: types.Message):
    await message.answer("Отправьте мне изображение.")

@dp.message(lambda msg: msg.photo)
async def handle_photo(message: types.Message):
    photo = message.photo[-1]  # самая большая версия
    file = await bot.get_file(photo.file_id)
    image_data = await bot.download_file(file.file_path)
    
    # Открываем изображение через PIL
    image = Image.open(io.BytesIO(image_data.read()))
    draw = ImageDraw.Draw(image)
    font = ImageFont.load_default()
    draw.text((10, 10), "Ваше фото", fill='red', font=font)
    
    # Сохраняем в буфер
    buf = io.BytesIO()
    image.save(buf, format='JPEG')
    buf.seek(0)
    
    # Отправляем обратно
    await message.reply_photo(photo=buf, caption="Вот ваше фото с подписью!")

Результат: пользователь отправляет фото, бот добавляет красную надпись и возвращает изменённое изображение.