129 lines
3.8 KiB
Python
129 lines
3.8 KiB
Python
from statemachine import StateMachine, State
|
|
from pytgbot.api_types.sendable.reply_markup import InlineKeyboardButton,InlineKeyboardMarkup
|
|
|
|
import logging
|
|
logger=logging.getLogger()
|
|
|
|
class TelegramStateMachine(StateMachine):
|
|
def __init__(self,chat_id=None, bot=None,*args,**kwargs):
|
|
self.chat_id = chat_id
|
|
self.user_id = user_id
|
|
# self.context=context
|
|
self.bot = bot
|
|
super().__init__(*args,**kwargs)
|
|
|
|
def send(self,text:str):
|
|
self.bot.send_message(chat_id=self.chat_id, text=text)
|
|
|
|
def send_confirm(self, text:str):
|
|
keyboard = [[InlineKeyboardButton("OK", callback_data='ok'),
|
|
InlineKeyboardButton("Repeat", callback_data='retry')],
|
|
[ InlineKeyboardButton("Cancel", callback_data='cancel')] ]
|
|
#,
|
|
# InlineKeyboardButton("Cancel", callback_data='3')]
|
|
#keyboard = [[InlineKeyboardButton("OK", callback_data='1'),
|
|
#]]
|
|
|
|
self.bot.send_message(chat_id=self.chat_id,
|
|
text=text,
|
|
reply_markup=InlineKeyboardMarkup(keyboard))
|
|
|
|
|
|
|
|
class CreatePostWorkflow(TelegramStateMachine):
|
|
init =State('init', initial=True)
|
|
|
|
wait = State('wait',value=0)
|
|
confirm =State('confirm')
|
|
finished = State('finished')
|
|
|
|
initialize = init.to(wait)
|
|
entered = wait.to(confirm)
|
|
next=confirm.to(wait)
|
|
retry=confirm.to(wait)
|
|
|
|
steps={
|
|
0:'title',
|
|
1:'text'
|
|
}
|
|
|
|
finish =confirm.to(finished)
|
|
cancel = finished.from_(wait,confirm)
|
|
@property
|
|
def value(self):
|
|
return self.p[self.steps[self.step]]
|
|
|
|
def __init__(self,chat_id=None, context=None,*args, **kwargs):
|
|
super().__init__(chat_id, context, *args, **kwargs)
|
|
self.p=dict()
|
|
self.step=0
|
|
self.initialize()
|
|
|
|
|
|
def confirmed(self):
|
|
if self.step >= len(self.steps)-1:
|
|
self.finish()
|
|
else:
|
|
self.next()
|
|
|
|
# processing messages
|
|
def process_message(self, text, is_callback=False):
|
|
if self.current_state==self.wait:
|
|
self.p[self.steps[self.step]]=text
|
|
self.send(str(self .p))
|
|
self.entered()
|
|
elif self.current_state==self.confirm:
|
|
if text=="ok":
|
|
self.confirmed()
|
|
elif query.data=='cancel':
|
|
self.cancel()
|
|
else:
|
|
self.retry()
|
|
|
|
def button_handler(self, update, context):
|
|
query = update.callback_query
|
|
if self.current_state ==self.confirm:
|
|
if query.data=='ok':
|
|
self.confirmed()
|
|
elif query.data=='cancel':
|
|
self.cancel()
|
|
else:
|
|
self.retry()
|
|
query.edit_message_text(text="Selected option: {}".format(query.data))
|
|
|
|
def command_handler(self,update,context):
|
|
logger.info("Processing Command: %s" % update.message.text)
|
|
|
|
def message_handler(self,update,context):
|
|
if self.current_state==self.wait:
|
|
self.p[self.steps[self.step]]=update.message.text
|
|
self.send(str(self.p))
|
|
self.entered()
|
|
elif self.current_state==self.confirm:
|
|
if update.message.text=="ok":
|
|
self.confirmed()
|
|
else:
|
|
self.retry()
|
|
#file = bot.getFile(update.message.photo[-1].file_id
|
|
|
|
|
|
def on_next(self):
|
|
self.step+=1
|
|
|
|
def on_cancel(self):
|
|
self.p={}
|
|
self.step=0
|
|
self.send("Canceled")
|
|
|
|
def on_finish(self):
|
|
self.step=0
|
|
self.send("Eingabe fertig:%s" % str(self.p))
|
|
|
|
def on_enter_confirm(self):
|
|
self.send_confirm("Bitte die Eingabe von %s bestaetigen" % self.value)
|
|
|
|
def on_enter_wait(self):
|
|
self.send("Bitte folgendes das Attribut %s eingeben " % self.steps[self.step])
|
|
|
|
|