Cap 7: Comandos Avanzados

Por: Artiko
telegraminline-keyboardvisiongruposarchivosclaude

Teclados inline

Los teclados inline aparecen debajo del mensaje y ejecutan acciones al presionar:

from telegram import InlineKeyboardButton, InlineKeyboardMarkup

async def cmd_menu(update: Update, context: ContextTypes.DEFAULT_TYPE):
    teclado = InlineKeyboardMarkup([
        [
            InlineKeyboardButton("📖 Explicar código", callback_data="mode:explain"),
            InlineKeyboardButton("🔍 Revisar código", callback_data="mode:review"),
        ],
        [
            InlineKeyboardButton("⚡ Modo conciso", callback_data="mode:concise"),
            InlineKeyboardButton("📚 Modo detallado", callback_data="mode:detailed"),
        ],
        [InlineKeyboardButton("🗑 Limpiar historial", callback_data="action:reset")],
    ])
    await update.message.reply_text("¿Qué modo quieres usar?", reply_markup=teclado)

Handler para los botones:

from telegram.ext import CallbackQueryHandler

async def handle_callback(update: Update, context: ContextTypes.DEFAULT_TYPE):
    query = update.callback_query
    await query.answer()  # OBLIGATORIO: confirmar recepción a Telegram

    data = query.data
    user_id = query.from_user.id

    if data.startswith("mode:"):
        modo = data.split(":")[1]
        context.user_data["modo"] = modo
        await query.edit_message_text(f"Modo cambiado a: *{modo}*", parse_mode="Markdown")

    elif data == "action:reset":
        clear_history(user_id)
        await query.edit_message_text("Historial borrado.")

# Registrar en bot.py
app.add_handler(CallbackQueryHandler(handle_callback))

Modo de código — /code

Comando especializado para tareas de programación con output formateado:

async def cmd_code(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """
    /code <tarea> — responde con enfoque en código, sin explicaciones largas
    Ejemplo: /code función Python que valida emails con regex
    """
    if not context.args:
        await update.message.reply_text("Uso: /code <descripción de lo que necesitas>")
        return

    tarea = " ".join(context.args)
    user_id = update.effective_user.id

    # System prompt específico para código
    prompt_codigo = [{"role": "user", "content":
        f"Escribe SOLO el código sin explicación larga. "
        f"Incluye comentarios inline donde sea necesario. "
        f"Tarea: {tarea}"
    }]

    stop = asyncio.Event()
    typing_task = asyncio.create_task(typing_loop(context.bot, update.effective_chat.id, stop))

    try:
        respuesta = ask_claude(prompt_codigo)
    finally:
        stop.set()
        typing_task.cancel()

    # Intentar formatear como código si detectamos bloque
    await update.message.reply_text(respuesta, parse_mode="Markdown")

Recibir y procesar archivos

Documentos (código, logs, JSON)

from telegram.ext import MessageHandler, filters

async def handle_document(update: Update, context: ContextTypes.DEFAULT_TYPE):
    doc = update.message.document

    # Limitar tamaño: máximo 1MB para texto
    if doc.file_size > 1_000_000:
        await update.message.reply_text("Archivo muy grande (máx 1MB).")
        return

    # Descargar el archivo
    file = await context.bot.get_file(doc.file_id)
    contenido = bytes()
    import io
    buffer = io.BytesIO()
    await file.download_to_memory(buffer)
    contenido = buffer.getvalue().decode("utf-8", errors="replace")

    user_id = update.effective_user.id
    caption = update.message.caption or "Analiza este archivo."

    add_message(user_id, "user", f"{caption}\n\nContenido del archivo `{doc.file_name}`:\n```\n{contenido[:8000]}\n```")
    history = get_history(user_id)
    respuesta = ask_claude(history)
    add_message(user_id, "assistant", respuesta)

    await send_long_message(update, respuesta)

# Registrar
app.add_handler(MessageHandler(filters.Document.ALL, handle_document))

Imágenes con Claude Vision

Claude puede analizar imágenes. Envías una captura de pantalla de un error y Claude lo explica:

import base64
import io

async def handle_photo(update: Update, context: ContextTypes.DEFAULT_TYPE):
    # Tomar la foto de mayor resolución
    photo = update.message.photo[-1]
    caption = update.message.caption or "Describe y analiza esta imagen."

    file = await context.bot.get_file(photo.file_id)
    buffer = io.BytesIO()
    await file.download_to_memory(buffer)
    img_data = base64.standard_b64encode(buffer.getvalue()).decode()

    # Mensaje multimodal para Claude
    mensaje_vision = {
        "role": "user",
        "content": [
            {
                "type": "image",
                "source": {
                    "type": "base64",
                    "media_type": "image/jpeg",
                    "data": img_data,
                },
            },
            {"type": "text", "text": caption},
        ],
    }

    user_id = update.effective_user.id
    history = get_history(user_id)
    # Las imágenes no se guardan en historial — solo el texto
    messages_con_imagen = history + [mensaje_vision]

    stop = asyncio.Event()
    typing_task = asyncio.create_task(typing_loop(context.bot, update.effective_chat.id, stop))
    try:
        respuesta = ask_claude(messages_con_imagen)
    finally:
        stop.set()
        typing_task.cancel()

    add_message(user_id, "user", f"[imagen enviada] {caption}")
    add_message(user_id, "assistant", respuesta)
    await send_long_message(update, respuesta)

app.add_handler(MessageHandler(filters.PHOTO, handle_photo))

Casos de uso con imágenes:

Soporte para grupos

En grupos, el bot solo responde cuando es mencionado o cuando el mensaje empieza con /:

# Filtro para grupos: solo responder si mencionan al bot
async def handle_group_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
    bot_username = context.bot.username
    text = update.message.text or ""

    # Responder solo si mencionan al bot
    if f"@{bot_username}" not in text:
        return

    # Limpiar la mención del texto
    prompt = text.replace(f"@{bot_username}", "").strip()
    if not prompt:
        return

    # Historial separado por chat (no por usuario en grupos)
    chat_id = update.effective_chat.id
    add_message(chat_id, "user", prompt)
    history = get_history(chat_id)
    respuesta = ask_claude(history)
    add_message(chat_id, "assistant", respuesta)

    await update.message.reply_text(respuesta)

# Registrar solo para grupos
app.add_handler(MessageHandler(
    filters.TEXT & filters.ChatType.GROUPS & ~filters.COMMAND,
    handle_group_message
))

Comando /export — Exportar conversación

async def cmd_export(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Exporta el historial de conversación como archivo de texto."""
    user_id = update.effective_user.id
    history = get_history(user_id)

    if not history:
        await update.message.reply_text("No hay conversación para exportar.")
        return

    contenido = "\n\n".join(
        f"{'Usuario' if m['role'] == 'user' else 'Claude'}: {m['content']}"
        for m in history
    )

    buffer = io.BytesIO(contenido.encode())
    buffer.name = "conversacion.txt"
    await update.message.reply_document(buffer, caption="Tu conversación con Claude.")

Rate limiting por usuario

Evita abuso con un límite de requests por minuto:

from collections import defaultdict
from time import time

_rate_limits: dict[int, list[float]] = defaultdict(list)
MAX_REQUESTS_PER_MINUTE = 10

def check_rate_limit(user_id: int) -> bool:
    """Retorna True si el usuario puede hacer más requests."""
    now = time()
    recent = [t for t in _rate_limits[user_id] if now - t < 60]
    _rate_limits[user_id] = recent

    if len(recent) >= MAX_REQUESTS_PER_MINUTE:
        return False

    _rate_limits[user_id].append(now)
    return True

# En el handler de mensajes:
async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
    user_id = update.effective_user.id
    if not check_rate_limit(user_id):
        await update.message.reply_text("Demasiadas requests. Espera un momento.")
        return
    # ... resto del handler