对话持久性
我想实现它的最简单方法是查看PicklePersistence()。
我见过的字典的唯一示例是conversations = { name : { (user_id,user_id): state} }，其中name 是给ConversationHandler() 的那个，元组即键(user_id,user_id) 是你的机器人正在与之交谈的user_id state 是对话的状态。好吧，也许不是user_id，也许是chat_id，但我不能肯定地说，我需要更多的豚鼠。
为了处理元组密钥，python-telegram-bot 包含一些工具来帮助您处理：encode_conversations_to_json 和 decode_conversations_from_json。
这里，on_flush 是一个变量，用于告诉代码是否要在每次调用 update_conversation() 时保存，当设置为 False 或仅在退出程序时设置为 True
最后一个细节：目前以下代码仅保存和从数据库中检索，但没有替换或删除。
from telegram.ext import BasePersistence
from config import mongo_URI
from copy import deepcopy
from telegram.utils.helpers import decode_conversations_from_json, encode_conversations_to_json
import mongoengine
import json
from bson import json_util
class Conversations(mongoengine.Document):
obj = mongoengine.DictField()
meta = { 'collection': 'Conversations', 'ordering': ['-id']}
class MongoPersistence(BasePersistence):
def __init__(self):
super(MongoPersistence, self).__init__(store_user_data=False,
store_chat_data=False,
store_bot_data=False)
dbname = "persistencedb"
mongoengine.connect(host=mongo_URI, db=dbname)
self.conversation_collection = "Conversations"
self.conversations = None
self.on_flush = False
def get_conversations(self, name):
if self.conversations:
pass
else:
document = Conversations.objects()
if document.first() == None:
document = {}
else:
document = document.first()['obj']
conversations_json = json_util.dumps(document)
self.conversations = decode_conversations_from_json(conversations_json)
return self.conversations.get(name, {}).copy()
def update_conversation(self, name, key, new_state):
if self.conversations.setdefault(name, {}).get(key) == new_state:
return
self.conversations[name][key] = new_state
if not self.on_flush:
conversations_dic = json_util.loads(encode_conversations_to_json(self.conversations))
document = Conversations(obj=conversations_dic)
document.save()
def flush(self):
conversations_dic = json_util.loads(encode_conversations_to_json(self.conversations))
document = Conversations(obj=conversations_dic)
document.save()
mongoengine.disconnect()
注意！有时对话需要用户预先设置user_data，而此代码未按要求提供。
所有持久性
这是一个更完整的代码（数据库中仍然缺少替换文档）。
from telegram.ext import BasePersistence
from collections import defaultdict
from config import mongo_URI
from copy import deepcopy
from telegram.utils.helpers import decode_user_chat_data_from_json, decode_conversations_from_json, encode_conversations_to_json
import mongoengine
import json
from bson import json_util
class Conversations(mongoengine.Document):
obj = mongoengine.DictField()
meta = { 'collection': 'Conversations', 'ordering': ['-id']}
class UserData(mongoengine.Document):
obj = mongoengine.DictField()
meta = { 'collection': 'UserData', 'ordering': ['-id']}
class ChatData(mongoengine.Document):
obj = mongoengine.DictField()
meta = { 'collection': 'ChatData', 'ordering': ['-id']}
class BotData(mongoengine.Document):
obj = mongoengine.DictField()
meta = { 'collection': 'BotData', 'ordering': ['-id']}
class DBHelper():
"""Class to add and get documents from a mongo database using mongoengine
"""
def __init__(self, dbname="persistencedb"):
mongoengine.connect(host=mongo_URI, db=dbname)
def add_item(self, data, collection):
if collection == "Conversations":
document = Conversations(obj=data)
elif collection == "UserData":
document = UserData(obj=data)
elif collection == "chat_data_collection":
document = ChatData(obj=data)
else:
document = BotData(obj=data)
document.save()
def get_item(self, collection):
if collection == "Conversations":
document = Conversations.objects()
elif collection == "UserData":
document = UserData.objects()
elif collection == "ChatData":
document = ChatData.objects()
else:
document = BotData.objects()
if document.first() == None:
document = {}
else:
document = document.first()['obj']
return document
def close(self):
mongoengine.disconnect()
class DBPersistence(BasePersistence):
"""Uses DBHelper to make the bot persistant on a database.
It's heavily inspired on PicklePersistence from python-telegram-bot
"""
def __init__(self):
super(DBPersistence, self).__init__(store_user_data=True,
store_chat_data=True,
store_bot_data=True)
self.persistdb = "persistancedb"
self.conversation_collection = "Conversations"
self.user_data_collection = "UserData"
self.chat_data_collection = "ChatData"
self.bot_data_collection = "BotData"
self.db = DBHelper()
self.user_data = None
self.chat_data = None
self.bot_data = None
self.conversations = None
self.on_flush = False
def get_conversations(self, name):
if self.conversations:
pass
else:
conversations_json = json_util.dumps(self.db.get_item(self.conversation_collection))
self.conversations = decode_conversations_from_json(conversations_json)
return self.conversations.get(name, {}).copy()
def update_conversation(self, name, key, new_state):
if self.conversations.setdefault(name, {}).get(key) == new_state:
return
self.conversations[name][key] = new_state
if not self.on_flush:
conversations_json = json_util.loads(encode_conversations_to_json(self.conversations))
self.db.add_item(conversations_json, self.conversation_collection)
def get_user_data(self):
if self.user_data:
pass
else:
user_data_json = json_util.dumps(self.db.get_item(self.user_data_collection))
if user_data_json != '{}':
self.user_data = decode_user_chat_data_from_json(user_data_json)
else:
self.user_data = defaultdict(dict,{})
return deepcopy(self.user_data)
def update_user_data(self, user_id, data):
if self.user_data is None:
self.user_data = defaultdict(dict)
# comment next line if you want to save to db every time this function is called
if self.user_data.get(user_id) == data:
return
self.user_data[user_id] = data
if not self.on_flush:
user_data_json = json_util.loads(json.dumps(self.user_data))
self.db.add_item(user_data_json, self.user_data_collection)
def get_chat_data(self):
if self.chat_data:
pass
else:
chat_data_json = json_util.dumps(self.db.get_item(self.chat_data_collection))
if chat_data_json != "{}":
self.chat_data = decode_user_chat_data_from_json(chat_data_json)
else:
self.chat_data = defaultdict(dict,{})
return deepcopy(self.chat_data)
def update_chat_data(self, chat_id, data):
if self.chat_data is None:
self.chat_data = defaultdict(dict)
# comment next line if you want to save to db every time this function is called
if self.chat_data.get(chat_id) == data:
return
self.chat_data[chat_id] = data
if not self.on_flush:
chat_data_json = json_util.loads(json.dumps(self.chat_data))
self.db.add_item(chat_data_json, self.chat_data_collection)
def get_bot_data(self):
if self.bot_data:
pass
else:
bot_data_json = json_util.dumps(self.db.get_item(self.bot_data_collection))
self.bot_data = json.loads(bot_data_json)
return deepcopy(self.bot_data)
def update_bot_data(self, data):
if self.bot_data == data:
return
self.bot_data = data.copy()
if not self.on_flush:
bot_data_json = json_util.loads(json.dumps(self.bot_data))
self.db.add_item(self.bot_data, self.bot_data_collection)
def flush(self):
if self.conversations:
conversations_json = json_util.loads(encode_conversations_to_json(self.conversations))
self.db.add_item(conversations_json, self.conversation_collection)
if self.user_data:
user_data_json = json_util.loads(json.dumps(self.user_data))
self.db.add_item(user_data_json, self.user_data_collection)
if self.chat_data:
chat_data_json = json_util.loads(json.dumps(self.chat_data))
self.db.add_item(chat_data_json, self.chat_data_collection)
if self.bot_data:
bot_data_json = json_util.loads(json.dumps(self.bot_data))
self.db.add_item(self.bot_data, self.bot_data_collection)
self.db.close()
两个细节：
- Chat_data 持久性暂时没有保存到数据库中。需要更多的测试。也许那部分代码有错误。
- 目前，
on_flush = False 工作的唯一代码部分是在对话中。在所有其他更新中，似乎调用是在分配之后完成的，所以 if variable[key] == data 始终是 True 并在保存到数据库之前完成代码，这就是为什么有评论说 # comment next line if you want to save to db every time this function is called 但可以节省很多钱。如果您设置 on_flush = True 并且代码提前停止（例如进程被终止），您将不会在数据库中保存任何内容。