Создание Webhook для Telegram-ботов на Aiogram3

# Создание виртуального окружения
python -m venv venv

# Активация (Windows)
venv\Scripts\activate

# Активация (Linux/Mac)
source venv/bin/activate

aiogram==3.22.0
pip-tools==7.5.2
python-dotenv==1.2.1
aiohttp==3.11.13

# Установка pip-tools
pip install pip-tools==7.5.2

# Компиляция зависимостей
pip-compile requirements.in

# Установка всех зависимостей
pip install -r requirements.txt

project/
├── .env
├── .gitignore
├── requirements.in
├── requirements.txt
├── config.py
├── commands.py
├── dp.py
├── middleware.py
├── run_webhook.py
├── run_polling.py
├── handlers/
│   ├── __init__.py
│   ├── h_common.py
│   └── h_weather.py
└── middleware/
    ├── __init__.py
    └── test_middlware.py

BOT_TOKEN=ваш_токен_бота
WH_BASE_URL=https://ваш-домен.ngrok.io
WH_PATH=/webhook

import os
from pathlib import Path
from dotenv import load_dotenv

BASE_DIR = Path(__file__).resolve().parent

env_file = BASE_DIR / '.env'
load_dotenv(dotenv_path=env_file)

TOKEN = os.getenv('BOT_TOKEN')
WH_BASE_URL = os.getenv('WH_BASE_URL')
WH_MAIN_BOT_PATH = os.getenv('WH_PATH')

# Проверка наличия токена
if not TOKEN:
    raise ValueError("BOT_TOKEN не найден в .env файле")

from aiogram.types import BotCommand

commands = [
    BotCommand(
        command="/start",
        description="📋 Приветствие",
        full_description="Начать работу с ботом"
    ),
    BotCommand(
        command="/weather",
        description="📋 Погода",
        full_description="Показать погоду в Минске"
    ),
]

from aiogram import Router
from aiogram.filters import Command
from aiogram.types import Message

router = Router()

@router.message(Command("start"))
async def cmd_start(message: Message):
    await message.answer(
        "👋 Привет! Я бот, который работает через вебхук.\n"
        "Доступные команды:\n"
        "/start - Приветствие\n"
        "/weather - Узнать погоду"
    )

import aiohttp
from aiogram import Router, F
from aiogram.filters import Command
from aiogram.types import Message
from aiogram.fsm.context import FSMContext

router = Router()


@router.message(Command(commands=["weather"]))
async def cmd_weather(message: Message, state: FSMContext):
    await state.clear()  # Очищаем состояние

    try:
        # Получаем погоду для Минска
        weather_data = await get_weather("Minsk")

        if weather_data:
            response = (
                f"🌤 Погода в Минске:\n"
                f"📍 Температура: {weather_data['temperature']}°C\n"
                f"💧 Влажность: {weather_data['humidity']}%\n"
                f"🌬 Давление: {weather_data['pressure']} hPa\n"
                f"🌀 Ветер: {weather_data['wind_speed']} м/с\n"
                f"📝 Описание: {weather_data['description']}"
            )
        else:
            response = "❌ Не удалось получить данные о погоде"

    except Exception as e:
        response = f"⚠️ Произошла ошибка: {str(e)}"

    await message.answer(response)


async def get_weather(city: str) -> dict:
    url = f"https://wttr.in/{city}?format=j1"

    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            if response.status == 200:
                data = await response.json()
                current = data['current_condition'][0]
                return {
                    'temperature': current['temp_C'],
                    'humidity': current['humidity'],
                    'pressure': current['pressure'],
                    'wind_speed': current['windspeedKmph'],
                    'description': current['weatherDesc'][0]['value']
                }
            return None

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


class TestMiddleware(BaseMiddleware):
    async def __call__(
        self,
        handler: Callable[[TelegramObject, Dict[str, Any]], Awaitable[Any]],
        event: TelegramObject,
        data: Dict[str, Any]
    ) -> Any:
        # Логирование входящих обновлений
        print(f"Получено обновление: {event}")
        
        # Вызов следующего обработчика
        result = await handler(event, data)
        
        return result

import logging
from typing import Any, Dict, Union

from aiogram import Bot, Dispatcher
from aiogram.client.session.aiohttp import AiohttpSession
from aiogram.fsm.storage.memory import MemoryStorage
from aiogram.fsm.strategy import FSMStrategy
from aiogram.utils.token import TokenValidationError, validate_token
from aiogram.webhook.aiohttp_server import SimpleRequestHandler, setup_application
from aiogram.client.bot import DefaultBotProperties
from aiogram.enums import ParseMode
from aiohttp import web

from config import TOKEN, WH_BASE_URL, WH_MAIN_BOT_PATH
from commands import commands
from dp import dp_connector


logging.basicConfig(
    level=logging.INFO,
    format="[%(asctime)s] [%(levelname)s] [%(lineno)d] [%(name)s] [%(message)s]",
)

PORT = 8888

def is_bot_token(value: str) -> Union[bool, Dict[str, Any]]:
    """Валидация токена бота"""
    try:
        validate_token(value)
    except TokenValidationError:
        return False
    return True

async def on_startup(bot: Bot):
    """Настройка вебхука при старте"""
    webhook_url = f"{WH_BASE_URL}{WH_MAIN_BOT_PATH}"
    await bot.set_webhook(webhook_url)
    await bot.set_my_commands(commands=commands)
    logging.info(f"Вебхук установлен: {webhook_url}")

def webhook():
    """Запуск вебхук сервера"""
    # Создание приложения aiohttp
    app = web.Application()
    
    # Настройка сессии
    session = AiohttpSession()
    
    # Настройка бота
    bot_settings = {
        "token": TOKEN,
        "session": session,
        "default": DefaultBotProperties(
            parse_mode=ParseMode.HTML,
        )
    }
    bot = Bot(**bot_settings)
    
    # Настройка диспетчера с FSM
    dp = Dispatcher(
        storage=MemoryStorage(),
        fsm_strategy=FSMStrategy.USER_IN_CHAT
    )
    
    # Регистрация обработчиков старта
    dp.startup.register(on_startup)
    
    # Подключение всех обработчиков
    dp_connector(dp)
    
    # Регистрация вебхук обработчика
    SimpleRequestHandler(
        dispatcher=dp,
        bot=bot
    ).register(app, path=WH_MAIN_BOT_PATH)
    
    # Настройка приложения
    setup_application(app, dp, bot=bot)
    
    # Запуск сервера
    web.run_app(app, host='0.0.0.0', port=PORT)

if __name__ == "__main__":
    # Проверка токена
    if not is_bot_token(TOKEN):
        logging.error("Неверный токен бота")
        exit(1)
    
    logging.info(f"Запуск вебхук сервера на порту {PORT}")
    webhook()

from aiogram.dispatcher.dispatcher import Dispatcher

# Импортируем middleware и handlers
from middleware import test_middlware
from handlers import h_common, h_weather


def dp_connector(dp: Dispatcher) -> Dispatcher:
    """Подключение всех компонентов к диспетчеру"""
    
    # Подключение middleware
    dp.update.middleware(test_middlware.TestMiddleware())
    
    # Подключение роутеров
    dp.include_router(h_common.router)
    dp.include_router(h_weather.router)
    
    return dp

import asyncio
import logging
from aiogram import Dispatcher
from aiogram.fsm.storage.memory import MemoryStorage
from aiogram.fsm.strategy import FSMStrategy
from aiogram.types import BotCommandScopeDefault

from config import bot
from commands import commands
from dp import dp_connector


async def main():
    logging.basicConfig(
        level=logging.INFO,
        format="[%(asctime)s] [%(levelname)s] [%(lineno)d] [%(name)s] [%(message)s]",
    )
    
    dp = Dispatcher(
        storage=MemoryStorage(),
        fsm_strategy=FSMStrategy.USER_IN_CHAT
    )
    
    # Установка команд
    await bot.set_my_commands(
        commands=commands,
        scope=BotCommandScopeDefault()
    )
    
    # Подключение всех компонентов
    dp_connector(dp)
    
    # Очистка вебхука перед запуском polling
    await bot.delete_webhook(drop_pending_updates=True)
    
    # Запуск polling
    await dp.start_polling(
        bot,
        allowed_updates=dp.resolve_used_update_types()
    )

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except (KeyboardInterrupt, SystemExit):
        print("Бот остановлен")

# Установка localtunnel глобально
npm install -g localtunnel

# Создание туннеля
lt --port 8888

# Пример полученного URL:
# https://wild-pumas-turn.loca.lt

# Запуск туннеля
ngrok http 8888

# Пример полученного URL:
# https://abc123.ngrok.io

1. HTTPS: Вебхук должен работать по HTTPS
2. Доменное имя: Собственный домен
3. Стабильный хостинг: Сервер должен быть всегда доступен
4. Nginx + SSL: Настройка прокси и SSL сертификатов

server {
    listen 443 ssl http2;
    server_name your-domain.com;

    ssl_certificate /path/to/cert.pem;
    ssl_certificate_key /path/to/key.pem;

    location /webhook {
        proxy_pass http://127.0.0.1:8888;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
    }
}

logging.basicConfig(
    level=logging.INFO,
    format="[%(asctime)s] [%(levelname)s] [%(lineno)d] [%(name)s] [%(message)s]",
    filename='webhook.log',  # Раскомментировать для записи в файл
    filemode='a'             # Режим добавления
)

@router.message(Command("webhook_status"))
async def cmd_webhook_status(message: Message, bot: Bot):
    webhook_info = await bot.get_webhook_info()
    await message.answer(
        f"📊 Статус вебхука:\n"
        f"URL: {webhook_info.url}\n"
        f"Активен: {webhook_info.pending_update_count}\n"
        f"Последняя ошибка: {webhook_info.last_error_message or 'Нет'}"
    )