Перейти к содержанию
  • Категории
  • Последние
  • Метки
  • Популярные
  • Пользователи
  • Группы
Свернуть
Логотип бренда
Категории
  1. Главная
  2. Категории
  3. Языки программирования
  4. Python
  5. Полный гайд по разработке бота на Aiogram | Как создать бота в ТГ

Полный гайд по разработке бота на Aiogram | Как создать бота в ТГ

Запланировано Прикреплена Закрыта Перенесена Python
2 Сообщения 1 Постеры 51 Просмотры
  • Сначала старые
  • Сначала новые
  • По количеству голосов
Ответить
  • Ответить, создав новую тему
Авторизуйтесь, чтобы ответить
Эта тема была удалена. Только пользователи с правом управления темами могут её видеть.
  • kirilljsK Не в сети
    kirilljsK Не в сети
    kirilljs
    js
    написал отредактировано kirilljs
    #1

    Так ну что ж, давайте без какого либо вступления сразу в бой 😁
    Будем создавать простого бота на Aiogram, а также хранить пользователей в Sqlite для сбора id.

    Нам понадобятся следующие библиотеки:

    • aiogram (документация тут - https://aiogram.dev)
    • dotenv
    • os
    • aiosqlite
    • asyncio

    И так архитектура у нас будет такая:
    bot.py
    config.py
    commands.py
    sqlite_db.py
    .env

    Начнем с файла bot.py:

    Ах, да не забудьте установить все библиотеки через pip install aiogram dotenv aiosqlite asyncio, os - уже встроен в python.

    Если кому в падлу читать тут, я выложил готовый код на GitHub:

    Link Preview Image
    GitHub - KirillEvo/defaultbot: Default telegram bot + db Sqlite

    Default telegram bot + db Sqlite. Contribute to KirillEvo/defaultbot development by creating an account on GitHub.

    favicon

    GitHub (github.com)

    Файл: bot.py

    import asyncio
    import logging
    
    from aiogram import Bot, Dispatcher, Router, types
    from aiogram.filters import Command
    
    from sqlite_db import SqliteUserDataManager
    
    # Создаем экземпляр базы данных
    data = SqliteUserDataManager()
    
    # Тут импортируем все команды, нам понадобится только старт
    
    from commands import (
        start_command,
        allusers_command
    )
    
    # Также добавим логирование что бы видеть ошибки
    # Настройка логирования
    logging.basicConfig(
        format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", level=logging.INFO
    )
    logger = logging.getLogger(__name__)
    
    # Импорт конфигурации
    from config import TOKEN
    
    # Создаем основной роутер
    main_router = Router()
    
    
    # Регистрируем роутер для вызова команды /start
    @main_router.message(Command("start"))
    async def handle_start(message: types.Message):
        await start_command(message, data)
    
    @main_router.message(Command("users"))
    async def handle_users(message: types.Message):
        await allusers_command(message, data)
    
    
    async def main():
        # Инициализация бота
        global bot
        bot = Bot(TOKEN)
        dp = Dispatcher()
    
        # Регистрируем роутер
        dp.include_router(main_router)
    
        # Запуск бота
        logger.info("🤖 Бот запущен")
        await bot.delete_webhook(drop_pending_updates=True)
        await dp.start_polling(bot)
    
    
    if __name__ == "__main__":
        asyncio.run(main())
    

    Я оставил комментарии в коде, надеюсь всем будет понятно что куда и к чему.

    После того как мы заполним bot.py приступаем к написанию кода для хранения базы данных:
    Файл: sqlite_db.py

    import asyncio
    import aiosqlite
    from pathlib import Path
    
    # Создаем класс нашей БД с методами
    class SqliteUserDataManager:
        def __init__(self, db_file="users.db"):
            self.db_file = Path(db_file)
            self.lock = asyncio.Lock()
            self.initialized = False
    
    
        # Этот код представляет собой асинхронный метод _ensure_db_exists, который проверяет и при необходимости создает базу данных SQLite с таблицей users.
        async def _ensure_db_exists(self):
            if self.initialized:
                return
                
            # Использование асинхронного блокировщика для предотвращения race condition (состояния гонки), когда несколько потоков/корутин могут попытаться инициализировать базу одновременно
            async with self.lock:
                if self.initialized:  # Дважды проверьте блокировку
                    return
                
                # Подключение к SQLite базе данных с использованием асинхронной библиотеки aiosqlite.
                async with aiosqlite.connect(self.db_file) as db:
                    await db.execute('''
                        CREATE TABLE IF NOT EXISTS users (
                            user_id INTEGER PRIMARY KEY,
                            username TEXT,
                            subscribed BOOLEAN DEFAULT 1,
                            created_at REAL
                        )
                    ''')
                    await db.commit()
                self.initialized = True
            
        # Этот код представляет собой асинхронный метод add_user, который добавляет пользователя в базу данных SQLite.
        async def add_user(self, user_id, username=None):
            await self._ensure_db_exists()
            async with self.lock:
                async with aiosqlite.connect(self.db_file) as db:
                    await db.execute(
                        '''
                        INSERT OR IGNORE INTO users (user_id, username, created_at)
                        VALUES (?, ?, ?)
                        ''',
                        (user_id, username, asyncio.get_event_loop().time())
                    )
                    # Обновляем имя пользователя, если оно изменилось
                    if username is not None:
                        await db.execute(
                            "UPDATE users SET username = ? WHERE user_id = ?",
                            (username, user_id)
                        )
                    await db.commit()
        
        # Этот код представляет собой асинхронный метод get_user, который получает информацию о пользователе из базы данных SQLite по его ID.
        async def get_all_users(self):
            await self._ensure_db_exists()
            async with self.lock:
                async with aiosqlite.connect(self.db_file) as db:
                    cursor = await db.execute("SELECT * FROM users")
                    rows = await cursor.fetchall()
                    return {
                        row[0]: {
                            "username": row[1],
                            "subscribed": bool(row[2]),
                            "created_at": row[3]
                        } for row in rows
                    }
    

    После написание методов для Sqlite переходим к commands.py:

    # Импортируем настройки ID админа из нашего файла config.py
    from config import ADMIN_ID
    
    # Простая команда которая отдает приветствие при выполнении /start в самом боте
    async def start_command(message, data):
        user = message.from_user
        await data.add_user(user.id, user.username)
        await message.answer("Прииииивеееееет!",)
    
    # Команда для получения списка всех пользователей, но только для админа
    async def allusers_command(message, data):
        user_id = message.from_user.id
        admin_id = int(ADMIN_ID)
    
        if user_id != admin_id:
            await message.answer("❌ У вас нет прав для выполнения этой команды.")
            return
    
        try:
            users = await data.get_all_users()
            count = len(users)
            await message.answer(f"Количество пользователей: {count}")
        except Exception as e:
            await message.answer("Ошибка при получении списка пользователей.")
            print(f"[ERROR] {e}")
    

    Ну и конечно же не забываем о файле: config.py

    import os
    from dotenv import load_dotenv
    load_dotenv()
    
    TOKEN = os.getenv("TOKEN")
    ADMIN_ID = os.getenv("ADMIN_ID")
    

    На этом все! Запускайте своих ботов и тестируйте.
    Если будут вопросы пишите в комментариях на форуме.

    defbot.png


    Кстати не забудь подписаться на мой ТГ канал и еще оставлю ссылку на своего бота для скачивания клипов из ТикТока 😊

    Link Preview Image
    Kirilljs

    Обо всем и не о чем сразу 👨🏻‍💻

    favicon

    Telegram (t.me)

    Сам бот:

    Link Preview Image
    Скачать видео ВК Клипы | ТикТок

    🤖 Умею скачивать видео из VK Clips, TikTok, Yappy. Интеграция: https://telega.in/tg_bots/vkclipssavebot/card

    favicon

    Telegram (telegram.me)

    1 ответ Последний ответ
    0
    • kirilljsK Не в сети
      kirilljsK Не в сети
      kirilljs
      js
      написал отредактировано
      #2

      Дополню статью, как в sqlite выводить список id пользователей, а точнее файл.

      Вот отдельная команда, аналогично как и другим

      async def ids_command(message, bot, user_db):
          user_id = message.from_user.id
          admin_id = int(os.getenv("ADMIN_ID"))
      
          if user_id != admin_id:
              await bot.reply_to(message, "❌ У вас нет прав для выполнения этой команды.")
              return
      
          try:
              # Получение списка ID
              ids = await user_db.get_all_ids()
              
              if not ids:
                  await bot.reply_to(message, "ℹ️ Нет пользователей в базе данных.")
                  return
                  
              # Определяем тип БД для названия файла
              db_type = "SQLite" if isinstance(user_db, SqliteUserDataManager) else "JSON"
              filename = f"user_ids_{db_type.lower()}.txt"
              
              # Создаем временный файл
              with open(filename, "w") as f:
                  for user_id in ids:
                      f.write(f"{user_id}\n")  # Каждый ID на новой строке
              
              # Отправляем файл через FSInputFile
              document = FSInputFile(path=filename)
              await bot.send_document(
                  chat_id=message.chat.id,
                  document=document,
                  caption=f"📋 Список ID пользователей ({db_type})"
              )
              
              # Удаляем временный файл
              os.remove(filename)
              
          except Exception as e:
              logger.error(f"Ошибка при получении ID: {e}")
              await bot.reply_to(message, f"❌ Ошибка: {str(e)}")
      

      И не забудьте добавить метод в файле sqlite_db.py

      async def get_all_ids(self):
              await self._ensure_db_exists()
              async with self.lock:
                  async with aiosqlite.connect(self.db_file) as db:
                      cursor = await db.execute("SELECT user_id FROM users")
                      rows = await cursor.fetchall()
                      return [row[0] for row in rows]
      

      И в самом bot.py:

      @main_router.message(Command("ids"))
      async def handle_ids(message: types.Message):
          await ids_command(message, bot, user_db)
      

      А вот как выводится файл:
      Снимок экрана 2025-06-30 в 18.15.47.png

      В основном это необходимо для рассылки пользователям уведомлений или рекламных интеграций.

      1 ответ Последний ответ
      0

      Категории

      • Главная
      • Новости
      • Фронтенд
      • Бекенд
      • Языки программирования

      Контакты

      • Сотрудничество
      • info@rosdesk.ru
      • Наш чат
      • Наш ТГ канал

      © 2024 - 2025 RosDesk, Inc. Все права защищены.

      Политика конфиденциальности
      • Войти

      • Нет учётной записи? Зарегистрироваться

      • Войдите или зарегистрируйтесь для поиска.
      • Первое сообщение
        Последнее сообщение
      0
      • Категории
      • Последние
      • Метки
      • Популярные
      • Пользователи
      • Группы