Создаем Telegram-бот на Django: инструкция для разработчиков

# Создание виртуальной среды
python -m venv venv

# Активация виртуальной среды (для Windows)
venv\Scripts\activate

# Активация виртуальной среды (для Linux/Mac)
source venv/bin/activate

# Установка Django и python-telegram-bot
pip install django python-telegram-bot

# Создание проекта
django-admin startproject telegrambot

# Переход в директорию проекта
cd telegrambot

# Создание приложения
python manage.py startapp bot

INSTALLED_APPS = [
'django.contrib.admin',
'django.contrib.auth',
'django.contrib.contenttypes',
'django.contrib.sessions',
'django.contrib.messages',
'django.contrib.staticfiles',
'bot', # Добавляем наше приложение
]

# .env
TELEGRAM_BOT_TOKEN=ваш_токен_бота

pip install python-dotenv

import os
from dotenv import load_dotenv

# Загрузка переменных окружения
load_dotenv()

# ... остальные настройки ...

# Настройки Telegram бота
TELEGRAM_BOT_TOKEN = os.getenv('TELEGRAM_BOT_TOKEN')

# Создаем директории, если их нет
import os
os.makedirs('bot/management/commands', exist_ok=True)

# Пустой файл для обозначения пакета Python

from django.core.management.base import BaseCommand
from telegram import Update
from telegram.ext import ApplicationBuilder, CommandHandler, MessageHandler, filters, CallbackContext
import logging
import os
from django.conf import settings

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

class Command(BaseCommand):
help = 'Запускает Telegram бота'

async def start(self, update: Update, context: CallbackContext) -> None:
user = update.effective_user
await update.message.reply_html(
f"Привет, {user.mention_html()}! Я бот, созданный с использованием Django.",
)

async def help_command(self, update: Update, context: CallbackContext) -> None:
await update.message.reply_text("Я могу выполнять различные команды. Используйте /start для начала.")

async def echo(self, update: Update, context: CallbackContext) -> None:
await update.message.reply_text(update.message.text)

def handle(self, *args, **options):
application = ApplicationBuilder().token(settings.TELEGRAM_BOT_TOKEN).build()

# Добавление обработчиков
application.add_handler(CommandHandler("start", self.start))
application.add_handler(CommandHandler("help", self.help_command))

# Обработчик для обычных текстовых сообщений
application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, self.echo))

# Запуск бота
self.stdout.write(self.style.SUCCESS('Запуск бота...'))
application.run_polling()

from django.core.management.base import BaseCommand
from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup
from telegram.ext import ApplicationBuilder, CommandHandler, MessageHandler, filters, CallbackContext, CallbackQueryHandler
import logging
from django.conf import settings
from bot.models import TelegramUser

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

class Command(BaseCommand):
help = 'Запускает Telegram бота'

async def start(self, update: Update, context: CallbackContext) -> None:
user = update.effective_user

# Создаем или обновляем пользователя в базе данных
telegram_user, created = TelegramUser.objects.get_or_create(
telegram_id=user.id,
defaults={
'username': user.username,
'first_name': user.first_name,
'last_name': user.last_name
}
)

# Если пользователь существует, обновляем его данные
if not created:
telegram_user.username = user.username
telegram_user.first_name = user.first_name
telegram_user.last_name = user.last_name
telegram_user.save()

# Создаем клавиатуру
keyboard = [
[InlineKeyboardButton("Помощь", callback_data='help')],
[InlineKeyboardButton("О боте", callback_data='about')]
]
reply_markup = InlineKeyboardMarkup(keyboard)

await update.message.reply_html(
f"Привет, {user.mention_html()}! Я бот, созданный с использованием Django.",
reply_markup=reply_markup
)

async def button_callback(self, update: Update, context: CallbackContext) -> None:
query = update.callback_query
await query.answer()

if query.data == 'help':
await query.message.reply_text("Я могу выполнять различные команды. Используйте /start для начала.")
elif query.data == 'about':
await query.message.reply_text("Я бот, созданный с использованием Django и python-telegram-bot.")

async def help_command(self, update: Update, context: CallbackContext) -> None:
await update.message.reply_text("Я могу выполнять различные команды. Используйте /start для начала.")

async def echo(self, update: Update, context: CallbackContext) -> None:
# Получаем или создаем пользователя
user = update.effective_user
telegram_user, _ = TelegramUser.objects.get_or_create(
telegram_id=user.id,
defaults={
'username': user.username,
'first_name': user.first_name,
'last_name': user.last_name
}
)

# Сохраняем сообщение в историю
telegram_user.message_history += f"\n{update.message.text}"
telegram_user.save()

await update.message.reply_text(f"Вы сказали: {update.message.text}")

def handle(self, *args, **options):
application = ApplicationBuilder().token(settings.TELEGRAM_BOT_TOKEN).build()

# Добавление обработчиков
application.add_handler(CommandHandler("start", self.start))
application.add_handler(CommandHandler("help", self.help_command))
application.add_handler(CallbackQueryHandler(self.button_callback))

# Обработчик для обычных текстовых сообщений
application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, self.echo))

# Запуск бота
self.stdout.write(self.style.SUCCESS('Запуск бота...'))
application.run_polling()

from django.db import models

class TelegramUser(models.Model):
telegram_id = models.BigIntegerField(primary_key=True)
username = models.CharField(max_length=255, null=True, blank=True)
first_name = models.CharField(max_length=255, null=True, blank=True)
last_name = models.CharField(max_length=255, null=True, blank=True)
message_history = models.TextField(blank=True, default="")
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)

def __str__(self):
return f"{self.first_name} {self.last_name} (@{self.username})"

python manage.py makemigrations
python manage.py migrate

from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup
from telegram.ext import CallbackContext
from .models import TelegramUser
from django.utils import timezone
import json

async def save_user_data(update: Update) -> TelegramUser:
"""Сохраняет или обновляет данные пользователя Telegram в базе данных."""
user = update.effective_user
telegram_user, created = TelegramUser.objects.get_or_create(
telegram_id=user.id,
defaults={
'username': user.username,
'first_name': user.first_name,
'last_name': user.last_name,
}
)

# Обновляем данные, если пользователь уже существует
if not created:
telegram_user.username = user.username
telegram_user.first_name = user.first_name
telegram_user.last_name = user.last_name
telegram_user.save()

return telegram_user

def create_menu_keyboard(items):
"""Создает клавиатуру на основе списка элементов."""
keyboard = []
for item in items:
keyboard.append([InlineKeyboardButton(item['text'], callback_data=item['callback_data'])])
return InlineKeyboardMarkup(keyboard)

async def log_user_activity(update: Update, activity_type: str, data: dict = None):
"""Логирует активность пользователя."""
user = await save_user_data(update)

# Создаем запись в истории активности
activity_data = {
'timestamp': timezone.now().isoformat(),
'type': activity_type,
'data': data or {}
}

# Добавляем к существующей истории или создаем новую
if user.message_history:
history = json.loads(user.message_history)
else:
history = []

history.append(activity_data)
user.message_history = json.dumps(history)
user.save()

return user

from django.core.management.base import BaseCommand
from telegram import Update
from telegram.ext import ApplicationBuilder, CommandHandler, MessageHandler, filters, CallbackContext, CallbackQueryHandler, ConversationHandler
import logging
from django.conf import settings
from bot.utils import save_user_data, create_menu_keyboard, log_user_activity

# Определение состояний для ConversationHandler
CHOOSING, TYPING_REPLY = range(2)

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

class Command(BaseCommand):
help = 'Запускает Telegram бота'

async def start(self, update: Update, context: CallbackContext) -> None:
user = update.effective_user
await save_user_data(update)

# Логируем активность
await log_user_activity(update, 'command', {'command': 'start'})

# Создаем меню
menu_items = [
{'text': 'Помощь', 'callback_data': 'help'},
{'text': 'О боте', 'callback_data': 'about'},
{'text': 'Начать диалог', 'callback_data': 'start_dialog'}
]
reply_markup = create_menu_keyboard(menu_items)

await update.message.reply_html(
f"Привет, {user.mention_html()}! Я бот, созданный с использованием Django.",
reply_markup=reply_markup
)

return CHOOSING

async def button_callback(self, update: Update, context: CallbackContext) -> int:
query = update.callback_query
await query.answer()

# Логируем активность
await log_user_activity(update, 'button_click', {'button': query.data})

if query.data == 'help':
await query.message.reply_text("Я могу выполнять различные команды. Используйте /start для начала.")
return CHOOSING
elif query.data == 'about':
await query.message.reply_text("Я бот, созданный с использованием Django и python-telegram-bot.")
return CHOOSING
elif query.data == 'start_dialog':
await query.message.reply_text("Пожалуйста, введите ваше имя:")
return TYPING_REPLY

return CHOOSING

async def handle_reply(self, update: Update, context: CallbackContext) -> int:
user_input = update.message.text

# Логируем активность
await log_user_activity(update, 'dialog_reply', {'input': user_input})

# Здесь можно обрабатывать ответы пользователя
await update.message.reply_text(f"Спасибо за ответ: {user_input}. Что еще вы хотите сделать?")

# Возвращаем меню
menu_items = [
{'text': 'Помощь', 'callback_data': 'help'},
{'text': 'О боте', 'callback_data': 'about'},
{'text': 'Начать диалог заново', 'callback_data': 'start_dialog'}
]
reply_markup = create_menu_keyboard(menu_items)

await update.message.reply_text("Выберите опцию:", reply_markup=reply_markup)

return CHOOSING

async def help_command(self, update: Update, context: CallbackContext) -> None:
# Логируем активность
await log_user_activity(update, 'command', {'command': 'help'})

await update.message.reply_text("Я могу выполнять различные команды. Используйте /start для начала.")

async def echo(self, update: Update, context: CallbackContext) -> None:
# Логируем активность
await log_user_activity(update, 'message', {'text': update.message.text})

await update.message.reply_text(f"Вы сказали: {update.message.text}")

def handle(self, *args, **options):
application = ApplicationBuilder().token(settings.TELEGRAM_BOT_TOKEN).build()

# Создаем ConversationHandler для управления диалогом
conv_handler = ConversationHandler(
entry_points=[CommandHandler("start", self.start)],
states={
CHOOSING: [
CallbackQueryHandler(self.button_callback)
],
TYPING_REPLY: [
MessageHandler(filters.TEXT & ~filters.COMMAND, self.handle_reply)
]
},
fallbacks=[CommandHandler("help", self.help_command)]
)

# Добавление обработчиков
application.add_handler(conv_handler)
application.add_handler(CommandHandler("help", self.help_command))

# Обработчик для обычных текстовых сообщений
application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, self.echo))

# Запуск бота
self.stdout.write(self.style.SUCCESS('Запуск бота...'))
application.run_polling()

from django.contrib import admin
from .models import TelegramUser

@admin.register(TelegramUser)
class TelegramUserAdmin(admin.ModelAdmin):
list_display = ('telegram_id', 'username', 'first_name', 'last_name', 'created_at', 'updated_at')
search_fields = ('telegram_id', 'username', 'first_name', 'last_name')
readonly_fields = ('created_at', 'updated_at')

python manage.py bot

pip install supervisor

[program:telegram_bot]
command=/path/to/venv/bin/python /path/to/project/manage.py bot
directory=/path/to/project
user=username
autostart=true
autorestart=true
stderr_logfile=/var/log/telegram_bot.err.log
stdout_logfile=/var/log/telegram_bot.out.log

supervisorctl reread
supervisorctl update
supervisorctl status telegram_bot

FROM python:3.10

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

CMD ["python", "manage.py", "bot"]

Django>=4.2.0,<4.3.0
python-telegram-bot>=20.0,<21.0
python-dotenv>=1.0.0,<2.0.0

docker build -t telegram-django-bot .
docker run -d --name telegram-bot --env-file .env telegram-django-bot

from django.test import TestCase
from .models import TelegramUser
import json

class TelegramBotTests(TestCase):
def setUp(self):
# Создаем тестового пользователя
self.user = TelegramUser.objects.create(
telegram_id=123456789,
username='test_user',
first_name='Test',
last_name='User'
)

def test_user_creation(self):
# Проверяем, что пользователь корректно создан
user = TelegramUser.objects.get(telegram_id=123456789)
self.assertEqual(user.username, 'test_user')
self.assertEqual(user.first_name, 'Test')
self.assertEqual(user.last_name, 'User')

def test_log_activity(self):
# Тестируем логирование активности
self.user.message_history = json.dumps([
{
'timestamp': '2023-01-01T12:00:00',
'type': 'command',
'data': {'command': 'start'}
}
])
self.user.save()

# Проверяем, что активность сохранена
user = TelegramUser.objects.get(telegram_id=123456789)
history = json.loads(user.message_history)
self.assertEqual(len(history), 1)
self.assertEqual(history[0]['type'], 'command')
self.assertEqual(history[0]['data']['command'], 'start')

python manage.py test bot

mkdir -p bot/templates/bot

<!DOCTYPE html>
<html>
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Telegram Web App</title>
<script src="https://telegram.org/js/telegram-web-app.js"></script>
<style>
body {
font-family: Arial, sans-serif;
margin: 0;
padding: 20px;
color: var(--tg-theme-text-color);
background-color: var(--tg-theme-bg-color);
}
.container {
max-width: 600px;
margin: 0 auto;
}
.form-group {
margin-bottom: 15px;
}
label {
display: block;
margin-bottom: 5px;
}
input, textarea {
width: 100%;
padding: 8px;
box-sizing: border-box;
border: 1px solid var(--tg-theme-hint-color);
border-radius: 4px;
background-color: var(--tg-theme-secondary-bg-color);
color: var(--tg-theme-text-color);
}
button {
padding: 10px 15px;
background-color: var(--tg-theme-button-color);
color: var(--tg-theme-button-text-color);
border: none;
border-radius: 4px;
cursor: pointer;
}
</style>
</head>
<body>
<div class="container">
<h1>Форма обратной связи</h1>
<div class="form-group">
<label for="name">Имя:</label>
<input type="text" id="name" name="name">
</div>
<div class="form-group">
<label for="email">Email:</label>
<input type="email" id="email" name="email">
</div>
<div class="form-group">
<label for="message">Сообщение:</label>
<textarea id="message" name="message" rows="4"></textarea>
</div>
<button id="submit-btn">Отправить</button>
</div>

<script>
// Инициализация Web App
let tg = window.Telegram.WebApp;
tg.expand();

// Получаем данные пользователя
let user = tg.initDataUnsafe.user;

// Заполняем форму данными пользователя, если они доступны
if (user) {
document.getElementById('name').value = user.first_name + ' ' + (user.last_name || '');
}

// Обработчик отправки формы
document.getElementById('submit-btn').addEventListener('click', function() {
let name = document.getElementById('name').value;
let email = document.getElementById('email').value;
let message = document.getElementById('message').value;

// Валидация формы
if (!name || !email || !message) {
tg.showPopup({
title: 'Ошибка',
message: 'Пожалуйста, заполните все поля',
buttons: [
{id: 'ok', type: 'ok', text: 'OK'}
]
});
return;
}

// Отправляем данные на сервер
fetch('/bot/webapp/submit/', {
method: 'POST',
headers: {
'Content-Type': 'application/json',
'X-CSRFToken': getCookie('csrftoken')
},
body: JSON.stringify({
name: name,
email: email,
message: message,
telegram_id: user ? user.id : null
})
})
.then(response => response.json())
.then(data => {
if (data.success) {
tg.showAlert('Спасибо! Ваше сообщение отправлено.');
tg.close();
} else {
tg.showAlert('Произошла ошибка: ' + data.error);
}
})
.catch(error => {
tg.showAlert('Произошла ошибка при отправке данных.');
console.error('Error:', error);
});
});

// Функция для получения CSRF токена из cookies
function getCookie(name) {
let cookieValue = null;
if (document.cookie && document.cookie !== '') {
const cookies = document.cookie.split(';');
for (let i = 0; i < cookies.length; i++) {
const cookie = cookies[i].trim();
if (cookie.substring(0, name.length + 1) === (name + '=')) {
cookieValue = decodeURIComponent(cookie.substring(name.length + 1));
break;
}
}
}
return cookieValue;
}
</script>
</body>
</html>

from django.shortcuts import render
from django.http import JsonResponse
from django.views.decorators.csrf import csrf_exempt
import json
from .models import TelegramUser
import logging

logger = logging.getLogger(__name__)

def webapp_view(request):
"""Отображает Web App страницу."""
return render(request, 'bot/webapp.html')

@csrf_exempt
def webapp_submit(request):
"""Обрабатывает отправку формы из Web App."""
if request.method == 'POST':
try:
data = json.loads(request.body)
name = data.get('name')
email = data.get('email')
message = data.get('message')
telegram_id = data.get('telegram_id')

# Сохраняем данные формы
if telegram_id:
try:
user = TelegramUser.objects.get(telegram_id=telegram_id)

# Обновляем историю сообщений пользователя
if user.message_history:
history = json.loads(user.message_history)
else:
history = []

history.append({
'type': 'webapp_form',
'data': {
'name': name,
'email': email,
'message': message
}
})

user.message_history = json.dumps(history)
user.save()

except TelegramUser.DoesNotExist:
logger.warning(f"User with telegram_id {telegram_id} not found")

# Здесь можно добавить логику для отправки данных формы
# например, по электронной почте или сохранения в другую модель

return JsonResponse({'success': True})
except Exception as e:
logger.error(f"Error processing form data: {e}")
return JsonResponse({'success': False, 'error': str(e)})

return JsonResponse({'success': False, 'error': 'Invalid request method'})

from django.urls import path
from . import views

app_name = 'bot'

urlpatterns = [
path('webapp/', views.webapp_view, name='webapp'),
path('webapp/submit/', views.webapp_submit, name='webapp_submit'),
]

from django.contrib import admin
from django.urls import path, include

urlpatterns = [
path('admin/', admin.site.urls),
path('bot/', include('bot.urls', namespace='bot')),
]

from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup, WebAppInfo
from telegram.ext import ApplicationBuilder, CommandHandler, MessageHandler, filters, CallbackContext, CallbackQueryHandler

# ... остальной код ...

async def start(self, update: Update, context: CallbackContext) -> None:
user = update.effective_user
await save_user_data(update)

# Логируем активность
await log_user_activity(update, 'command', {'command': 'start'})

# URL вашего веб-приложения
webapp_url = 'https://your-domain.com/bot/webapp/'

# Создаем кнопки, включая кнопку для Web App
keyboard = [
[InlineKeyboardButton("Помощь", callback_data='help')],
[InlineKeyboardButton("О боте", callback_data='about')],
[InlineKeyboardButton("Заполнить форму", web_app=WebAppInfo(url=webapp_url))]
]
reply_markup = InlineKeyboardMarkup(keyboard)

await update.message.reply_html(
f"Привет, {user.mention_html()}! Я бот, созданный с использованием Django.",
reply_markup=reply_markup
)

return CHOOSING