Практически все боты используют какое-нибудь хранилище информации. Чаще всего применяются базы данных, но иногда их использование может быть избыточным, особенно если вам не нужны ACID-транзакции и есть желание менять данные руками в максимально простом интерфейсе.
Для хранения информации можно использовать гугл-таблицы.
Сегодня мы разберём пример телеграмм бота для проведения тестов, где вопросы и ответы хранятся в разных вкладках одной Google-таблицы. Почему одной? Опять же, для облегчения доступа для потенциальных пользователей.
Создав бота, мы задеплоем его в облачном сервисе Amvera.
Это даст удобную механику обновлений через Git. Буквально за 3 команды, не выходя из IDE, можно доставить обновление, что проще настройки VPS в несколько раз. И после регистрации будет начислен бесплатный баланс на первые недели работы бота.
Структура Google-таблицы
Перед тем, как начать работать с гугл-таблицей в нашем боте, необходимо определиться с тем, как будет выглядеть структура нашего документа, для того, что бы мы могли написать парсер, который и будем использовать.
В нашем примере мы будем работать с тестами, в которых к каждому вопросу, существует 4 варианта ответа (1 правильный, 3 ошибочные).
Значит, нам нужно хранить следующую информацию о вопросе:
1. Непосредственно текст вопроса
2. Четыре варианта ответа на вопрос
3. Указание, какой из ответов является правильным
Было решено использовать следующую структуру:
- В столбце
A
— находится текст вопроса - В столбце
B
— правильный ответ - В столбцах
C
,D
,E
— неправильные ответы
Такая структура позволит с одной стороны явно указать какой ответ правильный, а с другой мы избежим дублирования ответов.
Кроме этого, можно средствами Google-таблиц реализовать проверки на корректность данных. Например, что среди неправильных ответов нет повторов, или что ни один из неправильных не совпадает с корректным ответом.
И таких листов, с таблицами такого формата может быть сколько угодно, каждая страница — это свой отдельный тест. Можно разделить их, как мы увидим позже, по предметам(Химия/История), а можно даже в рамках одного предмета делать несколько тестов.
Поскольку мы хотим использовать нашу таблицу и для хранения ответов — разберём формат и этой страницы:
В нашей таблице мы будем хранить ID пользователя, вопрос, ответ пользователя, правильный ответ и временную метку.
Зачем? Ведь это явно дублирование информации, по тексту вопроса можно получить ответ из другого листа? Дело в том, что в случае, если лист будет удалён, или исправлен — то и данные в таблице ответов исправятся, что может изменить, например, итоговый бал.
Важно сказать, что бот будет использоваться только для сбора ответов, все проверки могут быть реализованы средствами Google.
Взаимодействие с таблицей из Python
Наша программа на Python будет состоять из двух частей: первая — объекты и методы для работы с Google-таблицей, а вторая — непосредственно бот. Разумеется мы будем использовать первую часть в нашем боте, но для начала разберёмся как работать с гугл-таблицами из Python.
Для работы с API таблиц существует множество библиотек, мы будем использовать gspread
. Подробнее о ней можно почитать на сайте с официальной документацией.
Перед началом работы нам нужно создать аккаунт и получить ключи для взаимодействия.
Вот подробная инструкция
о том, как это сделать. В результате — у вас появится .json
файл следующего вида:
{
"type": "service_account",
"project_id": "pythonapi-433820",
"private_key_id": "7080a92d01c73eaf214379bb171093",
"private_key": "-----BEGIN PRIVATE KEY-----\nNrDyLw … jINQh/9\n-----END PRIVATE KEY-----\n",
"client_email": "api-123@pythonapi-433820.iam.gserviceaccount.com",
"client_id": "473 … hd.apps.googleusercontent.com"
}
Теперь можно и подключаться.
Для начала создадим файл config.py
, в нём мы будем хранить конфигурационную информацию: путь к нашему .json
— файлу, ссылку на файл и токен для бота:
CREDENTIALS_FILENAME = "credentials.json"
QUESTIONS_SPREADSHEET_URL = "https://docs.google.com/spreadsheets/d/15dL4-HSC7VjjnQHnppJMQZ"
BOT_TOKEN = "6941340264:BCGscBGpPZIuI-1sOyIgv-rzPgkyrhNt12c"
А теперь перейдём непосредственно к коду нашего парсера. Для начала импортируем все необходимые нам библиотеки:
import gspread
from config import CREDENTIALS_FILENAME, QUESTIONS_SPREADSHEET_URL
from random import shuffle
from datetime import datetime
А затем создадим класс, который будет отображением нашей таблицы в Python:
class Quizzer:
def __init__(self, question_spreadsheet_url=QUESTIONS_SPREADSHEET_URL):
self.account = gspread.service_account(filename=CREDENTIALS_FILENAME)
self.spreadsheet = self.account.open_by_url(question_spreadsheet_url)
self.topics = {
elem.title: elem.id for elem in self.spreadsheet.worksheets()
}
self.answers = self.spreadsheet.get_worksheet_by_id(self.topics.get("Results"))
Для начала мы создаём необходимые поля класса:
self.account
— аккаунт который будет использоваться для доступа к таблице. Подтягивается из нашего файлаself.url
— путь к файлу, с которым мы будем работатьself.spreadsheet
— непосредственно сам файл (уже открытый)self.topics
— словарь пар "Заголовок листа": "ID листа" — всех листов в нашем файлеself.answers
— открытый лист с заголовком "Results", куда мы будем записывать ответы
Теперь реализуем необходимые методы:
Получение списка тестов. Список тестов — это все страницы файла, кроме страницы с заголовком "Results"
def get_topics(self): return {key: value for key, value in self.topics.items() if key != "Results"}
Получение списка вопросов в тесте. Мы получим сырые данные со страницы, нужно будет их преобразовать в удобный формат
def get_question_by_topic(self, topic_name): if topic_name in self.topics: worksheet = self.spreadsheet.get_worksheet_by_id(self.topics.get(topic_name)) return worksheet.get_all_records() return [] def questions_and_answers(self, topic_name): questions = self.get_question_by_topic(topic_name) result = [] for elem in questions: answers = [elem["correct_answer"], elem["wrong_answer_1"], elem["wrong_answer_2"], elem["wrong_answer_3"]] shuffle(answers) new_format = { "question": elem["question"], "correct_answer": elem["correct_answer"], "answers": answers } result.append(new_format) return result
Так что сначала в методе
get_question_by_topic
мы получили сырые данные использовав методget_all_records
, а затем в методеquestions_and_answers
мы собрали список из правильного и неправильных ответов, перемешали его и затем уже вернули список вопросов в том формате, в котором предполагаем его использовать.
А кроме получения информации необходимо ещё и записывать ответы в таблицу. Так что реализуем ещё один метод, для записи:
def write_answer_to_result_cell(self, user_id, question, answer, correct_answer):
index = len(list(filter(None, self.answers.col_values(1)))) + 1
self.answers.update(f"A{index}:E{index}", [[
user_id, question, answer, correct_answer, f"{datetime.now()}"
]])
В принципе, на этом функционал по взаимодействию с гугл-таблицей реализован. И можно переходить к написанию собственно бота.
Телеграм-бот
Для реализации телеграмм бота будем использован асинхронный движок — aiogram
. Подробнее о реализации ботов с помощью этой библиотеки можно прочитать в этом учебнике.
Для начала импортируем все необходимые библиотеки:
import asyncio
from aiogram import Bot, Dispatcher, F, Router, types
from aiogram.filters.command import Command
from aiogram.fsm.context import FSMContext
from aiogram.fsm.state import State, StatesGroup
from config import BOT_TOKEN
from questionExtractor import Quizzer
asyncio
— библиотека для работы с асинхронным подходомaiogram
— непосредственно библиотека для создания ботов, извлекаем основные понятия и необходимый функционал для
работы со Стейт-Машинойconfig
— наш файл конфигурации,BOT_TOKEN
— токен для взаимодействия с ботом. Получить можно с помощью 'BotFather'questionExtractor
— наш модуль для взаимодействия с гугл-таблицей
Затем создаём необходимые базовые объекты:
router = Router()
— роутер, для распределения сообщений по обработчикамbot = Bot(token=BOT_TOKEN)
— непосредственно сам ботdp = Dispatcher()
— диспетчер сообщенийquizzer = Quizzer()
— объект для взаимодействия с таблицей
а затем создаём класс, описывающий все возможные состояния нашего бота:
class CurrentQuiz(StatesGroup):
start = State()
choosing_test = State()
question = State()
start
— бот только запущен(или перезапущен) — нужно выбрать тестchoosing_test
— процесс выбора теста — отправка первого вопросаquestion
— отправка вопросов и приём ответов
Перед непосредственной обработкой входящих сообщений напищем две вспомогательные функции.
Одна будет использоваться для создания кастомной клавиатуры из списка возможных ответов:
def create_keyboard(options):
"""Функция для создания клавиатуры из списка возможных вариантов"""
return types.ReplyKeyboardMarkup(keyboard=[[types.KeyboardButton(text=f"{elem}")] for elem in options])
А вторая, для формирования сообщения с вопросом и клавиатурой. Важно сказать, что мы храним всё наше текущее состояние в стейте. Стейт(состояние) — это некий словарь, где мы можем хранить какую-то информацию, причём этот словарь можно передавать между обработчиками.
В поле current_question
— мы храним тот вопрос, который хотим отправить, а в поле choosing_test
— список вопросов текущего теста.
async def ask_question(message: types.Message, state: FSMContext):
"""Функция для отправки вопроса с формированием клавиатуры ответов"""
data = await state.get_data()
question = data["current_question"]
keyboard = create_keyboard(question["answers"])
await message.answer(question["question"], reply_markup=keyboard)
await state.update_data(current_question=question)
await state.update_data(choosing_test=data["choosing_test"][1:])
В нашей функции мы получаем текущий вопрос из стейта, отправляем пользователю(создав клавиатуру с ответом), а затем обновляем данные в стейте.
Теперь пришло время написать обработчики пользовательских сообщений.
# Обработчик на команду старт. Стейт CurrentQuiz.start
@router.message(CurrentQuiz.start)
@router.message(Command("start"))
async def cmd_start(message: types.Message, state: FSMContext):
keyboard = create_keyboard(quizzer.get_topics().keys())
await message.answer("Привет, я бот Quizzer. Вот доступные темы для тестов. Выбери любую", reply_markup=keyboard)
await state.set_state(CurrentQuiz.choosing_test)
Это обработчик первой команды которую отправляет пользователь, запуская новый бот — /start
. В ней мы получаем список тем из нашей таблицы, а затем предлагаем пользователю выбрать одну из них.
# Обработчик стейта выбора теста
@router.message(CurrentQuiz.choosing_test, F.text.in_(quizzer.get_topics().keys()))
async def start_quizz(message: types.Message, state: FSMContext):
chosen_test_title = message.text
choosing_test = quizzer.questions_and_answers(message.text)
await state.update_data(
choosing_test=choosing_test,
current_question=choosing_test[0]
)
await message.answer(f"Выбрана тема: {chosen_test_title}")
await state.set_state(CurrentQuiz.question)
await ask_question(message, state)
После того как пользователь выберет тест, мы сохраняем в стейт вопросы из выбранного теста:
await state.update_data(
choosing_test=choosing_test,
current_question=choosing_test[0]
)
И задаём пользователю первый вопрос из него: await ask_question(message, state)
Самый важный обработчик — это обработчик получения ответа от пользователя:
@router.message(CurrentQuiz.question)
async def getting_answer(message: types.Message, state: FSMContext):
data = await state.get_data()
quizzer.write_answer_to_result_cell(
message.from_user.username,
data["current_question"]["question"],
message.text,
f'{data["current_question"]["correct_answer"]}'
)
remaining_questions = data["choosing_test"]
if remaining_questions:
await state.update_data(choosing_test=remaining_questions, current_question=remaining_questions[0])
await ask_question(message, state)
else:
await state.clear()
await message.answer("Все вопросы закончились", reply_markup=create_keyboard(["Выбрать новый квиз"]))
await state.set_state(CurrentQuiz.start)
Во-первых мы сохраняем ответ пользователя в таблицу:
quizzer.write_answer_to_result_cell(
message.from_user.username,
data["current_question"]["question"],
message.text,
f'{data["current_question"]["correct_answer"]}'
)
А затем проверяем, остались ли в тесте незаданные вопросы. Если остались — задаём следующий:
await state.update_data(choosing_test=remaining_questions, current_question=remaining_questions[0])
await ask_question(message, state)
А если нет — очищаем память стейта и возвращаемся к исходному стейту — запуску бота:
await state.clear()
await message.answer("Все вопросы закончились", reply_markup=create_keyboard(["Выбрать новый квиз"]))
await state.set_state(CurrentQuiz.start)
Теперь остаётся только запустить нашего бота:
# Запуск процесса поллинга новых апдейтов
async def main():
dp.include_router(router)
await dp.start_polling(bot)
if __name__ == "__main__":
asyncio.run(main())
Наш бот заработал на локальном компьютере, но в тот момент, когда мы его выключим наш бот отключится. Что бы работа бота не зависела от состояния вашего компьютера — принято загружать(деплоить) их в облако. Этим мы сейчас и займёмся.
Деплой
В качестве облака для деплоя будем использовать Amvera.
Сервис позволит осуществить деплой простым перетягиванием файлов в интерфейсе (или через команду git push amvera master в IDE) и предоставит стартовый баланс на первые недели бесплатного использования.
Создаём проект.
Загружаем необходимые файлы проекта, включая конфигурационные, а также файл с зависимостями. Его можно сгенерировать автоматически.
Используя например команду: pip freeze > requirements.txt
.
Но лучше написать руками, pip freeze генерирует много лишних зависимостей, которые замедляют сборку.
Затем конфигурируем: указываем версию python, механизм умправления зависимостями и основной файл программы:
Запускаем сборку и наслаждаемся работой нашего бота.
Если вы хотите использовать для деплоя Git, рекомендую ознакомиться со статьей.
В гугл-таблице у нас вопросы и ответы:
А всё взаимодействие — через бота:
В этом уроке мы познакомились с тем как использовать Google-таблицу в качестве "БД" для телеграм бота. Полный код проекта доступен в репозитории на GitHub