Разработка ботов с использованием Python

Расширенные примеры ботов

Пример 1: Бот с инлайн-клавиатурой и запросом внешнего API

Этот бот по команде /fact отправляет случайный факт с использованием API (uselessfacts.jsph.pl). Клавиатура позволяет получить новый факт без ввода команды.

import asyncio
import aiohttp
from aiogram import Bot, Dispatcher, types
from aiogram.filters import Command
from aiogram.types import InlineKeyboardButton, InlineKeyboardMarkup

API_TOKEN = 'YOUR_TOKEN'
bot = Bot(token=API_TOKEN)
dp = Dispatcher()

async def get_fact():
    async with aiohttp.ClientSession() as session:
        async with session.get('https://uselessfacts.jsph.pl/random.json?language=en') as resp:
            data = await resp.json()
            return data['text']

@dp.message(Command("fact"))
async def send_fact(message: types.Message):
    fact = await get_fact()
    keyboard = InlineKeyboardMarkup(
        inline_keyboard=[[InlineKeyboardButton(text='Ещё факт', callback_data='new_fact')]]
    )
    await message.answer(fact, reply_markup=keyboard)

@dp.callback_query(lambda c: c.data == 'new_fact')
async def new_fact(callback: types.CallbackQuery):
    fact = await get_fact()
    keyboard = InlineKeyboardMarkup(
        inline_keyboard=[[InlineKeyboardButton(text='Ещё факт', callback_data='new_fact')]]
    )
    await callback.message.edit_text(fact, reply_markup=keyboard)
    await callback.answer()

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

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

Пользователь отправляет /fact
Бот: Did you know that a group of flamingos is called a 'flamboyance'? (с кнопкой 'Ещё факт')
При нажатии на кнопку бот заменяет сообщение новым фактом.

Пояснение: используется aiohttp для асинхронного запроса, InlineKeyboardMarkup для кнопок. CallbackQuery обрабатывается отдельным хендлером.

Пример 2: Бот с FSM и сохранением данных в SQLite

Бот собирает заказ пиццы: выбирает размер, топпинги, адрес доставки. Данные сохраняются в базу SQLite.

import sqlite3
from aiogram import Bot, Dispatcher, types
from aiogram.filters import Command
from aiogram.fsm.state import State, StatesGroup
from aiogram.fsm.context import FSMContext
from aiogram.types import ReplyKeyboardMarkup, KeyboardButton

API_TOKEN = 'YOUR_TOKEN'
bot = Bot(token=API_TOKEN)
dp = Dispatcher()

# Инициализация БД
conn = sqlite3.connect('orders.db')
cursor = conn.cursor()
cursor.execute('''CREATE TABLE IF NOT EXISTS orders (
    user_id INTEGER,
    size TEXT,
    toppings TEXT,
    address TEXT
)''')
conn.commit()

class Order(StatesGroup):
    size = State()
    toppings = State()
    address = State()

size_kb = ReplyKeyboardMarkup(
    keyboard=[[KeyboardButton(text='Маленькая'), KeyboardButton(text='Большая')]],
    resize_keyboard=True
)

@dp.message(Command("order"))
async def start_order(message: types.Message, state: FSMContext):
    await state.set_state(Order.size)
    await message.answer("Выберите размер пиццы:", reply_markup=size_kb)

@dp.message(Order.size)
async def process_size(message: types.Message, state: FSMContext):
    if message.text not in ['Маленькая', 'Большая']:
        await message.answer("Пожалуйста, используйте кнопки")
        return
    await state.update_data(size=message.text)
    await state.set_state(Order.toppings)
    await message.answer("Какие топпинги? (введите через запятую)", reply_markup=types.ReplyKeyboardRemove())

@dp.message(Order.toppings)
async def process_toppings(message: types.Message, state: FSMContext):
    await state.update_data(toppings=message.text)
    await state.set_state(Order.address)
    await message.answer("Введите адрес доставки:")

@dp.message(Order.address)
async def process_address(message: types.Message, state: FSMContext):
    data = await state.update_data(address=message.text)
    # Сохранение в БД
    cursor.execute('INSERT INTO orders (user_id, size, toppings, address) VALUES (?, ?, ?, ?)',
                   (message.from_user.id, data['size'], data['toppings'], data['address']))
    conn.commit()
    await state.clear()
    await message.answer("Заказ принят! Спасибо.")

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

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

Пользователь вводит /order
Бот: Выберите размер пиццы: (кнопки Маленькая/Большая)
Пользователь: Большая
Бот: Какие топпинги? (введите через запятую)
Пользователь: сыр, грибы, оливки
Бот: Введите адрес доставки:
Пользователь: ул. Ленина, д.10
Бот: Заказ принят! Спасибо.

Пояснение: SQLite используется для хранения заказов. ReplyKeyboardMarkup для выбора размера. Валидация входа на каждом шаге.

Пример 3: Бот для загрузки изображений с наложением водяного знака

Бот принимает изображение, накладывает водяной знак (текст) и возвращает обратно.

import asyncio
from io import BytesIO
from PIL import Image, ImageDraw, ImageFont
from aiogram import Bot, Dispatcher, types
from aiogram.filters import Command

API_TOKEN = 'YOUR_TOKEN'
bot = Bot(token=API_TOKEN)
dp = Dispatcher()

async def add_watermark(image_bytes: bytes, text: str) -> bytes:
    image = Image.open(BytesIO(image_bytes))
    draw = ImageDraw.Draw(image)
    try:
        font = ImageFont.truetype("arial.ttf", 36)
    except:
        font = ImageFont.load_default()
    # Позиция водяного знака - правый нижний угол
    width, height = image.size
    text_width, text_height = draw.textsize(text, font)
    x = width - text_width - 10
    y = height - text_height - 10
    draw.text((x, y), text, fill=(255, 255, 255, 128), font=font)
    output = BytesIO()
    image.save(output, format='PNG')
    output.seek(0)
    return output.read()

@dp.message(Command("watermark"))
async def start_watermark(message: types.Message):
    await message.answer("Отправьте изображение, и я добавлю водяной знак.")

@dp.message(content_types=['photo'])
async def handle_photo(message: types.Message):
    # Получаем фото в максимальном разрешении
    photo = message.photo[-1]
    file_info = await bot.get_file(photo.file_id)
    downloaded_file = await bot.download_file(file_info.file_path)
    watermarked = await add_watermark(downloaded_file.read(), "@your_bot")
    await message.reply_photo(types.BufferedInputFile(watermarked, filename="watermarked.png"),
                              caption="Изображение с водяным знаком")

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

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

Пользователь отправляет фото
Бот: (возвращает то же фото с текстом "@your_bot" в правом нижнем углу)

Пояснение: используются библиотеки Pillow для обработки изображений. Водяной знак добавляется полупрозрачным. Обработка ошибок при загрузке шрифта.