«База данных» в гугл таблицах для телеграм-бота

Практически все боты используют какое-нибудь хранилище информации. Чаще всего применяются базы данных, но иногда их использование может быть избыточным, особенно если вам не нужны ACID-транзакции и есть желание менять данные руками в максимально простом интерфейсе.

Для хранения информации можно использовать гугл-таблицы.

Сегодня мы разберём пример телеграмм бота для проведения тестов, где вопросы и ответы хранятся в разных вкладках одной Google-таблицы. Почему одной? Опять же, для облегчения доступа для потенциальных пользователей.

Создав бота, мы задеплоем его в облачном сервисе Amvera.

Это даст удобную механику обновлений через Git. Буквально за 3 команды, не выходя из IDE, можно доставить обновление, что проще настройки VPS в несколько раз. И после регистрации будет начислен бесплатный баланс на первые недели работы бота.

Структура Google-таблицы

Перед тем, как начать работать с гугл-таблицей в нашем боте, необходимо определиться с тем, как будет выглядеть структура нашего документа, для того, что бы мы могли написать парсер, который и будем использовать.

В нашем примере мы будем работать с тестами, в которых к каждому вопросу, существует 4 варианта ответа (1 правильный, 3 ошибочные).
Значит, нам нужно хранить следующую информацию о вопросе:

1. Непосредственно текст вопроса
2. Четыре варианта ответа на вопрос
3. Указание, какой из ответов является правильным

Было решено использовать следующую структуру:

1. В столбце A — находится текст вопроса
2. В столбце B — правильный ответ
3. В столбцах 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": "[e-mail]",
  "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", куда мы будем записывать ответы

Теперь реализуем необходимые методы:

1. Получение списка тестов. Список тестов — это все страницы файла, кроме страницы с заголовком "Results"

   
   

   def get_topics(self):
       return {key: value for key, value in self.topics.items() if key != "Results"}

   
   

2. Получение списка вопросов в тесте. Мы получим сырые данные со страницы, нужно будет их преобразовать в удобный формат

   
   

   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 мы собрали список из правильного и неправильных ответов, перемешали его и затем уже вернули список вопросов в том формате, в котором предполагаем его использовать.

   
   

3. А кроме получения информации необходимо ещё и записывать ответы в таблицу. Так что реализуем ещё один метод, для записи:

   
   

    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