Cap 7: Comandos Avanzados
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:
- Screenshot de error → Claude explica qué falló
- Diagrama UML → Claude genera código
- UI screenshot → Claude sugiere mejoras
- Código fotografiado → Claude lo transcribe y refactoriza
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