Полный гайд по разработке бота на Aiogram | Как создать бота в ТГ
-
Так ну что ж, давайте без какого либо вступления сразу в бой
Будем создавать простого бота на 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:
GitHub - KirillEvo/defaultbot: Default telegram bot + db Sqlite
Default telegram bot + db Sqlite. Contribute to KirillEvo/defaultbot development by creating an account on GitHub.
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")
На этом все! Запускайте своих ботов и тестируйте.
Если будут вопросы пишите в комментариях на форуме.
Кстати не забудь подписаться на мой ТГ канал и еще оставлю ссылку на своего бота для скачивания клипов из ТикТока
Сам бот:
Скачать видео ВК Клипы | ТикТок
🤖 Умею скачивать видео из VK Clips, TikTok, Yappy. Интеграция: https://telega.in/tg_bots/vkclipssavebot/card
Telegram (telegram.me)
-
Дополню статью, как в 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)
А вот как выводится файл:
В основном это необходимо для рассылки пользователям уведомлений или рекламных интеграций.
© 2024 - 2025 RosDesk, Inc. Все права защищены.