Crear un Bot con Telegram

Crear un Bot con Telegram es demasiado facil, lo dificil es que sea un bot funcional. Para que me puedas entender mejor, si usas telegram buscar al usuario @botfather que se ve como en la siguiente imagen:

Luego vas a chatear con el indicandole /new y comenzará a preguntarte nombre para el bot y luego el usuario:

Como se puede apreciar en la imagen anterior elegi como nombre del bot webzeta, y como usuario webzeta_bot y ahi mismo te vuelve un DONE! lo crea de toque y te despues tambien fijate que dice: Use this token to access the http api………. lo que sigue es el token que te da acceso a las funcionalidades.

Mi bot lo tengo configurado para que responda solo a una serie de usuarios conocidos, y esto se debe a que esta funcionando en una raspberry y no quiero que nadie que no sea yo o mi familia vea una foto de mi casa (por ejemplo).

Este bot lo voy a configurar en mi #raspberry pi3b+ a la cual tengo una webcam conectada. Les cuento mi paso a paso:

sudo apt update && sudo apt upgrade -y
sudo apt install python3 python3-pip python3-venv ffmpeg fswebcam git -y
mkdir bot
cd bot
python3 -m venv bot
source bot/bin/activate

Hasta aca instalamos paquetes que vamos a necesitar y armamos un entorno virtual para el bot y para finalizar lo activamos. Ahora tenemos que instalar las cosas que va a utilizar el bot, pero de aca en más todo lo que sea una instalación de pip quedará dentro del entorno virtual.

pip install python-telegram-bot requests ansible beautifulsoup4

Otra pausa aca: Ansible no hace falta que lo instalen, yo lo uso mucho, pero no hace falta para las cosas sencillas.

Bien ahora como les dije yo tengo una webcam conectada a la raspberry y si quiero una foto le mando el comando:

fswebcam --no-banner /tmp/test.jpg && ls -lh /tmp/test.jpg

Si esto funciona es que esta andando correctamente!

Ahora si creamos un archivo bot.py y dentro:

nano bot.py

import os
import subprocess
import requests
from telegram import Update
from telegram.ext import ApplicationBuilder, CommandHandler, ContextTypes

# CONFIGURACIÓN (REMPLAZAR CON TUS DATOS)
BOT_TOKEN = "TU_TOKEN_AQUÍ"  # Reemplaza con tu token de BotFather
USER_ID_PERMITIDO = 123456789  # Reemplaza con tu ID de Telegram (obténlo con @userinfobot)

# --- FUNCIONES PRINCIPALES --- #
def get_dolar_rates():
    try:
        response = requests.get("https://api.bluelytics.com.ar/v2/latest", timeout=10)
        data = response.json()
        return {
            'oficial_compra': data['oficial']['value_buy'],
            'oficial_venta': data['oficial']['value_sell'],
            'blue_compra': data['blue']['value_buy'],
            'blue_venta': data['blue']['value_sell']
        }
    except Exception as e:
        raise ValueError(f"Error API: {str(e)}")

def get_weather(ciudad):
    try:
        response = requests.get(f"https://wttr.in/{ciudad}?format=3", timeout=5)
        return response.text if response.status_code == 200 else "Error al obtener clima"
    except:
        return "Servicio de clima no disponible"

# --- COMANDOS DEL BOT --- #
async def restricted_access(update: Update):
    if update.effective_user.id != USER_ID_PERMITIDO:
        await update.message.reply_text("Comando restringido")
        return False
    return True

async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
    if not await restricted_access(update):
        return
    await update.message.reply_text(
        "Bot Funcional\n"
        "/dolar - Cotización del dólar\n"
        "/clima <ciudad> - Estado del tiempo\n"
        "/foto - Tomar una foto"
    )

async def dolar(update: Update, context: ContextTypes.DEFAULT_TYPE):
    if not await restricted_access(update):
        return
    try:
        rates = get_dolar_rates()
        await update.message.reply_text(
            "Dólar (API Bluelytics):\n\n"
            f"Oficial: ${rates['oficial_compra']:.2f} / ${rates['oficial_venta']:.2f}\n"
            f"Blue: ${rates['blue_compra']:.2f} / ${rates['blue_venta']:.2f}\n"
            f"Spread: ${rates['oficial_venta']-rates['oficial_compra']:.2f} (Oficial)"
        )
    except Exception as e:
        await update.message.reply_text(f"Error: {str(e)}")

async def clima(update: Update, context: ContextTypes.DEFAULT_TYPE):
    if not await restricted_access(update):
        return
    if not context.args:
        await update.message.reply_text("Uso: /clima <ciudad>")
        return
    ciudad = " ".join(context.args)
    await update.message.reply_text(get_weather(ciudad))

async def foto(update: Update, context: ContextTypes.DEFAULT_TYPE):
    if not await restricted_access(update):
        return
    
    filename = f"/tmp/foto_{update.message.chat_id}.jpg"
    
    try:
        # Captura de foto con parámetros optimizados
        subprocess.run([
            'fswebcam',
            '-d', '/dev/video0',
            '-r', '1280x720',
            '--no-banner',
            '--jpeg', '85',
            '--skip', '3',
            filename
        ], check=True, timeout=10)
        
        if os.path.getsize(filename) > 1024:  # Verifica que la foto no esté vacía
            with open(filename, 'rb') as photo:
                await update.message.reply_photo(photo)
        else:
            await update.message.reply_text("Foto vacía o corrupta")
        
    except subprocess.TimeoutExpired:
        await update.message.reply_text("Tiempo de espera agotado")
    except subprocess.CalledProcessError:
        await update.message.reply_text(
            "Error técnico con la cámara:\n"
            "1. Verifica que esté conectada\n"
            "2. Prueba ejecutar manualmente:\n"
            "   fswebcam --no-banner /tmp/test.jpg"
        )
    except Exception as e:
        await update.message.reply_text(f"Error inesperado: {str(e)}")
    finally:
        if os.path.exists(filename):
            os.remove(filename)

# --- INICIO DEL BOT --- #
if __name__ == "__main__":
    print("Iniciando bot...")
    app = ApplicationBuilder().token(BOT_TOKEN).build()
    
    app.add_handler(CommandHandler("start", start))
    app.add_handler(CommandHandler("dolar", dolar))
    app.add_handler(CommandHandler("clima", clima))
    app.add_handler(CommandHandler("foto", foto))
    
    print("Bot listo. Presiona Ctrl+C para detenerlo")
    app.run_polling()

Luego guardamos y le damos permisos:

chmod +x bot.py

Variables que tenes que poner las tuyas:

TOKEN: Lo sacas del bot que creaste

USER_ID_PERMITIDO: Podes chatear con @userinfobot que al hacerlo te dará tu user_id

Luego lo ejecutamos con:

python3 bot.py

Ahora es cuestion de que vos chatees con tu bot preguntandole:

/clima ciudad

/dolar

/foto

Te va a responder usando las apis gratuitas que consumimos. Asi podes programar muchas acciones, en mi caso por ejemplo uso un comando para que ansible me diga si todos los dispositivos que quiero estan conectados, uso de recursos, memoria etc, pero eso ya lo podes ver en el blog buscando las entradas de ansible

Otra cosa importante, cuando se reinicia la maquina o la raspberry en este caso no va a estar funcionando, y hay varias formas de hacerlo para que si arranque al inicio. Una que me gusta es crear un servicio:

sudo nano /etc/systemd/system/telegram_bot.service

Adentro del archivo:

[Unit]
Description=Telegram Bot Service
After=network.target

[Service]
User=luiz  # Tu usuario
WorkingDirectory=/home/luiz/bot  # Ruta donde está tu bot.py
Environment="PATH=/home/luiz/bot/bin:/usr/bin:/bin"  # Ruta al venv
ExecStart=/home/luiz/bot/bin/python /home/luiz/bot/bot.py
Restart=always
RestartSec=5s

[Install]
WantedBy=multi-user.target

Luego lo guardas y ejecutas estos comandos:

sudo systemctl daemon-reload
sudo systemctl enable telegram_bot.service  # Para que inicie automáticamente
sudo systemctl start telegram_bot.service   # Iniciar ahora

y luego reinicia y ve si te funciono!

De lo contrario s epodria hacer un script que active segun tu version el source del venv y ejecute el comando python3 bot.py