Bezpłatny bot do generowania leadów w Telegramie na Pythonie z Yandex Cloud Functions
Rozwiązania dla samodzielnych projektów i małych zespołów pozwalają stworzyć prostą CRM bez kosztów. Bot kwalifikuje nowe leady poprzez dwa pytania z przyciskami inline, tworzy osobny temat w supergrupie dla każdego klienta oraz zapewnia dwukierunkowy przekaz wiadomości. Wdrożenie na Yandex Cloud Functions z darmowym poziomem: 1 milion wywołań miesięcznie i 1 GB Object Storage.
Schemat działania: użytkownik kliknie link głęboki ze strony → bot prosi o rodzaj działalności i cel → tworzy temat typu «Ivan — E-commerce — agent AI» → cała rozmowa jest agregowana w grupie z możliwością wyszukiwania i powiadomień.
Minimalna obudowa nad Telegram Bot API
Dla Cloud Functions idealne są lekkie rozwiązania bez frameworków takich jak aiogram. Zmniejsza to czas startu chłodnego do minimum, używając czystego httpx. Pełny warstwy API zajmuje tylko 57 linii:
"""telegram_api.py — cienka warstwa nad Bot API."""
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)Funkcja copy_message jest lepsza niż forward_message: eliminuje znacznik «Przesłane od…», a wiadomość wygląda naturalnie.
Handler webhook z walidacją
Yandex Cloud Function obsługuje bezpośrednie żądania HTTPS od Telegrama (allow-unauthenticated-invoke). Walidacja odbywa się za pomocą X-Telegram-Bot-Api-Secret-Token:
def handler(event, context):
"""Punkt wejścia funkcji 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"}Podział na _handle_callback i _handle_message upraszcza logikę.
Bezstanowa kwalifikacja leadów
Komenda /start uruchamia ankietę: pierwsze pytanie – rodzaj działalności, drugie – cel. Dane callback kodują oba odpowiedzi: q2:sales:ecommerce. Limit 64 bajtów został zachowany dla ustawień domyślnych.
Przykład pierwszego pytania:
def _handle_start(chat_id, from_user):
if storage.get_topic_id_by_user(chat_id):
tg.send_message(chat_id,
"Już podałeś zgłoszenie. Skontaktujemy się z Tobą w najbliższym czasie.\n\n"
"Jeśli chcesz coś dodać — po prostu napisz tu wiadomość.")
return
keyboard = {"inline_keyboard": [
[{"text": "E-commerce / Platformy handlowe", "callback_data": "q1:ecommerce"}],
[{"text": "Usługi / Serwis", "callback_data": "q1:services"}],
[{"text": "Produkcja", "callback_data": "q1:production"}],
[{"text": "IT / Startup", "callback_data": "q1:it"}],
[{"text": "Inne", "callback_data": "q1:other"}],
]}
tg.send_message(chat_id, f"Cześć, {from_user.get('first_name', '')}! Jestem bot VexAI.\n\n"
"Odpowiedz na 2 krótkie pytania, a skontaktujemy się z Tobą.")
tg.send_message(chat_id, "<b>Jak wygląda Twoja działalność?</b>", reply_markup=keyboard)Ochrona przed duplikatami: sprawdzenie mapowania w Object Storage. Dla opcji «Inne» — tymczasowy plik stanu.
Drugie pytanie działa podobnie, z przekazaniem q1_key w danych.
Tworzenie tematu i relay
Po zakończeniu ankiety generowany jest temat o limicie 128 znaków:
def _finalize(chat_id, from_user, q1_answer, q2_answer):
tg.send_message(chat_id, "Dziękujemy! Skontaktujemy się z Tobą wkrótce.")
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>Nowy lead</b>\n\n"
f"<b>Imię:</b> {full_name}\n"
f"<b>Działalność:</b> {q1_answer}\n"
f"<b>Zadanie:</b> {q2_answer}\n"
f"<b>Źródło:</b> website")
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)Relay w _handle_message:
- Administrator pisze w temacie → kopiowane do klienta.
- Klient pisze do bota → kopiowane do tematu.
Obsługa linków głębokich: /start website do śledzenia źródła.
Konfiguracja i ograniczenia
- Utwórz supergrupę z włączonymi tematami.
- Dodaj bota jako administratora: zarządzanie tematami i wysyłanie wiadomości.
- Skonfiguruj Yandex Cloud: usługa konta
storage.editor, kontener, klucze dostępu. - Ustaw webhook z tokenem tajnym.
Otrzymujesz:
- Wyszukiwanie leadów.
- Powiadomienia.
- Przypięte karty.
Ograniczenia: brak funneli, analizy, współpracy zespołowej. Idealne dla 20 leadów/miesiąc.
Co ważne
- Bezpłatne wdrożenie: Yandex Cloud Functions + Object Storage w warstwie darmowej.
- Projekt bezstanowy: dane callback przechowują stan, minimalne magazynowanie.
- Funkcja relay: pełna komunikacja w tematach bez dodatkowych aplikacji.
- Ochrona: token tajny + ochrona przed duplikatami.
- Skalowalność: do 20 leadów/miesiąc bez płatnej CRM.
— Editorial Team
Brak komentarzy.