Crea e implementa un bot de Telegram con memoria a corto y largo plazo

Resumen

En este artículo, usando Telegram y Python, mostraré cómo construir un Bot amigable con múltiples funciones que puede chatear con conversaciones de respuesta a preguntas (información a corto plazo) y almacenar datos del usuario para recordarlos en el futuro (información a largo plazo).

Ejemplo de chat (Imagen del autor)

Todo esto empezó porque una amiga mía me gritó por no recordar su cumpleaños. No sé si eso te ha pasado alguna vez. Así que pensé que podía fingir que recordaba los cumpleaños mientras en realidad tengo un Bot haciéndolo por mí. Ahora sé lo que estás pensando, ¿por qué crear algo desde cero en lugar de usar una de las millones de aplicaciones de calendario que existen? Y tienes razón, pero para nerds como nosotros … ¿qué hay de divertido en eso?

A través de este tutorial, explicaré paso a paso cómo construir un Telegram Bot inteligente con Python y MongoDB y cómo implementarlo gratis con Heroku y Cron-Job, usando mi Dates Reminder Bot como ejemplo (enlace a continuación).

Presentaré un código Python útil que se puede aplicar fácilmente en otros casos similares (simplemente copie, pegue, ejecute) y recorreré cada línea de código con comentarios para que pueda replicar este ejemplo (enlace al código completo a continuación).

En particular, pasaré por:

• Configuración: descripción general de la arquitectura, nueva generación de Telegram Bot, conexión MongoDB, entorno Python.
• Front-end: codifica los comandos del Bot para la interacción del usuario con pyTelegramBotAPI.
• Back-end: crea la aplicación del lado del servidor con flask y threading.
• Implementar el bot a través de Heroku y Cron-Job

Configuración

Quiero comenzar con la arquitectura general de esta aplicación para que todos tengamos una idea clara de cuál es el destino final de este breve pero divertido viaje. Voy a crear una aplicación Python, implementada en Heroku, que interactúa con los usuarios a través de un Telegram Bot, almacena información en un MongoDB y les envía mensajes recurrentes gracias a un Cron-Job.

Esta arquitectura es bastante estándar para aplicaciones web / móviles. De hecho, estás desarrollando una aplicación adecuada que utiliza Telegram como front-end.

Antes de que empieces a codificar, tienes que generar el Bot con Telegram y crear la instancia de MongoDB.

Telegram te permite generar un nuevo bot en menos de un minuto, simplemente envía un mensaje a BotFather y sigue tus instrucciones, así:

BotFather proporciona un TOKEN de API secreto que usarás más adelante. En cuanto a los posibles comandos de nuestro Bot, estableceré las siguientes opciones:

• función de inicio (esto se ejecuta cuando los usuarios inician el bot)
• guardar y eliminar eventos
• consultar los eventos de hoy y ver todos mis eventos guardados
• función de ayuda

Eso te permitirá visualizar y elegir los comandos en el chat con el Bot cuando esté en funcionamiento:

Pasa a la configuración de MongoDB y regístrate / inicie sesión en MongoDB Atlas, un servicio de base de datos en la nube. Te dan una cuenta gratuita con 500 MB de almacenamiento (selecciona la opción M0 Sandbox al crear el clúster).

En “colecciones”, puedes crear tu base de datos y colección:

En “conectar”, puedes crear tu usuario y establecer tu contraseña, la cual deberás ingresar en la cadena que te dan para conectar tu aplicación a la base de datos. Guarda la cadena completa porque la necesitarás más tarde.

Para codificar la aplicación, necesitas configurar el entorno Python. Instalarás las siguientes bibliotecas a través de la terminal:

pip install pyTelegramBotAPI  pip install pymongo  pip install dnspython  pip install dateparser  pip install flask

Creo que pyTelegramBotAPI es la mejor biblioteca de Python para las API de Telegram, ya que proporciona todas las herramientas para construir fácilmente un chatbot complejo. Se podría trabajar directamente con las API de Telegram, pero no es necesario reinventar la rueda aquí. Los paquetes pymongo y dnspython son necesarios para conectarse a MongoDB con Python. Para que el Bot comprenda las fechas, usaré dateparser.

Si estás leyendo esto, supongo que ya conoces Flask (ten en cuenta que si no planea implementar su bot, no lo vas a necesitar).

Ahora que está todo listo, toca comenzar.

Interfaz (Telegram Bot)

Puedes empezar a codificar el Bot. En primer lugar, llama a la instancia de pyTelegramBotAPI e inserta el TOKEN de la API de Telegram proporcionado por BotFather:

import telebotbot = telebot.TeleBot("insert the Telegram API TOKEN")

En segundo lugar, conéctate a la base de datos usando pymongo y la cadena de MongoDB Atlas:

import pymongoclient = pymongo.MongoClient("mongodb+srv://...")  db_name = "Telegram_DatesReminderBot"  collection_name = "users"  db = client[db_name][collection_name]

Luego, definirás un diccionario vacío que tiene la tarea de almacenar información temporal del usuario (como la identificación del usuario):

dic_user = {}

Ahora, revisarás los comandos principales previamente definidos (iniciar, guardar, …) y escribir las acciones apropiadas para el Bot. Comienza con “/ start”, el comando que se ejecuta automáticamente cuando comienzas a chatear con el Bot por primera vez. Deja que el Bot comience con un mensaje de bienvenida y una sugerencia de qué hacer a continuación:

@bot.message_handler(commands=['start'])  def _start(message):   msg = "Hello "+str(message.chat.username)+\   ", I'm a date reminder. Tell me birthdays and events to remind you. To learn how to use me, use \n/help"   bot.send_message(message.chat.id, msg)

¿Hay detalles que no reconoces? Deja que te ayude:

Fácil ¿verdad? Pero eso fue solo un calentamiento. Ahora, el comando “/ save” será bastante más complicado, ya que implica una conversación de preguntas y respuestas:

@bot.message_handler(commands=['save'])  def _save(message):   msg = "Set an event in the format 'month dd', for example: \n\   xmas day: Dec 25 \n\  I also understand: \n\  today, tomorrow, in 3 days, in 1 week, in 6 months, yesterday, 3 days ago ... so you can do: \n\   meeting: tomorrow"   message = bot.reply_to(message, msg)   bot.register_next_step_handler(message, save_event)

Como puedes ver, esta vez utilicé bot.reply_to () en lugar de bot.send_message (). Mientras que el segundo envía un mensaje general en el chat, el primero envía un mensaje en respuesta a una entrada específica. Lo estoy usando porque esta es una conversación real entre el usuario y el Bot: el usuario pide guardar información, el Bot responde a eso y espera que el usuario responda. El nuevo mensaje del usuario será procesado por bot.register_next_step_handler () como entrada para la función save_event (), que ahora definirás y luego codificarás.

Siguiendo el ejemplo anterior, quiero salvar el cumpleaños de mi madre. Con ese fin, escribo “mamá: 15 de octubre”. El Bot tiene que reconocer el nombre del evento, “mamá” y la fecha “15 de octubre” en el texto. Después de eso, la información debe almacenarse, considerando tanto los casos de usuarios que regresan (ya en la base de datos, por lo que necesitamos actualizar los eventos del usuario) como los nuevos (deben agregarse en la base de datos).

En términos de Python:

import dateparserdef save_event(message):   dic_user["id"] = str(message.chat.id)     ## get text   txt = message.text   name= txt.split(":")[0].strip()    date = txt.split(":")[1].strip()     ## check date   date = dateparser.parse(date).strftime('%b %d')     ## save   lst_users = db.distinct(key="id")   if dic_user["id"] not in lst_users:   db.insert_one({"id":dic_user["id"], "events":{name:date}})   else:   dic_events = db.find_one({"id":dic_user["id"]})["events"]   dic_events.update({name:date})   db.update_one({"id":dic_user["id"]}, {"$set":   {"events":dic_events}})    ## send done   msg = name+": "+date+" saved."   bot.send_message(message.chat.id, msg)

Ten en cuenta que utilicé dateparser para reconocer fechas legibles por humanos. Esto permite al usuario escribir la fecha en diferentes formatos:

De manera similar, puede crear las otras funciones para verificar y eliminar los eventos en la base de datos (proporcionaré el código completo al final del artículo, o puede consultar el enlace de GitHub al comienzo de este tutorial).

¿Qué pasa si el usuario envía algunos textos aleatorios no vinculados a ningún comando específico? El Bot no hará nada si no incluimos este tipo de evento. Por lo tanto, anticiparé 2 tipos de situaciones a las que el Bot responderá adecuadamente (saludos y mensajes de agradecimiento) y todos los demás casos a los que el Bot responderá con un mensaje estándar.

@bot.message_handler(func=lambda m: True)  def chat(message):   txt = message.text   if any(x in txt.lower() for x in ["thank","thx","cool"]):   msg = "anytime"   elif any(x in txt.lower() for x in ["hi","hello","yo","hey"]):   msg = "yo" if str(message.chat.username) == "none" else    "yo "+str(message.chat.username)   else:   msg = "save a date with \n/save"   bot.send_message(message.chat.id, msg)

Manten esta parte bastante básica porque no es el enfoque principal de este tutorial, pero se podrían usar técnicas de PNL más avanzadas para hacer que el chatbot sea más inteligente y conversador.

Dicho esto, nuestro Bot ya tiene la habilidad suficiente para ejecutarlo y probarlo. Puedes simplemente agregar la siguiente línea al final de tu código y ejecutar todo el script.

bot.infinity_polling(True)

Ahora tu Bot está en funcionamiento, alojado en tu computadora portátil. Puedes jugar con él en Telegram. Si no planeas implementar tu bot, puedes detenerte aquí, ya que el resto del artículo explica los pasos para la implementación.

Back-end (aplicación Flask)

Comienza creando un nuevo proyecto de aplicación en Heroku, una plataforma en cloud como servicio que permite implementar una aplicación PoC con solo una cuenta gratuita. Puedes vincular tu proyecto de GitHub e implementar una de las ramas.

Por el momento, aún no estamos listos para implementar la aplicación. Solo necesitamos la dirección web de la aplicación generada por Heroku. Guárdalo; lo necesitarás pronto.

Vuelve a la codificación y comienza agregando una nueva aplicación del lado del servidor con flask al script de Python que has editado hasta ahora:

import flaskserver = flask.Flask(__name__)

Tienes que configurar un webhook, una metodología común en el desarrollo web para pasar información en tiempo real de una aplicación a otra. Gracias al webhook, Telegram Bot envía datos en tiempo real a la aplicación Heroku.

@server.route('/Telegram API TOKEN', methods=['POST'])  def getMessage():   bot.process_new_updates([telebot.types.Update.de_json(  flask.request.stream.read().decode("utf-8"))])   return "!", 200@server.route("/")  def webhook():   bot.remove_webhook()   bot.set_webhook(url='Heroku App Web Address /'   +'Telegram API TOKEN')   return "!", 200

Entonces ahora tu Bot lo tiene casi todo, todavía necesita la última y más importante función: el recordatorio. Chatea y guarda la información, pero el objetivo principal de esta aplicación es enviar alertas automáticamente. Por lo tanto, voy a crear un programador que verifique los eventos de hoy cada vez que se ejecute y envíe recordatorios de eventos.

import datetimedef scheduler():   lst_users = db.distinct(key="id")   for user in lst_users:   dic_events = db.find_one({"id":user})["events"]   today = datetime.datetime.today().strftime('%b %d')   res = [k for k,v in dic_events.items() if v == today]   if len(res) > 0:   msg = "Today's events: "+", ".join(res)   bot.send_message(user, msg)

Cuando se ejecuta la aplicación, este programador se ejecutará en un thread diferente al de la aplicación principal del matraz. En otras palabras, la aplicación tendrá dos cosas sucediendo a la vez: las funcionalidades del chatbot y el recordatorio programado. Puede poner esas 2 acciones principales al final de la secuencia de comandos, y cuando se ejecute el código completo, harán que todo funcione.

import threading  import osif __name__ == "__main__":  threading.Thread(target=scheduler).start()   server.run(host="0.0.0.0",    port=int(os.environ.get("PORT", 5000)))

Desplegar

Finalmente, es el momento de implementar la aplicación. Antes de continuar, mostraré la estructura de mi repositorio:

No necesariamente tienes que mantener la misma estructura de repositorio, pero definitivamente necesitas los 2 archivos que Heroku requiere: los requisitos.txt, que puedes crear desde la terminal

pip freeze > requirements.txt

y el Procfile (sin extensión), que contiene el comando que se ejecutará después del despliegue

web: python telegram_bot.py

Con respecto a las claves de Telegram y MongoDB, debes ponerlas en Heroku como variables de configuración (las llamaré “telegram_key” y “mongodb_key”):

El archivo config.py en la carpeta de configuración tiene el trabajo de recuperar y usar las claves cuando la aplicación se ejecuta en Heroku. Aquí está el código completo en config.py:

import osENV = "PROD"#<--- change here DEV or PROD## keys  if ENV == "DEV":  from settings import keys   telegram_key = keys.telegram_key   mongodb_key = keys.mongodb_keyelif ENV == "PROD":   import ast   telegram_key = ast.literal_eval(os.environ["telegram_key"])   mongodb_key = ast.literal_eval(os.environ["mongodb_key"])## server  host = "0.0.0.0"  port = int(os.environ.get("PORT", 5000))

Y aquí está el código completo en telegram_bot.py, el script principal con el bot y la aplicación en el que has estado trabajando en todo este tutorial:

###############################################################################  #                            RUN MAIN                                         #  ###############################################################################    # setup  ## pkg  import telebot  import logging  import datetime  import dateparser  import pymongo  from settings import config    ## bot  bot = telebot.TeleBot(config.telegram_key)  dic_user = {}    ## setup db  client = pymongo.MongoClient(config.mongodb_key)  db_name = "Telegram_DatesReminderBot"  collection_name = "users"  db = client[db_name][collection_name]    # logging  logging.basicConfig(format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', level=logging.INFO)  logger = logging.getLogger(__name__)        # /start  @bot.message_handler(commands=['start'])  def _start(message):      ## reset      dic_user["id"] = str(message.chat.id)      db.delete_one({'id':dic_user["id"]})      logging.info(str(message.chat.username)+" - "+str(message.chat.id)+" --- START")        ## send first msg      msg = "Hello "+str(message.chat.username)+\            ", I'm a date reminder. Tell me birthdays and events to remind you. To learn how to use me, use \n/help"      bot.send_message(message.chat.id, msg)        # /help  @bot.message_handler(commands=['help'])  def _help(message):      msg = "Set an event, like someone's birthday, for example: \n\              Nadia: Oct 25 \nAnd I'm gonna save the date and every Oct 25 I will remind you of Nadia's birthday. \n\  Save a date with \n/save \n\  You can always check for today's event with \n/check"      bot.send_message(message.chat.id, msg)        # /save  @bot.message_handler(commands=['save'])  def _save(message):      msg = "Set an event in the format 'month dd', for example: \n\              xmas day: Dec 25 \n\  I also understand: \n\  today, tomorrow, in 3 days, in 1 week, in 6 months, yesterday, 3 days ago ... so you can do: \n\              meeting: tomorrow"      message = bot.reply_to(message, msg)      bot.register_next_step_handler(message, save_event)      def save_event(message):      dic_user["id"] = str(message.chat.id)        ## get text      txt = message.text      logging.info(str(message.chat.username)+" - "+str(message.chat.id)+" --- SAVE - "+txt)      name, date = txt.split(":")[0].strip(), txt.split(":")[1].strip()        ## check date      date = dateparser.parse(date).strftime('%b %d')        ## save      lst_users = db.distinct(key="id")      if dic_user["id"] not in lst_users:          db.insert_one({"id":dic_user["id"], "events":{name:date}})      else:          dic_events = db.find_one({"id":dic_user["id"]})["events"]          dic_events.update({name:date})          db.update_one({"id":dic_user["id"]}, {"$set":{"events":dic_events}})            ## send done      msg = name+": "+date+" saved."      bot.send_message(message.chat.id, msg)        # /check  @bot.message_handler(commands=['check'])  def _check(message):      dic_user["id"] = str(message.chat.id)         ## error      lst_users = db.distinct(key="id")      if dic_user["id"] not in lst_users:          msg = "First you need to save an event with \n/save"        ## query      else:          dic_events = db.find_one({"id":dic_user["id"]})["events"]          today = datetime.datetime.today().strftime('%b %d')          logging.info(str(message.chat.username)+" - "+str(message.chat.id)+" --- CHECKING")          res = [k for k,v in dic_events.items() if v == today]          msg = "Today's events: "+", ".join(res) if len(res) > 0 else "No events today"            bot.send_message(message.chat.id, msg)        # /view  @bot.message_handler(commands=['view'])  def _view(message):      dic_user["id"] = str(message.chat.id)         ## error      lst_users = db.distinct(key="id")      if dic_user["id"] not in lst_users:          msg = "You have no events. Save an event with \n/save"        ## query      else:          dic_events = db.find_one({"id":dic_user["id"]})["events"]          logging.info(str(message.chat.username)+" - "+str(message.chat.id)+" --- VIEW ALL")          msg = "\n".join(k+": "+v for k,v in dic_events.items())            bot.send_message(message.chat.id, msg)        # /delete  @bot.message_handler(commands=['delete'])  def _delete(message):      ## ask name      msg = "Tell me the event to Delete, for example: \n\              xmas day \nAnd I'm gonna stop the reminder."      message = bot.reply_to(message, msg)      bot.register_next_step_handler(message, delete_event)      def delete_event(message):      dic_user["id"] = str(message.chat.id)        ## get text      txt = message.text      logging.info(str(message.chat.username)+" - "+str(message.chat.id)+" --- DELETE - "+txt)        ## delete      dic_events = db.find_one({"id":dic_user["id"]})["events"]      dic_events.pop(txt)      db.update_one({"id":dic_user["id"]}, {"$set":{"events":dic_events}})            ## send done      msg = txt+" deleted."      bot.send_message(message.chat.id, msg)        # non-command message  @bot.message_handler(func=lambda m: True)  def chat(message):      txt = message.text      if any(x in txt.lower() for x in ["thank","thx","cool"]):          msg = "anytime"      elif any(x in txt.lower() for x in ["hi","hello","yo"]):          msg = "yo" if str(message.chat.username) is "none" else "yo "+str(message.chat.username)      else:          msg = "save a date with \n/save"      bot.send_message(message.chat.id, msg)        # scheduler  def scheduler():      lst_users = db.distinct(key="id")      logging.info("--- SCHEDULER for "+str(len(lst_users))+" users ---")      for user in lst_users:          #res = requests.get("https://api.telegram.org/bot1494658770:"+config.telegram_key+"/sendMessage?chat_id="+user+"&text=yo")          dic_events = db.find_one({"id":user})["events"]          today = datetime.datetime.today().strftime('%b %d')          res = [k for k,v in dic_events.items() if v == today]          if len(res) > 0:              msg = "Today's events: "+", ".join(res)              bot.send_message(user, msg)        # run  if config.ENV == "DEV":      bot.infinity_polling(True)  #bot.polling()      elif config.ENV == "PROD":      import flask      import threading        server = flask.Flask(__name__)        @server.route('/'+config.telegram_key, methods=['POST'])      def getMessage():          bot.process_new_updates([telebot.types.Update.de_json(flask.request.stream.read().decode("utf-8"))])          return "!", 200        @server.route("/")      def webhook():          bot.remove_webhook()          bot.set_webhook(url='https://botdatereminder.herokuapp.com/'+config.telegram_key)          return 'Chat with the Bot  <a href ="https://t.me/DatesReminderBot">here</a> \            or   Check the project code <a href ="https://github.com/mdipietro09/Bot_TelegramDatesReminder">here</a>', 200        if __name__ == "__main__":          threading.Thread(target=scheduler).start()          server.run(host=config.host, port=config.port)

Genial, ahora lo despliegas en Heroku:

Ahora deberías tener la aplicación web Heroku en funcionamiento y poder usar el Bot en Telegram. ¡Pruébalo!

Por último, pero no menos importante, establecerás un Cron-Job para hacer ping a la dirección web de Heroku. Eso hará que la página se vuelva a cargar, la aplicación se reinicie y el recordatorio programado se ejecutará en su thread.

Así que mañana, cuando nos despertemos, encontraremos un mensaje con los eventos del día.

Ejemplo de chat (Imagen del autor)

Conclusión

Este artículo ha sido un tutorial para mostrar cómo construir un bot de Telegram con Python que almacena datos de usuario en MongoDB y recupera información a través de Cron-Job. Ahora que sabe cómo funciona, puede desarrollar su propio chatbot y jugar con él.

¡Espero que lo hayan disfrutado! No dudéis en ponerse en contacto conmigo en LinkedIn y Twitter si tenéis preguntas y comentarios o simplemente para compartir los proyectos interesantes.

Many thanks to Planeta Chatbot for translating and publishing this article. You can find the original piece (English) here: