Skip to content
Back to top
View this page

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