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()