Python 기반 Telegram Lead Bot: 무료 CRM

파이썬과 얀덱스 클라우드로 무료 텔레그램 리드봇 만들기

텔레그램 봇 하나로 간단한 CRM 시스템을 구축할 수 있습니다. 비용은 0원입니다. 봇은 단 두 가지 인라인 질문으로 들어오는 리드를 자동 분류하고, 각 고객마다 전용 주제를 슈퍼그룹에 생성하며, 양방향 메시지 전달을 완전히 지원합니다. 얀덱스 클라우드 함수의 무료 티어를 활용하면 월 100만 회 호출 및 1GB 객체 저장소를 무료로 사용할 수 있습니다.

작업 흐름: 사용자가 웹사이트에서 깊은 링크를 따라 접속 → 봇이 사업 유형과 목표를 묻는다 → "아이반 — 전자상거래 — AI 에이전트"처럼 주제를 생성 → 모든 소통은 그룹 내에서 중앙집중화되며, 검색과 알림 기능도 제공됩니다.

텔레그램 봇 API의 가벼운 래퍼

클라우드 함수에서는 aiogram 같은 무거운 프레임워크보다 가볍고 경량화된 솔루션이 최적입니다. 콜드 스타트 시간을 최소화하기 위해 순수 httpx를 사용합니다. 전체 API 계층은 단 57줄로 구성됩니다:

"""telegram_api.py — 최소한의 봇 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)

copy_message는 forward_message보다 선호됩니다. 이는 "전송됨"이라는 레이블을 제거해 메시지가 더 깔끔하게 유지되기 때문입니다.

웹훅 핸들러와 유효성 검사

얀덱스 클라우드 함수는 텔레그램에서 오는 HTTPS 요청을 직접 처리합니다 (allow-unauthenticated-invoke). 유효성 검사는 X-Telegram-Bot-Api-Secret-Token을 통해 이루어집니다:

def handler(event, context):
    """얀덱스 클라우드 함수 진입점."""
    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"}

_handle_callback과 _handle_message로 로직을 분리하면 코드가 깔끔하고 유지보수가 쉬워집니다.

상태 없는 리드 질의 과정

/start 명령어가 실행되면 두 가지 질문으로 구성된 설문조사가 시작됩니다. 첫 번째는 사업 유형, 두 번째는 목표입니다. 콜백 데이터에는 두 답변이 함께 포함됩니다: q2:sales:ecommerce. 사전 설정된 옵션은 64바이트 제한을 지키도록 설계되었습니다.

첫 번째 질문 예시:

def _handle_start(chat_id, from_user):
    if storage.get_topic_id_by_user(chat_id):
        tg.send_message(chat_id,
            "이미 요청을 제출하셨습니다. 곧 연락드리겠습니다.\n\n"
            "추가로 말씀하고 싶은 내용이 있다면 여기에 답장해 주세요.")
        return

    keyboard = {"inline_keyboard": [
        [{"text": "전자상거래 / 마켓플레이스", "callback_data": "q1:ecommerce"}],
        [{"text": "서비스 / SaaS",           "callback_data": "q1:services"}],
        [{"text": "제조업",               "callback_data": "q1:production"}],
        [{"text": "IT / 스타트업",               "callback_data": "q1:it"}],
        [{"text": "기타",                     "callback_data": "q1:other"}],
    ]}
    tg.send_message(chat_id, f"안녕하세요, {from_user.get('first_name', '')}님! 저는 VexAI 봇입니다.\n\n"
                      "빠른 두 가지 질문에 답하시면 연락드리겠습니다.")
    tg.send_message(chat_id, "<b>어떤 사업을 운영하시나요?</b>", reply_markup=keyboard)

중복 방지: 객체 저장소에 매핑 정보를 확인합니다. "기타" 선택 시 임시 상태 파일을 사용합니다.

두 번째 질문도 유사하게 동작하며, q1_key를 콜백 데이터에 포함시킵니다.

주제 생성 및 메시지 릴레이

설문 조사 완료 후, 주제가 생성되며 이름 길이는 128자까지 가능합니다:

def _finalize(chat_id, from_user, q1_answer, q2_answer):
    tg.send_message(chat_id, "감사합니다! 곧 연락드리겠습니다.")

    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>신규 리드</b>\n\n"
               f"<b>이름:</b> {full_name}\n"
               f"<b>사업 유형:</b> {q1_answer}\n"
               f"<b>목표:</b> {q2_answer}\n"
               f"<b>소스:</b> 웹사이트")
    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)

_handle_message에서의 메시지 릴레이:

• 관리자가 주제에 답장 → 클라이언트에게 복사됨.
• 클라이언트가 봇에 메시지 → 주제로 복사됨.

딥링크 지원: /start website로 소스 추적 가능.

설치 및 제약 사항

• 슈퍼그룹 생성 후 주제 기능 활성화.
• 봇을 관리자로 추가: 주제 관리 및 메시지 전송 권한 부여.
• 얀덱스 클라우드 설정: 서비스 계정에 storage.editor 권한, 버킷 및 액세스 키 구성.
• 웹훅 설정 및 시크릿 토큰 등록.

얻을 수 있는 기능:

• 검색 가능한 리드 기록
• 실시간 알림
• 고정된 리드 카드

제약 사항: 판매 파이프라인, 분석 도구, 팀 협업 기능 없음. 월 20건까지 적합합니다.

핵심 요약

• 무료 배포: 얀덱스 클라우드 함수 + 객체 저장소 무료 티어 사용 가능.
• 상태 없는 설계: 콜백 데이터가 상태를 유지하므로 저장소 사용 최소화.
• 완전한 릴레이: 모든 대화는 주제 내에서 완전히 처리되며 외부 도구 필요 없음.
• 보안 강화: 시크릿 토큰 + 중복 방지 기능.
• 확장성: 유료 CRM 없이 월 20건까지 확장 가능.

— Editorial Team