🕵🏻 🧘🏽 🔚 电报作为IT项目的数据仓库 🔇 🥇 👩🏼‍🤝‍👩🏻

下午好，今天，我想与大家分享编写小型IT项目时遇到的问题和不寻常的解决方案。我必须马上说，这篇文章是为那些精通Bot，数据库，SQL和python编程语言的电报开发人员而写的。

整个项目发布在github上，链接将在本文的结尾。

主要问题

最初，我想为自己编写一个简单的电报机器人卡路里计数器，该计数器从用户处接收一个数字，并返回当天剩余多少卡路里。也就是说，您需要为每个用户存储大约两个变量。

最后，您必须选择一种存储此数据的方法。

选项- 全局变量 ，随机存取存储器。该选项立即失败，因为当程序崩溃时，我们将丢失所有内容
选项- 写入磁盘上的文件 。对于这样的项目，它可能会起作用，但是我计划在heroku上部署一个机器人，该机器人每天都会擦除磁盘上的所有数据。所以这个选项不合适
选项-Google电子表格 。最初，我想讲解这个选项，但是我开始理解并意识到对表的查询数量是有限制的，为了开始使用该表，您需要编写一堆代码并找出它们不太简单的api
选项是数据库 。是的，这是所有方面的最佳选择。但是对于这样的项目，使用起来甚至很有趣。另外，在第三方服务器上部署和支持数据库将花费大量资金。

结果，这些选项都没有出现。当然，还有其他数十种方法，但是我希望它是免费，快速且最少的代码。

解决方案

这个想法很简单，对于数据存储，我们将在内存中使用sqllite数据库，因为它已经内置在python 3中，并且我们将表以很小的间隔（大约每30秒一次）备份到Telegram服务器，并在程序进程关闭时进行备份。

如果服务器崩溃，则在第一个请求时，我们将自动从Telegram服务器下载表并将数据恢复到sqllite。

您可以根据需要在内存数据库中使用其他任何数据库。

优点

性能 -由于使用主内存中的数据，程序的执行速度甚至比在第三方服务器上使用数据库时要快（执行和测试的速度图表将在结尾）
免费 -无需购买第三方数据库服务器，所有数据作为备份免费存储在Telegram服务器上
相对可靠 -如果服务器由于未知原因而崩溃，那么我们将丢失最近30秒钟（备份间隔时间）的最大数据，对于一个工作中的原型或小型项目来说，这就足够了。
切换到常规数据库时的最低成本 -您需要替换连接数据，删除备份代码并将表数据从备份转移到新数据库。

缺点

缺乏水平缩放
您需要在Telegram中使用两个帐户（一个用于管理员，另一个用于测试用户）
由于锁定，服务器无法在俄罗斯运行
在评论中，我想您还会发现其他许多细微差别。

该死

让我们编写一个简单的答题器并运行速度测试。

该机器人将使用与api电报航迹图交互的异步库以python编程语言编写。

首先要做的是填写机器人的设置，我不会告诉您如何从BotFather获取令牌，关于该主题的文章已有数百篇。

我们还需要管理员的第二个帐户来保存管理员的备份。

为了获取admin_id和config_id，我们需要从管理员帐户启动bot并将其写入“ 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”字样的消息，则我们将使用以下数据模型创建一个用户表：

chatid-唯一的聊天ID用户
名称-用户名
click-点击次数
state-状态机的值未在该项目中使用，但是如果没有更复杂的值就不能做

添加测试用户，并将文档与我们的表一起发送到Telegram服务器。我们还以消息形式将admin_id和config_id发送给管理员。收到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))

用户逻辑

首先，我们尝试从内存数据库中获取发送消息的用户的数据。如果发现错误，请从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()

如果我们在数据库中找到该用户，那么我们将处理以下按钮：

通过点击“点击”，我们更新此用户的点击次数
当您点击“评分”时，我们将显示点击次数最多的15个人的列表。

如果找不到用户，请给他写一个错误。

  #      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)

因此，最有趣的是。

从Telergam服务器保存和检索数据

我们从用户表中卸载所有数据，将字典转换为字符串，然后修改存储在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的链接。

怎么跑

从github下载项目。我们在任何python开发环境中启动该项目（例如：PyCharm）。
开发环境将自动从需求文件中加载必要的库。
从main.py中的BotFather替换令牌
我们开始这个项目
在第二个帐户中，单击/开始并输入单词“ admin”
关闭项目，然后在main.py文件中填写admin_id和config_id
我们启动项目，然后从用户帐户中按开始
获利

测试和图表

在具有最小实例特征的heroku服务器上进行了测试。因此，我们可以假设所有测试都在大致相同的条件下执行。

这些图是由〜100个请求答案的样本制成的。并给出了样本的平均值。

具有最小特征的Amazon RDS上的PostgreSQL被用作第三方服务器上的数据库。

对于一百万用户，备份时间成为一个问题。

备份的大小完全取决于您的数据模型，在我的情况下，拥有一百万个用户，就获得了一个21 MB的数据文件。

结论

这种数据存储方法适用于多达一百万个用户的项目。也就是说，对于原型或个人创业公司，此方法具有生命权。

结果，我们得到了一个完全独立的答题器，独立于远程数据库。

这是上述部署到heroku的项目：@Clicker_fast_bot

我还用这种思想实施了一个更复杂的项目：@Random_friend_bot

两人聊天室和聊天室的相似之处，但仅限于电报中。

他正在寻找与异性相距100公里的范围内的人进行交流，并与新的对话者进行了密谈。

如果有趣的话，我可以放弃该项目的源代码。另外，如果这个主题相关，那么在下一篇文章中，我将描述没有外部数据库的Rest api的创建。这就是django-sqllite-Telegram堆栈。

任何批评我都会很高兴，谢谢您的关注！

电报作为IT项目的数据仓库