Bot de Leads Gratis en Telegram con Python y Yandex Cloud
Flujo: el usuario sigue un enlace profundo desde el sitio web → el bot pregunta sobre el tipo de negocio y el objetivo → crea un tema como "Iván — E-commerce — Agente de IA" → toda la comunicación se centraliza en el grupo con búsqueda y notificaciones.
Envoltorio ligero sobre la API de Telegram
Para funciones en la nube, las soluciones ligeras sin marcos pesados como aiogram funcionan mejor. Se minimiza el tiempo de inicio frío, usando únicamente httpx. La capa completa de API ocupa solo 57 líneas:
"""telegram_api.py — envoltorio mínimo sobre la API del bot."""
import os
import httpx
BOT_TOKEN = os.environ.get("BOT_TOKEN", "")
BASE_URL = f"https://api.telegram.org/bot{BOT_TOKEN}"
def _call(method: str, **kwargs) -> dict:
resp = httpx.post(f"{BASE_URL}/{method}", json=kwargs, timeout=10)
return resp.json()
def send_message(chat_id, text, reply_markup=None, message_thread_id=None):
params = {"chat_id": chat_id, "text": text, "parse_mode": "HTML"}
if reply_markup:
params["reply_markup"] = reply_markup
if message_thread_id:
params["message_thread_id"] = message_thread_id
return _call("sendMessage", **params)
def create_forum_topic(chat_id, name):
return _call("createForumTopic", chat_id=chat_id, name=name)
def copy_message(chat_id, from_chat_id, message_id, message_thread_id=None):
params = {"chat_id": chat_id, "from_chat_id": from_chat_id, "message_id": message_id}
if message_thread_id:
params["message_thread_id"] = message_thread_id
return _call("copyMessage", **params)copy_message es preferible a forward_message: elimina la etiqueta "Enviado desde...", manteniendo los mensajes limpios.
Manejador de Webhook con Validación
La función de Yandex Cloud maneja directamente las solicitudes HTTPS de Telegram (allow-unauthenticated-invoke). La validación usa X-Telegram-Bot-Api-Secret-Token:
def handler(event, context):
"""Punto de entrada de la función en Yandex Cloud."""
headers = event.get("headers", {})
secret = headers.get("X-Telegram-Bot-Api-Secret-Token") or \
headers.get("x-telegram-bot-api-secret-token", "")
if secret != WEBHOOK_SECRET:
return {"statusCode": 403, "body": "Forbidden"}
body = event.get("body", "{}")
update = json.loads(body) if isinstance(body, str) else body
if "callback_query" in update:
_handle_callback(update["callback_query"])
elif "message" in update:
_handle_message(update["message"])
return {"statusCode": 200, "body": "ok"}Separar la lógica en _handle_callback y _handle_message mantiene el código limpio y mantenible.
Calificación de leads sin estado
El comando /start desencadena una encuesta de dos preguntas: primero, tipo de negocio; segundo, objetivo. Los datos de callback codifican ambas respuestas: q2:sales:ecommerce. Se respeta el límite de 64 bytes para opciones predefinidas.
Ejemplo de la primera pregunta:
def _handle_start(chat_id, from_user):
if storage.get_topic_id_by_user(chat_id):
tg.send_message(chat_id,
"Ya has enviado una solicitud. Nos pondremos en contacto pronto.\n\n"
"Si quieres agregar algo, responde aquí.")
return
keyboard = {"inline_keyboard": [
[{"text": "E-commerce / Marketplaces", "callback_data": "q1:ecommerce"}],
[{"text": "Servicios / SaaS", "callback_data": "q1:services"}],
[{"text": "Manufactura", "callback_data": "q1:production"}],
[{"text": "IT / Startup", "callback_data": "q1:it"}],
[{"text": "Otro", "callback_data": "q1:other"}],
]}
tg.send_message(chat_id, f"Hola, {from_user.get('first_name', '')}! Soy el bot VexAI.\n\n"
"Responde 2 preguntas rápidas, y te contactaremos.")
tg.send_message(chat_id, "<b>¿Qué tipo de negocio tienes?</b>", reply_markup=keyboard)Prevención de duplicados: verifica el mapa en Almacenamiento de Objetos. Para "Otro", se usa un archivo de estado temporal.
La segunda pregunta funciona igual, pasando el q1_key en los datos de callback.
Creación de temas y retransmisión de mensajes
Después de la encuesta, se genera un tema con límite de 128 caracteres:
def _finalize(chat_id, from_user, q1_answer, q2_answer):
tg.send_message(chat_id, "¡Gracias! Te contactaremos pronto.")
full_name = f"{from_user.get('first_name', '')} {from_user.get('last_name', '')}".strip()
topic_title = f"{full_name} — {q1_answer} — {q2_answer}"
if len(topic_title) > 128:
topic_title = topic_title[:125] + "..."
result = tg.create_forum_topic(ADMIN_GROUP_ID, topic_title)
topic_id = result["result"]["message_thread_id"]
summary = (f"<b>Nuevo Lead</b>\n\n"
f"<b>Nombre:</b> {full_name}\n"
f"<b>Negocio:</b> {q1_answer}\n"
f"<b>Objetivo:</b> {q2_answer}\n"
f"<b>Fuente:</b> sitio web")
tg.send_message(ADMIN_GROUP_ID, summary, message_thread_id=topic_id)
storage.save_topic_mapping(topic_id, chat_id)
storage.save_user_mapping(chat_id, topic_id)Retransmisión de mensajes en _handle_message:
- El administrador responde en el tema → copiado al cliente.
- El cliente envía mensaje al bot → copiado al tema.
Se soportan enlaces profundos: /start website para rastrear la fuente.
Configuración y limitaciones
- Crea un supergrupo y activa los temas.
- Añade el bot como administrador: gestiona temas, envía mensajes.
- Configura Yandex Cloud: cuenta de servicio con
storage.editor, bucket, claves de acceso. - Establece el webhook con token secreto.
Obtienes:
- Historial de leads buscable.
- Notificaciones en tiempo real.
- Tarjetas de lead fijadas.
Limitaciones: no hay embudo de ventas, análisis ni colaboración en equipo. Ideal para hasta 20 leads/mes.
Conclusiones clave
- Despliegue gratuito: Yandex Cloud Functions + Almacenamiento de Objetos en modo gratuito.
- Diseño sin estado: los datos de callback contienen el estado; se necesita almacenamiento mínimo.
- Retransmisión completa: conversación completa en temas — sin herramientas externas.
- Seguridad: token secreto + protección contra duplicados.
- Escalabilidad: soporta hasta 20 leads/mes sin CRM de pago.
— Editorial Team
Aún no hay comentarios.