Телеграмм бот - как создать

from pydantic_settings import BaseSettings
from typing import List

# Класс настроек с автоматической загрузкой из .env
class Settings(BaseSettings):
    # Токен бота из BotFather
    bot_token: str
    
    # URL подключения к базе данных
    database_url: str
    
    # Список ID администраторов бота
    admin_ids: List[int]
    
    # Режим отладки (расширенное логирование)
    debug: bool = False

    class Config:
        # Указываем файл с переменными
        env_file = ".env"
        env_file_encoding = "utf-8"

# Создаём глобальный экземпляр настроек
settings = Settings()

import asyncio
import logging
from aiogram import Bot, Dispatcher
from aiogram.enums import ParseMode
from aiogram.client.default import DefaultBotProperties

# Импорт локальных модулей
from config import settings
from handlers import start, callbacks

# Настройка логирования для отладки
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)

async def main():
    # Создание экземпляра бота с настройками
    bot = Bot(
        token=settings.bot_token,
        default=DefaultBotProperties(
            parse_mode=ParseMode.HTML
        )
    )
    
    # Диспетчер управляет маршрутизацией обновлений
    dp = Dispatcher()
    
    # Регистрация роутеров с обработчиками
    dp.include_router(start.router)
    dp.include_router(callbacks.router)
    
    # Удаление необработанных обновлений при старте
    await bot.delete_webhook(drop_pending_updates=True)
    
    # Запуск long-polling (получение обновлений)
    await dp.start_polling(bot)

# Точка входа при запуске скрипта
if __name__ == "__main__":
    asyncio.run(main())

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

# Импорт клавиатуры главного меню
from keyboards.inline import get_main_menu

# Создание роутера для группировки обработчиков
router = Router()

# Декоратор регистрирует функцию как обработчик /start
@router.message(CommandStart())
async def cmd_start(message: Message):
    # Получаем имя пользователя для персонализации
    user_name = message.from_user.first_name
    
    # Формируем приветственное сообщение с HTML-разметкой
    welcome_text = f"""
Привет, {user_name}!

Я бот-помощник компании. Вот что я умею:

- Отвечать на частые вопросы
- Принимать заявки
- Показывать каталог товаров
- Связывать с оператором

Выберите нужный раздел:
"""
    
    # Отправка сообщения с inline-клавиатурой
    await message.answer(
        text=welcome_text,
        reply_markup=get_main_menu()
    )

from aiogram import Router, F
from aiogram.types import CallbackQuery

router = Router()

# Обработчик нажатия кнопки "Каталог"
@router.callback_query(F.data == "catalog")
async def show_catalog(callback: CallbackQuery):
    # answer() убирает "часики" загрузки на кнопке
    await callback.answer()
    
    # Редактируем существующее сообщение
    await callback.message.edit_text(
        text="Выберите категорию товаров:",
        reply_markup=get_catalog_keyboard()
    )

# Обработчик с динамическим параметром
@router.callback_query(F.data.startswith("product_"))
async def show_product(callback: CallbackQuery):
    # Извлекаем ID товара из callback_data
    product_id = callback.data.split("_")[1]
    
    # Получаем данные из базы (пример)
    product = await get_product_by_id(product_id)
    
    await callback.answer()
    await callback.message.edit_text(
        text=f"{product.name}\n\nЦена: {product.price} руб."
    )

from aiogram.types import InlineKeyboardMarkup, InlineKeyboardButton
from aiogram.utils.keyboard import InlineKeyboardBuilder

def get_main_menu() -> InlineKeyboardMarkup:
    # Builder упрощает создание клавиатур
    builder = InlineKeyboardBuilder()
    
    # Добавляем кнопки с callback_data
    builder.row(
        InlineKeyboardButton(
            text="Каталог товаров",
            callback_data="catalog"
        )
    )
    builder.row(
        InlineKeyboardButton(
            text="Частые вопросы",
            callback_data="faq"
        ),
        InlineKeyboardButton(
            text="О компании",
            callback_data="about"
        )
    )
    # Кнопка со ссылкой на внешний сайт
    builder.row(
        InlineKeyboardButton(
            text="Перейти на сайт",
            url="https://example.com"
        )
    )
    
    return builder.as_markup()

from sqlalchemy import Column, Integer, String, DateTime, Boolean
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import declarative_base, sessionmaker
from datetime import datetime

from config import settings

# Базовый класс для всех моделей
Base = declarative_base()

# Модель пользователя бота
class User(Base):
    __tablename__ = "users"
    
    # Telegram ID как первичный ключ
    id = Column(Integer, primary_key=True)
    
    # Имя пользователя из профиля
    username = Column(String(64), nullable=True)
    first_name = Column(String(128))
    
    # Дата первого обращения к боту
    created_at = Column(DateTime, default=datetime.utcnow)
    
    # Флаг активной подписки
    is_premium = Column(Boolean, default=False)
    
    # Языковой код для локализации
    language = Column(String(10), default="ru")

# Асинхронный движок для работы с БД
engine = create_async_engine(settings.database_url)

# Фабрика сессий для запросов
async_session = sessionmaker(
    engine, 
    class_=AsyncSession, 
    expire_on_commit=False
)

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from .models import User, async_session

async def get_or_create_user(
    user_id: int,
    username: str,
    first_name: str
) -> User:
    """Получает пользователя из БД или создаёт нового"""
    
    async with async_session() as session:
        # Поиск существующего пользователя
        query = select(User).where(User.id == user_id)
        result = await session.execute(query)
        user = result.scalar_one_or_none()
        
        if user is None:
            # Создание нового пользователя
            user = User(
                id=user_id,
                username=username,
                first_name=first_name
            )
            session.add(user)
            await session.commit()
        
        return user

async def update_user_premium(
    user_id: int, 
    is_premium: bool
) -> None:
    """Обновляет статус подписки пользователя"""
    
    async with async_session() as session:
        query = select(User).where(User.id == user_id)
        result = await session.execute(query)
        user = result.scalar_one()
        
        user.is_premium = is_premium
        await session.commit()

import logging
from aiogram import Router
from aiogram.types import ErrorEvent

router = Router()
logger = logging.getLogger(__name__)

# Глобальный обработчик всех ошибок
@router.error()
async def error_handler(event: ErrorEvent):
    # Логируем полную информацию об ошибке
    logger.error(
        f"Ошибка при обработке обновления: {event.exception}",
        exc_info=True
    )
    
    # Получаем объект обновления для контекста
    update = event.update
    
    # Пытаемся уведомить пользователя об ошибке
    if update.message:
        try:
            await update.message.answer(
                "Произошла ошибка. Попробуйте позже."
            )
        except:
            pass
    
    # Возвращаем True чтобы ошибка считалась обработанной
    return True