Añadido helpers para funciones extra. Añadida función para escapar carácteres para la API de Telegram

wallamanta/alert.py

@@ -3,6 +3,7 @@ import os
 import threading
 import logging
 import prettytable
+import helpers
 
 from worker import Worker
 from telegram import Update
@@ -37,13 +38,13 @@ Ejemplo 2: `/add cpu;10;30;;intel,core 2 duo,celeron;;;100`\n
 Los campos opcionales que se dejen vacíos tomarán el valor configurado en el archivo `\.env`\n
 Lista los productos con `/list` o obtén la información de uno en concreto con `/list nombre del producto`\n
 Borra un producto con `/remove nombre del producto`"""
-    await update.message.reply_markdown_v2(message)
+    await update.message.reply_markdown_v2(helpers.telegram_escape_characters(message))
 
 async def add_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
     message = """Tienes que pasar el número correcto de parámetros: `/add producto;precio_mínimo;precio_máximo,excluir_título(opcional, separado por comas);excluir_descripción_y_título(opciona, separado por comas);latitud(opcional);longitud(opcional),distancia(opcional)`\n
 Ejemplo: `/add placa base itx;0;150`\n
 Ejemplo 2: `/add cpu;10;30;;intel,core 2 duo,celeron;;;100`\n
-Los campos opcionales que se dejen vacíos tomarán el valor configurado en el archivo `\.env`"""
+Los campos opcionales que se dejen vacíos tomarán el valor configurado en el archivo `.env`"""
 
     args = update.message.text.split("/add ")
     if len(args) == 1:
@@ -87,8 +88,8 @@ Los campos opcionales que se dejen vacíos tomarán el valor configurado en el a
             p.start()
             message = f"Añadido {product_name} a seguimiento"
         else:
-            message = f"{product_name} ya está en seguimiento\!"
-    await update.message.reply_markdown_v2(message)
+            message = f"{product_name} ya está en seguimiento!"
+    await update.message.reply_markdown_v2(helpers.telegram_escape_characters(message))
 
 async def remove_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
     product_to_remove = update.message.text[len('/remove '):]
@@ -123,7 +124,7 @@ async def list_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> No
         table.align['Máx'] = 'r'
         for product in products:
             table.add_row([product['product_name'], f"{product['min_price']}€", f"{product['max_price']}€"])
-    await update.message.reply_markdown_v2(f'```{table}```')
+    await update.message.reply_markdown_v2(f'```{helpers.telegram_escape_characters(table)}```')
 
 def main()->None:
     products = parse_json_file()

wallamanta/helpers.py

@@ -0,0 +1,6 @@
+TELEGRAM_ESCAPE_CHARACTERS = ['_', '*', '[', ']', '(', ')', '~', '`', '>', '#', '+', '-', '=', '|', '{', '}', '.', '!']
+
+def telegram_escape_characters(text):
+    for character in TELEGRAM_ESCAPE_CHARACTERS:
+        text = text.replace(character, f'\{character}')
+    return text

wallamanta/worker.py

@@ -4,6 +4,7 @@ import telegram
 import os
 import logging
 import json
+import helpers
 
 TELEGRAM_CHANNEL_ID = os.getenv("TELEGRAM_CHANNEL_ID") 
 TELEGRAM_TOKEN = os.getenv("TELEGRAM_TOKEN")
@@ -11,6 +12,7 @@ LATITUDE = os.getenv("LATITUDE")
 LONGITUDE = os.getenv("LONGITUDE")
 SLEEP_TIME = int(os.getenv("SLEEP_TIME"))
 
+
 # Enable logging
 logging.basicConfig(
     format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', level=logging.INFO
@@ -75,11 +77,11 @@ class Worker:
                         if not self.has_excluded_words(article['title'].lower(), article['description'].lower(), product['exclude']) and not self.is_title_key_word_excluded(article['title'].lower(), product['title_keyword_exclude']):
                             try:
                                 text =  f"*Artículo*: {article['title']}\n*Descripción*: {article['description']}\n*Precio*: {article['price']} {article['currency']}\n[Ir al anuncio](https://es.wallapop.com/item/{article['web_slug']})".replace(".", "\.")
-                                url = f"https://api.telegram.org/bot{TELEGRAM_TOKEN}/sendMessage?chat_id={TELEGRAM_CHANNEL_ID}&text={text}&parse_mode=MarkdownV2"
+                                url = f"https://api.telegram.org/bot{TELEGRAM_TOKEN}/sendMessage?chat_id={TELEGRAM_CHANNEL_ID}&text={helpers.telegram_escape_characters(text)}&parse_mode=MarkdownV2"
                                 logging.info(requests.get(url).json())
                             except:
                                 text =  f"*Artículo*: {article['title']}\n*Descripción*: {article['description']}\n*Precio*: {article['price']} {article['currency']}\n[Ir al anuncio](https://es.wallapop.com/item/{article['web_slug']})".replace(".", "\.")
-                                url = f"https://api.telegram.org/bot{TELEGRAM_TOKEN}/sendMessage?chat_id={TELEGRAM_CHANNEL_ID}&text={text}&parse_mode=MarkdownV2"
+                                url = f"https://api.telegram.org/bot{TELEGRAM_TOKEN}/sendMessage?chat_id={TELEGRAM_CHANNEL_ID}&text={helpers.telegram_escape_characters(text)text}&parse_mode=MarkdownV2"
                                 requests.get(url)
                             time.sleep(1) # Avoid Telegram flood restriction
                         list.insert(0, article['id'])