mirror of
https://github.com/python-telegram-bot/python-telegram-bot.git
synced 2024-11-29 03:32:49 +01:00
f77f4b0cf7
Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com> Co-authored-by: Harshil <37377066+harshil21@users.noreply.github.com>
154 lines
4.9 KiB
Python
154 lines
4.9 KiB
Python
#!/usr/bin/env python
|
|
# pylint: disable=unused-argument
|
|
# This program is dedicated to the public domain under the CC0 license.
|
|
|
|
"""
|
|
First, a few callback functions are defined. Then, those functions are passed to
|
|
the Application and registered at their respective places.
|
|
Then, the bot is started and runs until we press Ctrl-C on the command line.
|
|
|
|
Usage:
|
|
Example of a bot-user conversation using ConversationHandler.
|
|
Send /start to initiate the conversation.
|
|
Press Ctrl-C on the command line or send a signal to the process to stop the
|
|
bot.
|
|
"""
|
|
|
|
import logging
|
|
from typing import Dict
|
|
|
|
from telegram import ReplyKeyboardMarkup, ReplyKeyboardRemove, Update
|
|
from telegram.ext import (
|
|
Application,
|
|
CommandHandler,
|
|
ContextTypes,
|
|
ConversationHandler,
|
|
MessageHandler,
|
|
filters,
|
|
)
|
|
|
|
# Enable logging
|
|
logging.basicConfig(
|
|
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", level=logging.INFO
|
|
)
|
|
# set higher logging level for httpx to avoid all GET and POST requests being logged
|
|
logging.getLogger("httpx").setLevel(logging.WARNING)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
CHOOSING, TYPING_REPLY, TYPING_CHOICE = range(3)
|
|
|
|
reply_keyboard = [
|
|
["Age", "Favourite colour"],
|
|
["Number of siblings", "Something else..."],
|
|
["Done"],
|
|
]
|
|
markup = ReplyKeyboardMarkup(reply_keyboard, one_time_keyboard=True)
|
|
|
|
|
|
def facts_to_str(user_data: Dict[str, str]) -> str:
|
|
"""Helper function for formatting the gathered user info."""
|
|
facts = [f"{key} - {value}" for key, value in user_data.items()]
|
|
return "\n".join(facts).join(["\n", "\n"])
|
|
|
|
|
|
async def start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int:
|
|
"""Start the conversation and ask user for input."""
|
|
await update.message.reply_text(
|
|
"Hi! My name is Doctor Botter. I will hold a more complex conversation with you. "
|
|
"Why don't you tell me something about yourself?",
|
|
reply_markup=markup,
|
|
)
|
|
|
|
return CHOOSING
|
|
|
|
|
|
async def regular_choice(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int:
|
|
"""Ask the user for info about the selected predefined choice."""
|
|
text = update.message.text
|
|
context.user_data["choice"] = text
|
|
await update.message.reply_text(f"Your {text.lower()}? Yes, I would love to hear about that!")
|
|
|
|
return TYPING_REPLY
|
|
|
|
|
|
async def custom_choice(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int:
|
|
"""Ask the user for a description of a custom category."""
|
|
await update.message.reply_text(
|
|
'Alright, please send me the category first, for example "Most impressive skill"'
|
|
)
|
|
|
|
return TYPING_CHOICE
|
|
|
|
|
|
async def received_information(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int:
|
|
"""Store info provided by user and ask for the next category."""
|
|
user_data = context.user_data
|
|
text = update.message.text
|
|
category = user_data["choice"]
|
|
user_data[category] = text
|
|
del user_data["choice"]
|
|
|
|
await update.message.reply_text(
|
|
"Neat! Just so you know, this is what you already told me:"
|
|
f"{facts_to_str(user_data)}You can tell me more, or change your opinion"
|
|
" on something.",
|
|
reply_markup=markup,
|
|
)
|
|
|
|
return CHOOSING
|
|
|
|
|
|
async def done(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int:
|
|
"""Display the gathered info and end the conversation."""
|
|
user_data = context.user_data
|
|
if "choice" in user_data:
|
|
del user_data["choice"]
|
|
|
|
await update.message.reply_text(
|
|
f"I learned these facts about you: {facts_to_str(user_data)}Until next time!",
|
|
reply_markup=ReplyKeyboardRemove(),
|
|
)
|
|
|
|
user_data.clear()
|
|
return ConversationHandler.END
|
|
|
|
|
|
def main() -> None:
|
|
"""Run the bot."""
|
|
# Create the Application and pass it your bot's token.
|
|
application = Application.builder().token("TOKEN").build()
|
|
|
|
# Add conversation handler with the states CHOOSING, TYPING_CHOICE and TYPING_REPLY
|
|
conv_handler = ConversationHandler(
|
|
entry_points=[CommandHandler("start", start)],
|
|
states={
|
|
CHOOSING: [
|
|
MessageHandler(
|
|
filters.Regex("^(Age|Favourite colour|Number of siblings)$"), regular_choice
|
|
),
|
|
MessageHandler(filters.Regex("^Something else...$"), custom_choice),
|
|
],
|
|
TYPING_CHOICE: [
|
|
MessageHandler(
|
|
filters.TEXT & ~(filters.COMMAND | filters.Regex("^Done$")), regular_choice
|
|
)
|
|
],
|
|
TYPING_REPLY: [
|
|
MessageHandler(
|
|
filters.TEXT & ~(filters.COMMAND | filters.Regex("^Done$")),
|
|
received_information,
|
|
)
|
|
],
|
|
},
|
|
fallbacks=[MessageHandler(filters.Regex("^Done$"), done)],
|
|
)
|
|
|
|
application.add_handler(conv_handler)
|
|
|
|
# Run the bot until the user presses Ctrl-C
|
|
application.run_polling(allowed_updates=Update.ALL_TYPES)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|