Hide navigation sidebar
Hide table of contents sidebar
PTB has undergone significant changes in v20. Please read the documentation carefully and also check out the transition guide in the wiki.
Back to top
Toggle table of contents sidebar

# `persistentconversationbot.py`

      1#!/usr/bin/env python
      2# pylint: disable=unused-argument
      3# This program is dedicated to the public domain under the CC0 license.
      4
      5"""
      6First, a few callback functions are defined. Then, those functions are passed to
      7the Application and registered at their respective places.
      8Then, the bot is started and runs until we press Ctrl-C on the command line.
      9
     10Usage:
     11Example of a bot-user conversation using ConversationHandler.
     12Send /start to initiate the conversation.
     13Press Ctrl-C on the command line or send a signal to the process to stop the
     14bot.
     15"""
     16
     17import logging
     18from typing import Dict
     19
     20from telegram import ReplyKeyboardMarkup, ReplyKeyboardRemove, Update
     21from telegram.ext import (
     22    Application,
     23    CommandHandler,
     24    ContextTypes,
     25    ConversationHandler,
     26    MessageHandler,
     27    PicklePersistence,
     28    filters,
     29)
     30
     31# Enable logging
     32logging.basicConfig(
     33    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", level=logging.INFO
     34)
     35# set higher logging level for httpx to avoid all GET and POST requests being logged
     36logging.getLogger("httpx").setLevel(logging.WARNING)
     37
     38logger = logging.getLogger(__name__)
     39
     40CHOOSING, TYPING_REPLY, TYPING_CHOICE = range(3)
     41
     42reply_keyboard = [
     43    ["Age", "Favourite colour"],
     44    ["Number of siblings", "Something else..."],
     45    ["Done"],
     46]
     47markup = ReplyKeyboardMarkup(reply_keyboard, one_time_keyboard=True)
     48
     49
     50def facts_to_str(user_data: Dict[str, str]) -> str:
     51    """Helper function for formatting the gathered user info."""
     52    facts = [f"{key} - {value}" for key, value in user_data.items()]
     53    return "\n".join(facts).join(["\n", "\n"])
     54
     55
     56async def start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int:
     57    """Start the conversation, display any stored data and ask user for input."""
     58    reply_text = "Hi! My name is Doctor Botter."
     59    if context.user_data:
     60        reply_text += (
     61            f" You already told me your {', '.join(context.user_data.keys())}. Why don't you "
     62            "tell me something more about yourself? Or change anything I already know."
     63        )
     64    else:
     65        reply_text += (
     66            " I will hold a more complex conversation with you. Why don't you tell me "
     67            "something about yourself?"
     68        )
     69    await update.message.reply_text(reply_text, reply_markup=markup)
     70
     71    return CHOOSING
     72
     73
     74async def regular_choice(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int:
     75    """Ask the user for info about the selected predefined choice."""
     76    text = update.message.text.lower()
     77    context.user_data["choice"] = text
     78    if context.user_data.get(text):
     79        reply_text = (
     80            f"Your {text}? I already know the following about that: {context.user_data[text]}"
     81        )
     82    else:
     83        reply_text = f"Your {text}? Yes, I would love to hear about that!"
     84    await update.message.reply_text(reply_text)
     85
     86    return TYPING_REPLY
     87
     88
     89async def custom_choice(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int:
     90    """Ask the user for a description of a custom category."""
     91    await update.message.reply_text(
     92        'Alright, please send me the category first, for example "Most impressive skill"'
     93    )
     94
     95    return TYPING_CHOICE
     96
     97
     98async def received_information(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int:
     99    """Store info provided by user and ask for the next category."""
    100    text = update.message.text
    101    category = context.user_data["choice"]
    102    context.user_data[category] = text.lower()
    103    del context.user_data["choice"]
    104
    105    await update.message.reply_text(
    106        "Neat! Just so you know, this is what you already told me:"
    107        f"{facts_to_str(context.user_data)}"
    108        "You can tell me more, or change your opinion on something.",
    109        reply_markup=markup,
    110    )
    111
    112    return CHOOSING
    113
    114
    115async def show_data(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    116    """Display the gathered info."""
    117    await update.message.reply_text(
    118        f"This is what you already told me: {facts_to_str(context.user_data)}"
    119    )
    120
    121
    122async def done(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int:
    123    """Display the gathered info and end the conversation."""
    124    if "choice" in context.user_data:
    125        del context.user_data["choice"]
    126
    127    await update.message.reply_text(
    128        f"I learned these facts about you: {facts_to_str(context.user_data)}Until next time!",
    129        reply_markup=ReplyKeyboardRemove(),
    130    )
    131    return ConversationHandler.END
    132
    133
    134def main() -> None:
    135    """Run the bot."""
    136    # Create the Application and pass it your bot's token.
    137    persistence = PicklePersistence(filepath="conversationbot")
    138    application = Application.builder().token("TOKEN").persistence(persistence).build()
    139
    140    # Add conversation handler with the states CHOOSING, TYPING_CHOICE and TYPING_REPLY
    141    conv_handler = ConversationHandler(
    142        entry_points=[CommandHandler("start", start)],
    143        states={
    144            CHOOSING: [
    145                MessageHandler(
    146                    filters.Regex("^(Age|Favourite colour|Number of siblings)$"), regular_choice
    147                ),
    148                MessageHandler(filters.Regex("^Something else...$"), custom_choice),
    149            ],
    150            TYPING_CHOICE: [
    151                MessageHandler(
    152                    filters.TEXT & ~(filters.COMMAND | filters.Regex("^Done$")), regular_choice
    153                )
    154            ],
    155            TYPING_REPLY: [
    156                MessageHandler(
    157                    filters.TEXT & ~(filters.COMMAND | filters.Regex("^Done$")),
    158                    received_information,
    159                )
    160            ],
    161        },
    162        fallbacks=[MessageHandler(filters.Regex("^Done$"), done)],
    163        name="my_conversation",
    164        persistent=True,
    165    )
    166
    167    application.add_handler(conv_handler)
    168
    169    show_data_handler = CommandHandler("show_data", show_data)
    170    application.add_handler(show_data_handler)
    171
    172    # Run the bot until the user presses Ctrl-C
    173    application.run_polling(allowed_updates=Update.ALL_TYPES)
    174
    175
    176if __name__ == "__main__":
    177    main()