- PVSM.RU - https://www.pvsm.ru -
Добрый день, сегодня я хотел бы поделится с Вами проблемами и их необычными решениями, которые встретились при написании небольших IT проектов. Сразу скажу, что статья для тех, кто хоть немного разбирается в разработке телеграмм ботов, баз данных, SQL и в языке программировании python.
Весь проект выложен на github, ссылка будет в конце статьи.
Изначально я хотел для себя написать простенького телеграмм бота счетчика калорий, который получает число от пользователя и возвращает сколько калорий осталось до нормы на день. То есть нужно хранить грубо говоря пару переменных для каждого пользователя.
В итоге нужно было выбрать способ хранить эти данные.
В итоге ни один из этих вариантов не подошел. Конечно же есть и десятки других способов, но мне хотелось бы, чтобы было бесплатно, быстро и минимум кода.
Идея очень простая, для хранения данных мы будем использовать in memory базу данных sqllite, так как она уже встроена в python 3 и будем делать бэкапы нашей таблицы на сервера Telegram с небольшим интервалом (примерно каждые 30 секунд) и бэкап при закрытие процесса программы.
Если сервер упал, то при первом запросе мы автоматически загрузим нашу таблицу с сервера Telegram и восстановим данные в sqllite.
Можно использовать и любую другую in memory бд, кому как нравится.
Напишем простой кликер и проведем тесты на скорость выполнения.
Бот будет написан на языке программирования python с использованием асинхронной библиотеки взаимодействия с api телеграмма aiogram.
Первым делом нужно заполнить настройки бота, не буду рассказывать как получить токен от BotFather, уже сотни статей есть на эту тему.
Также нам нужен второй аккаунт в телеграмме для админа, в котором будут сохраняться наши бекапы.
Для того, чтобы получить admin_id и config_id нам нужно запустить бота с аккаунта администратора и написать боту «admin», после чего он создаст первый бекап, и напишет ваш admin_id, config_id. Заменяем и запускаем бота заново.
#--------------------Настройки бота-------------------------
# Ваш токен от BotFather
TOKEN = '1234567:your_token'
# Логирование
logging.basicConfig(level=logging.INFO)
bot = Bot(token=TOKEN)
dp = Dispatcher(bot)
# Ваш айди аккаунта администратора и айди сообщения где хранится файл с данными
admin_id=12345678
config_id=12345
conn = sqlite3.connect(":memory:") # настройки in memory бд
cursor = conn.cursor()
Если боту приходит сообщение со словом «admin», то мы создаем таблицу пользователей с такой моделью данных:
Добавляем тестового пользователя, и отправляем документ на сервер Telegram с нашей таблицей. Так же отправляем admin_id и config_id администратору в виде сообщений. После получения айдишников, данный код нужно закомментировать.
# Логика для администратора
if message.text == 'admin':
cursor.execute("CREATE TABLE users (chatid INTEGER , name TEXT, click INTEGER, state INTEGER)")
cursor.execute("INSERT INTO users VALUES (1234, 'eee', 1,0)")
conn.commit()
sql = "SELECT * FROM users "
cursor.execute(sql)
data = cursor.fetchall()
str_data = json.dumps(data)
await bot.send_document(message.chat.id, io.StringIO(str_data))
await bot.send_message(message.chat.id, 'admin_id = {}'.format(message.chat.id))
await bot.send_message(message.chat.id, 'config_id = {}'.format(message.message_id+1))
Первым делом пытаемся получить из in memory бд данные пользователя, который отправил сообщение. Если ловим ошибку, то загружаем данные с бекапа сервера Telergam, заполняем нашу бд данными с бекапа и повторно пытаемся найти пользователя.
# Логика для пользователя
try:
sql = "SELECT * FROM users where chatid={}".format(message.chat.id)
cursor.execute(sql)
data = cursor.fetchone() # or use fetchone()
except Exception:
data = await get_data()
cursor.execute("CREATE TABLE users (chatid INTEGER , name TEXT, click INTEGER, state INTEGER)")
cursor.executemany("INSERT INTO users VALUES (?,?,?,?)", data)
conn.commit()
sql = "SELECT * FROM users where chatid={}".format(message.chat.id)
cursor.execute(sql)
data = cursor.fetchone() # or use fetchone()
Если мы нашли пользователя в бд, то обрабатываем кнопки:
Если не нашли пользователя, то написать ему ошибку.
#При нажатии кнопки клик увеличиваем значение click на один и сохраняем
if data is not None:
if message.text == 'Клик':
sql = "UPDATE users SET click = {} WHERE chatid = {}".format(data[2]+1,message.chat.id)
cursor.execute(sql)
conn.commit()
await bot.send_message(message.chat.id, 'Кликов: {} '.format(data[2]+1))
# При нажатии кнопки Рейтинг выводим пользователю топ 10
if message.text == 'Рейтинг':
sql = "SELECT * FROM users ORDER BY click DESC LIMIT 15"
cursor.execute(sql)
newlist = cursor.fetchall() # or use fetchone()
sql_count = "SELECT COUNT(chatid) FROM users"
cursor.execute(sql_count)
count=cursor.fetchone()
rating='Всего: {}n'.format(count[0])
i=1
for user in newlist:
rating=rating+str(i)+': '+user[1]+' - '+str(user[2])+'n'
i+=1
await bot.send_message(message.chat.id, rating)
else:
await bot.send_message(message.chat.id, 'Вы не зарегистрированы')
Пытаемся найти пользователя в бд, если его нет, то добавляем новую строку в таблицу и делаем бэкап.
Если ловим ошибку, то подгружаем последний бэкап, заполняем таблицу и повторяем попытку регистрации.
sql_select = "SELECT * FROM users where chatid={}".format(message.chat.id)
sql_insert = "INSERT INTO users VALUES ({}, '{}', {},{})".format(message.chat.id,message.chat.first_name, 0, 0)
try:
cursor.execute(sql_select)
data = cursor.fetchone()
if data is None:
cursor.execute(sql_insert)
conn.commit()
await save_data()
except Exception:
data = await get_data()
cursor.execute("CREATE TABLE users (chatid INTEGER , name TEXT, click INTEGER, state INTEGER)")
cursor.executemany("INSERT INTO users VALUES (?,?,?,?)", data)
conn.commit()
cursor.execute(sql_select)
data = cursor.fetchone()
if data is None:
cursor.execute(sql_insert)
conn.commit()
await save_data()
# Создаем кнопки
button = KeyboardButton('Клик')
button2 = KeyboardButton('Рейтинг')
# Добавляем
kb = ReplyKeyboardMarkup(resize_keyboard=True).add(button).add(button2)
# Отправляем сообщение с кнопкой
await bot.send_message(message.chat.id,'Приветствую {}'.format(message.chat.first_name),reply_markup=kb)
Так, ну и самое интересное.
Мы выгружаем все данные с таблицы пользователей, переводим словарь в строку и изменяем наш файл, который хранится на серверах Telegram.
#--------------------Сохранение данных-------------------------
async def save_data():
sql = "SELECT * FROM users "
cursor.execute(sql)
data = cursor.fetchall() # or use fetchone()
try:
# Переводим словарь в строку
str_data=json.dumps(data)
# Обновляем наш файл с данными
await bot.edit_message_media(InputMediaDocument(io.StringIO(str_data)), admin_id, config_id)
except Exception as ex:
print(ex)
Для того, чтобы получить бэкап нам нужно переслать сообщение с файлом от админа к админу. Затем получить путь к файл, считать данные по url и вернуть весь бэкап.
# #--------------------Получение данных-------------------------
async def get_data():
# Пересылаем сообщение в данными от админа к админу
forward_data = await bot.forward_message(admin_id, admin_id, config_id)
# Получаем путь к файлу, который переслали
file_data = await bot.get_file(forward_data.document.file_id)
# Получаем файл по url
file_url_data = bot.get_file_url(file_data.file_path)
# Считываем данные с файла
json_file= urlopen(file_url_data).read()
# Переводим данные из json в словарь и возвращаем
return json.loads(json_file)
Ну вот почти и все, осталось только написать таймер, чтобы делал бэкапы и протестировать бота.
Создаем поток, который каждые 30 секунд выполняет наш метод save_data()
def timer_start():
threading.Timer(30.0, timer_start).start()
try:
asyncio.run_coroutine_threadsafe(save_data(),bot.loop)
except Exception as exc:
pass
Ну и в главной программе мы запускаем таймер и самого бота.
#--------------------Запуск бота-------------------------
if __name__ == '__main__':
timer_start()
executor.start_polling(dp, skip_updates=True)
Так с кодом вроде бы разобрались, вот ссылка рабочего проекта на github [1].
Тесты проводились на серверах heroku с минимальными характеристиками инстансов. Так, что можно считать, что все тесты были выполнены в более менее равных условиях.
Графики сделаны по выборкам из ~100 запрос-ответов. И представлены средние показатели выборки.
В качестве базы данных на стороннем сервере использовался PostgreSQL на Amazon RDS с минимальными характеристиками.
При одном миллионе пользователей время бэкапов становится проблемой.
Размер бэкапа полностью зависит от вашей модели данных, в моем случае при одном миллионе пользователей получился файл с данными на 21 мегабайт.
Данный метод хранения данных имеет смысл для проектов до миллиона пользователей. То есть для прототипа или личного стартапа данный способ имеет право на жизнь.
В итоге мы получили полностью автономного кликера, независящий от удаленных баз данных.
Вот выше описанный проект, развернутый на heroku: @Clicker_fast_bot
Так же я реализовал более сложный проект с данной идеологией: @Random_friend_bot
Подобие чатвдвоем и чатрулет, но только в телеграмме.
Он ищет в радиусе 100 км человека противоположного пола для общения и реализует закрытый чат с новым собеседником.
Если будет интересно могу скинуть исходный код проекта. Так же если данная тема будет актуальна, то в следующей статье могу описать создание Rest api без внешних БД. То есть такой стек django-sqllite-Telegram.
Буду рад любой критике, спасибо за внимание!
Автор: LilHack
Источник [2]
Сайт-источник PVSM.RU: https://www.pvsm.ru
Путь до страницы источника: https://www.pvsm.ru/python/328063
Ссылки в тексте:
[1] ссылка рабочего проекта на github: https://github.com/akashkinKV/click-bot-telegram
[2] Источник: https://habr.com/ru/post/464865/?utm_source=habrahabr&utm_medium=rss&utm_campaign=464865
Нажмите здесь для печати.