persistentconversationbot.pyΒΆ

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