Telegram Python 线索机器人：免费 CRM

用Python和Yandex云在Telegram上免费搭建线索机器人

无需任何费用，独立开发者和小型团队即可通过Telegram机器人构建简易客户关系管理系统。机器人仅需两个内联问题即可完成线索初筛，为每位客户创建专属话题，并实现双向消息互通。部署于Yandex云函数免费套餐：每月100万次调用和1GB对象存储空间。

流程：用户通过网站深链接进入 → 机器人询问业务类型与目标 → 自动生成如「Ivan — 电商 — AI助手」的话题 → 所有沟通集中于群组，支持搜索与通知。

轻量级Telegram Bot API封装

对于云函数而言，轻量级方案优于使用aiogram等重型框架。冷启动时间最小化，采用纯httpx实现。完整API层仅57行代码：

"""telegram_api.py — Telegram 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)

copy_message优于forward_message：可移除「转发自…」标签，保持消息整洁。

带验证的Webhook处理器

Yandex云函数直接处理来自Telegram的HTTPS请求（允许未认证调用）。验证使用X-Telegram-Bot-Api-Secret-Token：

def handler(event, context):
    """Yandex云函数入口点。"""
    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"Hi, {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用于追踪来源。

部署与限制条件

• 创建超级群组并启用话题功能。
• 添加机器人为管理员：具备管理话题和发送消息权限。
• 配置Yandex云：服务账户需拥有storage.editor权限，设置存储桶及访问密钥。
• 设置带密钥的Webhook。

您将获得：

• 可搜索的线索历史记录；
• 实时通知提醒；
• 固定的线索卡片展示。

限制说明：不支持销售漏斗、数据分析或团队协作。适合每月最多20个线索。

核心要点

• 零成本部署：Yandex云函数 + 对象存储免费套餐。
• 无状态设计：回调数据承载状态，所需存储极少。
• 全链路中继：完整对话集中于话题，无需外部工具。
• 安全防护：密钥验证 + 重复提交拦截。
• 可扩展性：每月支持20个线索，无需付费CRM。

— Editorial Team