State and Borg design patterns used in a Telegram wizard bot
up vote
8
down vote
favorite
1
I'm making a sort of a user interface for a telegram bot and my idea was to use State design pattern for making sort of a wizard for all the process, with user inputting data in each step or pressing /skip for skipping a single step or /cancel for ending everything.
It's working pretty well. I put the states/steps in a list so I can always get to the next one easily by a common method in the parent. Every state is a Borg, so I can instantiate a state everywhere and will always have the same values and methods.
I also made a Singleton class named Progress for putting all the input in the wizard process so I can store it and send it altogether at the end. I know you can use modules in Python as singleton, but I'm more used to this and I prefer not polluting the namespace. I also think this is easier for testing and using.
Being a Python beginner, I'm proud with this code, but I think it has some code smells and it can be improved in terms of design and best practices. I also made it for a single user, but haven't tested two persons accessing the bot simultaneously. I'd like to restrict transitions among states too.
step_states.py:
from abc import ABCMeta, abstractmethod
from telegram import InlineKeyboardButton, InlineKeyboardMarkup, ReplyKeyboardRemove
from telegram.ext import ConversationHandler
from my_package.amazon.search_indexes import SEARCH_INDEXES
from environments import get_blogs
OK = 'OK'
CANCEL = 'CANCEL'
DELETE = 'DELETE'
class StepState(metaclass=ABCMeta):
"""
This is a Borg class which shares state and methods so all the instances created for it are different references
but containing the same values. This is also true for children of the same class, so:
Children1() == Children1() # False, because references are different, but the values are the same.
Also if you change one, you change the other
Children1() == Children2() # False, references are different and you can change anyone without affecting the other
"""
states =
name = "state" # This is not used yet. Thinking of deleting it everywhere
allowed = # This is not used, but could be interesting to allow only some transitions. Just a sketch
# State is a Borg design pattern class
__shared_state = {}
def __init__(self):
self.__dict__ = self.__shared_state
def __eq__(self, other):
return self.__hash__() == other.__hash__()
def message(self, update):
"""
Returns the object holding the info coming from Telegram with the methods for replying and looking into this info
:param update: the Telegram update, probably a text message
:return: the update or query object
"""
if update.callback_query:
return update.callback_query.message
else:
return update.message
def draw_ui(self, bot, update):
"""
Draws the UI in Telegram (usually a custom keyboard) for the user to input data
:param bot: the Telegram bot
:param update: the Telegram update, probably a text message
"""
pass
@abstractmethod
def handle(self, bot=None, update=None):
"""
Handles all the input info into the Progress object and transitions to the next state
:param bot: the Telegram bot
:param update: the Telegram update, probably a text message
"""
self.progress = Progress()
# TODO How to make cancel() common to all states
#query = update.callback_query
#if query.data == CANCEL:
# return cancel(bot, update)
def next_state(self, new_state=None):
"""
Inherited method which decides to move to the next state, or to the specified state if any
:param new_state: the next state, if any
:return: the next state
"""
# TODO Check if the transition is allowed?
if new_state is None:
default_next_state_index = (self.index + 1) % len(self.states)
next_new_state = self.states[default_next_state_index]
# TODO logging
logger.info('Current state:', self, ' => switched to new state', next_new_state)
return next_new_state
else:
return new_state
@property
def index(self):
"""
Returns the index of this state within the list with all states
:return: the index as an int
"""
return StepState.states.index(self)
def __hash__(self):
return hash(self.__class__.name)
def __str__(self):
return self.__class__.name
class NoneState(StepState):
name = 'NoneState'
def draw_ui(self, bot, update):
pass
def handle(self, bot=None, update=None):
pass
class DepState(StepState):
name = 'DepState'
def draw_ui(self, bot, update):
reply_keyboard = [
]
num_cols = 3
current_row =
# Think of SEARCH_INDEXES as a list with objects having two strings: spanish_description and browse_node_id
for dep in SEARCH_INDEXES:
current_row.append(InlineKeyboardButton(dep.spanish_desc,
callback_data="{}={}".format(dep.browse_node_id, dep.spanish_desc)))
if len(current_row) >= num_cols:
reply_keyboard.append(current_row)
current_row =
reply_markup = InlineKeyboardMarkup(reply_keyboard)
self.message(update).reply_text('Enter the department or /cancel', reply_markup=reply_markup)
def handle(self, bot=None, update=None):
super().handle()
query = update.callback_query
if query.data:
current_input_dep, text = query.data.split('=')
bot.edit_message_text(text="Departament chosen: {}".format(text), chat_id=query.message.chat_id,
message_id=query.message.message_id)
# We optionally log anything
Progress().input_dep = current_input_dep
Progress().next_state()
Progress().state.draw_ui(bot, update)
return Progress().state
class NodeState(StepState):
name = 'NodeState'
def draw_ui(self, bot, update):
reply_keyboard = [
[InlineKeyboardButton(str(number), callback_data=str(number)) for number in range(5)],
[InlineKeyboardButton(str(number), callback_data=str(number)) for number in range(5, 10)],
[InlineKeyboardButton('Accept', callback_data=OK),
InlineKeyboardButton('Cancel', callback_data=CANCEL)],
]
reply_markup = InlineKeyboardMarkup(reply_keyboard)
self.message(update).reply_text('Input numbers for the node or /cancel', reply_markup=reply_markup)
# We get the reference to the message we will use to update the value and set into Progress
Progress().tracking_message = self.message(update).reply_text("Typed data : {}".format(''))
def handle(self, bot=None, update=None):
super().handle()
query = update.callback_query
if query.data == OK:
# TODO Validate input data
#logger.info('Input data: {}'.format(self.progress.input_node))
bot.edit_message_text(text="Chosen node: {}".format(self.progress.input_node), chat_id=query.message.chat_id,
message_id=query.message.message_id)
bot.delete_message(chat_id=Progress().tracking_message.chat_id,
message_id=Progress().tracking_message.message_id)
Progress().next_state()
Progress().state.draw_ui(bot, update)
return Progress().state
else:
if not self.progress.input_node:
self.progress.input_node = ''
if query.data == DELETE:
if len(self.progress.input_node):
self.progress.input_node = self.progress.input_node[:-1]
else:
self.progress.input_node += query.data
# We update the output so the user sees if he types correctly
bot.edit_message_text(text="Input data: {}".format(self.progress.input_node),
chat_id=query.message.chat_id,
message_id=Progress().tracking_message.message_id)
class BlogState(StepState):
name = 'BlogState'
def draw_ui(self, bot, update):
reply_keyboard = [
[InlineKeyboardButton(blog, callback_data='{}={}'.format(blog, bid)) for blog, bid in get_blogs().items()]
]
reply_markup = InlineKeyboardMarkup(reply_keyboard)
self.message(update).reply_text('Choose an option, /skip or /cancel', reply_markup=reply_markup)
def handle(self, bot=None, update=None):
super().handle()
query = update.callback_query
if query.data:
bname, bid = query.data.split('=')
bot.edit_message_text(text="Chosen option: {}".format(bname), chat_id=query.message.chat_id,
message_id=query.message.message_id)
# We optionally log anything
# We text the user the requirements for next state
# query.message.reply_text('Departamento elegido. ', reply_markup=ReplyKeyboardRemove())
Progress().input_blog = (bname,bid)
Progress().next_state()
Progress().state.draw_ui(bot, update)
return Progress().state
class StartTimeState(StepState):
name = 'StartTimeState'
def draw_ui(self, bot, update):
reply_keyboard = [
[InlineKeyboardButton('Delete', callback_data=DELETE)],
[InlineKeyboardButton(str(number), callback_data=str(number)) for number in range(5)],
[InlineKeyboardButton(str(number), callback_data=str(number)) for number in range(5, 10)],
[InlineKeyboardButton(':', callback_data=':')],
[InlineKeyboardButton('Accept', callback_data=OK),
InlineKeyboardButton('Cancel', callback_data=CANCEL)],
]
reply_markup = InlineKeyboardMarkup(reply_keyboard)
self.message(update).reply_text('Input start time or /cancel', reply_markup=reply_markup)
# We get the reference to the message we will use to update the value and set into Progress
Progress().tracking_message = self.message(update).reply_text("Start time: {}".format(''))
def handle(self, bot=None, update=None):
super().handle()
query = update.callback_query
if query.data == OK:
# TODO Validate input data
#logger.info('Intpu data: {}'.format(self.progress.input_start_time))
bot.delete_message(chat_id=query.message.chat_id, message_id=query.message.message_id)
#bot.edit_message_text(text="Input start time: {}".format(self.progress.input_start_time),chat_id=query.message.chat_id,message_id=query.message.message_id)
#bot.delete_message(chat_id=Progress().tracking_message.chat_id,message_id=Progress().tracking_message.message_id)
Progress().next_state()
Progress().state.draw_ui(bot, update)
return Progress().state
else:
if not self.progress.input_start_time:
self.progress.input_start_time = ''
if query.data == DELETE:
if len(self.progress.input_start_time):
self.progress.input_start_time = self.progress.input_start_time[:-1]
else:
self.progress.input_start_time += query.data
# We update the output so the user sees if he types correctly
bot.edit_message_text(text="Start time: {}".format(self.progress.input_start_time),
chat_id=query.message.chat_id,
message_id=Progress().tracking_message.message_id)
# ... more states
# ... more and more states
# ...and so on, you can imagine
class KeywordsState(StepState):
name = 'KeywordsState'
def draw_ui(self, bot, update):
self.message(update).reply_text('Enter keywords', reply_markup=ReplyKeyboardRemove())
def handle(self, bot=None, update=None):
#logger.info("Keywords: %s", update.message.text)
# TODO Store in a variable or similar the keywords
update.message.reply_text(
'Thanks. The keywods are: {}'.format(update.message.text))
Progress().next_state()
Progress().state.draw_ui(bot, update)
return Progress().state
class ConfirmState(StepState):
name = 'ConfirmState'
def draw_ui(self, bot, update):
self.message(update).reply_text('Confirm all this info: {!s}'.format(Progress()))
reply_keyboard = [
[InlineKeyboardButton('Accept', callback_data=OK),
InlineKeyboardButton('Cancel', callback_data=CANCEL)],
]
reply_markup = InlineKeyboardMarkup(reply_keyboard)
self.message(update).reply_text('Accep or cancel (/cancel)', reply_markup=reply_markup)
def handle(self, bot=None, update=None):
query = update.callback_query
self.message(update).reply_text('Finalizado')
# Se lanza el procesamiento
# Se resetea el progreso
self.progress.clear()
Progress().next_state(NoneState())
return ConversationHandler.END
# This is important. It's intended for setting the steps in an ordered-sorted way
StepState.states = [NoneState(), DepState(), NodeState(), BlogState(), IntervalState(), StartTimeState(), EndTimeState(), RepeatsState(), LabelsState(), KeywordsState(), ConfirmState()]
class Progress(object):
"""
This singleton class contains the whole information collected along all the wizard process
"""
__instance = None
def __new__(cls):
if Progress.__instance is None:
Progress.__instance = object.__new__(cls)
Progress.__instance.input_blog = None
Progress.__instance.state = NoneState()
Progress.__instance.tracking_message = None
Progress.__instance.input_dep = None
Progress.__instance.input_node = None
Progress.__instance.input_minutes = None
Progress.__instance.input_start_time = None
Progress.__instance.input_end_time = None
return Progress.__instance
def clear(self):
"""
Resets the progress
"""
self.input_blog = None
self.state = NoneState()
self.tracking_message = None
self.input_dep = None
self.input_node = None
self.input_minutes = None
self.input_start_time = None
self.input_end_time = None
def next_state(self, new_state=None):
"""
Moves to the next state or to the specified state if any
:param new_state: the next state to move to
:return: the next state, already set to the Progress object
"""
if new_state is not None:
if self.state:
self.state = self.state.next_state(new_state)
else:
self.state = NoneState
#logger('Couldnt go to next state')
else:
self.state = self.state.next_state()
def __str__(self):
return '{}: dep={}, node={}, start={},end={}'.format(self.__class__.__name__, self.input_dep, self.input_node, self.input_start_time, self.input_end_time)
Now we use the previous code here:
bot_prototype:
#!/usr/bin/python3
from telegram import ReplyKeyboardMarkup, ReplyKeyboardRemove
from telegram.ext import (Updater, Filters, RegexHandler, CommandHandler,
CallbackQueryHandler, ConversationHandler, MessageHandler)
from my_package.step_states import DepState, NodeState, StartTimeState, EndTimeState, KeywordsState, ConfirmState,
BlogState, IntervalState
from my_package.step_states import Progress
from environments import get_bot_token
import logging
logging.basicConfig(format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
level=logging.INFO)
logger = logging.getLogger(__name__)
OK = 'OK'
CANCEL = 'CANCEL'
DELETE = 'DELETE'
# Updater is telegram code and the bot_token is an id string
updater = Updater(get_bot_token())
def start(bot, update):
"""
Starts the wizard planning process
:param bot: the Telegram bot
:param update: the Telegram update, probably a text message
:return: the state for the ConversationHandler
"""
p = Progress()
p.next_state()
p.state.draw_ui(bot, update)
return p.state
def error(bot, update, error):
"""
Logs errors
:param bot: the Telegram bot
:param update: the Telegram update, probably a text message
"""
logger.error('Update "%s" caused error "%s"', update, error)
def skip(bot, update):
"""
Skips this step in the wizard process
:param bot: the Telegram bot
:param update: the Telegram update, probably a text message
:return: the next state skipping the present one
"""
user = update.message.from_user
logger.info("%s skipped a step.", user.first_name)
update.message.reply_text("Skipping steps is not supported yet. Process is done")
# If we don't cancel at the end, we should remove any keyboard which could be present
# TODO Right now we don't accept skips and we cancel everything.
# TODO In the future, we will look at present state and choose the next state
return cancel(bot, update)
def cancel(bot, update):
"""
Cancels the whole wizard process
:param bot: the Telegram bot
:param update: the Telegram update, probably a text message
:return: the END state for he ConversationHandler
"""
Progress().clear()
user = update.message.from_user
logger.info("{} cancelled the process.".format(user.first_name))
update.message.reply_text('Process ended.', reply_markup=ReplyKeyboardRemove())
return ConversationHandler.END
def main():
global updater
dp = updater.dispatcher
# Add conversation handler with the states
# The telegram conversation handler needs a list of handlers with function so it can execute desired code in each state/step
conv_handler = ConversationHandler(
entry_points=[ CommandHandler('start', start) ],
states={
DepState(): [CallbackQueryHandler(DepState().handle), CommandHandler('skip', skip)],
NodeState(): [CallbackQueryHandler(NodeState().handle), CommandHandler('skip', skip)],
BlogState(): [CallbackQueryHandler(BlogState().handle), CommandHandler('skip', skip)],
IntervalState(): [CallbackQueryHandler(IntervalState().handle), CommandHandler('skip', skip)],
StartTimeState(): [CallbackQueryHandler(StartTimeState().handle), CommandHandler('skip', skip)],
EndTimeState(): [CallbackQueryHandler(EndTimeState().handle), CommandHandler('skip', skip)],
KeywordsState(): [MessageHandler(Filters.text, KeywordsState().handle), CommandHandler('skip', skip)],
ConfirmState(): [CallbackQueryHandler(ConfirmState().handle)],
},
fallbacks=[CommandHandler('cancel', cancel)]
)
dp.add_handler(conv_handler)
# dp.add_handler(CallbackQueryHandler(button_callback))
updater.dispatcher.add_error_handler(error)
# Start the Bot
updater.start_polling()
# Run the bot until you press Ctrl-C or the process receives SIGINT,
# SIGTERM or SIGABRT. This should be used most of the time, since
# start_polling() is non-blocking and will stop the bot gracefully.
updater.idle()
if __name__ == '__main__':
main()
python beginner python-3.x design-patterns
edited Jan 26 at 13:41
Mast
7,43963686
asked Jan 16 at 15:52
madtyn
1415
bumped to the homepage by Community♦ 11 mins ago
This question has answers that may be good or bad; the system has marked it active so that they can be reviewed.
add a comment |
up vote
8
down vote
favorite
1
I'm making a sort of a user interface for a telegram bot and my idea was to use State design pattern for making sort of a wizard for all the process, with user inputting data in each step or pressing /skip for skipping a single step or /cancel for ending everything.
It's working pretty well. I put the states/steps in a list so I can always get to the next one easily by a common method in the parent. Every state is a Borg, so I can instantiate a state everywhere and will always have the same values and methods.
I also made a Singleton class named Progress for putting all the input in the wizard process so I can store it and send it altogether at the end. I know you can use modules in Python as singleton, but I'm more used to this and I prefer not polluting the namespace. I also think this is easier for testing and using.
Being a Python beginner, I'm proud with this code, but I think it has some code smells and it can be improved in terms of design and best practices. I also made it for a single user, but haven't tested two persons accessing the bot simultaneously. I'd like to restrict transitions among states too.
step_states.py:
from abc import ABCMeta, abstractmethod
from telegram import InlineKeyboardButton, InlineKeyboardMarkup, ReplyKeyboardRemove
from telegram.ext import ConversationHandler
from my_package.amazon.search_indexes import SEARCH_INDEXES
from environments import get_blogs
OK = 'OK'
CANCEL = 'CANCEL'
DELETE = 'DELETE'
class StepState(metaclass=ABCMeta):
"""
This is a Borg class which shares state and methods so all the instances created for it are different references
but containing the same values. This is also true for children of the same class, so:
Children1() == Children1() # False, because references are different, but the values are the same.
Also if you change one, you change the other
Children1() == Children2() # False, references are different and you can change anyone without affecting the other
"""
states =
name = "state" # This is not used yet. Thinking of deleting it everywhere
allowed = # This is not used, but could be interesting to allow only some transitions. Just a sketch
# State is a Borg design pattern class
__shared_state = {}
def __init__(self):
self.__dict__ = self.__shared_state
def __eq__(self, other):
return self.__hash__() == other.__hash__()
def message(self, update):
"""
Returns the object holding the info coming from Telegram with the methods for replying and looking into this info
:param update: the Telegram update, probably a text message
:return: the update or query object
"""
if update.callback_query:
return update.callback_query.message
else:
return update.message
def draw_ui(self, bot, update):
"""
Draws the UI in Telegram (usually a custom keyboard) for the user to input data
:param bot: the Telegram bot
:param update: the Telegram update, probably a text message
"""
pass
@abstractmethod
def handle(self, bot=None, update=None):
"""
Handles all the input info into the Progress object and transitions to the next state
:param bot: the Telegram bot
:param update: the Telegram update, probably a text message
"""
self.progress = Progress()
# TODO How to make cancel() common to all states
#query = update.callback_query
#if query.data == CANCEL:
# return cancel(bot, update)
def next_state(self, new_state=None):
"""
Inherited method which decides to move to the next state, or to the specified state if any
:param new_state: the next state, if any
:return: the next state
"""
# TODO Check if the transition is allowed?
if new_state is None:
default_next_state_index = (self.index + 1) % len(self.states)
next_new_state = self.states[default_next_state_index]
# TODO logging
logger.info('Current state:', self, ' => switched to new state', next_new_state)
return next_new_state
else:
return new_state
@property
def index(self):
"""
Returns the index of this state within the list with all states
:return: the index as an int
"""
return StepState.states.index(self)
def __hash__(self):
return hash(self.__class__.name)
def __str__(self):
return self.__class__.name
class NoneState(StepState):
name = 'NoneState'
def draw_ui(self, bot, update):
pass
def handle(self, bot=None, update=None):
pass
class DepState(StepState):
name = 'DepState'
def draw_ui(self, bot, update):
reply_keyboard = [
]
num_cols = 3
current_row =
# Think of SEARCH_INDEXES as a list with objects having two strings: spanish_description and browse_node_id
for dep in SEARCH_INDEXES:
current_row.append(InlineKeyboardButton(dep.spanish_desc,
callback_data="{}={}".format(dep.browse_node_id, dep.spanish_desc)))
if len(current_row) >= num_cols:
reply_keyboard.append(current_row)
current_row =
reply_markup = InlineKeyboardMarkup(reply_keyboard)
self.message(update).reply_text('Enter the department or /cancel', reply_markup=reply_markup)
def handle(self, bot=None, update=None):
super().handle()
query = update.callback_query
if query.data:
current_input_dep, text = query.data.split('=')
bot.edit_message_text(text="Departament chosen: {}".format(text), chat_id=query.message.chat_id,
message_id=query.message.message_id)
# We optionally log anything
Progress().input_dep = current_input_dep
Progress().next_state()
Progress().state.draw_ui(bot, update)
return Progress().state
class NodeState(StepState):
name = 'NodeState'
def draw_ui(self, bot, update):
reply_keyboard = [
[InlineKeyboardButton(str(number), callback_data=str(number)) for number in range(5)],
[InlineKeyboardButton(str(number), callback_data=str(number)) for number in range(5, 10)],
[InlineKeyboardButton('Accept', callback_data=OK),
InlineKeyboardButton('Cancel', callback_data=CANCEL)],
]
reply_markup = InlineKeyboardMarkup(reply_keyboard)
self.message(update).reply_text('Input numbers for the node or /cancel', reply_markup=reply_markup)
# We get the reference to the message we will use to update the value and set into Progress
Progress().tracking_message = self.message(update).reply_text("Typed data : {}".format(''))
def handle(self, bot=None, update=None):
super().handle()
query = update.callback_query
if query.data == OK:
# TODO Validate input data
#logger.info('Input data: {}'.format(self.progress.input_node))
bot.edit_message_text(text="Chosen node: {}".format(self.progress.input_node), chat_id=query.message.chat_id,
message_id=query.message.message_id)
bot.delete_message(chat_id=Progress().tracking_message.chat_id,
message_id=Progress().tracking_message.message_id)
Progress().next_state()
Progress().state.draw_ui(bot, update)
return Progress().state
else:
if not self.progress.input_node:
self.progress.input_node = ''
if query.data == DELETE:
if len(self.progress.input_node):
self.progress.input_node = self.progress.input_node[:-1]
else:
self.progress.input_node += query.data
# We update the output so the user sees if he types correctly
bot.edit_message_text(text="Input data: {}".format(self.progress.input_node),
chat_id=query.message.chat_id,
message_id=Progress().tracking_message.message_id)
class BlogState(StepState):
name = 'BlogState'
def draw_ui(self, bot, update):
reply_keyboard = [
[InlineKeyboardButton(blog, callback_data='{}={}'.format(blog, bid)) for blog, bid in get_blogs().items()]
]
reply_markup = InlineKeyboardMarkup(reply_keyboard)
self.message(update).reply_text('Choose an option, /skip or /cancel', reply_markup=reply_markup)
def handle(self, bot=None, update=None):
super().handle()
query = update.callback_query
if query.data:
bname, bid = query.data.split('=')
bot.edit_message_text(text="Chosen option: {}".format(bname), chat_id=query.message.chat_id,
message_id=query.message.message_id)
# We optionally log anything
# We text the user the requirements for next state
# query.message.reply_text('Departamento elegido. ', reply_markup=ReplyKeyboardRemove())
Progress().input_blog = (bname,bid)
Progress().next_state()
Progress().state.draw_ui(bot, update)
return Progress().state
class StartTimeState(StepState):
name = 'StartTimeState'
def draw_ui(self, bot, update):
reply_keyboard = [
[InlineKeyboardButton('Delete', callback_data=DELETE)],
[InlineKeyboardButton(str(number), callback_data=str(number)) for number in range(5)],
[InlineKeyboardButton(str(number), callback_data=str(number)) for number in range(5, 10)],
[InlineKeyboardButton(':', callback_data=':')],
[InlineKeyboardButton('Accept', callback_data=OK),
InlineKeyboardButton('Cancel', callback_data=CANCEL)],
]
reply_markup = InlineKeyboardMarkup(reply_keyboard)
self.message(update).reply_text('Input start time or /cancel', reply_markup=reply_markup)
# We get the reference to the message we will use to update the value and set into Progress
Progress().tracking_message = self.message(update).reply_text("Start time: {}".format(''))
def handle(self, bot=None, update=None):
super().handle()
query = update.callback_query
if query.data == OK:
# TODO Validate input data
#logger.info('Intpu data: {}'.format(self.progress.input_start_time))
bot.delete_message(chat_id=query.message.chat_id, message_id=query.message.message_id)
#bot.edit_message_text(text="Input start time: {}".format(self.progress.input_start_time),chat_id=query.message.chat_id,message_id=query.message.message_id)
#bot.delete_message(chat_id=Progress().tracking_message.chat_id,message_id=Progress().tracking_message.message_id)
Progress().next_state()
Progress().state.draw_ui(bot, update)
return Progress().state
else:
if not self.progress.input_start_time:
self.progress.input_start_time = ''
if query.data == DELETE:
if len(self.progress.input_start_time):
self.progress.input_start_time = self.progress.input_start_time[:-1]
else:
self.progress.input_start_time += query.data
# We update the output so the user sees if he types correctly
bot.edit_message_text(text="Start time: {}".format(self.progress.input_start_time),
chat_id=query.message.chat_id,
message_id=Progress().tracking_message.message_id)
# ... more states
# ... more and more states
# ...and so on, you can imagine
class KeywordsState(StepState):
name = 'KeywordsState'
def draw_ui(self, bot, update):
self.message(update).reply_text('Enter keywords', reply_markup=ReplyKeyboardRemove())
def handle(self, bot=None, update=None):
#logger.info("Keywords: %s", update.message.text)
# TODO Store in a variable or similar the keywords
update.message.reply_text(
'Thanks. The keywods are: {}'.format(update.message.text))
Progress().next_state()
Progress().state.draw_ui(bot, update)
return Progress().state
class ConfirmState(StepState):
name = 'ConfirmState'
def draw_ui(self, bot, update):
self.message(update).reply_text('Confirm all this info: {!s}'.format(Progress()))
reply_keyboard = [
[InlineKeyboardButton('Accept', callback_data=OK),
InlineKeyboardButton('Cancel', callback_data=CANCEL)],
]
reply_markup = InlineKeyboardMarkup(reply_keyboard)
self.message(update).reply_text('Accep or cancel (/cancel)', reply_markup=reply_markup)
def handle(self, bot=None, update=None):
query = update.callback_query
self.message(update).reply_text('Finalizado')
# Se lanza el procesamiento
# Se resetea el progreso
self.progress.clear()
Progress().next_state(NoneState())
return ConversationHandler.END
# This is important. It's intended for setting the steps in an ordered-sorted way
StepState.states = [NoneState(), DepState(), NodeState(), BlogState(), IntervalState(), StartTimeState(), EndTimeState(), RepeatsState(), LabelsState(), KeywordsState(), ConfirmState()]
class Progress(object):
"""
This singleton class contains the whole information collected along all the wizard process
"""
__instance = None
def __new__(cls):
if Progress.__instance is None:
Progress.__instance = object.__new__(cls)
Progress.__instance.input_blog = None
Progress.__instance.state = NoneState()
Progress.__instance.tracking_message = None
Progress.__instance.input_dep = None
Progress.__instance.input_node = None
Progress.__instance.input_minutes = None
Progress.__instance.input_start_time = None
Progress.__instance.input_end_time = None
return Progress.__instance
def clear(self):
"""
Resets the progress
"""
self.input_blog = None
self.state = NoneState()
self.tracking_message = None
self.input_dep = None
self.input_node = None
self.input_minutes = None
self.input_start_time = None
self.input_end_time = None
def next_state(self, new_state=None):
"""
Moves to the next state or to the specified state if any
:param new_state: the next state to move to
:return: the next state, already set to the Progress object
"""
if new_state is not None:
if self.state:
self.state = self.state.next_state(new_state)
else:
self.state = NoneState
#logger('Couldnt go to next state')
else:
self.state = self.state.next_state()
def __str__(self):
return '{}: dep={}, node={}, start={},end={}'.format(self.__class__.__name__, self.input_dep, self.input_node, self.input_start_time, self.input_end_time)
Now we use the previous code here:
bot_prototype:
#!/usr/bin/python3
from telegram import ReplyKeyboardMarkup, ReplyKeyboardRemove
from telegram.ext import (Updater, Filters, RegexHandler, CommandHandler,
CallbackQueryHandler, ConversationHandler, MessageHandler)
from my_package.step_states import DepState, NodeState, StartTimeState, EndTimeState, KeywordsState, ConfirmState,
BlogState, IntervalState
from my_package.step_states import Progress
from environments import get_bot_token
import logging
logging.basicConfig(format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
level=logging.INFO)
logger = logging.getLogger(__name__)
OK = 'OK'
CANCEL = 'CANCEL'
DELETE = 'DELETE'
# Updater is telegram code and the bot_token is an id string
updater = Updater(get_bot_token())
def start(bot, update):
"""
Starts the wizard planning process
:param bot: the Telegram bot
:param update: the Telegram update, probably a text message
:return: the state for the ConversationHandler
"""
p = Progress()
p.next_state()
p.state.draw_ui(bot, update)
return p.state
def error(bot, update, error):
"""
Logs errors
:param bot: the Telegram bot
:param update: the Telegram update, probably a text message
"""
logger.error('Update "%s" caused error "%s"', update, error)
def skip(bot, update):
"""
Skips this step in the wizard process
:param bot: the Telegram bot
:param update: the Telegram update, probably a text message
:return: the next state skipping the present one
"""
user = update.message.from_user
logger.info("%s skipped a step.", user.first_name)
update.message.reply_text("Skipping steps is not supported yet. Process is done")
# If we don't cancel at the end, we should remove any keyboard which could be present
# TODO Right now we don't accept skips and we cancel everything.
# TODO In the future, we will look at present state and choose the next state
return cancel(bot, update)
def cancel(bot, update):
"""
Cancels the whole wizard process
:param bot: the Telegram bot
:param update: the Telegram update, probably a text message
:return: the END state for he ConversationHandler
"""
Progress().clear()
user = update.message.from_user
logger.info("{} cancelled the process.".format(user.first_name))
update.message.reply_text('Process ended.', reply_markup=ReplyKeyboardRemove())
return ConversationHandler.END
def main():
global updater
dp = updater.dispatcher
# Add conversation handler with the states
# The telegram conversation handler needs a list of handlers with function so it can execute desired code in each state/step
conv_handler = ConversationHandler(
entry_points=[ CommandHandler('start', start) ],
states={
DepState(): [CallbackQueryHandler(DepState().handle), CommandHandler('skip', skip)],
NodeState(): [CallbackQueryHandler(NodeState().handle), CommandHandler('skip', skip)],
BlogState(): [CallbackQueryHandler(BlogState().handle), CommandHandler('skip', skip)],
IntervalState(): [CallbackQueryHandler(IntervalState().handle), CommandHandler('skip', skip)],
StartTimeState(): [CallbackQueryHandler(StartTimeState().handle), CommandHandler('skip', skip)],
EndTimeState(): [CallbackQueryHandler(EndTimeState().handle), CommandHandler('skip', skip)],
KeywordsState(): [MessageHandler(Filters.text, KeywordsState().handle), CommandHandler('skip', skip)],
ConfirmState(): [CallbackQueryHandler(ConfirmState().handle)],
},
fallbacks=[CommandHandler('cancel', cancel)]
)
dp.add_handler(conv_handler)
# dp.add_handler(CallbackQueryHandler(button_callback))
updater.dispatcher.add_error_handler(error)
# Start the Bot
updater.start_polling()
# Run the bot until you press Ctrl-C or the process receives SIGINT,
# SIGTERM or SIGABRT. This should be used most of the time, since
# start_polling() is non-blocking and will stop the bot gracefully.
updater.idle()
if __name__ == '__main__':
main()
python beginner python-3.x design-patterns
edited Jan 26 at 13:41
Mast
7,43963686
asked Jan 16 at 15:52
madtyn
1415
bumped to the homepage by Community♦ 11 mins ago
This question has answers that may be good or bad; the system has marked it active so that they can be reviewed.
Please do not update the code in your question to incorporate feedback from answers, doing so goes against the Question + Answer style of Code Review. This is not a forum where you should keep the most updated version in your question. Please see what you may and may not do after receiving answers.
– Mast
Jan 26 at 13:41
@Mast I was only formatting and changing indentation. Also I deleted some comments. All of it was done to make it easy to read and understand, but I didn't any refactoring, changed any method or added anything.
– madtyn
Jan 27 at 11:35
Feel free to post a follow-up question when you've made significant changes. Leave the code in this question as-is.
– Mast
Jan 27 at 13:50
add a comment |
up vote
8
down vote
favorite
1
up vote
8
down vote
favorite
1
1
I'm making a sort of a user interface for a telegram bot and my idea was to use State design pattern for making sort of a wizard for all the process, with user inputting data in each step or pressing /skip for skipping a single step or /cancel for ending everything.
It's working pretty well. I put the states/steps in a list so I can always get to the next one easily by a common method in the parent. Every state is a Borg, so I can instantiate a state everywhere and will always have the same values and methods.
I also made a Singleton class named Progress for putting all the input in the wizard process so I can store it and send it altogether at the end. I know you can use modules in Python as singleton, but I'm more used to this and I prefer not polluting the namespace. I also think this is easier for testing and using.
Being a Python beginner, I'm proud with this code, but I think it has some code smells and it can be improved in terms of design and best practices. I also made it for a single user, but haven't tested two persons accessing the bot simultaneously. I'd like to restrict transitions among states too.
step_states.py:
from abc import ABCMeta, abstractmethod
from telegram import InlineKeyboardButton, InlineKeyboardMarkup, ReplyKeyboardRemove
from telegram.ext import ConversationHandler
from my_package.amazon.search_indexes import SEARCH_INDEXES
from environments import get_blogs
OK = 'OK'
CANCEL = 'CANCEL'
DELETE = 'DELETE'
class StepState(metaclass=ABCMeta):
"""
This is a Borg class which shares state and methods so all the instances created for it are different references
but containing the same values. This is also true for children of the same class, so:
Children1() == Children1() # False, because references are different, but the values are the same.
Also if you change one, you change the other
Children1() == Children2() # False, references are different and you can change anyone without affecting the other
"""
states =
name = "state" # This is not used yet. Thinking of deleting it everywhere
allowed = # This is not used, but could be interesting to allow only some transitions. Just a sketch
# State is a Borg design pattern class
__shared_state = {}
def __init__(self):
self.__dict__ = self.__shared_state
def __eq__(self, other):
return self.__hash__() == other.__hash__()
def message(self, update):
"""
Returns the object holding the info coming from Telegram with the methods for replying and looking into this info
:param update: the Telegram update, probably a text message
:return: the update or query object
"""
if update.callback_query:
return update.callback_query.message
else:
return update.message
def draw_ui(self, bot, update):
"""
Draws the UI in Telegram (usually a custom keyboard) for the user to input data
:param bot: the Telegram bot
:param update: the Telegram update, probably a text message
"""
pass
@abstractmethod
def handle(self, bot=None, update=None):
"""
Handles all the input info into the Progress object and transitions to the next state
:param bot: the Telegram bot
:param update: the Telegram update, probably a text message
"""
self.progress = Progress()
# TODO How to make cancel() common to all states
#query = update.callback_query
#if query.data == CANCEL:
# return cancel(bot, update)
def next_state(self, new_state=None):
"""
Inherited method which decides to move to the next state, or to the specified state if any
:param new_state: the next state, if any
:return: the next state
"""
# TODO Check if the transition is allowed?
if new_state is None:
default_next_state_index = (self.index + 1) % len(self.states)
next_new_state = self.states[default_next_state_index]
# TODO logging
logger.info('Current state:', self, ' => switched to new state', next_new_state)
return next_new_state
else:
return new_state
@property
def index(self):
"""
Returns the index of this state within the list with all states
:return: the index as an int
"""
return StepState.states.index(self)
def __hash__(self):
return hash(self.__class__.name)
def __str__(self):
return self.__class__.name
class NoneState(StepState):
name = 'NoneState'
def draw_ui(self, bot, update):
pass
def handle(self, bot=None, update=None):
pass
class DepState(StepState):
name = 'DepState'
def draw_ui(self, bot, update):
reply_keyboard = [
]
num_cols = 3
current_row =
# Think of SEARCH_INDEXES as a list with objects having two strings: spanish_description and browse_node_id
for dep in SEARCH_INDEXES:
current_row.append(InlineKeyboardButton(dep.spanish_desc,
callback_data="{}={}".format(dep.browse_node_id, dep.spanish_desc)))
if len(current_row) >= num_cols:
reply_keyboard.append(current_row)
current_row =
reply_markup = InlineKeyboardMarkup(reply_keyboard)
self.message(update).reply_text('Enter the department or /cancel', reply_markup=reply_markup)
def handle(self, bot=None, update=None):
super().handle()
query = update.callback_query
if query.data:
current_input_dep, text = query.data.split('=')
bot.edit_message_text(text="Departament chosen: {}".format(text), chat_id=query.message.chat_id,
message_id=query.message.message_id)
# We optionally log anything
Progress().input_dep = current_input_dep
Progress().next_state()
Progress().state.draw_ui(bot, update)
return Progress().state
class NodeState(StepState):
name = 'NodeState'
def draw_ui(self, bot, update):
reply_keyboard = [
[InlineKeyboardButton(str(number), callback_data=str(number)) for number in range(5)],
[InlineKeyboardButton(str(number), callback_data=str(number)) for number in range(5, 10)],
[InlineKeyboardButton('Accept', callback_data=OK),
InlineKeyboardButton('Cancel', callback_data=CANCEL)],
]
reply_markup = InlineKeyboardMarkup(reply_keyboard)
self.message(update).reply_text('Input numbers for the node or /cancel', reply_markup=reply_markup)
# We get the reference to the message we will use to update the value and set into Progress
Progress().tracking_message = self.message(update).reply_text("Typed data : {}".format(''))
def handle(self, bot=None, update=None):
super().handle()
query = update.callback_query
if query.data == OK:
# TODO Validate input data
#logger.info('Input data: {}'.format(self.progress.input_node))
bot.edit_message_text(text="Chosen node: {}".format(self.progress.input_node), chat_id=query.message.chat_id,
message_id=query.message.message_id)
bot.delete_message(chat_id=Progress().tracking_message.chat_id,
message_id=Progress().tracking_message.message_id)
Progress().next_state()
Progress().state.draw_ui(bot, update)
return Progress().state
else:
if not self.progress.input_node:
self.progress.input_node = ''
if query.data == DELETE:
if len(self.progress.input_node):
self.progress.input_node = self.progress.input_node[:-1]
else:
self.progress.input_node += query.data
# We update the output so the user sees if he types correctly
bot.edit_message_text(text="Input data: {}".format(self.progress.input_node),
chat_id=query.message.chat_id,
message_id=Progress().tracking_message.message_id)
class BlogState(StepState):
name = 'BlogState'
def draw_ui(self, bot, update):
reply_keyboard = [
[InlineKeyboardButton(blog, callback_data='{}={}'.format(blog, bid)) for blog, bid in get_blogs().items()]
]
reply_markup = InlineKeyboardMarkup(reply_keyboard)
self.message(update).reply_text('Choose an option, /skip or /cancel', reply_markup=reply_markup)
def handle(self, bot=None, update=None):
super().handle()
query = update.callback_query
if query.data:
bname, bid = query.data.split('=')
bot.edit_message_text(text="Chosen option: {}".format(bname), chat_id=query.message.chat_id,
message_id=query.message.message_id)
# We optionally log anything
# We text the user the requirements for next state
# query.message.reply_text('Departamento elegido. ', reply_markup=ReplyKeyboardRemove())
Progress().input_blog = (bname,bid)
Progress().next_state()
Progress().state.draw_ui(bot, update)
return Progress().state
class StartTimeState(StepState):
name = 'StartTimeState'
def draw_ui(self, bot, update):
reply_keyboard = [
[InlineKeyboardButton('Delete', callback_data=DELETE)],
[InlineKeyboardButton(str(number), callback_data=str(number)) for number in range(5)],
[InlineKeyboardButton(str(number), callback_data=str(number)) for number in range(5, 10)],
[InlineKeyboardButton(':', callback_data=':')],
[InlineKeyboardButton('Accept', callback_data=OK),
InlineKeyboardButton('Cancel', callback_data=CANCEL)],
]
reply_markup = InlineKeyboardMarkup(reply_keyboard)
self.message(update).reply_text('Input start time or /cancel', reply_markup=reply_markup)
# We get the reference to the message we will use to update the value and set into Progress
Progress().tracking_message = self.message(update).reply_text("Start time: {}".format(''))
def handle(self, bot=None, update=None):
super().handle()
query = update.callback_query
if query.data == OK:
# TODO Validate input data
#logger.info('Intpu data: {}'.format(self.progress.input_start_time))
bot.delete_message(chat_id=query.message.chat_id, message_id=query.message.message_id)
#bot.edit_message_text(text="Input start time: {}".format(self.progress.input_start_time),chat_id=query.message.chat_id,message_id=query.message.message_id)
#bot.delete_message(chat_id=Progress().tracking_message.chat_id,message_id=Progress().tracking_message.message_id)
Progress().next_state()
Progress().state.draw_ui(bot, update)
return Progress().state
else:
if not self.progress.input_start_time:
self.progress.input_start_time = ''
if query.data == DELETE:
if len(self.progress.input_start_time):
self.progress.input_start_time = self.progress.input_start_time[:-1]
else:
self.progress.input_start_time += query.data
# We update the output so the user sees if he types correctly
bot.edit_message_text(text="Start time: {}".format(self.progress.input_start_time),
chat_id=query.message.chat_id,
message_id=Progress().tracking_message.message_id)
# ... more states
# ... more and more states
# ...and so on, you can imagine
class KeywordsState(StepState):
name = 'KeywordsState'
def draw_ui(self, bot, update):
self.message(update).reply_text('Enter keywords', reply_markup=ReplyKeyboardRemove())
def handle(self, bot=None, update=None):
#logger.info("Keywords: %s", update.message.text)
# TODO Store in a variable or similar the keywords
update.message.reply_text(
'Thanks. The keywods are: {}'.format(update.message.text))
Progress().next_state()
Progress().state.draw_ui(bot, update)
return Progress().state
class ConfirmState(StepState):
name = 'ConfirmState'
def draw_ui(self, bot, update):
self.message(update).reply_text('Confirm all this info: {!s}'.format(Progress()))
reply_keyboard = [
[InlineKeyboardButton('Accept', callback_data=OK),
InlineKeyboardButton('Cancel', callback_data=CANCEL)],
]
reply_markup = InlineKeyboardMarkup(reply_keyboard)
self.message(update).reply_text('Accep or cancel (/cancel)', reply_markup=reply_markup)
def handle(self, bot=None, update=None):
query = update.callback_query
self.message(update).reply_text('Finalizado')
# Se lanza el procesamiento
# Se resetea el progreso
self.progress.clear()
Progress().next_state(NoneState())
return ConversationHandler.END
# This is important. It's intended for setting the steps in an ordered-sorted way
StepState.states = [NoneState(), DepState(), NodeState(), BlogState(), IntervalState(), StartTimeState(), EndTimeState(), RepeatsState(), LabelsState(), KeywordsState(), ConfirmState()]
class Progress(object):
"""
This singleton class contains the whole information collected along all the wizard process
"""
__instance = None
def __new__(cls):
if Progress.__instance is None:
Progress.__instance = object.__new__(cls)
Progress.__instance.input_blog = None
Progress.__instance.state = NoneState()
Progress.__instance.tracking_message = None
Progress.__instance.input_dep = None
Progress.__instance.input_node = None
Progress.__instance.input_minutes = None
Progress.__instance.input_start_time = None
Progress.__instance.input_end_time = None
return Progress.__instance
def clear(self):
"""
Resets the progress
"""
self.input_blog = None
self.state = NoneState()
self.tracking_message = None
self.input_dep = None
self.input_node = None
self.input_minutes = None
self.input_start_time = None
self.input_end_time = None
def next_state(self, new_state=None):
"""
Moves to the next state or to the specified state if any
:param new_state: the next state to move to
:return: the next state, already set to the Progress object
"""
if new_state is not None:
if self.state:
self.state = self.state.next_state(new_state)
else:
self.state = NoneState
#logger('Couldnt go to next state')
else:
self.state = self.state.next_state()
def __str__(self):
return '{}: dep={}, node={}, start={},end={}'.format(self.__class__.__name__, self.input_dep, self.input_node, self.input_start_time, self.input_end_time)
Now we use the previous code here:
bot_prototype:
#!/usr/bin/python3
from telegram import ReplyKeyboardMarkup, ReplyKeyboardRemove
from telegram.ext import (Updater, Filters, RegexHandler, CommandHandler,
CallbackQueryHandler, ConversationHandler, MessageHandler)
from my_package.step_states import DepState, NodeState, StartTimeState, EndTimeState, KeywordsState, ConfirmState,
BlogState, IntervalState
from my_package.step_states import Progress
from environments import get_bot_token
import logging
logging.basicConfig(format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
level=logging.INFO)
logger = logging.getLogger(__name__)
OK = 'OK'
CANCEL = 'CANCEL'
DELETE = 'DELETE'
# Updater is telegram code and the bot_token is an id string
updater = Updater(get_bot_token())
def start(bot, update):
"""
Starts the wizard planning process
:param bot: the Telegram bot
:param update: the Telegram update, probably a text message
:return: the state for the ConversationHandler
"""
p = Progress()
p.next_state()
p.state.draw_ui(bot, update)
return p.state
def error(bot, update, error):
"""
Logs errors
:param bot: the Telegram bot
:param update: the Telegram update, probably a text message
"""
logger.error('Update "%s" caused error "%s"', update, error)
def skip(bot, update):
"""
Skips this step in the wizard process
:param bot: the Telegram bot
:param update: the Telegram update, probably a text message
:return: the next state skipping the present one
"""
user = update.message.from_user
logger.info("%s skipped a step.", user.first_name)
update.message.reply_text("Skipping steps is not supported yet. Process is done")
# If we don't cancel at the end, we should remove any keyboard which could be present
# TODO Right now we don't accept skips and we cancel everything.
# TODO In the future, we will look at present state and choose the next state
return cancel(bot, update)
def cancel(bot, update):
"""
Cancels the whole wizard process
:param bot: the Telegram bot
:param update: the Telegram update, probably a text message
:return: the END state for he ConversationHandler
"""
Progress().clear()
user = update.message.from_user
logger.info("{} cancelled the process.".format(user.first_name))
update.message.reply_text('Process ended.', reply_markup=ReplyKeyboardRemove())
return ConversationHandler.END
def main():
global updater
dp = updater.dispatcher
# Add conversation handler with the states
# The telegram conversation handler needs a list of handlers with function so it can execute desired code in each state/step
conv_handler = ConversationHandler(
entry_points=[ CommandHandler('start', start) ],
states={
DepState(): [CallbackQueryHandler(DepState().handle), CommandHandler('skip', skip)],
NodeState(): [CallbackQueryHandler(NodeState().handle), CommandHandler('skip', skip)],
BlogState(): [CallbackQueryHandler(BlogState().handle), CommandHandler('skip', skip)],
IntervalState(): [CallbackQueryHandler(IntervalState().handle), CommandHandler('skip', skip)],
StartTimeState(): [CallbackQueryHandler(StartTimeState().handle), CommandHandler('skip', skip)],
EndTimeState(): [CallbackQueryHandler(EndTimeState().handle), CommandHandler('skip', skip)],
KeywordsState(): [MessageHandler(Filters.text, KeywordsState().handle), CommandHandler('skip', skip)],
ConfirmState(): [CallbackQueryHandler(ConfirmState().handle)],
},
fallbacks=[CommandHandler('cancel', cancel)]
)
dp.add_handler(conv_handler)
# dp.add_handler(CallbackQueryHandler(button_callback))
updater.dispatcher.add_error_handler(error)
# Start the Bot
updater.start_polling()
# Run the bot until you press Ctrl-C or the process receives SIGINT,
# SIGTERM or SIGABRT. This should be used most of the time, since
# start_polling() is non-blocking and will stop the bot gracefully.
updater.idle()
if __name__ == '__main__':
main()
python beginner python-3.x design-patterns
edited Jan 26 at 13:41
Mast
7,43963686
asked Jan 16 at 15:52
madtyn
1415
I'm making a sort of a user interface for a telegram bot and my idea was to use State design pattern for making sort of a wizard for all the process, with user inputting data in each step or pressing /skip for skipping a single step or /cancel for ending everything.
It's working pretty well. I put the states/steps in a list so I can always get to the next one easily by a common method in the parent. Every state is a Borg, so I can instantiate a state everywhere and will always have the same values and methods.
I also made a Singleton class named Progress for putting all the input in the wizard process so I can store it and send it altogether at the end. I know you can use modules in Python as singleton, but I'm more used to this and I prefer not polluting the namespace. I also think this is easier for testing and using.
Being a Python beginner, I'm proud with this code, but I think it has some code smells and it can be improved in terms of design and best practices. I also made it for a single user, but haven't tested two persons accessing the bot simultaneously. I'd like to restrict transitions among states too.
step_states.py:
from abc import ABCMeta, abstractmethod
from telegram import InlineKeyboardButton, InlineKeyboardMarkup, ReplyKeyboardRemove
from telegram.ext import ConversationHandler
from my_package.amazon.search_indexes import SEARCH_INDEXES
from environments import get_blogs
OK = 'OK'
CANCEL = 'CANCEL'
DELETE = 'DELETE'
class StepState(metaclass=ABCMeta):
"""
This is a Borg class which shares state and methods so all the instances created for it are different references
but containing the same values. This is also true for children of the same class, so:
Children1() == Children1() # False, because references are different, but the values are the same.
Also if you change one, you change the other
Children1() == Children2() # False, references are different and you can change anyone without affecting the other
"""
states =
name = "state" # This is not used yet. Thinking of deleting it everywhere
allowed = # This is not used, but could be interesting to allow only some transitions. Just a sketch
# State is a Borg design pattern class
__shared_state = {}
def __init__(self):
self.__dict__ = self.__shared_state
def __eq__(self, other):
return self.__hash__() == other.__hash__()
def message(self, update):
"""
Returns the object holding the info coming from Telegram with the methods for replying and looking into this info
:param update: the Telegram update, probably a text message
:return: the update or query object
"""
if update.callback_query:
return update.callback_query.message
else:
return update.message
def draw_ui(self, bot, update):
"""
Draws the UI in Telegram (usually a custom keyboard) for the user to input data
:param bot: the Telegram bot
:param update: the Telegram update, probably a text message
"""
pass
@abstractmethod
def handle(self, bot=None, update=None):
"""
Handles all the input info into the Progress object and transitions to the next state
:param bot: the Telegram bot
:param update: the Telegram update, probably a text message
"""
self.progress = Progress()
# TODO How to make cancel() common to all states
#query = update.callback_query
#if query.data == CANCEL:
# return cancel(bot, update)
def next_state(self, new_state=None):
"""
Inherited method which decides to move to the next state, or to the specified state if any
:param new_state: the next state, if any
:return: the next state
"""
# TODO Check if the transition is allowed?
if new_state is None:
default_next_state_index = (self.index + 1) % len(self.states)
next_new_state = self.states[default_next_state_index]
# TODO logging
logger.info('Current state:', self, ' => switched to new state', next_new_state)
return next_new_state
else:
return new_state
@property
def index(self):
"""
Returns the index of this state within the list with all states
:return: the index as an int
"""
return StepState.states.index(self)
def __hash__(self):
return hash(self.__class__.name)
def __str__(self):
return self.__class__.name
class NoneState(StepState):
name = 'NoneState'
def draw_ui(self, bot, update):
pass
def handle(self, bot=None, update=None):
pass
class DepState(StepState):
name = 'DepState'
def draw_ui(self, bot, update):
reply_keyboard = [
]
num_cols = 3
current_row =
# Think of SEARCH_INDEXES as a list with objects having two strings: spanish_description and browse_node_id
for dep in SEARCH_INDEXES:
current_row.append(InlineKeyboardButton(dep.spanish_desc,
callback_data="{}={}".format(dep.browse_node_id, dep.spanish_desc)))
if len(current_row) >= num_cols:
reply_keyboard.append(current_row)
current_row =
reply_markup = InlineKeyboardMarkup(reply_keyboard)
self.message(update).reply_text('Enter the department or /cancel', reply_markup=reply_markup)
def handle(self, bot=None, update=None):
super().handle()
query = update.callback_query
if query.data:
current_input_dep, text = query.data.split('=')
bot.edit_message_text(text="Departament chosen: {}".format(text), chat_id=query.message.chat_id,
message_id=query.message.message_id)
# We optionally log anything
Progress().input_dep = current_input_dep
Progress().next_state()
Progress().state.draw_ui(bot, update)
return Progress().state
class NodeState(StepState):
name = 'NodeState'
def draw_ui(self, bot, update):
reply_keyboard = [
[InlineKeyboardButton(str(number), callback_data=str(number)) for number in range(5)],
[InlineKeyboardButton(str(number), callback_data=str(number)) for number in range(5, 10)],
[InlineKeyboardButton('Accept', callback_data=OK),
InlineKeyboardButton('Cancel', callback_data=CANCEL)],
]
reply_markup = InlineKeyboardMarkup(reply_keyboard)
self.message(update).reply_text('Input numbers for the node or /cancel', reply_markup=reply_markup)
# We get the reference to the message we will use to update the value and set into Progress
Progress().tracking_message = self.message(update).reply_text("Typed data : {}".format(''))
def handle(self, bot=None, update=None):
super().handle()
query = update.callback_query
if query.data == OK:
# TODO Validate input data
#logger.info('Input data: {}'.format(self.progress.input_node))
bot.edit_message_text(text="Chosen node: {}".format(self.progress.input_node), chat_id=query.message.chat_id,
message_id=query.message.message_id)
bot.delete_message(chat_id=Progress().tracking_message.chat_id,
message_id=Progress().tracking_message.message_id)
Progress().next_state()
Progress().state.draw_ui(bot, update)
return Progress().state
else:
if not self.progress.input_node:
self.progress.input_node = ''
if query.data == DELETE:
if len(self.progress.input_node):
self.progress.input_node = self.progress.input_node[:-1]
else:
self.progress.input_node += query.data
# We update the output so the user sees if he types correctly
bot.edit_message_text(text="Input data: {}".format(self.progress.input_node),
chat_id=query.message.chat_id,
message_id=Progress().tracking_message.message_id)
class BlogState(StepState):
name = 'BlogState'
def draw_ui(self, bot, update):
reply_keyboard = [
[InlineKeyboardButton(blog, callback_data='{}={}'.format(blog, bid)) for blog, bid in get_blogs().items()]
]
reply_markup = InlineKeyboardMarkup(reply_keyboard)
self.message(update).reply_text('Choose an option, /skip or /cancel', reply_markup=reply_markup)
def handle(self, bot=None, update=None):
super().handle()
query = update.callback_query
if query.data:
bname, bid = query.data.split('=')
bot.edit_message_text(text="Chosen option: {}".format(bname), chat_id=query.message.chat_id,
message_id=query.message.message_id)
# We optionally log anything
# We text the user the requirements for next state
# query.message.reply_text('Departamento elegido. ', reply_markup=ReplyKeyboardRemove())
Progress().input_blog = (bname,bid)
Progress().next_state()
Progress().state.draw_ui(bot, update)
return Progress().state
class StartTimeState(StepState):
name = 'StartTimeState'
def draw_ui(self, bot, update):
reply_keyboard = [
[InlineKeyboardButton('Delete', callback_data=DELETE)],
[InlineKeyboardButton(str(number), callback_data=str(number)) for number in range(5)],
[InlineKeyboardButton(str(number), callback_data=str(number)) for number in range(5, 10)],
[InlineKeyboardButton(':', callback_data=':')],
[InlineKeyboardButton('Accept', callback_data=OK),
InlineKeyboardButton('Cancel', callback_data=CANCEL)],
]
reply_markup = InlineKeyboardMarkup(reply_keyboard)
self.message(update).reply_text('Input start time or /cancel', reply_markup=reply_markup)
# We get the reference to the message we will use to update the value and set into Progress
Progress().tracking_message = self.message(update).reply_text("Start time: {}".format(''))
def handle(self, bot=None, update=None):
super().handle()
query = update.callback_query
if query.data == OK:
# TODO Validate input data
#logger.info('Intpu data: {}'.format(self.progress.input_start_time))
bot.delete_message(chat_id=query.message.chat_id, message_id=query.message.message_id)
#bot.edit_message_text(text="Input start time: {}".format(self.progress.input_start_time),chat_id=query.message.chat_id,message_id=query.message.message_id)
#bot.delete_message(chat_id=Progress().tracking_message.chat_id,message_id=Progress().tracking_message.message_id)
Progress().next_state()
Progress().state.draw_ui(bot, update)
return Progress().state
else:
if not self.progress.input_start_time:
self.progress.input_start_time = ''
if query.data == DELETE:
if len(self.progress.input_start_time):
self.progress.input_start_time = self.progress.input_start_time[:-1]
else:
self.progress.input_start_time += query.data
# We update the output so the user sees if he types correctly
bot.edit_message_text(text="Start time: {}".format(self.progress.input_start_time),
chat_id=query.message.chat_id,
message_id=Progress().tracking_message.message_id)
# ... more states
# ... more and more states
# ...and so on, you can imagine
class KeywordsState(StepState):
name = 'KeywordsState'
def draw_ui(self, bot, update):
self.message(update).reply_text('Enter keywords', reply_markup=ReplyKeyboardRemove())
def handle(self, bot=None, update=None):
#logger.info("Keywords: %s", update.message.text)
# TODO Store in a variable or similar the keywords
update.message.reply_text(
'Thanks. The keywods are: {}'.format(update.message.text))
Progress().next_state()
Progress().state.draw_ui(bot, update)
return Progress().state
class ConfirmState(StepState):
name = 'ConfirmState'
def draw_ui(self, bot, update):
self.message(update).reply_text('Confirm all this info: {!s}'.format(Progress()))
reply_keyboard = [
[InlineKeyboardButton('Accept', callback_data=OK),
InlineKeyboardButton('Cancel', callback_data=CANCEL)],
]
reply_markup = InlineKeyboardMarkup(reply_keyboard)
self.message(update).reply_text('Accep or cancel (/cancel)', reply_markup=reply_markup)
def handle(self, bot=None, update=None):
query = update.callback_query
self.message(update).reply_text('Finalizado')
# Se lanza el procesamiento
# Se resetea el progreso
self.progress.clear()
Progress().next_state(NoneState())
return ConversationHandler.END
# This is important. It's intended for setting the steps in an ordered-sorted way
StepState.states = [NoneState(), DepState(), NodeState(), BlogState(), IntervalState(), StartTimeState(), EndTimeState(), RepeatsState(), LabelsState(), KeywordsState(), ConfirmState()]
class Progress(object):
"""
This singleton class contains the whole information collected along all the wizard process
"""
__instance = None
def __new__(cls):
if Progress.__instance is None:
Progress.__instance = object.__new__(cls)
Progress.__instance.input_blog = None
Progress.__instance.state = NoneState()
Progress.__instance.tracking_message = None
Progress.__instance.input_dep = None
Progress.__instance.input_node = None
Progress.__instance.input_minutes = None
Progress.__instance.input_start_time = None
Progress.__instance.input_end_time = None
return Progress.__instance
def clear(self):
"""
Resets the progress
"""
self.input_blog = None
self.state = NoneState()
self.tracking_message = None
self.input_dep = None
self.input_node = None
self.input_minutes = None
self.input_start_time = None
self.input_end_time = None
def next_state(self, new_state=None):
"""
Moves to the next state or to the specified state if any
:param new_state: the next state to move to
:return: the next state, already set to the Progress object
"""
if new_state is not None:
if self.state:
self.state = self.state.next_state(new_state)
else:
self.state = NoneState
#logger('Couldnt go to next state')
else:
self.state = self.state.next_state()
def __str__(self):
return '{}: dep={}, node={}, start={},end={}'.format(self.__class__.__name__, self.input_dep, self.input_node, self.input_start_time, self.input_end_time)
Now we use the previous code here:
bot_prototype:
#!/usr/bin/python3
from telegram import ReplyKeyboardMarkup, ReplyKeyboardRemove
from telegram.ext import (Updater, Filters, RegexHandler, CommandHandler,
CallbackQueryHandler, ConversationHandler, MessageHandler)
from my_package.step_states import DepState, NodeState, StartTimeState, EndTimeState, KeywordsState, ConfirmState,
BlogState, IntervalState
from my_package.step_states import Progress
from environments import get_bot_token
import logging
logging.basicConfig(format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
level=logging.INFO)
logger = logging.getLogger(__name__)
OK = 'OK'
CANCEL = 'CANCEL'
DELETE = 'DELETE'
# Updater is telegram code and the bot_token is an id string
updater = Updater(get_bot_token())
def start(bot, update):
"""
Starts the wizard planning process
:param bot: the Telegram bot
:param update: the Telegram update, probably a text message
:return: the state for the ConversationHandler
"""
p = Progress()
p.next_state()
p.state.draw_ui(bot, update)
return p.state
def error(bot, update, error):
"""
Logs errors
:param bot: the Telegram bot
:param update: the Telegram update, probably a text message
"""
logger.error('Update "%s" caused error "%s"', update, error)
def skip(bot, update):
"""
Skips this step in the wizard process
:param bot: the Telegram bot
:param update: the Telegram update, probably a text message
:return: the next state skipping the present one
"""
user = update.message.from_user
logger.info("%s skipped a step.", user.first_name)
update.message.reply_text("Skipping steps is not supported yet. Process is done")
# If we don't cancel at the end, we should remove any keyboard which could be present
# TODO Right now we don't accept skips and we cancel everything.
# TODO In the future, we will look at present state and choose the next state
return cancel(bot, update)
def cancel(bot, update):
"""
Cancels the whole wizard process
:param bot: the Telegram bot
:param update: the Telegram update, probably a text message
:return: the END state for he ConversationHandler
"""
Progress().clear()
user = update.message.from_user
logger.info("{} cancelled the process.".format(user.first_name))
update.message.reply_text('Process ended.', reply_markup=ReplyKeyboardRemove())
return ConversationHandler.END
def main():
global updater
dp = updater.dispatcher
# Add conversation handler with the states
# The telegram conversation handler needs a list of handlers with function so it can execute desired code in each state/step
conv_handler = ConversationHandler(
entry_points=[ CommandHandler('start', start) ],
states={
DepState(): [CallbackQueryHandler(DepState().handle), CommandHandler('skip', skip)],
NodeState(): [CallbackQueryHandler(NodeState().handle), CommandHandler('skip', skip)],
BlogState(): [CallbackQueryHandler(BlogState().handle), CommandHandler('skip', skip)],
IntervalState(): [CallbackQueryHandler(IntervalState().handle), CommandHandler('skip', skip)],
StartTimeState(): [CallbackQueryHandler(StartTimeState().handle), CommandHandler('skip', skip)],
EndTimeState(): [CallbackQueryHandler(EndTimeState().handle), CommandHandler('skip', skip)],
KeywordsState(): [MessageHandler(Filters.text, KeywordsState().handle), CommandHandler('skip', skip)],
ConfirmState(): [CallbackQueryHandler(ConfirmState().handle)],
},
fallbacks=[CommandHandler('cancel', cancel)]
)
dp.add_handler(conv_handler)
# dp.add_handler(CallbackQueryHandler(button_callback))
updater.dispatcher.add_error_handler(error)
# Start the Bot
updater.start_polling()
# Run the bot until you press Ctrl-C or the process receives SIGINT,
# SIGTERM or SIGABRT. This should be used most of the time, since
# start_polling() is non-blocking and will stop the bot gracefully.
updater.idle()
if __name__ == '__main__':
main()
python beginner python-3.x design-patterns
python beginner python-3.x design-patterns
edited Jan 26 at 13:41
Mast
7,43963686
asked Jan 16 at 15:52
madtyn
1415
edited Jan 26 at 13:41
Mast
7,43963686
asked Jan 16 at 15:52
madtyn
1415
edited Jan 26 at 13:41
Mast
7,43963686
edited Jan 26 at 13:41
Mast
7,43963686
edited Jan 26 at 13:41
Mast
7,43963686
7,43963686
asked Jan 16 at 15:52
madtyn
1415
asked Jan 16 at 15:52
madtyn
1415
asked Jan 16 at 15:52
madtyn
1415
1415
bumped to the homepage by Community♦ 11 mins ago
This question has answers that may be good or bad; the system has marked it active so that they can be reviewed.
bumped to the homepage by Community♦ 11 mins ago
This question has answers that may be good or bad; the system has marked it active so that they can be reviewed.
Please do not update the code in your question to incorporate feedback from answers, doing so goes against the Question + Answer style of Code Review. This is not a forum where you should keep the most updated version in your question. Please see what you may and may not do after receiving answers.
– Mast
Jan 26 at 13:41
@Mast I was only formatting and changing indentation. Also I deleted some comments. All of it was done to make it easy to read and understand, but I didn't any refactoring, changed any method or added anything.
– madtyn
Jan 27 at 11:35
Feel free to post a follow-up question when you've made significant changes. Leave the code in this question as-is.
– Mast
Jan 27 at 13:50
add a comment |
Please do not update the code in your question to incorporate feedback from answers, doing so goes against the Question + Answer style of Code Review. This is not a forum where you should keep the most updated version in your question. Please see what you may and may not do after receiving answers.
– Mast
Jan 26 at 13:41
@Mast I was only formatting and changing indentation. Also I deleted some comments. All of it was done to make it easy to read and understand, but I didn't any refactoring, changed any method or added anything.
– madtyn
Jan 27 at 11:35
Feel free to post a follow-up question when you've made significant changes. Leave the code in this question as-is.
– Mast
Jan 27 at 13:50
Please do not update the code in your question to incorporate feedback from answers, doing so goes against the Question + Answer style of Code Review. This is not a forum where you should keep the most updated version in your question. Please see what you may and may not do after receiving answers.
– Mast
Jan 26 at 13:41
Please do not update the code in your question to incorporate feedback from answers, doing so goes against the Question + Answer style of Code Review. This is not a forum where you should keep the most updated version in your question. Please see what you may and may not do after receiving answers.
– Mast
Jan 26 at 13:41
@Mast I was only formatting and changing indentation. Also I deleted some comments. All of it was done to make it easy to read and understand, but I didn't any refactoring, changed any method or added anything.
– madtyn
Jan 27 at 11:35
@Mast I was only formatting and changing indentation. Also I deleted some comments. All of it was done to make it easy to read and understand, but I didn't any refactoring, changed any method or added anything.
– madtyn
Jan 27 at 11:35
Feel free to post a follow-up question when you've made significant changes. Leave the code in this question as-is.
– Mast
Jan 27 at 13:50
Feel free to post a follow-up question when you've made significant changes. Leave the code in this question as-is.
– Mast
Jan 27 at 13:50
add a comment |
1 Answer
1
active
oldest
votes
up vote
0
down vote
I mainly decoupled the states from the bots putting a handler_list method inside each state.
Briefly I comment some suggestions and good points Gareeth Reese made:
- Cutting source code lines to 80 chars. Still pending. I'm doing it the next time I post
- The use of self.states and the index method to choose a next state is inefficient (proportional to number of states). The antipattern here is over-encapsulation, the co-ordination needs to be handled at a higher level,for example in a state machine class like your Progress class. Still pending
The data structure that's needed is a mapping from state to next state, for example, in the Progress class you could have:
_state_order = [NoneState(), DepState(), NodeState(), ...]
_next_state = dict(zip(_state_order[:-1], _state_order[1:]))
and then in the Progress.next_state method you'd write:
self.state = self._next_state[self.state]
The message method is only used in the context
self.message(update).reply_text(...)
The duplicated code could be eliminated like this:
def reply_text(self, update, *args, *kwargs):
"""Reply to a Telegram update ..."""
if update.callback_query:
message = update.callback_query.message
else:
message = update.message
message.reply_text(*args, **kwargs)
And since this does not use self, it does not need to be a method on a class, it could be a @staticmethod or, better, an ordinary function. Tried it but some parameter was giving an error
Borg pattern is unnecessary and its machinery can be dropped from the StepState class. Still thinking about it
After removing all these attributes and methods, the only attribute remaining is progress. This is only used in the handle method, and so it would make sense to pass it as a parameter to that method and avoid the need for the attribute. Still thinking about it
Still pending as well. After removing the progress attribute, none of the remaining methods use self, so there's no need to create state objects, you can just use the state classes as namespaces for the draw_ui and handle functions:
_state_order = [NoneState, DepState, NodeState, ...]
I evolved a little bit the code until I came to this below. I know that many changes are still pending and DRY principle is broken, but the main things are done:
step_states:
class StepState(metaclass=ABCMeta):
"""
This is a Borg class which shares state and methods so all the instances created for it are different references
but containing the same values. This is also true for children of the same class, so:
Children1() == Children1() # False, because references are different, but the values are the same.
Also if you change one, you change the other
Children1() == Children2() # False, references are different and you can change anyone without affecting others
"""
states =
# State is a Borg design pattern class
__shared_state = {}
def __init__(self):
self.__dict__ = self.__shared_state
@staticmethod
def message(update):
"""
Returns the object holding the info coming from Telegram with the methods for replying
and looking into this info
:param update: the Telegram update, probably a text message
:return: the update or query object
"""
if update.callback_query:
return update.callback_query.message
else:
return update.message
@abstractmethod
def draw_ui(self, bot, update):
"""
Draws the UI in Telegram (usually a custom keyboard) for the user to input data
:param bot: the Telegram bot
:param update: the Telegram update, probably a text message
"""
pass
@abstractmethod
def handle(self, bot=None, update=None):
"""
Handles all the input info into the Progress object and transitions to the next state
:param bot: the Telegram bot
:param update: the Telegram update, probably a text message
"""
logger.info('Handling state {} in parent'.format(self.__class__.__name__))
self.progress = Progress()
if update and update.callback_query:
query = update.callback_query
if query.data == CANCEL:
return self.cancel(bot, update)
@abstractmethod
def handler_list(self):
"""
Returns the list with the handlers for managing the events for this state
Example: return [ CallbackQueryHandler(callbackFunc), CommandHandler('cmd', commandFunc)]
Will apply the callbackFunc for managing a query-like update, but if not,
will try to apply commandFunc for an incoming command /cmd
:return: the handlers list
"""
pass
def next_state(self, new_state=None):
"""
Inherited method which decides to move to the next state, or to the specified state if any
:param new_state: the next state, if any
:return: the next state
"""
# TODO Check if the transition is allowed?
next_new_state = None
if new_state is None:
default_next_state_index = (self.index + 1) % len(self.states)
next_new_state = self.states[default_next_state_index]
else:
next_new_state = new_state
logger.info('Actual state: {} => switched to new state {}'.format(self, next_new_state))
return next_new_state
@staticmethod
def skip(bot, update):
"""
Skips this step in the wizard process
:param bot: the Telegram bot
:param update: the Telegram update, probably a text message
:return: the next state skipping the present one
"""
logger.info('Skipping state {}'.format(Progress().state))
update.message.reply_text("Se salta este paso.")
# If we don't cancel at the end, we should remove any keyboard which could be present
Progress().next_state()
Progress().state.draw_ui(bot, update)
return Progress().state
@staticmethod
def cancel(bot, update):
"""
Cancels the whole wizard process
:param bot: the Telegram bot
:param update: the Telegram update, probably a text message
:return: the END state for he ConversationHandler
"""
Progress().clear()
user = StepState.message(update).from_user
logger.info("{} canceled the process.".format(user.first_name))
StepState.message(update).reply_text('Proceso finalizado.', reply_markup=ReplyKeyboardRemove())
return ConversationHandler.END
def validate_input(self, input_data):
"""
Validates the input data for this step/state
:param input_data: the input data
:return: True if input data is valid, otherwise False
"""
return True
@property
def index(self):
"""
Returns the index of this state within the list with all states
:return: the index as an int
"""
return StepState.states.index(self)
def __eq__(self, other):
return self.__hash__() == other.__hash__()
def __hash__(self):
return hash(self.__class__.__name__)
def __str__(self):
return self.__class__.__name__
class DepState(StepState):
def draw_ui(self, bot, update):
reply_keyboard = [
]
num_cols = 3
current_row =
# Think of SEARCH_INDEXES as a list with objects having two strings: spanish_description and browse_node_id
for dep in SEARCH_INDEXES:
current_row.append(InlineKeyboardButton(dep.spanish_desc,
callback_data="{}={}".format(dep.browse_node_id, dep.spanish_desc)))
if len(current_row) >= num_cols:
reply_keyboard.append(current_row)
current_row =
reply_markup = InlineKeyboardMarkup(reply_keyboard)
self.message(update).reply_text('Enter the *departament*, do /skip or /cancel',
parse_mode='Markdown', reply_markup=reply_markup)
def handle(self, bot=None, update=None):
parent_result = super().handle(bot, update)
if parent_result is not None:
return parent_result
query = update.callback_query
if query.data:
current_input_dep, text = query.data.split('=')
bot.edit_message_text(text="Departament chosen: {}".format(text), chat_id=query.message.chat_id,
message_id=query.message.message_id)
# We optionally log anything
Progress().input_dep = current_input_dep
Progress().next_state()
Progress().state.draw_ui(bot, update)
return Progress().state
class NodeState(StepState):
def draw_ui(self, bot, update):
reply_keyboard = [
[InlineKeyboardButton('Delete', callback_data=DELETE)],
[InlineKeyboardButton(str(number), callback_data=str(number)) for number in range(5)],
[InlineKeyboardButton(str(number), callback_data=str(number)) for number in range(5, 10)],
[InlineKeyboardButton('Accept', callback_data=OK),
InlineKeyboardButton('Cancel', callback_data=CANCEL)],
]
reply_markup = InlineKeyboardMarkup(reply_keyboard)
self.message(update).reply_text('Input numbers for the node, do /skip or /cancel',
parse_mode='Markdown', reply_markup=reply_markup)
# We get the reference to the message we will use to update the value and set into Progress
Progress().tracking_message = self.message(update).reply_text("Typed data : {}".format(''))
def handle(self, bot=None, update=None):
parent_result = super().handle(bot, update)
if parent_result is not None:
return parent_result
query = update.callback_query
if hasattr(query, 'data'):
if query.data == OK:
if self.validate_input(self.progress.input_node):
bot.edit_message_text(text="Chosen node: {}".format(self.progress.input_node),
chat_id=query.message.chat_id, message_id=query.message.message_id)
bot.delete_message(chat_id=Progress().tracking_message.chat_id,
message_id=Progress().tracking_message.message_id)
Progress().next_state()
else:
no_keywords = self.progress.input_keywords is None or not self.progress.input_keywords.strip()
if self.progress.input_node:
self.message(update).reply_text('Not allowed value')
self.progress.input_node = ''
elif no_keywords:
# If there are no keywords, then the node is mandatory
self.message(update).reply_text('This input is mandatory. Enter an allowed value')
Progress().next_state(self)
Progress().state.draw_ui(bot, update)
return Progress().state
else:
if not self.progress.input_node:
self.progress.input_node = ''
if query.data == DELETE:
if len(self.progress.input_node):
self.progress.input_node = self.progress.input_node[:-1]
else:
self.progress.input_node += query.data
# We update the output so the user sees if he types correctly
bot.edit_message_text(text="Input data: {}".format(self.progress.input_node),
chat_id=query.message.chat_id,
message_id=Progress().tracking_message.message_id)
else:
if update.message.text and self.validate_input(update.message.text):
self.progress.input_node = update.message.text
Progress().next_state()
Progress().state.draw_ui(bot, update)
elif not self.validate_input(update.message.text):
self.message(update).reply_text('Valor no admitido')
self.message(update).reply_text('Este dato es obligatorio. Introduzca un valor valido')
self.progress.input_node = ''
Progress().next_state(self)
Progress().state.draw_ui(bot, update)
return Progress().state
@staticmethod
def skip(bot, update):
logger.info('Skipping state {}'.format(Progress().state))
# If we don't cancel at the end, we should remove any keyboard which could be present
if Progress().input_keywords is None or not Progress().input_keywords.strip():
update.message.reply_text('Este dato es obligatorio. Introduzca un valor valido')
else:
Progress().input_node = ''
Progress().next_state()
update.message.reply_text("Se salta este paso.")
Progress().state.draw_ui(bot, update)
return Progress().state
def validate_input(self, input_data):
data = input_data
if type(input_data) is str:
data = int(data.strip())
response = requests.get('https://www.amazon.es/exec/obidos/tg/browse/-/{}'.format(data))
return response.status_code == 200
def handler_list(self):
return [RegexHandler('^(d+)$', self.handle),
CallbackQueryHandler(self.handle), CommandHandler('skip', self.skip)]
class BlogState(StepState):
def draw_ui(self, bot, update):
reply_keyboard = [
[InlineKeyboardButton(blog, callback_data='{}={}'.format(blog, bid)) for blog, bid in sorted(get_blogs().items())]
]
reply_markup = InlineKeyboardMarkup(reply_keyboard)
self.message(update).reply_text('Choose an option, /skip or /cancel', parse_mode='Markdown',reply_markup=reply_markup)
def handle(self, bot=None, update=None):
parent_result = super().handle(bot, update)
if parent_result is not None:
return parent_result
query = update.callback_query
if query.data:
bname, bid = query.data.split('=')
bot.edit_message_text(text="Chosen option: {}".format(bname), chat_id=query.message.chat_id,
message_id=query.message.message_id)
# We optionally log anything
# We text the user the requirements for next state
# query.message.reply_text('Departamento elegido. ', reply_markup=ReplyKeyboardRemove())
Progress().input_blog = (bname,bid)
if bname == 'xxx':
Progress().input_dep = HEALTH.browse_node_id
Progress().next_state(KeywordsState())
elif bname == 'books':
Progress().input_dep = BOOKS.browse_node_id
Progress().next_state(KeywordsState())
else:
Progress().next_state()
Progress().state.draw_ui(bot, update)
return Progress().state
def handler_list(self):
return [CallbackQueryHandler(self.handle), CommandHandler('skip', self.skip)]
class StartTimeState(StepState):
def draw_ui(self, bot, update):
reply_keyboard = [
[InlineKeyboardButton('Delete', callback_data=DELETE)],
[InlineKeyboardButton(str(number), callback_data=str(number)) for number in range(5)],
[InlineKeyboardButton(str(number), callback_data=str(number)) for number in range(5, 10)],
[InlineKeyboardButton(':', callback_data=':')],
[InlineKeyboardButton('Accept', callback_data=OK),
InlineKeyboardButton('Cancel', callback_data=CANCEL)],
]
reply_markup = InlineKeyboardMarkup(reply_keyboard)
self.message(update).reply_text('Input start time or /cancel',
parse_mode='Markdown', reply_markup=reply_markup)
# We get the reference to the message we will use to update the value and set into Progress
Progress().tracking_message = self.message(update).reply_text("Start time: {}".format(''))
def handle(self, bot=None, update=None):
parent_result = super().handle(bot, update)
if parent_result is not None:
return parent_result
query = update.callback_query
if hasattr(query, 'data'):
if query.data == OK:
bot.delete_message(chat_id=query.message.chat_id, message_id=query.message.message_id)
Progress().next_state()
Progress().state.draw_ui(bot, update)
return Progress().state
else:
if not self.progress.input_start_time:
self.progress.input_start_time = ''
if query.data == DELETE:
if len(self.progress.input_start_time):
self.progress.input_start_time = self.progress.input_start_time[:-1]
else:
self.progress.input_start_time += query.data
# We update the output so the user sees if he types correctly
bot.edit_message_text(text="Start time: {}".format(self.progress.input_start_time),
chat_id=query.message.chat_id,
message_id=Progress().tracking_message.message_id)
else:
if update.message.text and self.validate_input(update.message.text):
self.progress.input_start_time = update.message.text
Progress().next_state()
Progress().state.draw_ui(bot, update)
elif not self.validate_input(update.message.text):
self.message(update).reply_text('Valor no admitido')
self.message(update).reply_text('Este dato es obligatorio. Introduzca un valor valido')
self.progress.input_start_time = ''
Progress().next_state(self)
Progress().state.draw_ui(bot, update)
return Progress().state
def handler_list(self):
return [RegexHandler(TIME_REGEX, self.handle),
CallbackQueryHandler(self.handle), CommandHandler('skip', self.skip)]
# ... more states
# ... more and more states
# ...and so on, you can imagine
class KeywordsState(StepState):
name = 'KeywordsState'
def draw_ui(self, bot, update):
self.message(update).reply_text('Enter keywords', parse_mode='Markdown',
reply_markup=ReplyKeyboardRemove())
def handle(self, bot=None, update=None):
parent_result = super().handle(bot, update)
if parent_result is not None:
return parent_result
keywords = update.message.text.strip()
if keywords:
Progress().input_keywords = keywords
update.message.reply_text('Thanks. The keywords are: {}'.format(keywords))
# When SearchIndex equals All, BrowseNode cannot be present
if Progress().input_dep is None:
Progress().next_state(StartTimeState())
else:
Progress().next_state()
else:
update.message.reply_text('No ha introducido palabras. Introdúzcalas, haga /skip o /cancel')
Progress().next_state(self)
Progress().state.draw_ui(bot, update)
return Progress().state
@staticmethod
def skip(bot, update):
logger.info('Skipping state {}'.format(Progress().state))
# If we don't cancel at the end, we should remove any keyboard which could be present
if Progress().input_dep is None:
update.message.reply_text('Este dato es obligatorio. Introduzca un valor valido')
else:
Progress().input_keywords = None
Progress().next_state()
update.message.reply_text("Skipping this step.")
Progress().state.draw_ui(bot, update)
return Progress().state
def handler_list(self):
return [MessageHandler(Filters.text, self.handle), CommandHandler('skip', self.skip)]
class ConfirmState(StepState):
def draw_ui(self, bot, update):
self.message(update).reply_text('Confirm all this info: {!s}'.format(Progress()))
reply_keyboard = [
[InlineKeyboardButton('Accept', callback_data=OK),
InlineKeyboardButton('Cancel', callback_data=CANCEL)],
]
reply_markup = InlineKeyboardMarkup(reply_keyboard)
self.message(update).reply_text('Accept or cancel (/cancel)', reply_markup=reply_markup)
# This returns a generator with a data list to consume
self.generator = search_asins(Progress().input_dep, Progress().input_node, Progress().input_keywords)
def handle(self, bot=None, update=None):
parent_result = super().handle(bot, update)
if parent_result is not None:
return parent_result
# OK granted, CANCEL is managed in parent
query = update.callback_query
# We finished with the wizard and launch everything
# Launching stuff here...
# ...
# Progress should be reset
self.progress.clear()
Progress().next_state(NoneState())
return ConversationHandler.END
def handler_list(self):
return [CallbackQueryHandler(self.handle)]
# This is intended for setting the steps in an ordered-sorted way
StepState.states = [
NoneState(), BlogState(), DepState(), KeywordsState(), NodeState(), StartTimeState(),
IntervalState(), EndTimeState(), RepeatsState(), ConfirmState()
]
class Progress(object):
"""
This singleton class contains the whole information collected along all the wizard process
"""
__instance = None
def __new__(cls):
if Progress.__instance is None:
Progress.__instance = object.__new__(cls)
Progress.__instance.input_blog = None
Progress.__instance.state = NoneState()
Progress.__instance.tracking_message = None
Progress.__instance.input_dep = None
Progress.__instance.input_node = None
Progress.__instance.input_minutes = '60'
Progress.__instance.input_start_time = None
Progress.__instance.input_end_time = None
Progress.__instance.input_keywords = None
Progress.__instance.input_repeats = None
return Progress.__instance
def clear(self):
"""
Resets the progress
"""
self.input_blog = None
self.state = NoneState()
self.tracking_message = None
self.input_dep = None
self.input_node = None
self.input_minutes = '60'
self.input_start_time = None
self.input_end_time = None
self.input_keywords = None
def next_state(self, new_state=None):
"""
Moves to the next state or to the specified state if any
:param new_state: the next state to move to
:return: the next state, already set to the Progress object
"""
if new_state is not None:
if self.state:
self.state = self.state.next_state(new_state)
else:
self.state = NoneState
logger.info("It was impossible o move to the next state")
else:
self.state = self.state.next_state()
def __str__(self):
return "{}: blog={}, dep={}, node={}, start={}, "
"interval={}', end={}, repeats={}, keywords={}".format(self.__class__.__name__,self.input_blog[0],self.input_dep,self.input_node,self.input_start_time,self.input_minutes,self.input_end_time,self.input_repeats,self.input_keywords)
bot.py:
#!/usr/bin/python3
from telegram.ext import (Updater, Filters, CommandHandler, ConversationHandler, MessageHandler)
from my_package.step_states import Progress, StepState
from environments import get_bot_token, getLogger
# Enable logging
logger = getLogger(__name__)
updater = Updater(get_bot_token())
def start(bot, update):
"""
Sends a message when the command /start is issued.
:param bot: the bot
:param update: the update info from Telegram for this command
"""
update.message.reply_text('Bot started')
def error(bot, update, error):
"""
Logs errors
:param bot: the Telegram bot
:param update: the Telegram update, probably a text message
"""
logger.error('Update "%s" caused error "%s"', update, error)
def restricted(my_handler):
"""
Decorates a handler for restricting its use
:param my_handler: the handler to be restricted
:return: the restricted handler
"""
@wraps(my_handler)
def wrapped(bot, update, *args, **kwargs):
user_id, user_name = update.effective_user.id, update.effective_user.first_name
if user_id not in get_allowed_users():
update.message.reply_text("Unauthorized access {} con id {}.n".format(user_name, user_id))
return
logger.info('Entering {} '.format(my_handler.__name__))
return my_handler(bot, update, *args, **kwargs)
return wrapped
@restricted
def plan(bot, update): #, blog_id=get_blog_id(), interval=60, dep_param_id=None):
"""
Starts the wizard for scheduling item posts
:param bot: the bot
:param update: the update info from Telegram for this command
"""
p = Progress()
p.next_state()
p.state.draw_ui(bot, update)
return p.state
def main():
"""Start the bot."""
# Create the EventHandler and pass it your bot's token.
global updater
# Get the dispatcher to register handlers
dp = updater.dispatcher
# on different commands - answer in Telegram
dp.add_handler(CommandHandler('start', start))
# Add conversation handler with the states
# The telegram conversation handler needs a handler_list with functions
# so it can execute desired code in each state/step
conv_handler = ConversationHandler(
entry_points=[CommandHandler('plan', plan)],
# We enter all the states and all the configured handlers for each state
states={state: state.handler_list() for state in StepState.states},
fallbacks=[CommandHandler('cancel', StepState.cancel)]
)
dp.add_handler(conv_handler)
# log all errors
dp.add_error_handler(error)
# Start the Bot
updater.start_polling()
# Run the bot until you press Ctrl-C or the process receives SIGINT,
# SIGTERM or SIGABRT. This should be used most of the time, since
# start_polling() is non-blocking and will stop the bot gracefully.
updater.idle()
if __name__ == '__main__':
main()
answered Feb 12 at 19:05
madtyn
1415
add a comment |
1 Answer
1
active
oldest
votes
1 Answer
1
active
oldest
votes
active
oldest
votes
active
oldest
votes
up vote
0
down vote
I mainly decoupled the states from the bots putting a handler_list method inside each state.
Briefly I comment some suggestions and good points Gareeth Reese made:
- Cutting source code lines to 80 chars. Still pending. I'm doing it the next time I post
- The use of self.states and the index method to choose a next state is inefficient (proportional to number of states). The antipattern here is over-encapsulation, the co-ordination needs to be handled at a higher level,for example in a state machine class like your Progress class. Still pending
The data structure that's needed is a mapping from state to next state, for example, in the Progress class you could have:
_state_order = [NoneState(), DepState(), NodeState(), ...]
_next_state = dict(zip(_state_order[:-1], _state_order[1:]))
and then in the Progress.next_state method you'd write:
self.state = self._next_state[self.state]
The message method is only used in the context
self.message(update).reply_text(...)
The duplicated code could be eliminated like this:
def reply_text(self, update, *args, *kwargs):
"""Reply to a Telegram update ..."""
if update.callback_query:
message = update.callback_query.message
else:
message = update.message
message.reply_text(*args, **kwargs)
And since this does not use self, it does not need to be a method on a class, it could be a @staticmethod or, better, an ordinary function. Tried it but some parameter was giving an error
Borg pattern is unnecessary and its machinery can be dropped from the StepState class. Still thinking about it
After removing all these attributes and methods, the only attribute remaining is progress. This is only used in the handle method, and so it would make sense to pass it as a parameter to that method and avoid the need for the attribute. Still thinking about it
Still pending as well. After removing the progress attribute, none of the remaining methods use self, so there's no need to create state objects, you can just use the state classes as namespaces for the draw_ui and handle functions:
_state_order = [NoneState, DepState, NodeState, ...]
I evolved a little bit the code until I came to this below. I know that many changes are still pending and DRY principle is broken, but the main things are done:
step_states:
class StepState(metaclass=ABCMeta):
"""
This is a Borg class which shares state and methods so all the instances created for it are different references
but containing the same values. This is also true for children of the same class, so:
Children1() == Children1() # False, because references are different, but the values are the same.
Also if you change one, you change the other
Children1() == Children2() # False, references are different and you can change anyone without affecting others
"""
states =
# State is a Borg design pattern class
__shared_state = {}
def __init__(self):
self.__dict__ = self.__shared_state
@staticmethod
def message(update):
"""
Returns the object holding the info coming from Telegram with the methods for replying
and looking into this info
:param update: the Telegram update, probably a text message
:return: the update or query object
"""
if update.callback_query:
return update.callback_query.message
else:
return update.message
@abstractmethod
def draw_ui(self, bot, update):
"""
Draws the UI in Telegram (usually a custom keyboard) for the user to input data
:param bot: the Telegram bot
:param update: the Telegram update, probably a text message
"""
pass
@abstractmethod
def handle(self, bot=None, update=None):
"""
Handles all the input info into the Progress object and transitions to the next state
:param bot: the Telegram bot
:param update: the Telegram update, probably a text message
"""
logger.info('Handling state {} in parent'.format(self.__class__.__name__))
self.progress = Progress()
if update and update.callback_query:
query = update.callback_query
if query.data == CANCEL:
return self.cancel(bot, update)
@abstractmethod
def handler_list(self):
"""
Returns the list with the handlers for managing the events for this state
Example: return [ CallbackQueryHandler(callbackFunc), CommandHandler('cmd', commandFunc)]
Will apply the callbackFunc for managing a query-like update, but if not,
will try to apply commandFunc for an incoming command /cmd
:return: the handlers list
"""
pass
def next_state(self, new_state=None):
"""
Inherited method which decides to move to the next state, or to the specified state if any
:param new_state: the next state, if any
:return: the next state
"""
# TODO Check if the transition is allowed?
next_new_state = None
if new_state is None:
default_next_state_index = (self.index + 1) % len(self.states)
next_new_state = self.states[default_next_state_index]
else:
next_new_state = new_state
logger.info('Actual state: {} => switched to new state {}'.format(self, next_new_state))
return next_new_state
@staticmethod
def skip(bot, update):
"""
Skips this step in the wizard process
:param bot: the Telegram bot
:param update: the Telegram update, probably a text message
:return: the next state skipping the present one
"""
logger.info('Skipping state {}'.format(Progress().state))
update.message.reply_text("Se salta este paso.")
# If we don't cancel at the end, we should remove any keyboard which could be present
Progress().next_state()
Progress().state.draw_ui(bot, update)
return Progress().state
@staticmethod
def cancel(bot, update):
"""
Cancels the whole wizard process
:param bot: the Telegram bot
:param update: the Telegram update, probably a text message
:return: the END state for he ConversationHandler
"""
Progress().clear()
user = StepState.message(update).from_user
logger.info("{} canceled the process.".format(user.first_name))
StepState.message(update).reply_text('Proceso finalizado.', reply_markup=ReplyKeyboardRemove())
return ConversationHandler.END
def validate_input(self, input_data):
"""
Validates the input data for this step/state
:param input_data: the input data
:return: True if input data is valid, otherwise False
"""
return True
@property
def index(self):
"""
Returns the index of this state within the list with all states
:return: the index as an int
"""
return StepState.states.index(self)
def __eq__(self, other):
return self.__hash__() == other.__hash__()
def __hash__(self):
return hash(self.__class__.__name__)
def __str__(self):
return self.__class__.__name__
class DepState(StepState):
def draw_ui(self, bot, update):
reply_keyboard = [
]
num_cols = 3
current_row =
# Think of SEARCH_INDEXES as a list with objects having two strings: spanish_description and browse_node_id
for dep in SEARCH_INDEXES:
current_row.append(InlineKeyboardButton(dep.spanish_desc,
callback_data="{}={}".format(dep.browse_node_id, dep.spanish_desc)))
if len(current_row) >= num_cols:
reply_keyboard.append(current_row)
current_row =
reply_markup = InlineKeyboardMarkup(reply_keyboard)
self.message(update).reply_text('Enter the *departament*, do /skip or /cancel',
parse_mode='Markdown', reply_markup=reply_markup)
def handle(self, bot=None, update=None):
parent_result = super().handle(bot, update)
if parent_result is not None:
return parent_result
query = update.callback_query
if query.data:
current_input_dep, text = query.data.split('=')
bot.edit_message_text(text="Departament chosen: {}".format(text), chat_id=query.message.chat_id,
message_id=query.message.message_id)
# We optionally log anything
Progress().input_dep = current_input_dep
Progress().next_state()
Progress().state.draw_ui(bot, update)
return Progress().state
class NodeState(StepState):
def draw_ui(self, bot, update):
reply_keyboard = [
[InlineKeyboardButton('Delete', callback_data=DELETE)],
[InlineKeyboardButton(str(number), callback_data=str(number)) for number in range(5)],
[InlineKeyboardButton(str(number), callback_data=str(number)) for number in range(5, 10)],
[InlineKeyboardButton('Accept', callback_data=OK),
InlineKeyboardButton('Cancel', callback_data=CANCEL)],
]
reply_markup = InlineKeyboardMarkup(reply_keyboard)
self.message(update).reply_text('Input numbers for the node, do /skip or /cancel',
parse_mode='Markdown', reply_markup=reply_markup)
# We get the reference to the message we will use to update the value and set into Progress
Progress().tracking_message = self.message(update).reply_text("Typed data : {}".format(''))
def handle(self, bot=None, update=None):
parent_result = super().handle(bot, update)
if parent_result is not None:
return parent_result
query = update.callback_query
if hasattr(query, 'data'):
if query.data == OK:
if self.validate_input(self.progress.input_node):
bot.edit_message_text(text="Chosen node: {}".format(self.progress.input_node),
chat_id=query.message.chat_id, message_id=query.message.message_id)
bot.delete_message(chat_id=Progress().tracking_message.chat_id,
message_id=Progress().tracking_message.message_id)
Progress().next_state()
else:
no_keywords = self.progress.input_keywords is None or not self.progress.input_keywords.strip()
if self.progress.input_node:
self.message(update).reply_text('Not allowed value')
self.progress.input_node = ''
elif no_keywords:
# If there are no keywords, then the node is mandatory
self.message(update).reply_text('This input is mandatory. Enter an allowed value')
Progress().next_state(self)
Progress().state.draw_ui(bot, update)
return Progress().state
else:
if not self.progress.input_node:
self.progress.input_node = ''
if query.data == DELETE:
if len(self.progress.input_node):
self.progress.input_node = self.progress.input_node[:-1]
else:
self.progress.input_node += query.data
# We update the output so the user sees if he types correctly
bot.edit_message_text(text="Input data: {}".format(self.progress.input_node),
chat_id=query.message.chat_id,
message_id=Progress().tracking_message.message_id)
else:
if update.message.text and self.validate_input(update.message.text):
self.progress.input_node = update.message.text
Progress().next_state()
Progress().state.draw_ui(bot, update)
elif not self.validate_input(update.message.text):
self.message(update).reply_text('Valor no admitido')
self.message(update).reply_text('Este dato es obligatorio. Introduzca un valor valido')
self.progress.input_node = ''
Progress().next_state(self)
Progress().state.draw_ui(bot, update)
return Progress().state
@staticmethod
def skip(bot, update):
logger.info('Skipping state {}'.format(Progress().state))
# If we don't cancel at the end, we should remove any keyboard which could be present
if Progress().input_keywords is None or not Progress().input_keywords.strip():
update.message.reply_text('Este dato es obligatorio. Introduzca un valor valido')
else:
Progress().input_node = ''
Progress().next_state()
update.message.reply_text("Se salta este paso.")
Progress().state.draw_ui(bot, update)
return Progress().state
def validate_input(self, input_data):
data = input_data
if type(input_data) is str:
data = int(data.strip())
response = requests.get('https://www.amazon.es/exec/obidos/tg/browse/-/{}'.format(data))
return response.status_code == 200
def handler_list(self):
return [RegexHandler('^(d+)$', self.handle),
CallbackQueryHandler(self.handle), CommandHandler('skip', self.skip)]
class BlogState(StepState):
def draw_ui(self, bot, update):
reply_keyboard = [
[InlineKeyboardButton(blog, callback_data='{}={}'.format(blog, bid)) for blog, bid in sorted(get_blogs().items())]
]
reply_markup = InlineKeyboardMarkup(reply_keyboard)
self.message(update).reply_text('Choose an option, /skip or /cancel', parse_mode='Markdown',reply_markup=reply_markup)
def handle(self, bot=None, update=None):
parent_result = super().handle(bot, update)
if parent_result is not None:
return parent_result
query = update.callback_query
if query.data:
bname, bid = query.data.split('=')
bot.edit_message_text(text="Chosen option: {}".format(bname), chat_id=query.message.chat_id,
message_id=query.message.message_id)
# We optionally log anything
# We text the user the requirements for next state
# query.message.reply_text('Departamento elegido. ', reply_markup=ReplyKeyboardRemove())
Progress().input_blog = (bname,bid)
if bname == 'xxx':
Progress().input_dep = HEALTH.browse_node_id
Progress().next_state(KeywordsState())
elif bname == 'books':
Progress().input_dep = BOOKS.browse_node_id
Progress().next_state(KeywordsState())
else:
Progress().next_state()
Progress().state.draw_ui(bot, update)
return Progress().state
def handler_list(self):
return [CallbackQueryHandler(self.handle), CommandHandler('skip', self.skip)]
class StartTimeState(StepState):
def draw_ui(self, bot, update):
reply_keyboard = [
[InlineKeyboardButton('Delete', callback_data=DELETE)],
[InlineKeyboardButton(str(number), callback_data=str(number)) for number in range(5)],
[InlineKeyboardButton(str(number), callback_data=str(number)) for number in range(5, 10)],
[InlineKeyboardButton(':', callback_data=':')],
[InlineKeyboardButton('Accept', callback_data=OK),
InlineKeyboardButton('Cancel', callback_data=CANCEL)],
]
reply_markup = InlineKeyboardMarkup(reply_keyboard)
self.message(update).reply_text('Input start time or /cancel',
parse_mode='Markdown', reply_markup=reply_markup)
# We get the reference to the message we will use to update the value and set into Progress
Progress().tracking_message = self.message(update).reply_text("Start time: {}".format(''))
def handle(self, bot=None, update=None):
parent_result = super().handle(bot, update)
if parent_result is not None:
return parent_result
query = update.callback_query
if hasattr(query, 'data'):
if query.data == OK:
bot.delete_message(chat_id=query.message.chat_id, message_id=query.message.message_id)
Progress().next_state()
Progress().state.draw_ui(bot, update)
return Progress().state
else:
if not self.progress.input_start_time:
self.progress.input_start_time = ''
if query.data == DELETE:
if len(self.progress.input_start_time):
self.progress.input_start_time = self.progress.input_start_time[:-1]
else:
self.progress.input_start_time += query.data
# We update the output so the user sees if he types correctly
bot.edit_message_text(text="Start time: {}".format(self.progress.input_start_time),
chat_id=query.message.chat_id,
message_id=Progress().tracking_message.message_id)
else:
if update.message.text and self.validate_input(update.message.text):
self.progress.input_start_time = update.message.text
Progress().next_state()
Progress().state.draw_ui(bot, update)
elif not self.validate_input(update.message.text):
self.message(update).reply_text('Valor no admitido')
self.message(update).reply_text('Este dato es obligatorio. Introduzca un valor valido')
self.progress.input_start_time = ''
Progress().next_state(self)
Progress().state.draw_ui(bot, update)
return Progress().state
def handler_list(self):
return [RegexHandler(TIME_REGEX, self.handle),
CallbackQueryHandler(self.handle), CommandHandler('skip', self.skip)]
# ... more states
# ... more and more states
# ...and so on, you can imagine
class KeywordsState(StepState):
name = 'KeywordsState'
def draw_ui(self, bot, update):
self.message(update).reply_text('Enter keywords', parse_mode='Markdown',
reply_markup=ReplyKeyboardRemove())
def handle(self, bot=None, update=None):
parent_result = super().handle(bot, update)
if parent_result is not None:
return parent_result
keywords = update.message.text.strip()
if keywords:
Progress().input_keywords = keywords
update.message.reply_text('Thanks. The keywords are: {}'.format(keywords))
# When SearchIndex equals All, BrowseNode cannot be present
if Progress().input_dep is None:
Progress().next_state(StartTimeState())
else:
Progress().next_state()
else:
update.message.reply_text('No ha introducido palabras. Introdúzcalas, haga /skip o /cancel')
Progress().next_state(self)
Progress().state.draw_ui(bot, update)
return Progress().state
@staticmethod
def skip(bot, update):
logger.info('Skipping state {}'.format(Progress().state))
# If we don't cancel at the end, we should remove any keyboard which could be present
if Progress().input_dep is None:
update.message.reply_text('Este dato es obligatorio. Introduzca un valor valido')
else:
Progress().input_keywords = None
Progress().next_state()
update.message.reply_text("Skipping this step.")
Progress().state.draw_ui(bot, update)
return Progress().state
def handler_list(self):
return [MessageHandler(Filters.text, self.handle), CommandHandler('skip', self.skip)]
class ConfirmState(StepState):
def draw_ui(self, bot, update):
self.message(update).reply_text('Confirm all this info: {!s}'.format(Progress()))
reply_keyboard = [
[InlineKeyboardButton('Accept', callback_data=OK),
InlineKeyboardButton('Cancel', callback_data=CANCEL)],
]
reply_markup = InlineKeyboardMarkup(reply_keyboard)
self.message(update).reply_text('Accept or cancel (/cancel)', reply_markup=reply_markup)
# This returns a generator with a data list to consume
self.generator = search_asins(Progress().input_dep, Progress().input_node, Progress().input_keywords)
def handle(self, bot=None, update=None):
parent_result = super().handle(bot, update)
if parent_result is not None:
return parent_result
# OK granted, CANCEL is managed in parent
query = update.callback_query
# We finished with the wizard and launch everything
# Launching stuff here...
# ...
# Progress should be reset
self.progress.clear()
Progress().next_state(NoneState())
return ConversationHandler.END
def handler_list(self):
return [CallbackQueryHandler(self.handle)]
# This is intended for setting the steps in an ordered-sorted way
StepState.states = [
NoneState(), BlogState(), DepState(), KeywordsState(), NodeState(), StartTimeState(),
IntervalState(), EndTimeState(), RepeatsState(), ConfirmState()
]
class Progress(object):
"""
This singleton class contains the whole information collected along all the wizard process
"""
__instance = None
def __new__(cls):
if Progress.__instance is None:
Progress.__instance = object.__new__(cls)
Progress.__instance.input_blog = None
Progress.__instance.state = NoneState()
Progress.__instance.tracking_message = None
Progress.__instance.input_dep = None
Progress.__instance.input_node = None
Progress.__instance.input_minutes = '60'
Progress.__instance.input_start_time = None
Progress.__instance.input_end_time = None
Progress.__instance.input_keywords = None
Progress.__instance.input_repeats = None
return Progress.__instance
def clear(self):
"""
Resets the progress
"""
self.input_blog = None
self.state = NoneState()
self.tracking_message = None
self.input_dep = None
self.input_node = None
self.input_minutes = '60'
self.input_start_time = None
self.input_end_time = None
self.input_keywords = None
def next_state(self, new_state=None):
"""
Moves to the next state or to the specified state if any
:param new_state: the next state to move to
:return: the next state, already set to the Progress object
"""
if new_state is not None:
if self.state:
self.state = self.state.next_state(new_state)
else:
self.state = NoneState
logger.info("It was impossible o move to the next state")
else:
self.state = self.state.next_state()
def __str__(self):
return "{}: blog={}, dep={}, node={}, start={}, "
"interval={}', end={}, repeats={}, keywords={}".format(self.__class__.__name__,self.input_blog[0],self.input_dep,self.input_node,self.input_start_time,self.input_minutes,self.input_end_time,self.input_repeats,self.input_keywords)
bot.py:
#!/usr/bin/python3
from telegram.ext import (Updater, Filters, CommandHandler, ConversationHandler, MessageHandler)
from my_package.step_states import Progress, StepState
from environments import get_bot_token, getLogger
# Enable logging
logger = getLogger(__name__)
updater = Updater(get_bot_token())
def start(bot, update):
"""
Sends a message when the command /start is issued.
:param bot: the bot
:param update: the update info from Telegram for this command
"""
update.message.reply_text('Bot started')
def error(bot, update, error):
"""
Logs errors
:param bot: the Telegram bot
:param update: the Telegram update, probably a text message
"""
logger.error('Update "%s" caused error "%s"', update, error)
def restricted(my_handler):
"""
Decorates a handler for restricting its use
:param my_handler: the handler to be restricted
:return: the restricted handler
"""
@wraps(my_handler)
def wrapped(bot, update, *args, **kwargs):
user_id, user_name = update.effective_user.id, update.effective_user.first_name
if user_id not in get_allowed_users():
update.message.reply_text("Unauthorized access {} con id {}.n".format(user_name, user_id))
return
logger.info('Entering {} '.format(my_handler.__name__))
return my_handler(bot, update, *args, **kwargs)
return wrapped
@restricted
def plan(bot, update): #, blog_id=get_blog_id(), interval=60, dep_param_id=None):
"""
Starts the wizard for scheduling item posts
:param bot: the bot
:param update: the update info from Telegram for this command
"""
p = Progress()
p.next_state()
p.state.draw_ui(bot, update)
return p.state
def main():
"""Start the bot."""
# Create the EventHandler and pass it your bot's token.
global updater
# Get the dispatcher to register handlers
dp = updater.dispatcher
# on different commands - answer in Telegram
dp.add_handler(CommandHandler('start', start))
# Add conversation handler with the states
# The telegram conversation handler needs a handler_list with functions
# so it can execute desired code in each state/step
conv_handler = ConversationHandler(
entry_points=[CommandHandler('plan', plan)],
# We enter all the states and all the configured handlers for each state
states={state: state.handler_list() for state in StepState.states},
fallbacks=[CommandHandler('cancel', StepState.cancel)]
)
dp.add_handler(conv_handler)
# log all errors
dp.add_error_handler(error)
# Start the Bot
updater.start_polling()
# Run the bot until you press Ctrl-C or the process receives SIGINT,
# SIGTERM or SIGABRT. This should be used most of the time, since
# start_polling() is non-blocking and will stop the bot gracefully.
updater.idle()
if __name__ == '__main__':
main()
answered Feb 12 at 19:05
madtyn
1415
add a comment |
up vote
0
down vote
I mainly decoupled the states from the bots putting a handler_list method inside each state.
Briefly I comment some suggestions and good points Gareeth Reese made:
- Cutting source code lines to 80 chars. Still pending. I'm doing it the next time I post
- The use of self.states and the index method to choose a next state is inefficient (proportional to number of states). The antipattern here is over-encapsulation, the co-ordination needs to be handled at a higher level,for example in a state machine class like your Progress class. Still pending
The data structure that's needed is a mapping from state to next state, for example, in the Progress class you could have:
_state_order = [NoneState(), DepState(), NodeState(), ...]
_next_state = dict(zip(_state_order[:-1], _state_order[1:]))
and then in the Progress.next_state method you'd write:
self.state = self._next_state[self.state]
The message method is only used in the context
self.message(update).reply_text(...)
The duplicated code could be eliminated like this:
def reply_text(self, update, *args, *kwargs):
"""Reply to a Telegram update ..."""
if update.callback_query:
message = update.callback_query.message
else:
message = update.message
message.reply_text(*args, **kwargs)
And since this does not use self, it does not need to be a method on a class, it could be a @staticmethod or, better, an ordinary function. Tried it but some parameter was giving an error
Borg pattern is unnecessary and its machinery can be dropped from the StepState class. Still thinking about it
After removing all these attributes and methods, the only attribute remaining is progress. This is only used in the handle method, and so it would make sense to pass it as a parameter to that method and avoid the need for the attribute. Still thinking about it
Still pending as well. After removing the progress attribute, none of the remaining methods use self, so there's no need to create state objects, you can just use the state classes as namespaces for the draw_ui and handle functions:
_state_order = [NoneState, DepState, NodeState, ...]
I evolved a little bit the code until I came to this below. I know that many changes are still pending and DRY principle is broken, but the main things are done:
step_states:
class StepState(metaclass=ABCMeta):
"""
This is a Borg class which shares state and methods so all the instances created for it are different references
but containing the same values. This is also true for children of the same class, so:
Children1() == Children1() # False, because references are different, but the values are the same.
Also if you change one, you change the other
Children1() == Children2() # False, references are different and you can change anyone without affecting others
"""
states =
# State is a Borg design pattern class
__shared_state = {}
def __init__(self):
self.__dict__ = self.__shared_state
@staticmethod
def message(update):
"""
Returns the object holding the info coming from Telegram with the methods for replying
and looking into this info
:param update: the Telegram update, probably a text message
:return: the update or query object
"""
if update.callback_query:
return update.callback_query.message
else:
return update.message
@abstractmethod
def draw_ui(self, bot, update):
"""
Draws the UI in Telegram (usually a custom keyboard) for the user to input data
:param bot: the Telegram bot
:param update: the Telegram update, probably a text message
"""
pass
@abstractmethod
def handle(self, bot=None, update=None):
"""
Handles all the input info into the Progress object and transitions to the next state
:param bot: the Telegram bot
:param update: the Telegram update, probably a text message
"""
logger.info('Handling state {} in parent'.format(self.__class__.__name__))
self.progress = Progress()
if update and update.callback_query:
query = update.callback_query
if query.data == CANCEL:
return self.cancel(bot, update)
@abstractmethod
def handler_list(self):
"""
Returns the list with the handlers for managing the events for this state
Example: return [ CallbackQueryHandler(callbackFunc), CommandHandler('cmd', commandFunc)]
Will apply the callbackFunc for managing a query-like update, but if not,
will try to apply commandFunc for an incoming command /cmd
:return: the handlers list
"""
pass
def next_state(self, new_state=None):
"""
Inherited method which decides to move to the next state, or to the specified state if any
:param new_state: the next state, if any
:return: the next state
"""
# TODO Check if the transition is allowed?
next_new_state = None
if new_state is None:
default_next_state_index = (self.index + 1) % len(self.states)
next_new_state = self.states[default_next_state_index]
else:
next_new_state = new_state
logger.info('Actual state: {} => switched to new state {}'.format(self, next_new_state))
return next_new_state
@staticmethod
def skip(bot, update):
"""
Skips this step in the wizard process
:param bot: the Telegram bot
:param update: the Telegram update, probably a text message
:return: the next state skipping the present one
"""
logger.info('Skipping state {}'.format(Progress().state))
update.message.reply_text("Se salta este paso.")
# If we don't cancel at the end, we should remove any keyboard which could be present
Progress().next_state()
Progress().state.draw_ui(bot, update)
return Progress().state
@staticmethod
def cancel(bot, update):
"""
Cancels the whole wizard process
:param bot: the Telegram bot
:param update: the Telegram update, probably a text message
:return: the END state for he ConversationHandler
"""
Progress().clear()
user = StepState.message(update).from_user
logger.info("{} canceled the process.".format(user.first_name))
StepState.message(update).reply_text('Proceso finalizado.', reply_markup=ReplyKeyboardRemove())
return ConversationHandler.END
def validate_input(self, input_data):
"""
Validates the input data for this step/state
:param input_data: the input data
:return: True if input data is valid, otherwise False
"""
return True
@property
def index(self):
"""
Returns the index of this state within the list with all states
:return: the index as an int
"""
return StepState.states.index(self)
def __eq__(self, other):
return self.__hash__() == other.__hash__()
def __hash__(self):
return hash(self.__class__.__name__)
def __str__(self):
return self.__class__.__name__
class DepState(StepState):
def draw_ui(self, bot, update):
reply_keyboard = [
]
num_cols = 3
current_row =
# Think of SEARCH_INDEXES as a list with objects having two strings: spanish_description and browse_node_id
for dep in SEARCH_INDEXES:
current_row.append(InlineKeyboardButton(dep.spanish_desc,
callback_data="{}={}".format(dep.browse_node_id, dep.spanish_desc)))
if len(current_row) >= num_cols:
reply_keyboard.append(current_row)
current_row =
reply_markup = InlineKeyboardMarkup(reply_keyboard)
self.message(update).reply_text('Enter the *departament*, do /skip or /cancel',
parse_mode='Markdown', reply_markup=reply_markup)
def handle(self, bot=None, update=None):
parent_result = super().handle(bot, update)
if parent_result is not None:
return parent_result
query = update.callback_query
if query.data:
current_input_dep, text = query.data.split('=')
bot.edit_message_text(text="Departament chosen: {}".format(text), chat_id=query.message.chat_id,
message_id=query.message.message_id)
# We optionally log anything
Progress().input_dep = current_input_dep
Progress().next_state()
Progress().state.draw_ui(bot, update)
return Progress().state
class NodeState(StepState):
def draw_ui(self, bot, update):
reply_keyboard = [
[InlineKeyboardButton('Delete', callback_data=DELETE)],
[InlineKeyboardButton(str(number), callback_data=str(number)) for number in range(5)],
[InlineKeyboardButton(str(number), callback_data=str(number)) for number in range(5, 10)],
[InlineKeyboardButton('Accept', callback_data=OK),
InlineKeyboardButton('Cancel', callback_data=CANCEL)],
]
reply_markup = InlineKeyboardMarkup(reply_keyboard)
self.message(update).reply_text('Input numbers for the node, do /skip or /cancel',
parse_mode='Markdown', reply_markup=reply_markup)
# We get the reference to the message we will use to update the value and set into Progress
Progress().tracking_message = self.message(update).reply_text("Typed data : {}".format(''))
def handle(self, bot=None, update=None):
parent_result = super().handle(bot, update)
if parent_result is not None:
return parent_result
query = update.callback_query
if hasattr(query, 'data'):
if query.data == OK:
if self.validate_input(self.progress.input_node):
bot.edit_message_text(text="Chosen node: {}".format(self.progress.input_node),
chat_id=query.message.chat_id, message_id=query.message.message_id)
bot.delete_message(chat_id=Progress().tracking_message.chat_id,
message_id=Progress().tracking_message.message_id)
Progress().next_state()
else:
no_keywords = self.progress.input_keywords is None or not self.progress.input_keywords.strip()
if self.progress.input_node:
self.message(update).reply_text('Not allowed value')
self.progress.input_node = ''
elif no_keywords:
# If there are no keywords, then the node is mandatory
self.message(update).reply_text('This input is mandatory. Enter an allowed value')
Progress().next_state(self)
Progress().state.draw_ui(bot, update)
return Progress().state
else:
if not self.progress.input_node:
self.progress.input_node = ''
if query.data == DELETE:
if len(self.progress.input_node):
self.progress.input_node = self.progress.input_node[:-1]
else:
self.progress.input_node += query.data
# We update the output so the user sees if he types correctly
bot.edit_message_text(text="Input data: {}".format(self.progress.input_node),
chat_id=query.message.chat_id,
message_id=Progress().tracking_message.message_id)
else:
if update.message.text and self.validate_input(update.message.text):
self.progress.input_node = update.message.text
Progress().next_state()
Progress().state.draw_ui(bot, update)
elif not self.validate_input(update.message.text):
self.message(update).reply_text('Valor no admitido')
self.message(update).reply_text('Este dato es obligatorio. Introduzca un valor valido')
self.progress.input_node = ''
Progress().next_state(self)
Progress().state.draw_ui(bot, update)
return Progress().state
@staticmethod
def skip(bot, update):
logger.info('Skipping state {}'.format(Progress().state))
# If we don't cancel at the end, we should remove any keyboard which could be present
if Progress().input_keywords is None or not Progress().input_keywords.strip():
update.message.reply_text('Este dato es obligatorio. Introduzca un valor valido')
else:
Progress().input_node = ''
Progress().next_state()
update.message.reply_text("Se salta este paso.")
Progress().state.draw_ui(bot, update)
return Progress().state
def validate_input(self, input_data):
data = input_data
if type(input_data) is str:
data = int(data.strip())
response = requests.get('https://www.amazon.es/exec/obidos/tg/browse/-/{}'.format(data))
return response.status_code == 200
def handler_list(self):
return [RegexHandler('^(d+)$', self.handle),
CallbackQueryHandler(self.handle), CommandHandler('skip', self.skip)]
class BlogState(StepState):
def draw_ui(self, bot, update):
reply_keyboard = [
[InlineKeyboardButton(blog, callback_data='{}={}'.format(blog, bid)) for blog, bid in sorted(get_blogs().items())]
]
reply_markup = InlineKeyboardMarkup(reply_keyboard)
self.message(update).reply_text('Choose an option, /skip or /cancel', parse_mode='Markdown',reply_markup=reply_markup)
def handle(self, bot=None, update=None):
parent_result = super().handle(bot, update)
if parent_result is not None:
return parent_result
query = update.callback_query
if query.data:
bname, bid = query.data.split('=')
bot.edit_message_text(text="Chosen option: {}".format(bname), chat_id=query.message.chat_id,
message_id=query.message.message_id)
# We optionally log anything
# We text the user the requirements for next state
# query.message.reply_text('Departamento elegido. ', reply_markup=ReplyKeyboardRemove())
Progress().input_blog = (bname,bid)
if bname == 'xxx':
Progress().input_dep = HEALTH.browse_node_id
Progress().next_state(KeywordsState())
elif bname == 'books':
Progress().input_dep = BOOKS.browse_node_id
Progress().next_state(KeywordsState())
else:
Progress().next_state()
Progress().state.draw_ui(bot, update)
return Progress().state
def handler_list(self):
return [CallbackQueryHandler(self.handle), CommandHandler('skip', self.skip)]
class StartTimeState(StepState):
def draw_ui(self, bot, update):
reply_keyboard = [
[InlineKeyboardButton('Delete', callback_data=DELETE)],
[InlineKeyboardButton(str(number), callback_data=str(number)) for number in range(5)],
[InlineKeyboardButton(str(number), callback_data=str(number)) for number in range(5, 10)],
[InlineKeyboardButton(':', callback_data=':')],
[InlineKeyboardButton('Accept', callback_data=OK),
InlineKeyboardButton('Cancel', callback_data=CANCEL)],
]
reply_markup = InlineKeyboardMarkup(reply_keyboard)
self.message(update).reply_text('Input start time or /cancel',
parse_mode='Markdown', reply_markup=reply_markup)
# We get the reference to the message we will use to update the value and set into Progress
Progress().tracking_message = self.message(update).reply_text("Start time: {}".format(''))
def handle(self, bot=None, update=None):
parent_result = super().handle(bot, update)
if parent_result is not None:
return parent_result
query = update.callback_query
if hasattr(query, 'data'):
if query.data == OK:
bot.delete_message(chat_id=query.message.chat_id, message_id=query.message.message_id)
Progress().next_state()
Progress().state.draw_ui(bot, update)
return Progress().state
else:
if not self.progress.input_start_time:
self.progress.input_start_time = ''
if query.data == DELETE:
if len(self.progress.input_start_time):
self.progress.input_start_time = self.progress.input_start_time[:-1]
else:
self.progress.input_start_time += query.data
# We update the output so the user sees if he types correctly
bot.edit_message_text(text="Start time: {}".format(self.progress.input_start_time),
chat_id=query.message.chat_id,
message_id=Progress().tracking_message.message_id)
else:
if update.message.text and self.validate_input(update.message.text):
self.progress.input_start_time = update.message.text
Progress().next_state()
Progress().state.draw_ui(bot, update)
elif not self.validate_input(update.message.text):
self.message(update).reply_text('Valor no admitido')
self.message(update).reply_text('Este dato es obligatorio. Introduzca un valor valido')
self.progress.input_start_time = ''
Progress().next_state(self)
Progress().state.draw_ui(bot, update)
return Progress().state
def handler_list(self):
return [RegexHandler(TIME_REGEX, self.handle),
CallbackQueryHandler(self.handle), CommandHandler('skip', self.skip)]
# ... more states
# ... more and more states
# ...and so on, you can imagine
class KeywordsState(StepState):
name = 'KeywordsState'
def draw_ui(self, bot, update):
self.message(update).reply_text('Enter keywords', parse_mode='Markdown',
reply_markup=ReplyKeyboardRemove())
def handle(self, bot=None, update=None):
parent_result = super().handle(bot, update)
if parent_result is not None:
return parent_result
keywords = update.message.text.strip()
if keywords:
Progress().input_keywords = keywords
update.message.reply_text('Thanks. The keywords are: {}'.format(keywords))
# When SearchIndex equals All, BrowseNode cannot be present
if Progress().input_dep is None:
Progress().next_state(StartTimeState())
else:
Progress().next_state()
else:
update.message.reply_text('No ha introducido palabras. Introdúzcalas, haga /skip o /cancel')
Progress().next_state(self)
Progress().state.draw_ui(bot, update)
return Progress().state
@staticmethod
def skip(bot, update):
logger.info('Skipping state {}'.format(Progress().state))
# If we don't cancel at the end, we should remove any keyboard which could be present
if Progress().input_dep is None:
update.message.reply_text('Este dato es obligatorio. Introduzca un valor valido')
else:
Progress().input_keywords = None
Progress().next_state()
update.message.reply_text("Skipping this step.")
Progress().state.draw_ui(bot, update)
return Progress().state
def handler_list(self):
return [MessageHandler(Filters.text, self.handle), CommandHandler('skip', self.skip)]
class ConfirmState(StepState):
def draw_ui(self, bot, update):
self.message(update).reply_text('Confirm all this info: {!s}'.format(Progress()))
reply_keyboard = [
[InlineKeyboardButton('Accept', callback_data=OK),
InlineKeyboardButton('Cancel', callback_data=CANCEL)],
]
reply_markup = InlineKeyboardMarkup(reply_keyboard)
self.message(update).reply_text('Accept or cancel (/cancel)', reply_markup=reply_markup)
# This returns a generator with a data list to consume
self.generator = search_asins(Progress().input_dep, Progress().input_node, Progress().input_keywords)
def handle(self, bot=None, update=None):
parent_result = super().handle(bot, update)
if parent_result is not None:
return parent_result
# OK granted, CANCEL is managed in parent
query = update.callback_query
# We finished with the wizard and launch everything
# Launching stuff here...
# ...
# Progress should be reset
self.progress.clear()
Progress().next_state(NoneState())
return ConversationHandler.END
def handler_list(self):
return [CallbackQueryHandler(self.handle)]
# This is intended for setting the steps in an ordered-sorted way
StepState.states = [
NoneState(), BlogState(), DepState(), KeywordsState(), NodeState(), StartTimeState(),
IntervalState(), EndTimeState(), RepeatsState(), ConfirmState()
]
class Progress(object):
"""
This singleton class contains the whole information collected along all the wizard process
"""
__instance = None
def __new__(cls):
if Progress.__instance is None:
Progress.__instance = object.__new__(cls)
Progress.__instance.input_blog = None
Progress.__instance.state = NoneState()
Progress.__instance.tracking_message = None
Progress.__instance.input_dep = None
Progress.__instance.input_node = None
Progress.__instance.input_minutes = '60'
Progress.__instance.input_start_time = None
Progress.__instance.input_end_time = None
Progress.__instance.input_keywords = None
Progress.__instance.input_repeats = None
return Progress.__instance
def clear(self):
"""
Resets the progress
"""
self.input_blog = None
self.state = NoneState()
self.tracking_message = None
self.input_dep = None
self.input_node = None
self.input_minutes = '60'
self.input_start_time = None
self.input_end_time = None
self.input_keywords = None
def next_state(self, new_state=None):
"""
Moves to the next state or to the specified state if any
:param new_state: the next state to move to
:return: the next state, already set to the Progress object
"""
if new_state is not None:
if self.state:
self.state = self.state.next_state(new_state)
else:
self.state = NoneState
logger.info("It was impossible o move to the next state")
else:
self.state = self.state.next_state()
def __str__(self):
return "{}: blog={}, dep={}, node={}, start={}, "
"interval={}', end={}, repeats={}, keywords={}".format(self.__class__.__name__,self.input_blog[0],self.input_dep,self.input_node,self.input_start_time,self.input_minutes,self.input_end_time,self.input_repeats,self.input_keywords)
bot.py:
#!/usr/bin/python3
from telegram.ext import (Updater, Filters, CommandHandler, ConversationHandler, MessageHandler)
from my_package.step_states import Progress, StepState
from environments import get_bot_token, getLogger
# Enable logging
logger = getLogger(__name__)
updater = Updater(get_bot_token())
def start(bot, update):
"""
Sends a message when the command /start is issued.
:param bot: the bot
:param update: the update info from Telegram for this command
"""
update.message.reply_text('Bot started')
def error(bot, update, error):
"""
Logs errors
:param bot: the Telegram bot
:param update: the Telegram update, probably a text message
"""
logger.error('Update "%s" caused error "%s"', update, error)
def restricted(my_handler):
"""
Decorates a handler for restricting its use
:param my_handler: the handler to be restricted
:return: the restricted handler
"""
@wraps(my_handler)
def wrapped(bot, update, *args, **kwargs):
user_id, user_name = update.effective_user.id, update.effective_user.first_name
if user_id not in get_allowed_users():
update.message.reply_text("Unauthorized access {} con id {}.n".format(user_name, user_id))
return
logger.info('Entering {} '.format(my_handler.__name__))
return my_handler(bot, update, *args, **kwargs)
return wrapped
@restricted
def plan(bot, update): #, blog_id=get_blog_id(), interval=60, dep_param_id=None):
"""
Starts the wizard for scheduling item posts
:param bot: the bot
:param update: the update info from Telegram for this command
"""
p = Progress()
p.next_state()
p.state.draw_ui(bot, update)
return p.state
def main():
"""Start the bot."""
# Create the EventHandler and pass it your bot's token.
global updater
# Get the dispatcher to register handlers
dp = updater.dispatcher
# on different commands - answer in Telegram
dp.add_handler(CommandHandler('start', start))
# Add conversation handler with the states
# The telegram conversation handler needs a handler_list with functions
# so it can execute desired code in each state/step
conv_handler = ConversationHandler(
entry_points=[CommandHandler('plan', plan)],
# We enter all the states and all the configured handlers for each state
states={state: state.handler_list() for state in StepState.states},
fallbacks=[CommandHandler('cancel', StepState.cancel)]
)
dp.add_handler(conv_handler)
# log all errors
dp.add_error_handler(error)
# Start the Bot
updater.start_polling()
# Run the bot until you press Ctrl-C or the process receives SIGINT,
# SIGTERM or SIGABRT. This should be used most of the time, since
# start_polling() is non-blocking and will stop the bot gracefully.
updater.idle()
if __name__ == '__main__':
main()
answered Feb 12 at 19:05
madtyn
1415
add a comment |
up vote
0
down vote
up vote
0
down vote
I mainly decoupled the states from the bots putting a handler_list method inside each state.
Briefly I comment some suggestions and good points Gareeth Reese made:
- Cutting source code lines to 80 chars. Still pending. I'm doing it the next time I post
- The use of self.states and the index method to choose a next state is inefficient (proportional to number of states). The antipattern here is over-encapsulation, the co-ordination needs to be handled at a higher level,for example in a state machine class like your Progress class. Still pending
The data structure that's needed is a mapping from state to next state, for example, in the Progress class you could have:
_state_order = [NoneState(), DepState(), NodeState(), ...]
_next_state = dict(zip(_state_order[:-1], _state_order[1:]))
and then in the Progress.next_state method you'd write:
self.state = self._next_state[self.state]
The message method is only used in the context
self.message(update).reply_text(...)
The duplicated code could be eliminated like this:
def reply_text(self, update, *args, *kwargs):
"""Reply to a Telegram update ..."""
if update.callback_query:
message = update.callback_query.message
else:
message = update.message
message.reply_text(*args, **kwargs)
And since this does not use self, it does not need to be a method on a class, it could be a @staticmethod or, better, an ordinary function. Tried it but some parameter was giving an error
Borg pattern is unnecessary and its machinery can be dropped from the StepState class. Still thinking about it
After removing all these attributes and methods, the only attribute remaining is progress. This is only used in the handle method, and so it would make sense to pass it as a parameter to that method and avoid the need for the attribute. Still thinking about it
Still pending as well. After removing the progress attribute, none of the remaining methods use self, so there's no need to create state objects, you can just use the state classes as namespaces for the draw_ui and handle functions:
_state_order = [NoneState, DepState, NodeState, ...]
I evolved a little bit the code until I came to this below. I know that many changes are still pending and DRY principle is broken, but the main things are done:
step_states:
class StepState(metaclass=ABCMeta):
"""
This is a Borg class which shares state and methods so all the instances created for it are different references
but containing the same values. This is also true for children of the same class, so:
Children1() == Children1() # False, because references are different, but the values are the same.
Also if you change one, you change the other
Children1() == Children2() # False, references are different and you can change anyone without affecting others
"""
states =
# State is a Borg design pattern class
__shared_state = {}
def __init__(self):
self.__dict__ = self.__shared_state
@staticmethod
def message(update):
"""
Returns the object holding the info coming from Telegram with the methods for replying
and looking into this info
:param update: the Telegram update, probably a text message
:return: the update or query object
"""
if update.callback_query:
return update.callback_query.message
else:
return update.message
@abstractmethod
def draw_ui(self, bot, update):
"""
Draws the UI in Telegram (usually a custom keyboard) for the user to input data
:param bot: the Telegram bot
:param update: the Telegram update, probably a text message
"""
pass
@abstractmethod
def handle(self, bot=None, update=None):
"""
Handles all the input info into the Progress object and transitions to the next state
:param bot: the Telegram bot
:param update: the Telegram update, probably a text message
"""
logger.info('Handling state {} in parent'.format(self.__class__.__name__))
self.progress = Progress()
if update and update.callback_query:
query = update.callback_query
if query.data == CANCEL:
return self.cancel(bot, update)
@abstractmethod
def handler_list(self):
"""
Returns the list with the handlers for managing the events for this state
Example: return [ CallbackQueryHandler(callbackFunc), CommandHandler('cmd', commandFunc)]
Will apply the callbackFunc for managing a query-like update, but if not,
will try to apply commandFunc for an incoming command /cmd
:return: the handlers list
"""
pass
def next_state(self, new_state=None):
"""
Inherited method which decides to move to the next state, or to the specified state if any
:param new_state: the next state, if any
:return: the next state
"""
# TODO Check if the transition is allowed?
next_new_state = None
if new_state is None:
default_next_state_index = (self.index + 1) % len(self.states)
next_new_state = self.states[default_next_state_index]
else:
next_new_state = new_state
logger.info('Actual state: {} => switched to new state {}'.format(self, next_new_state))
return next_new_state
@staticmethod
def skip(bot, update):
"""
Skips this step in the wizard process
:param bot: the Telegram bot
:param update: the Telegram update, probably a text message
:return: the next state skipping the present one
"""
logger.info('Skipping state {}'.format(Progress().state))
update.message.reply_text("Se salta este paso.")
# If we don't cancel at the end, we should remove any keyboard which could be present
Progress().next_state()
Progress().state.draw_ui(bot, update)
return Progress().state
@staticmethod
def cancel(bot, update):
"""
Cancels the whole wizard process
:param bot: the Telegram bot
:param update: the Telegram update, probably a text message
:return: the END state for he ConversationHandler
"""
Progress().clear()
user = StepState.message(update).from_user
logger.info("{} canceled the process.".format(user.first_name))
StepState.message(update).reply_text('Proceso finalizado.', reply_markup=ReplyKeyboardRemove())
return ConversationHandler.END
def validate_input(self, input_data):
"""
Validates the input data for this step/state
:param input_data: the input data
:return: True if input data is valid, otherwise False
"""
return True
@property
def index(self):
"""
Returns the index of this state within the list with all states
:return: the index as an int
"""
return StepState.states.index(self)
def __eq__(self, other):
return self.__hash__() == other.__hash__()
def __hash__(self):
return hash(self.__class__.__name__)
def __str__(self):
return self.__class__.__name__
class DepState(StepState):
def draw_ui(self, bot, update):
reply_keyboard = [
]
num_cols = 3
current_row =
# Think of SEARCH_INDEXES as a list with objects having two strings: spanish_description and browse_node_id
for dep in SEARCH_INDEXES:
current_row.append(InlineKeyboardButton(dep.spanish_desc,
callback_data="{}={}".format(dep.browse_node_id, dep.spanish_desc)))
if len(current_row) >= num_cols:
reply_keyboard.append(current_row)
current_row =
reply_markup = InlineKeyboardMarkup(reply_keyboard)
self.message(update).reply_text('Enter the *departament*, do /skip or /cancel',
parse_mode='Markdown', reply_markup=reply_markup)
def handle(self, bot=None, update=None):
parent_result = super().handle(bot, update)
if parent_result is not None:
return parent_result
query = update.callback_query
if query.data:
current_input_dep, text = query.data.split('=')
bot.edit_message_text(text="Departament chosen: {}".format(text), chat_id=query.message.chat_id,
message_id=query.message.message_id)
# We optionally log anything
Progress().input_dep = current_input_dep
Progress().next_state()
Progress().state.draw_ui(bot, update)
return Progress().state
class NodeState(StepState):
def draw_ui(self, bot, update):
reply_keyboard = [
[InlineKeyboardButton('Delete', callback_data=DELETE)],
[InlineKeyboardButton(str(number), callback_data=str(number)) for number in range(5)],
[InlineKeyboardButton(str(number), callback_data=str(number)) for number in range(5, 10)],
[InlineKeyboardButton('Accept', callback_data=OK),
InlineKeyboardButton('Cancel', callback_data=CANCEL)],
]
reply_markup = InlineKeyboardMarkup(reply_keyboard)
self.message(update).reply_text('Input numbers for the node, do /skip or /cancel',
parse_mode='Markdown', reply_markup=reply_markup)
# We get the reference to the message we will use to update the value and set into Progress
Progress().tracking_message = self.message(update).reply_text("Typed data : {}".format(''))
def handle(self, bot=None, update=None):
parent_result = super().handle(bot, update)
if parent_result is not None:
return parent_result
query = update.callback_query
if hasattr(query, 'data'):
if query.data == OK:
if self.validate_input(self.progress.input_node):
bot.edit_message_text(text="Chosen node: {}".format(self.progress.input_node),
chat_id=query.message.chat_id, message_id=query.message.message_id)
bot.delete_message(chat_id=Progress().tracking_message.chat_id,
message_id=Progress().tracking_message.message_id)
Progress().next_state()
else:
no_keywords = self.progress.input_keywords is None or not self.progress.input_keywords.strip()
if self.progress.input_node:
self.message(update).reply_text('Not allowed value')
self.progress.input_node = ''
elif no_keywords:
# If there are no keywords, then the node is mandatory
self.message(update).reply_text('This input is mandatory. Enter an allowed value')
Progress().next_state(self)
Progress().state.draw_ui(bot, update)
return Progress().state
else:
if not self.progress.input_node:
self.progress.input_node = ''
if query.data == DELETE:
if len(self.progress.input_node):
self.progress.input_node = self.progress.input_node[:-1]
else:
self.progress.input_node += query.data
# We update the output so the user sees if he types correctly
bot.edit_message_text(text="Input data: {}".format(self.progress.input_node),
chat_id=query.message.chat_id,
message_id=Progress().tracking_message.message_id)
else:
if update.message.text and self.validate_input(update.message.text):
self.progress.input_node = update.message.text
Progress().next_state()
Progress().state.draw_ui(bot, update)
elif not self.validate_input(update.message.text):
self.message(update).reply_text('Valor no admitido')
self.message(update).reply_text('Este dato es obligatorio. Introduzca un valor valido')
self.progress.input_node = ''
Progress().next_state(self)
Progress().state.draw_ui(bot, update)
return Progress().state
@staticmethod
def skip(bot, update):
logger.info('Skipping state {}'.format(Progress().state))
# If we don't cancel at the end, we should remove any keyboard which could be present
if Progress().input_keywords is None or not Progress().input_keywords.strip():
update.message.reply_text('Este dato es obligatorio. Introduzca un valor valido')
else:
Progress().input_node = ''
Progress().next_state()
update.message.reply_text("Se salta este paso.")
Progress().state.draw_ui(bot, update)
return Progress().state
def validate_input(self, input_data):
data = input_data
if type(input_data) is str:
data = int(data.strip())
response = requests.get('https://www.amazon.es/exec/obidos/tg/browse/-/{}'.format(data))
return response.status_code == 200
def handler_list(self):
return [RegexHandler('^(d+)$', self.handle),
CallbackQueryHandler(self.handle), CommandHandler('skip', self.skip)]
class BlogState(StepState):
def draw_ui(self, bot, update):
reply_keyboard = [
[InlineKeyboardButton(blog, callback_data='{}={}'.format(blog, bid)) for blog, bid in sorted(get_blogs().items())]
]
reply_markup = InlineKeyboardMarkup(reply_keyboard)
self.message(update).reply_text('Choose an option, /skip or /cancel', parse_mode='Markdown',reply_markup=reply_markup)
def handle(self, bot=None, update=None):
parent_result = super().handle(bot, update)
if parent_result is not None:
return parent_result
query = update.callback_query
if query.data:
bname, bid = query.data.split('=')
bot.edit_message_text(text="Chosen option: {}".format(bname), chat_id=query.message.chat_id,
message_id=query.message.message_id)
# We optionally log anything
# We text the user the requirements for next state
# query.message.reply_text('Departamento elegido. ', reply_markup=ReplyKeyboardRemove())
Progress().input_blog = (bname,bid)
if bname == 'xxx':
Progress().input_dep = HEALTH.browse_node_id
Progress().next_state(KeywordsState())
elif bname == 'books':
Progress().input_dep = BOOKS.browse_node_id
Progress().next_state(KeywordsState())
else:
Progress().next_state()
Progress().state.draw_ui(bot, update)
return Progress().state
def handler_list(self):
return [CallbackQueryHandler(self.handle), CommandHandler('skip', self.skip)]
class StartTimeState(StepState):
def draw_ui(self, bot, update):
reply_keyboard = [
[InlineKeyboardButton('Delete', callback_data=DELETE)],
[InlineKeyboardButton(str(number), callback_data=str(number)) for number in range(5)],
[InlineKeyboardButton(str(number), callback_data=str(number)) for number in range(5, 10)],
[InlineKeyboardButton(':', callback_data=':')],
[InlineKeyboardButton('Accept', callback_data=OK),
InlineKeyboardButton('Cancel', callback_data=CANCEL)],
]
reply_markup = InlineKeyboardMarkup(reply_keyboard)
self.message(update).reply_text('Input start time or /cancel',
parse_mode='Markdown', reply_markup=reply_markup)
# We get the reference to the message we will use to update the value and set into Progress
Progress().tracking_message = self.message(update).reply_text("Start time: {}".format(''))
def handle(self, bot=None, update=None):
parent_result = super().handle(bot, update)
if parent_result is not None:
return parent_result
query = update.callback_query
if hasattr(query, 'data'):
if query.data == OK:
bot.delete_message(chat_id=query.message.chat_id, message_id=query.message.message_id)
Progress().next_state()
Progress().state.draw_ui(bot, update)
return Progress().state
else:
if not self.progress.input_start_time:
self.progress.input_start_time = ''
if query.data == DELETE:
if len(self.progress.input_start_time):
self.progress.input_start_time = self.progress.input_start_time[:-1]
else:
self.progress.input_start_time += query.data
# We update the output so the user sees if he types correctly
bot.edit_message_text(text="Start time: {}".format(self.progress.input_start_time),
chat_id=query.message.chat_id,
message_id=Progress().tracking_message.message_id)
else:
if update.message.text and self.validate_input(update.message.text):
self.progress.input_start_time = update.message.text
Progress().next_state()
Progress().state.draw_ui(bot, update)
elif not self.validate_input(update.message.text):
self.message(update).reply_text('Valor no admitido')
self.message(update).reply_text('Este dato es obligatorio. Introduzca un valor valido')
self.progress.input_start_time = ''
Progress().next_state(self)
Progress().state.draw_ui(bot, update)
return Progress().state
def handler_list(self):
return [RegexHandler(TIME_REGEX, self.handle),
CallbackQueryHandler(self.handle), CommandHandler('skip', self.skip)]
# ... more states
# ... more and more states
# ...and so on, you can imagine
class KeywordsState(StepState):
name = 'KeywordsState'
def draw_ui(self, bot, update):
self.message(update).reply_text('Enter keywords', parse_mode='Markdown',
reply_markup=ReplyKeyboardRemove())
def handle(self, bot=None, update=None):
parent_result = super().handle(bot, update)
if parent_result is not None:
return parent_result
keywords = update.message.text.strip()
if keywords:
Progress().input_keywords = keywords
update.message.reply_text('Thanks. The keywords are: {}'.format(keywords))
# When SearchIndex equals All, BrowseNode cannot be present
if Progress().input_dep is None:
Progress().next_state(StartTimeState())
else:
Progress().next_state()
else:
update.message.reply_text('No ha introducido palabras. Introdúzcalas, haga /skip o /cancel')
Progress().next_state(self)
Progress().state.draw_ui(bot, update)
return Progress().state
@staticmethod
def skip(bot, update):
logger.info('Skipping state {}'.format(Progress().state))
# If we don't cancel at the end, we should remove any keyboard which could be present
if Progress().input_dep is None:
update.message.reply_text('Este dato es obligatorio. Introduzca un valor valido')
else:
Progress().input_keywords = None
Progress().next_state()
update.message.reply_text("Skipping this step.")
Progress().state.draw_ui(bot, update)
return Progress().state
def handler_list(self):
return [MessageHandler(Filters.text, self.handle), CommandHandler('skip', self.skip)]
class ConfirmState(StepState):
def draw_ui(self, bot, update):
self.message(update).reply_text('Confirm all this info: {!s}'.format(Progress()))
reply_keyboard = [
[InlineKeyboardButton('Accept', callback_data=OK),
InlineKeyboardButton('Cancel', callback_data=CANCEL)],
]
reply_markup = InlineKeyboardMarkup(reply_keyboard)
self.message(update).reply_text('Accept or cancel (/cancel)', reply_markup=reply_markup)
# This returns a generator with a data list to consume
self.generator = search_asins(Progress().input_dep, Progress().input_node, Progress().input_keywords)
def handle(self, bot=None, update=None):
parent_result = super().handle(bot, update)
if parent_result is not None:
return parent_result
# OK granted, CANCEL is managed in parent
query = update.callback_query
# We finished with the wizard and launch everything
# Launching stuff here...
# ...
# Progress should be reset
self.progress.clear()
Progress().next_state(NoneState())
return ConversationHandler.END
def handler_list(self):
return [CallbackQueryHandler(self.handle)]
# This is intended for setting the steps in an ordered-sorted way
StepState.states = [
NoneState(), BlogState(), DepState(), KeywordsState(), NodeState(), StartTimeState(),
IntervalState(), EndTimeState(), RepeatsState(), ConfirmState()
]
class Progress(object):
"""
This singleton class contains the whole information collected along all the wizard process
"""
__instance = None
def __new__(cls):
if Progress.__instance is None:
Progress.__instance = object.__new__(cls)
Progress.__instance.input_blog = None
Progress.__instance.state = NoneState()
Progress.__instance.tracking_message = None
Progress.__instance.input_dep = None
Progress.__instance.input_node = None
Progress.__instance.input_minutes = '60'
Progress.__instance.input_start_time = None
Progress.__instance.input_end_time = None
Progress.__instance.input_keywords = None
Progress.__instance.input_repeats = None
return Progress.__instance
def clear(self):
"""
Resets the progress
"""
self.input_blog = None
self.state = NoneState()
self.tracking_message = None
self.input_dep = None
self.input_node = None
self.input_minutes = '60'
self.input_start_time = None
self.input_end_time = None
self.input_keywords = None
def next_state(self, new_state=None):
"""
Moves to the next state or to the specified state if any
:param new_state: the next state to move to
:return: the next state, already set to the Progress object
"""
if new_state is not None:
if self.state:
self.state = self.state.next_state(new_state)
else:
self.state = NoneState
logger.info("It was impossible o move to the next state")
else:
self.state = self.state.next_state()
def __str__(self):
return "{}: blog={}, dep={}, node={}, start={}, "
"interval={}', end={}, repeats={}, keywords={}".format(self.__class__.__name__,self.input_blog[0],self.input_dep,self.input_node,self.input_start_time,self.input_minutes,self.input_end_time,self.input_repeats,self.input_keywords)
bot.py:
#!/usr/bin/python3
from telegram.ext import (Updater, Filters, CommandHandler, ConversationHandler, MessageHandler)
from my_package.step_states import Progress, StepState
from environments import get_bot_token, getLogger
# Enable logging
logger = getLogger(__name__)
updater = Updater(get_bot_token())
def start(bot, update):
"""
Sends a message when the command /start is issued.
:param bot: the bot
:param update: the update info from Telegram for this command
"""
update.message.reply_text('Bot started')
def error(bot, update, error):
"""
Logs errors
:param bot: the Telegram bot
:param update: the Telegram update, probably a text message
"""
logger.error('Update "%s" caused error "%s"', update, error)
def restricted(my_handler):
"""
Decorates a handler for restricting its use
:param my_handler: the handler to be restricted
:return: the restricted handler
"""
@wraps(my_handler)
def wrapped(bot, update, *args, **kwargs):
user_id, user_name = update.effective_user.id, update.effective_user.first_name
if user_id not in get_allowed_users():
update.message.reply_text("Unauthorized access {} con id {}.n".format(user_name, user_id))
return
logger.info('Entering {} '.format(my_handler.__name__))
return my_handler(bot, update, *args, **kwargs)
return wrapped
@restricted
def plan(bot, update): #, blog_id=get_blog_id(), interval=60, dep_param_id=None):
"""
Starts the wizard for scheduling item posts
:param bot: the bot
:param update: the update info from Telegram for this command
"""
p = Progress()
p.next_state()
p.state.draw_ui(bot, update)
return p.state
def main():
"""Start the bot."""
# Create the EventHandler and pass it your bot's token.
global updater
# Get the dispatcher to register handlers
dp = updater.dispatcher
# on different commands - answer in Telegram
dp.add_handler(CommandHandler('start', start))
# Add conversation handler with the states
# The telegram conversation handler needs a handler_list with functions
# so it can execute desired code in each state/step
conv_handler = ConversationHandler(
entry_points=[CommandHandler('plan', plan)],
# We enter all the states and all the configured handlers for each state
states={state: state.handler_list() for state in StepState.states},
fallbacks=[CommandHandler('cancel', StepState.cancel)]
)
dp.add_handler(conv_handler)
# log all errors
dp.add_error_handler(error)
# Start the Bot
updater.start_polling()
# Run the bot until you press Ctrl-C or the process receives SIGINT,
# SIGTERM or SIGABRT. This should be used most of the time, since
# start_polling() is non-blocking and will stop the bot gracefully.
updater.idle()
if __name__ == '__main__':
main()
answered Feb 12 at 19:05
madtyn
1415
I mainly decoupled the states from the bots putting a handler_list method inside each state.
Briefly I comment some suggestions and good points Gareeth Reese made:
- Cutting source code lines to 80 chars. Still pending. I'm doing it the next time I post
- The use of self.states and the index method to choose a next state is inefficient (proportional to number of states). The antipattern here is over-encapsulation, the co-ordination needs to be handled at a higher level,for example in a state machine class like your Progress class. Still pending
The data structure that's needed is a mapping from state to next state, for example, in the Progress class you could have:
_state_order = [NoneState(), DepState(), NodeState(), ...]
_next_state = dict(zip(_state_order[:-1], _state_order[1:]))
and then in the Progress.next_state method you'd write:
self.state = self._next_state[self.state]
The message method is only used in the context
self.message(update).reply_text(...)
The duplicated code could be eliminated like this:
def reply_text(self, update, *args, *kwargs):
"""Reply to a Telegram update ..."""
if update.callback_query:
message = update.callback_query.message
else:
message = update.message
message.reply_text(*args, **kwargs)
And since this does not use self, it does not need to be a method on a class, it could be a @staticmethod or, better, an ordinary function. Tried it but some parameter was giving an error
Borg pattern is unnecessary and its machinery can be dropped from the StepState class. Still thinking about it
After removing all these attributes and methods, the only attribute remaining is progress. This is only used in the handle method, and so it would make sense to pass it as a parameter to that method and avoid the need for the attribute. Still thinking about it
Still pending as well. After removing the progress attribute, none of the remaining methods use self, so there's no need to create state objects, you can just use the state classes as namespaces for the draw_ui and handle functions:
_state_order = [NoneState, DepState, NodeState, ...]
I evolved a little bit the code until I came to this below. I know that many changes are still pending and DRY principle is broken, but the main things are done:
step_states:
class StepState(metaclass=ABCMeta):
"""
This is a Borg class which shares state and methods so all the instances created for it are different references
but containing the same values. This is also true for children of the same class, so:
Children1() == Children1() # False, because references are different, but the values are the same.
Also if you change one, you change the other
Children1() == Children2() # False, references are different and you can change anyone without affecting others
"""
states =
# State is a Borg design pattern class
__shared_state = {}
def __init__(self):
self.__dict__ = self.__shared_state
@staticmethod
def message(update):
"""
Returns the object holding the info coming from Telegram with the methods for replying
and looking into this info
:param update: the Telegram update, probably a text message
:return: the update or query object
"""
if update.callback_query:
return update.callback_query.message
else:
return update.message
@abstractmethod
def draw_ui(self, bot, update):
"""
Draws the UI in Telegram (usually a custom keyboard) for the user to input data
:param bot: the Telegram bot
:param update: the Telegram update, probably a text message
"""
pass
@abstractmethod
def handle(self, bot=None, update=None):
"""
Handles all the input info into the Progress object and transitions to the next state
:param bot: the Telegram bot
:param update: the Telegram update, probably a text message
"""
logger.info('Handling state {} in parent'.format(self.__class__.__name__))
self.progress = Progress()
if update and update.callback_query:
query = update.callback_query
if query.data == CANCEL:
return self.cancel(bot, update)
@abstractmethod
def handler_list(self):
"""
Returns the list with the handlers for managing the events for this state
Example: return [ CallbackQueryHandler(callbackFunc), CommandHandler('cmd', commandFunc)]
Will apply the callbackFunc for managing a query-like update, but if not,
will try to apply commandFunc for an incoming command /cmd
:return: the handlers list
"""
pass
def next_state(self, new_state=None):
"""
Inherited method which decides to move to the next state, or to the specified state if any
:param new_state: the next state, if any
:return: the next state
"""
# TODO Check if the transition is allowed?
next_new_state = None
if new_state is None:
default_next_state_index = (self.index + 1) % len(self.states)
next_new_state = self.states[default_next_state_index]
else:
next_new_state = new_state
logger.info('Actual state: {} => switched to new state {}'.format(self, next_new_state))
return next_new_state
@staticmethod
def skip(bot, update):
"""
Skips this step in the wizard process
:param bot: the Telegram bot
:param update: the Telegram update, probably a text message
:return: the next state skipping the present one
"""
logger.info('Skipping state {}'.format(Progress().state))
update.message.reply_text("Se salta este paso.")
# If we don't cancel at the end, we should remove any keyboard which could be present
Progress().next_state()
Progress().state.draw_ui(bot, update)
return Progress().state
@staticmethod
def cancel(bot, update):
"""
Cancels the whole wizard process
:param bot: the Telegram bot
:param update: the Telegram update, probably a text message
:return: the END state for he ConversationHandler
"""
Progress().clear()
user = StepState.message(update).from_user
logger.info("{} canceled the process.".format(user.first_name))
StepState.message(update).reply_text('Proceso finalizado.', reply_markup=ReplyKeyboardRemove())
return ConversationHandler.END
def validate_input(self, input_data):
"""
Validates the input data for this step/state
:param input_data: the input data
:return: True if input data is valid, otherwise False
"""
return True
@property
def index(self):
"""
Returns the index of this state within the list with all states
:return: the index as an int
"""
return StepState.states.index(self)
def __eq__(self, other):
return self.__hash__() == other.__hash__()
def __hash__(self):
return hash(self.__class__.__name__)
def __str__(self):
return self.__class__.__name__
class DepState(StepState):
def draw_ui(self, bot, update):
reply_keyboard = [
]
num_cols = 3
current_row =
# Think of SEARCH_INDEXES as a list with objects having two strings: spanish_description and browse_node_id
for dep in SEARCH_INDEXES:
current_row.append(InlineKeyboardButton(dep.spanish_desc,
callback_data="{}={}".format(dep.browse_node_id, dep.spanish_desc)))
if len(current_row) >= num_cols:
reply_keyboard.append(current_row)
current_row =
reply_markup = InlineKeyboardMarkup(reply_keyboard)
self.message(update).reply_text('Enter the *departament*, do /skip or /cancel',
parse_mode='Markdown', reply_markup=reply_markup)
def handle(self, bot=None, update=None):
parent_result = super().handle(bot, update)
if parent_result is not None:
return parent_result
query = update.callback_query
if query.data:
current_input_dep, text = query.data.split('=')
bot.edit_message_text(text="Departament chosen: {}".format(text), chat_id=query.message.chat_id,
message_id=query.message.message_id)
# We optionally log anything
Progress().input_dep = current_input_dep
Progress().next_state()
Progress().state.draw_ui(bot, update)
return Progress().state
class NodeState(StepState):
def draw_ui(self, bot, update):
reply_keyboard = [
[InlineKeyboardButton('Delete', callback_data=DELETE)],
[InlineKeyboardButton(str(number), callback_data=str(number)) for number in range(5)],
[InlineKeyboardButton(str(number), callback_data=str(number)) for number in range(5, 10)],
[InlineKeyboardButton('Accept', callback_data=OK),
InlineKeyboardButton('Cancel', callback_data=CANCEL)],
]
reply_markup = InlineKeyboardMarkup(reply_keyboard)
self.message(update).reply_text('Input numbers for the node, do /skip or /cancel',
parse_mode='Markdown', reply_markup=reply_markup)
# We get the reference to the message we will use to update the value and set into Progress
Progress().tracking_message = self.message(update).reply_text("Typed data : {}".format(''))
def handle(self, bot=None, update=None):
parent_result = super().handle(bot, update)
if parent_result is not None:
return parent_result
query = update.callback_query
if hasattr(query, 'data'):
if query.data == OK:
if self.validate_input(self.progress.input_node):
bot.edit_message_text(text="Chosen node: {}".format(self.progress.input_node),
chat_id=query.message.chat_id, message_id=query.message.message_id)
bot.delete_message(chat_id=Progress().tracking_message.chat_id,
message_id=Progress().tracking_message.message_id)
Progress().next_state()
else:
no_keywords = self.progress.input_keywords is None or not self.progress.input_keywords.strip()
if self.progress.input_node:
self.message(update).reply_text('Not allowed value')
self.progress.input_node = ''
elif no_keywords:
# If there are no keywords, then the node is mandatory
self.message(update).reply_text('This input is mandatory. Enter an allowed value')
Progress().next_state(self)
Progress().state.draw_ui(bot, update)
return Progress().state
else:
if not self.progress.input_node:
self.progress.input_node = ''
if query.data == DELETE:
if len(self.progress.input_node):
self.progress.input_node = self.progress.input_node[:-1]
else:
self.progress.input_node += query.data
# We update the output so the user sees if he types correctly
bot.edit_message_text(text="Input data: {}".format(self.progress.input_node),
chat_id=query.message.chat_id,
message_id=Progress().tracking_message.message_id)
else:
if update.message.text and self.validate_input(update.message.text):
self.progress.input_node = update.message.text
Progress().next_state()
Progress().state.draw_ui(bot, update)
elif not self.validate_input(update.message.text):
self.message(update).reply_text('Valor no admitido')
self.message(update).reply_text('Este dato es obligatorio. Introduzca un valor valido')
self.progress.input_node = ''
Progress().next_state(self)
Progress().state.draw_ui(bot, update)
return Progress().state
@staticmethod
def skip(bot, update):
logger.info('Skipping state {}'.format(Progress().state))
# If we don't cancel at the end, we should remove any keyboard which could be present
if Progress().input_keywords is None or not Progress().input_keywords.strip():
update.message.reply_text('Este dato es obligatorio. Introduzca un valor valido')
else:
Progress().input_node = ''
Progress().next_state()
update.message.reply_text("Se salta este paso.")
Progress().state.draw_ui(bot, update)
return Progress().state
def validate_input(self, input_data):
data = input_data
if type(input_data) is str:
data = int(data.strip())
response = requests.get('https://www.amazon.es/exec/obidos/tg/browse/-/{}'.format(data))
return response.status_code == 200
def handler_list(self):
return [RegexHandler('^(d+)$', self.handle),
CallbackQueryHandler(self.handle), CommandHandler('skip', self.skip)]
class BlogState(StepState):
def draw_ui(self, bot, update):
reply_keyboard = [
[InlineKeyboardButton(blog, callback_data='{}={}'.format(blog, bid)) for blog, bid in sorted(get_blogs().items())]
]
reply_markup = InlineKeyboardMarkup(reply_keyboard)
self.message(update).reply_text('Choose an option, /skip or /cancel', parse_mode='Markdown',reply_markup=reply_markup)
def handle(self, bot=None, update=None):
parent_result = super().handle(bot, update)
if parent_result is not None:
return parent_result
query = update.callback_query
if query.data:
bname, bid = query.data.split('=')
bot.edit_message_text(text="Chosen option: {}".format(bname), chat_id=query.message.chat_id,
message_id=query.message.message_id)
# We optionally log anything
# We text the user the requirements for next state
# query.message.reply_text('Departamento elegido. ', reply_markup=ReplyKeyboardRemove())
Progress().input_blog = (bname,bid)
if bname == 'xxx':
Progress().input_dep = HEALTH.browse_node_id
Progress().next_state(KeywordsState())
elif bname == 'books':
Progress().input_dep = BOOKS.browse_node_id
Progress().next_state(KeywordsState())
else:
Progress().next_state()
Progress().state.draw_ui(bot, update)
return Progress().state
def handler_list(self):
return [CallbackQueryHandler(self.handle), CommandHandler('skip', self.skip)]
class StartTimeState(StepState):
def draw_ui(self, bot, update):
reply_keyboard = [
[InlineKeyboardButton('Delete', callback_data=DELETE)],
[InlineKeyboardButton(str(number), callback_data=str(number)) for number in range(5)],
[InlineKeyboardButton(str(number), callback_data=str(number)) for number in range(5, 10)],
[InlineKeyboardButton(':', callback_data=':')],
[InlineKeyboardButton('Accept', callback_data=OK),
InlineKeyboardButton('Cancel', callback_data=CANCEL)],
]
reply_markup = InlineKeyboardMarkup(reply_keyboard)
self.message(update).reply_text('Input start time or /cancel',
parse_mode='Markdown', reply_markup=reply_markup)
# We get the reference to the message we will use to update the value and set into Progress
Progress().tracking_message = self.message(update).reply_text("Start time: {}".format(''))
def handle(self, bot=None, update=None):
parent_result = super().handle(bot, update)
if parent_result is not None:
return parent_result
query = update.callback_query
if hasattr(query, 'data'):
if query.data == OK:
bot.delete_message(chat_id=query.message.chat_id, message_id=query.message.message_id)
Progress().next_state()
Progress().state.draw_ui(bot, update)
return Progress().state
else:
if not self.progress.input_start_time:
self.progress.input_start_time = ''
if query.data == DELETE:
if len(self.progress.input_start_time):
self.progress.input_start_time = self.progress.input_start_time[:-1]
else:
self.progress.input_start_time += query.data
# We update the output so the user sees if he types correctly
bot.edit_message_text(text="Start time: {}".format(self.progress.input_start_time),
chat_id=query.message.chat_id,
message_id=Progress().tracking_message.message_id)
else:
if update.message.text and self.validate_input(update.message.text):
self.progress.input_start_time = update.message.text
Progress().next_state()
Progress().state.draw_ui(bot, update)
elif not self.validate_input(update.message.text):
self.message(update).reply_text('Valor no admitido')
self.message(update).reply_text('Este dato es obligatorio. Introduzca un valor valido')
self.progress.input_start_time = ''
Progress().next_state(self)
Progress().state.draw_ui(bot, update)
return Progress().state
def handler_list(self):
return [RegexHandler(TIME_REGEX, self.handle),
CallbackQueryHandler(self.handle), CommandHandler('skip', self.skip)]
# ... more states
# ... more and more states
# ...and so on, you can imagine
class KeywordsState(StepState):
name = 'KeywordsState'
def draw_ui(self, bot, update):
self.message(update).reply_text('Enter keywords', parse_mode='Markdown',
reply_markup=ReplyKeyboardRemove())
def handle(self, bot=None, update=None):
parent_result = super().handle(bot, update)
if parent_result is not None:
return parent_result
keywords = update.message.text.strip()
if keywords:
Progress().input_keywords = keywords
update.message.reply_text('Thanks. The keywords are: {}'.format(keywords))
# When SearchIndex equals All, BrowseNode cannot be present
if Progress().input_dep is None:
Progress().next_state(StartTimeState())
else:
Progress().next_state()
else:
update.message.reply_text('No ha introducido palabras. Introdúzcalas, haga /skip o /cancel')
Progress().next_state(self)
Progress().state.draw_ui(bot, update)
return Progress().state
@staticmethod
def skip(bot, update):
logger.info('Skipping state {}'.format(Progress().state))
# If we don't cancel at the end, we should remove any keyboard which could be present
if Progress().input_dep is None:
update.message.reply_text('Este dato es obligatorio. Introduzca un valor valido')
else:
Progress().input_keywords = None
Progress().next_state()
update.message.reply_text("Skipping this step.")
Progress().state.draw_ui(bot, update)
return Progress().state
def handler_list(self):
return [MessageHandler(Filters.text, self.handle), CommandHandler('skip', self.skip)]
class ConfirmState(StepState):
def draw_ui(self, bot, update):
self.message(update).reply_text('Confirm all this info: {!s}'.format(Progress()))
reply_keyboard = [
[InlineKeyboardButton('Accept', callback_data=OK),
InlineKeyboardButton('Cancel', callback_data=CANCEL)],
]
reply_markup = InlineKeyboardMarkup(reply_keyboard)
self.message(update).reply_text('Accept or cancel (/cancel)', reply_markup=reply_markup)
# This returns a generator with a data list to consume
self.generator = search_asins(Progress().input_dep, Progress().input_node, Progress().input_keywords)
def handle(self, bot=None, update=None):
parent_result = super().handle(bot, update)
if parent_result is not None:
return parent_result
# OK granted, CANCEL is managed in parent
query = update.callback_query
# We finished with the wizard and launch everything
# Launching stuff here...
# ...
# Progress should be reset
self.progress.clear()
Progress().next_state(NoneState())
return ConversationHandler.END
def handler_list(self):
return [CallbackQueryHandler(self.handle)]
# This is intended for setting the steps in an ordered-sorted way
StepState.states = [
NoneState(), BlogState(), DepState(), KeywordsState(), NodeState(), StartTimeState(),
IntervalState(), EndTimeState(), RepeatsState(), ConfirmState()
]
class Progress(object):
"""
This singleton class contains the whole information collected along all the wizard process
"""
__instance = None
def __new__(cls):
if Progress.__instance is None:
Progress.__instance = object.__new__(cls)
Progress.__instance.input_blog = None
Progress.__instance.state = NoneState()
Progress.__instance.tracking_message = None
Progress.__instance.input_dep = None
Progress.__instance.input_node = None
Progress.__instance.input_minutes = '60'
Progress.__instance.input_start_time = None
Progress.__instance.input_end_time = None
Progress.__instance.input_keywords = None
Progress.__instance.input_repeats = None
return Progress.__instance
def clear(self):
"""
Resets the progress
"""
self.input_blog = None
self.state = NoneState()
self.tracking_message = None
self.input_dep = None
self.input_node = None
self.input_minutes = '60'
self.input_start_time = None
self.input_end_time = None
self.input_keywords = None
def next_state(self, new_state=None):
"""
Moves to the next state or to the specified state if any
:param new_state: the next state to move to
:return: the next state, already set to the Progress object
"""
if new_state is not None:
if self.state:
self.state = self.state.next_state(new_state)
else:
self.state = NoneState
logger.info("It was impossible o move to the next state")
else:
self.state = self.state.next_state()
def __str__(self):
return "{}: blog={}, dep={}, node={}, start={}, "
"interval={}', end={}, repeats={}, keywords={}".format(self.__class__.__name__,self.input_blog[0],self.input_dep,self.input_node,self.input_start_time,self.input_minutes,self.input_end_time,self.input_repeats,self.input_keywords)
bot.py:
#!/usr/bin/python3
from telegram.ext import (Updater, Filters, CommandHandler, ConversationHandler, MessageHandler)
from my_package.step_states import Progress, StepState
from environments import get_bot_token, getLogger
# Enable logging
logger = getLogger(__name__)
updater = Updater(get_bot_token())
def start(bot, update):
"""
Sends a message when the command /start is issued.
:param bot: the bot
:param update: the update info from Telegram for this command
"""
update.message.reply_text('Bot started')
def error(bot, update, error):
"""
Logs errors
:param bot: the Telegram bot
:param update: the Telegram update, probably a text message
"""
logger.error('Update "%s" caused error "%s"', update, error)
def restricted(my_handler):
"""
Decorates a handler for restricting its use
:param my_handler: the handler to be restricted
:return: the restricted handler
"""
@wraps(my_handler)
def wrapped(bot, update, *args, **kwargs):
user_id, user_name = update.effective_user.id, update.effective_user.first_name
if user_id not in get_allowed_users():
update.message.reply_text("Unauthorized access {} con id {}.n".format(user_name, user_id))
return
logger.info('Entering {} '.format(my_handler.__name__))
return my_handler(bot, update, *args, **kwargs)
return wrapped
@restricted
def plan(bot, update): #, blog_id=get_blog_id(), interval=60, dep_param_id=None):
"""
Starts the wizard for scheduling item posts
:param bot: the bot
:param update: the update info from Telegram for this command
"""
p = Progress()
p.next_state()
p.state.draw_ui(bot, update)
return p.state
def main():
"""Start the bot."""
# Create the EventHandler and pass it your bot's token.
global updater
# Get the dispatcher to register handlers
dp = updater.dispatcher
# on different commands - answer in Telegram
dp.add_handler(CommandHandler('start', start))
# Add conversation handler with the states
# The telegram conversation handler needs a handler_list with functions
# so it can execute desired code in each state/step
conv_handler = ConversationHandler(
entry_points=[CommandHandler('plan', plan)],
# We enter all the states and all the configured handlers for each state
states={state: state.handler_list() for state in StepState.states},
fallbacks=[CommandHandler('cancel', StepState.cancel)]
)
dp.add_handler(conv_handler)
# log all errors
dp.add_error_handler(error)
# Start the Bot
updater.start_polling()
# Run the bot until you press Ctrl-C or the process receives SIGINT,
# SIGTERM or SIGABRT. This should be used most of the time, since
# start_polling() is non-blocking and will stop the bot gracefully.
updater.idle()
if __name__ == '__main__':
main()
answered Feb 12 at 19:05
madtyn
1415
answered Feb 12 at 19:05
madtyn
1415
answered Feb 12 at 19:05
madtyn
1415
answered Feb 12 at 19:05
madtyn
1415
1415
add a comment |
add a comment |
draft saved
draft discarded
Thanks for contributing an answer to Code Review Stack Exchange!
- Please be sure to answer the question. Provide details and share your research!
But avoid …
- Asking for help, clarification, or responding to other answers.
- Making statements based on opinion; back them up with references or personal experience.
Use MathJax to format equations. MathJax reference.
To learn more, see our tips on writing great answers.
Some of your past answers have not been well-received, and you're in danger of being blocked from answering.
Please pay close attention to the following guidance:
- Please be sure to answer the question. Provide details and share your research!
But avoid …
- Asking for help, clarification, or responding to other answers.
- Making statements based on opinion; back them up with references or personal experience.
To learn more, see our tips on writing great answers.
draft saved
draft discarded
Sign up or log in
StackExchange.ready(function () {
StackExchange.helpers.onClickDraftSave('#login-link');
});
Sign up using Google
Sign up using Facebook
Sign up using Email and Password
Post as a guest
Required, but never shown
StackExchange.ready(
function () {
StackExchange.openid.initPostLogin('.new-post-login', 'https%3a%2f%2fcodereview.stackexchange.com%2fquestions%2f185230%2fstate-and-borg-design-patterns-used-in-a-telegram-wizard-bot%23new-answer', 'question_page');
}
);
Post as a guest
Required, but never shown
Sign up or log in
StackExchange.ready(function () {
StackExchange.helpers.onClickDraftSave('#login-link');
});
Sign up using Google
Sign up using Facebook
Sign up using Email and Password
Post as a guest
Required, but never shown
Sign up or log in
StackExchange.ready(function () {
StackExchange.helpers.onClickDraftSave('#login-link');
});
Sign up using Google
Sign up using Facebook
Sign up using Email and Password
Post as a guest
Required, but never shown
Sign up or log in
StackExchange.ready(function () {
StackExchange.helpers.onClickDraftSave('#login-link');
});
Sign up using Google
Sign up using Facebook
Sign up using Email and Password
Sign up using Google
Sign up using Facebook
Sign up using Email and Password
Post as a guest
Required, but never shown
Required, but never shown
Required, but never shown
Required, but never shown
Required, but never shown
Required, but never shown
Required, but never shown
Required, but never shown
Required, but never shown
Please do not update the code in your question to incorporate feedback from answers, doing so goes against the Question + Answer style of Code Review. This is not a forum where you should keep the most updated version in your question. Please see what you may and may not do after receiving answers.
– Mast
Jan 26 at 13:41
@Mast I was only formatting and changing indentation. Also I deleted some comments. All of it was done to make it easy to read and understand, but I didn't any refactoring, changed any method or added anything.
– madtyn
Jan 27 at 11:35
Feel free to post a follow-up question when you've made significant changes. Leave the code in this question as-is.
– Mast
Jan 27 at 13:50