@afjs/telegram-bot-sdk

v1.0.0

Published

a month ago

Telegram Bot API的Node.js SDK

Downloads

0High
0Medium
0Low

tinsfox

telegram bot api sdk webhook messaging nodejs

Telegram Bot SDK

一个功能完整、类型安全的 Telegram Bot API Node.js SDK，基于 TypeScript 开发，提供简洁易用的 API 接口。

兼容性：支持 Telegram Bot API 7.0+ 的所有功能，包括最新的消息类型、键盘、Webhook 和文件处理功能。

✨ 特性

🚀 完整的 API 支持 - 支持 Telegram Bot API 的所有方法
📝 TypeScript 支持 - 完整的类型定义，提供优秀的开发体验
🔄 自动重试机制 - 内置请求重试和错误处理
📁 文件上传支持 - 支持各种文件类型的上传（图片、文档、音频等）
🎯 Webhook 支持 - 完整的 Webhook 设置和管理
🛠️ 实用工具函数 - 丰富的辅助函数简化开发
🎨 键盘构建器 - 便捷的内联键盘和回复键盘构建
🔐 安全性 - 内置 Token 验证和安全处理
📦 轻量级 - 最小化依赖，高性能

📦 安装

系统要求

Node.js >= 18.0.0
支持 ES Modules

安装命令

npm install @afjs/telegram-bot-sdk

yarn add @afjs/telegram-bot-sdk

pnpm add @afjs/telegram-bot-sdk

🚀 快速开始

基础使用

import { TelegramBot } from '@afjs/telegram-bot-sdk'

// 创建机器人实例
const bot = new TelegramBot({
  token: process.env.BOT_TOKEN!, // 从环境变量获取
  timeout: 30000,
  retries: 3
})

// 发送消息
async function sendMessage() {
  try {
    const chatId = process.env.CHAT_ID! // 从环境变量获取聊天 ID
    const message = await bot.sendMessage({
      chat_id: chatId,
      text: '你好！这是来自 Telegram Bot SDK 的消息。'
    })
    console.log('消息已发送:', message.message_id)
  } catch (error) {
    if (error instanceof Error) {
      console.error('发送失败:', error.message)

      // 处理常见错误
      if (error.message.includes('chat not found')) {
        console.error('错误：聊天不存在，请检查 CHAT_ID')
      } else if (error.message.includes('Unauthorized')) {
        console.error('错误：Bot Token 无效，请检查 BOT_TOKEN')
      }
    } else {
      console.error('未知错误:', error)
    }
  }
}

sendMessage()

获取机器人信息

async function getBotInfo() {
  try {
    const botInfo = await bot.getMe()
    console.log('机器人信息:', botInfo)
  } catch (error) {
    console.error('获取信息失败:', error.message)
  }
}

📚 详细文档

配置选项

interface BotConfig {
  token: string        // 机器人 Token（必需）
  baseUrl?: string     // API 基础 URL（可选，默认为官方 API）
  timeout?: number     // 请求超时时间（毫秒，默认 30000）
  retries?: number     // 重试次数（默认 0）
}

消息发送

发送文本消息

await bot.sendMessage({
  chat_id: chatId,
  text: '这是一条文本消息',
  parse_mode: 'HTML', // 支持 'HTML', 'MarkdownV2', 'Markdown'
  disable_notification: false,
  reply_markup: {
    inline_keyboard: [[
      { text: '按钮1', callback_data: 'btn1' },
      { text: '按钮2', callback_data: 'btn2' }
    ]]
  }
})

发送图片

// 发送网络图片
await bot.sendPhoto({
  chat_id: chatId,
  photo: 'https://example.com/image.jpg',
  caption: '图片说明'
})

// 发送本地文件
import { readFileSync } from 'fs'
const imageBuffer = readFileSync('./image.jpg')

await bot.sendPhoto({
  chat_id: chatId,
  photo: imageBuffer,
  caption: '本地图片'
})

发送文档

await bot.sendDocument({
  chat_id: chatId,
  document: documentBuffer, // Buffer 或文件 URL
  caption: '文档说明',
  disable_content_type_detection: false
})

发送其他媒体类型

// 发送音频
await bot.sendAudio({
  chat_id: chatId,
  audio: audioBuffer,
  duration: 180,
  title: '音频标题',
  performer: '演唱者'
})

// 发送视频
await bot.sendVideo({
  chat_id: chatId,
  video: videoBuffer,
  duration: 60,
  width: 1920,
  height: 1080,
  caption: '视频说明'
})

// 发送语音消息
await bot.sendVoice({
  chat_id: chatId,
  voice: voiceBuffer,
  duration: 30
})

键盘和按钮

内联键盘

import { createInlineKeyboard } from '@afjs/telegram-bot-sdk'

const keyboard = createInlineKeyboard([
  [
    { text: '选项1', callback_data: 'option1' },
    { text: '选项2', callback_data: 'option2' }
  ],
  [
    { text: '访问网站', url: 'https://example.com' }
  ],
  [
    { text: '分享', switch_inline_query: 'share_text' }
  ]
])

await bot.sendMessage({
  chat_id: chatId,
  text: '请选择一个选项:',
  reply_markup: keyboard
})

回复键盘

import { createReplyKeyboard } from '@afjs/telegram-bot-sdk'

const keyboard = createReplyKeyboard([
  [{ text: '🔍 搜索' }, { text: '⚙️ 设置' }],
  [{ text: '📊 统计' }, { text: '❓ 帮助' }],
  [{ text: '🚪 退出' }]
], {
  resize_keyboard: true,
  one_time_keyboard: true
})

await bot.sendMessage({
  chat_id: chatId,
  text: '选择功能:',
  reply_markup: keyboard
})

移除键盘

import { removeKeyboard } from '@afjs/telegram-bot-sdk'

await bot.sendMessage({
  chat_id: chatId,
  text: '键盘已移除',
  reply_markup: removeKeyboard()
})

处理更新

轮询方式

async function startPolling() {
  let offset = 0

  while (true) {
    try {
      const updates = await bot.getUpdates({
        offset,
        limit: 100,
        timeout: 30
      })

      for (const update of updates) {
        await handleUpdate(update)
        offset = update.update_id + 1
      }
    } catch (error) {
      console.error('获取更新失败:', error.message)
      await new Promise(resolve => setTimeout(resolve, 5000))
    }
  }
}

async function handleUpdate(update) {
  if (update.message) {
    await handleMessage(update.message)
  }

  if (update.callback_query) {
    await handleCallbackQuery(update.callback_query)
  }
}

Webhook 方式

// 设置 Webhook
await bot.setWebhook({
  url: 'https://yourdomain.com/webhook',
  secret_token: 'your_secret_token',
  max_connections: 100,
  allowed_updates: ['message', 'callback_query']
})

// 获取 Webhook 信息
const webhookInfo = await bot.getWebhookInfo()
console.log('Webhook 信息:', webhookInfo)

// 删除 Webhook
await bot.deleteWebhook()

实用工具函数

消息处理工具

import {
  getMessageFromUpdate,
  getUserIdFromUpdate,
  getChatIdFromUpdate,
  getTextFromMessage,
  isCommand,
  getCommandArgs
} from '@afjs/telegram-bot-sdk'

// 从更新中提取消息
const message = getMessageFromUpdate(update)

// 提取用户ID和聊天ID
const userId = getUserIdFromUpdate(update)
const chatId = getChatIdFromUpdate(update)

// 检查是否为命令
if (isCommand(message, 'start')) {
  const args = getCommandArgs(message)
  console.log('命令参数:', args)
}

文本格式化工具

import {
  escapeHTML,
  escapeMarkdownV2,
  formatMessage
} from '@afjs/telegram-bot-sdk'

// HTML 转义
const safeText = escapeHTML('<b>用户输入</b>')

// MarkdownV2 转义
const markdownText = escapeMarkdownV2('*特殊字符*')

// 自动格式化
const formatted = formatMessage('用户输入', 'HTML')

验证工具

import {
  isValidBotToken,
  getBotIdFromToken,
  isFileSizeValid
} from '@afjs/telegram-bot-sdk'

// 验证 Token 格式
if (isValidBotToken(token)) {
  const botId = getBotIdFromToken(token)
  console.log('Bot ID:', botId)
}

// 验证文件大小
if (isFileSizeValid(fileSize, 'photo')) {
  // 文件大小符合要求
}

其他实用工具

import {
  delay,
  chunk,
  truncateText,
  formatFileSize,
  generateRandomString
} from '@afjs/telegram-bot-sdk'

// 延迟执行
await delay(1000) // 等待1秒

// 数组分块
const chunks = chunk([1, 2, 3, 4, 5], 2) // [[1, 2], [3, 4], [5]]

// 截断文本
const short = truncateText('很长的文本...', 10) // "很长的文本..."

// 格式化文件大小
const size = formatFileSize(1024) // "1 KB"

// 生成随机字符串
const random = generateRandomString(10)

文件操作

获取文件信息

// 获取文件信息
const file = await bot.getFile('file_id')
console.log('文件路径:', file.file_path)

// 获取文件下载链接
const downloadUrl = bot.getFileUrl(file.file_path)
console.log('下载链接:', downloadUrl)

文件上传

import { readFileSync } from 'fs'

// 上传本地文件
const fileBuffer = readFileSync('./document.pdf')
await bot.sendDocument({
  chat_id: chatId,
  document: fileBuffer,
  caption: 'PDF 文档'
})

// 上传网络文件
await bot.sendDocument({
  chat_id: chatId,
  document: 'https://example.com/document.pdf',
  caption: '网络文档'
})

消息管理

编辑消息

// 编辑文本消息
await bot.editMessageText({
  chat_id: chatId,
  message_id: messageId,
  text: '更新后的消息内容',
  parse_mode: 'HTML'
})

删除消息

// 删除单条消息
await bot.deleteMessage({
  chat_id: chatId,
  message_id: messageId
})

// 删除多条消息
await bot.deleteMessages({
  chat_id: chatId,
  message_ids: [messageId1, messageId2, messageId3]
})

回调查询处理

async function handleCallbackQuery(callbackQuery) {
  // 应答回调查询（必需）
  await bot.answerCallbackQuery({
    callback_query_id: callbackQuery.id,
    text: '操作完成',
    show_alert: false
  })

  // 处理回调数据
  const data = callbackQuery.data
  switch (data) {
    case 'button1':
      // 处理按钮1点击
      break
    case 'button2':
      // 处理按钮2点击
      break
  }
}

内联查询处理

async function handleInlineQuery(inlineQuery) {
  const results = [
    {
      type: 'article',
      id: '1',
      title: '搜索结果1',
      description: '描述信息',
      input_message_content: {
        message_text: '这是搜索结果1的内容'
      }
    }
  ]

  await bot.answerInlineQuery({
    inline_query_id: inlineQuery.id,
    results: results,
    cache_time: 300
  })
}

投票功能

// 创建投票
await bot.sendPoll({
  chat_id: chatId,
  question: '你最喜欢哪种编程语言？',
  options: ['JavaScript', 'Python', 'Go', 'Rust'],
  is_anonymous: false,
  type: 'regular',
  allows_multiple_answers: true
})

// 创建测验
await bot.sendPoll({
  chat_id: chatId,
  question: '2 + 2 = ?',
  options: ['3', '4', '5', '6'],
  type: 'quiz',
  correct_option_id: 1,
  explanation: '2 + 2 等于 4'
})

位置和联系人

// 发送位置
await bot.sendLocation({
  chat_id: chatId,
  latitude: 39.9042,
  longitude: 116.4074,
  live_period: 3600 // 实时位置持续时间（秒）
})

// 发送联系人
await bot.sendContact({
  chat_id: chatId,
  phone_number: '+1234567890',
  first_name: 'John',
  last_name: 'Doe'
})

🔧 高级功能

错误处理

import { TelegramBot } from '@afjs/telegram-bot-sdk'

const bot = new TelegramBot({
  token: 'YOUR_TOKEN',
  retries: 3,
  timeout: 30000
})

try {
  await bot.sendMessage({
    chat_id: 'invalid_chat_id',
    text: 'Test message'
  })
} catch (error) {
  if (error.message.includes('chat not found')) {
    console.log('聊天不存在')
  } else if (error.message.includes('bot was blocked')) {
    console.log('机器人被用户屏蔽')
  } else {
    console.error('其他错误:', error.message)
  }
}

批量操作

import { chunk, delay } from '@afjs/telegram-bot-sdk'

// 批量发送消息（避免触发限制）
async function sendBulkMessages(chatIds, message) {
  const batches = chunk(chatIds, 30) // 每批30个

  for (const batch of batches) {
    const promises = batch.map(chatId =>
      bot.sendMessage({ chat_id: chatId, text: message })
        .catch(error => console.error(`发送到 ${chatId} 失败:`, error.message))
    )

    await Promise.allSettled(promises)
    await delay(1000) // 批次间延迟1秒
  }
}

深度链接

import { buildDeepLink } from '@afjs/telegram-bot-sdk'

// 创建深度链接
const deepLink = buildDeepLink('your_bot_username', 'start_parameter')
console.log('深度链接:', deepLink)
// 输出: https://t.me/your_bot_username?start=start_parameter

🌐 Serverless 环境使用

Cloudflare Workers

Telegram Bot SDK 完全兼容 Cloudflare Workers 环境。以下是完整的实现示例：

基础设置

// worker.ts
import { TelegramBot, isCommand, getCommandArgs, escapeHTML } from '@afjs/telegram-bot-sdk'

export interface Env {
  BOT_TOKEN: string
  WEBHOOK_SECRET?: string
}

export default {
  async fetch(request: Request, env: Env, ctx: ExecutionContext): Promise<Response> {
    const bot = new TelegramBot({
      token: env.BOT_TOKEN,
      timeout: 10000 // Cloudflare Workers 有执行时间限制
    })

    const url = new URL(request.url)

    // 处理 Webhook
    if (request.method === 'POST' && url.pathname === '/webhook') {
      return handleWebhook(request, bot, env)
    }

    // 设置 Webhook
    if (request.method === 'GET' && url.pathname === '/set-webhook') {
      return setWebhook(bot, url.origin, env)
    }

    return new Response('Telegram Bot is running!', { status: 200 })
  }
}

async function handleWebhook(request: Request, bot: TelegramBot, env: Env): Promise<Response> {
  try {
    // 验证 Webhook Secret（强烈推荐）
    if (env.WEBHOOK_SECRET) {
      const secretHeader = request.headers.get('X-Telegram-Bot-Api-Secret-Token')
      if (!secretHeader || secretHeader !== env.WEBHOOK_SECRET) {
        console.warn('Webhook 验证失败:', {
          expected: env.WEBHOOK_SECRET,
          received: secretHeader,
          headers: Object.fromEntries(request.headers.entries())
        })
        return new Response('Unauthorized - Invalid secret token', { status: 401 })
      }
      console.log('✅ Webhook Secret 验证成功')
    } else {
      console.warn('⚠️  未设置 WEBHOOK_SECRET，建议设置以提高安全性')
    }

    const update = await request.json()
    await handleUpdate(update, bot)

    return new Response('OK', { status: 200 })
  } catch (error) {
    console.error('Webhook 处理错误:', error)
    return new Response('Internal Server Error', { status: 500 })
  }
}

async function setWebhook(bot: TelegramBot, origin: string, env: Env): Promise<Response> {
  try {
    const webhookUrl = `${origin}/webhook`

    await bot.setWebhook({
      url: webhookUrl,
      secret_token: env.WEBHOOK_SECRET,
      allowed_updates: ['message', 'callback_query', 'inline_query']
    })

    return new Response(`Webhook 已设置: ${webhookUrl}`, { status: 200 })
  } catch (error) {
    return new Response(`设置 Webhook 失败: ${error.message}`, { status: 500 })
  }
}

async function handleUpdate(update: any, bot: TelegramBot) {
  if (update.message) {
    await handleMessage(update.message, bot)
  }

  if (update.callback_query) {
    await handleCallbackQuery(update.callback_query, bot)
  }

  if (update.inline_query) {
    await handleInlineQuery(update.inline_query, bot)
  }
}

async function handleMessage(message: any, bot: TelegramBot) {
  const chatId = message.chat.id
  const text = message.text || ''

  if (isCommand(message, 'start')) {
    await bot.sendMessage({
      chat_id: chatId,
      text: `你好 ${escapeHTML(message.from.first_name)}！\n\n这个机器人运行在 Cloudflare Workers 上。\n\n可用命令:\n/help - 帮助\n/ping - 测试响应\n/time - 当前时间`,
      parse_mode: 'HTML'
    })
  } else if (isCommand(message, 'help')) {
    await bot.sendMessage({
      chat_id: chatId,
      text: '🤖 <b>Serverless Telegram Bot</b>\n\n' +
            '这是一个运行在 Cloudflare Workers 上的 Telegram 机器人示例。\n\n' +
            '<b>可用命令:</b>\n' +
            '/start - 开始使用\n' +
            '/help - 显示帮助\n' +
            '/ping - 测试机器人响应\n' +
            '/time - 显示当前时间\n' +
            '/echo &lt;文本&gt; - 回显消息',
      parse_mode: 'HTML'
    })
  } else if (isCommand(message, 'ping')) {
    const startTime = Date.now()
    await bot.sendMessage({
      chat_id: chatId,
      text: `🏓 Pong! 响应时间: ${Date.now() - startTime}ms`
    })
  } else if (isCommand(message, 'time')) {
    await bot.sendMessage({
      chat_id: chatId,
      text: `🕐 当前时间: ${new Date().toISOString()}`
    })
  } else if (isCommand(message, 'echo')) {
    const args = getCommandArgs(message)
    const echoText = args.join(' ') || '请提供要回显的文本'
    await bot.sendMessage({
      chat_id: chatId,
      text: `📢 ${escapeHTML(echoText)}`,
      parse_mode: 'HTML'
    })
  } else if (text) {
    // 处理普通消息
    await bot.sendMessage({
      chat_id: chatId,
      text: '我收到了你的消息！使用 /help 查看可用命令。'
    })
  }
}

async function handleCallbackQuery(callbackQuery: any, bot: TelegramBot) {
  await bot.answerCallbackQuery({
    callback_query_id: callbackQuery.id,
    text: '操作完成'
  })

  // 处理回调数据
  const data = callbackQuery.data
  const chatId = callbackQuery.message.chat.id

  switch (data) {
    case 'button_clicked':
      await bot.sendMessage({
        chat_id: chatId,
        text: '按钮被点击了！'
      })
      break
  }
}

async function handleInlineQuery(inlineQuery: any, bot: TelegramBot) {
  const results = [
    {
      type: 'article',
      id: '1',
      title: 'Serverless Bot',
      description: '来自 Cloudflare Workers 的问候',
      input_message_content: {
        message_text: '🚀 这条消息来自运行在 Cloudflare Workers 上的 Telegram Bot！'
      }
    }
  ]

  await bot.answerInlineQuery({
    inline_query_id: inlineQuery.id,
    results: results,
    cache_time: 300
  })
}

wrangler.toml 配置

name = "telegram-bot"
main = "src/worker.ts"
compatibility_date = "2024-01-01"

# 环境变量通过 wrangler secret 命令设置，不在此文件中配置
# 使用命令：wrangler secret put BOT_TOKEN
# 使用命令：wrangler secret put WEBHOOK_SECRET

部署步骤

# 1. 安装 Wrangler CLI
npm install -g wrangler

# 2. 登录 Cloudflare
wrangler login

# 3. 创建项目
mkdir telegram-bot-worker
cd telegram-bot-worker
npm init -y

# 4. 安装依赖
npm install @afjs/telegram-bot-sdk
npm install -D @cloudflare/workers-types typescript

# 5. 设置环境变量
wrangler secret put BOT_TOKEN
wrangler secret put WEBHOOK_SECRET

# 6. 部署
wrangler deploy

# 7. 设置 Webhook（重要！）
# 部署完成后，访问你的 Worker URL 来设置 Webhook
# 例如：https://your-worker.your-subdomain.workers.dev/set-webhook
# 或者使用 curl 命令直接设置：
curl -X POST "https://api.telegram.org/bot<YOUR_BOT_TOKEN>/setWebhook" \
     -H "Content-Type: application/json" \
     -d '{
       "url": "https://your-worker.your-subdomain.workers.dev/webhook",
       "secret_token": "your_webhook_secret"
     }'

设置 Webhook 的几种方式

方式1：通过浏览器访问

https://your-worker.your-subdomain.workers.dev/set-webhook

方式2：使用 curl 命令

curl -X POST "https://api.telegram.org/bot<YOUR_BOT_TOKEN>/setWebhook" \
     -H "Content-Type: application/json" \
     -d '{
       "url": "https://your-worker.your-subdomain.workers.dev/webhook",
       "secret_token": "your_webhook_secret",
       "allowed_updates": ["message", "callback_query", "inline_query"]
     }'

方式3：使用 SDK 设置（推荐）

// setup-webhook.ts - 单独的设置脚本
import { TelegramBot } from '@afjs/telegram-bot-sdk'

const bot = new TelegramBot({
  token: process.env.BOT_TOKEN! // 从环境变量获取
})

async function setupWebhook() {
  try {
    const webhookUrl = process.env.WEBHOOK_URL! // 从环境变量获取 Webhook URL
    const webhookSecret = process.env.WEBHOOK_SECRET! // 从环境变量获取密钥

    await bot.setWebhook({
      url: webhookUrl,
      secret_token: webhookSecret,
      allowed_updates: ['message', 'callback_query', 'inline_query'],
      drop_pending_updates: true // 清除待处理的更新
    })

    console.log('✅ Webhook 设置成功')
    console.log('Webhook URL:', webhookUrl)

    // 验证 Webhook 设置
    const webhookInfo = await bot.getWebhookInfo()
    console.log('Webhook 信息:', {
      url: webhookInfo.url,
      pendingUpdateCount: webhookInfo.pending_update_count,
      lastErrorDate: webhookInfo.last_error_date,
      lastErrorMessage: webhookInfo.last_error_message
    })
  } catch (error) {
    if (error instanceof Error) {
      console.error('❌ Webhook 设置失败:', error.message)
    } else {
      console.error('❌ 未知错误:', error)
    }
  }
}

setupWebhook()

Vercel Edge Functions

// api/webhook.ts
import { TelegramBot } from '@afjs/telegram-bot-sdk'
import type { NextRequest } from 'next/server'

export const config = {
  runtime: 'edge'
}

export default async function handler(req: NextRequest) {
  if (req.method !== 'POST') {
    return new Response('Method not allowed', { status: 405 })
  }

  // 验证 Webhook Secret
  if (process.env.WEBHOOK_SECRET) {
    const secretHeader = req.headers.get('X-Telegram-Bot-Api-Secret-Token')
    if (!secretHeader || secretHeader !== process.env.WEBHOOK_SECRET) {
      console.warn('Webhook 验证失败:', {
        expected: process.env.WEBHOOK_SECRET,
        received: secretHeader
      })
      return new Response('Unauthorized - Invalid secret token', { status: 401 })
    }
    console.log('✅ Webhook Secret 验证成功')
  } else {
    console.warn('⚠️  未设置 WEBHOOK_SECRET，建议设置以提高安全性')
  }

  const bot = new TelegramBot({
    token: process.env.BOT_TOKEN!,
    timeout: 10000
  })

  try {
    const update = await req.json()

    if (update.message) {
      const chatId = update.message.chat.id
      const text = update.message.text || ''

      if (text.startsWith('/start')) {
        await bot.sendMessage({
          chat_id: chatId,
          text: '你好！这是运行在 Vercel Edge Functions 上的机器人。'
        })
      }
    }

    return new Response('OK')
  } catch (error) {
    console.error('处理更新失败:', error)
    return new Response('Internal Server Error', { status: 500 })
  }
}

// api/set-webhook.ts - 设置 Webhook 的端点
import { TelegramBot } from '@afjs/telegram-bot-sdk'
import type { NextRequest } from 'next/server'

export const config = {
  runtime: 'edge'
}

export default async function handler(req: NextRequest) {
  if (req.method !== 'GET') {
    return new Response('Method not allowed', { status: 405 })
  }

  const bot = new TelegramBot({
    token: process.env.BOT_TOKEN!
  })

  try {
    const url = new URL(req.url)
    const webhookUrl = `${url.origin}/api/webhook`

    await bot.setWebhook({
      url: webhookUrl,
      secret_token: process.env.WEBHOOK_SECRET,
      allowed_updates: ['message', 'callback_query', 'inline_query']
    })

    return new Response(`✅ Webhook 已设置: ${webhookUrl}`)
  } catch (error) {
    return new Response(`❌ 设置失败: ${error.message}`, { status: 500 })
  }
}

Vercel 部署步骤

# 1. 创建 Next.js 项目
npx create-next-app@latest telegram-bot-vercel
cd telegram-bot-vercel

# 2. 安装依赖
npm install @afjs/telegram-bot-sdk

# 3. 设置环境变量
vercel env add BOT_TOKEN
vercel env add WEBHOOK_SECRET

# 4. 部署
vercel deploy

# 5. 设置 Webhook
# 访问：https://your-app.vercel.app/api/set-webhook

AWS Lambda

// lambda/handler.ts
import { APIGatewayProxyEvent, APIGatewayProxyResult } from 'aws-lambda'
import { TelegramBot } from '@afjs/telegram-bot-sdk'

const bot = new TelegramBot({
  token: process.env.BOT_TOKEN!,
  timeout: 10000
})

export const webhook = async (
  event: APIGatewayProxyEvent
): Promise<APIGatewayProxyResult> => {
  try {
    if (event.httpMethod !== 'POST') {
      return {
        statusCode: 405,
        body: 'Method not allowed'
      }
    }

    // 验证 Webhook Secret
    if (process.env.WEBHOOK_SECRET) {
      const secretHeader = event.headers['X-Telegram-Bot-Api-Secret-Token'] ||
                          event.headers['x-telegram-bot-api-secret-token']
      if (!secretHeader || secretHeader !== process.env.WEBHOOK_SECRET) {
        console.warn('Webhook 验证失败:', {
          expected: process.env.WEBHOOK_SECRET,
          received: secretHeader,
          headers: event.headers
        })
        return {
          statusCode: 401,
          body: 'Unauthorized - Invalid secret token'
        }
      }
      console.log('✅ Webhook Secret 验证成功')
    } else {
      console.warn('⚠️  未设置 WEBHOOK_SECRET，建议设置以提高安全性')
    }

    const update = JSON.parse(event.body || '{}')

    if (update.message) {
      const chatId = update.message.chat.id
      const text = update.message.text || ''

      if (text.startsWith('/start')) {
        await bot.sendMessage({
          chat_id: chatId,
          text: '你好！这是运行在 AWS Lambda 上的机器人。'
        })
      }
    }

    return {
      statusCode: 200,
      body: 'OK'
    }
  } catch (error) {
    console.error('Lambda 处理错误:', error)
    return {
      statusCode: 500,
      body: 'Internal Server Error'
    }
  }
}

// 设置 Webhook 的 Lambda 函数
export const setWebhook = async (
  event: APIGatewayProxyEvent
): Promise<APIGatewayProxyResult> => {
  try {
    const headers = event.headers
    const host = headers.Host || headers.host
    const stage = event.requestContext.stage
    const webhookUrl = `https://${host}/${stage}/webhook`

    await bot.setWebhook({
      url: webhookUrl,
      secret_token: process.env.WEBHOOK_SECRET,
      allowed_updates: ['message', 'callback_query', 'inline_query']
    })

    return {
      statusCode: 200,
      body: JSON.stringify({
        message: '✅ Webhook 设置成功',
        url: webhookUrl
      })
    }
  } catch (error) {
    return {
      statusCode: 500,
      body: JSON.stringify({
        error: '❌ Webhook 设置失败',
        message: error.message
      })
    }
  }
}

serverless.yml 配置

service: telegram-bot

provider:
  name: aws
  runtime: nodejs18.x
  region: us-east-1
  environment:
    BOT_TOKEN: ${env:BOT_TOKEN}
    WEBHOOK_SECRET: ${env:WEBHOOK_SECRET}

functions:
  webhook:
    handler: lambda/handler.webhook
    events:
      - http:
          path: webhook
          method: post
          cors: true

  setWebhook:
    handler: lambda/handler.setWebhook
    events:
      - http:
          path: set-webhook
          method: get
          cors: true

plugins:
  - serverless-offline

AWS Lambda 部署步骤

# 1. 安装 Serverless Framework
npm install -g serverless

# 2. 创建项目
serverless create --template aws-nodejs-typescript --path telegram-bot-lambda
cd telegram-bot-lambda

# 3. 安装依赖
npm install @afjs/telegram-bot-sdk
npm install -D @types/aws-lambda

# 4. 配置 AWS 凭证
serverless config credentials --provider aws --key YOUR_ACCESS_KEY --secret YOUR_SECRET_KEY

# 5. 设置环境变量
export BOT_TOKEN="your_bot_token"
export WEBHOOK_SECRET="your_webhook_secret"

# 6. 部署
serverless deploy

# 7. 设置 Webhook
# 访问部署后的 URL：https://your-api-id.execute-api.region.amazonaws.com/dev/set-webhook

Serverless 环境最佳实践

0. Webhook 设置（必需步骤）

重要： 在 Serverless 环境中，必须先设置 Webhook，Telegram 才会将更新发送到你的函数。

// 验证 Webhook 是否设置成功
async function checkWebhook() {
  const bot = new TelegramBot({ token: process.env.BOT_TOKEN! })

  try {
    const webhookInfo = await bot.getWebhookInfo()
    console.log('Webhook 状态:', {
      url: webhookInfo.url,
      hasCustomCertificate: webhookInfo.has_custom_certificate,
      pendingUpdateCount: webhookInfo.pending_update_count,
      lastErrorDate: webhookInfo.last_error_date,
      lastErrorMessage: webhookInfo.last_error_message
    })

    if (!webhookInfo.url) {
      console.warn('⚠️  Webhook 未设置！机器人无法接收消息。')
    } else {
      console.log('✅ Webhook 已正确设置')
    }
  } catch (error) {
    console.error('检查 Webhook 失败:', error.message)
  }
}

删除 Webhook（切换到轮询模式时）

async function removeWebhook() {
  const bot = new TelegramBot({ token: process.env.BOT_TOKEN! })

  try {
    await bot.deleteWebhook({ drop_pending_updates: true })
    console.log('✅ Webhook 已删除')
  } catch (error) {
    console.error('删除 Webhook 失败:', error.message)
  }
}

1. 超时设置

// Serverless 环境通常有执行时间限制
const bot = new TelegramBot({
  token: process.env.BOT_TOKEN!,
  timeout: 8000, // 设置较短的超时时间
  retries: 1     // 减少重试次数
})

2. 错误处理

async function handleUpdate(update: any, bot: TelegramBot) {
  try {
    // 处理更新逻辑
    if (update.message) {
      await handleMessage(update.message, bot)
    }
  } catch (error) {
    console.error('处理更新失败:', error)
    // 在 Serverless 环境中，不要抛出错误，而是记录并继续
    // 避免导致 Webhook 重试
  }
}

3. 状态管理

// 在 Serverless 环境中，如果需要状态管理，使用外部存储
// 例如：Redis、DynamoDB、Cloudflare KV、PostgreSQL 等
// 注意：内存状态在每次函数调用后都会丢失

// 示例：使用外部存储接口
interface StateStorage {
  get(key: string): Promise<any>
  set(key: string, value: any): Promise<void>
}

class ServerlessBot {
  constructor(private bot: TelegramBot, private storage: StateStorage) {}

  async getUserState(userId: number): Promise<any> {
    return await this.storage.get(`state:${userId}`)
  }

  async setUserState(userId: number, state: any): Promise<void> {
    await this.storage.set(`state:${userId}`, state)
  }
}

4. 批量操作优化

// 在 Serverless 环境中避免长时间运行的批量操作
async function sendBulkMessages(chatIds: number[], message: string, bot: TelegramBot) {
  // 限制批量大小
  const maxBatch = 10
  const batch = chatIds.slice(0, maxBatch)

  const promises = batch.map(chatId =>
    bot.sendMessage({ chat_id: chatId, text: message })
      .catch(error => console.error(`发送失败 ${chatId}:`, error.message))
  )

  await Promise.allSettled(promises)

  // 如果还有更多消息，可以触发另一个函数调用
  if (chatIds.length > maxBatch) {
    // 触发队列或另一个函数处理剩余消息
  }
}

5. 冷启动优化

// 在模块级别初始化机器人实例（Lambda 等环境）
const bot = new TelegramBot({
  token: process.env.BOT_TOKEN!
})

export const handler = async (event: any) => {
  // 使用预初始化的机器人实例
  // 减少冷启动时间
}

环境变量配置

不同 Serverless 平台的环境变量设置：

# Cloudflare Workers
wrangler secret put BOT_TOKEN

# Vercel
vercel env add BOT_TOKEN

# AWS Lambda (使用 Serverless Framework)
# serverless.yml
environment:
  BOT_TOKEN: ${env:BOT_TOKEN}

# Netlify
netlify env:set BOT_TOKEN your_token_here

📝 示例项目

基础机器人

import { TelegramBot, isCommand, getCommandArgs } from '@afjs/telegram-bot-sdk'

const bot = new TelegramBot({
  token: process.env.BOT_TOKEN!
})

async function handleMessage(message) {
  const chatId = message.chat.id

  if (isCommand(message, 'start')) {
    await bot.sendMessage({
      chat_id: chatId,
      text: '欢迎使用机器人！发送 /help 查看帮助。'
    })
  } else if (isCommand(message, 'help')) {
    await bot.sendMessage({
      chat_id: chatId,
      text: '可用命令:\n/start - 开始\n/help - 帮助\n/echo <文本> - 回显'
    })
  } else if (isCommand(message, 'echo')) {
    const args = getCommandArgs(message)
    const text = args.join(' ') || '请提供要回显的文本'
    await bot.sendMessage({
      chat_id: chatId,
      text: `回显: ${text}`
    })
  }
}

// 启动轮询
async function startBot() {
  let offset = 0

  while (true) {
    try {
      const updates = await bot.getUpdates({ offset, timeout: 30 })

      for (const update of updates) {
        if (update.message) {
          await handleMessage(update.message)
        }
        offset = update.update_id + 1
      }
    } catch (error) {
      console.error('轮询错误:', error.message)
      await new Promise(resolve => setTimeout(resolve, 5000))
    }
  }
}

startBot()

完整示例

查看 examples/ 目录中的完整示例：

basic.js - 基础功能演示
advanced.js - 高级功能和完整机器人实现

🔒 安全注意事项

保护 Bot Token

// ❌ 不要在代码中硬编码 Token
const bot = new TelegramBot({ token: '123456:ABC-DEF...' })

// ✅ 使用环境变量
const bot = new TelegramBot({ token: process.env.BOT_TOKEN! })

验证 Webhook Secret（重要）

// 在 Serverless 环境中，必须验证 Webhook Secret
async function handleWebhook(request: Request) {
  // 获取 Telegram 发送的 Secret Token
  const secretHeader = request.headers.get('X-Telegram-Bot-Api-Secret-Token')

  if (!secretHeader || secretHeader !== process.env.WEBHOOK_SECRET) {
    console.warn('Webhook 验证失败 - 可能是恶意请求')
    return new Response('Unauthorized', { status: 401 })
  }

  // 验证通过，处理更新
  const update = await request.json()
  // ... 处理逻辑
}

为什么需要验证 Webhook Secret？

防止恶意用户向你的端点发送虚假更新
确保请求确实来自 Telegram 服务器
避免 DDoS 攻击和资源浪费

生成安全的 Webhook Secret

import { generateRandomString } from '@afjs/telegram-bot-sdk'

// 生成强随机字符串作为 Secret
const webhookSecret = generateRandomString(32)
console.log('Webhook Secret:', webhookSecret)

// 或者使用 Node.js crypto 模块
import { randomBytes } from 'crypto'
const secret = randomBytes(32).toString('hex')

验证用户输入

import { escapeHTML } from '@afjs/telegram-bot-sdk'

// 转义用户输入以防止注入攻击
const safeText = escapeHTML(userInput)

限制访问

const ALLOWED_USERS = [123456789, 987654321]

if (!ALLOWED_USERS.includes(message.from.id)) {
  return // 拒绝未授权用户
}

IP 白名单（可选）

// Telegram Webhook IP 范围（可选验证）
const TELEGRAM_IP_RANGES = [
  '149.154.160.0/20',
  '91.108.4.0/22'
]

function isValidTelegramIP(ip: string): boolean {
  // 实现 IP 范围检查逻辑
  // 注意：Telegram 的 IP 可能会变化，不建议严格依赖
  return true
}

📊 性能优化

请求限制

Telegram Bot API 有以下限制：

每秒最多 30 条消息
每分钟最多 20 条消息到同一个群组
每秒最多 1 条消息到同一个用户

import { delay } from '@afjs/telegram-bot-sdk'

// 添加延迟避免触发限制
for (const chatId of chatIds) {
  await bot.sendMessage({ chat_id: chatId, text: 'Hello' })
  await delay(100) // 延迟100ms
}

批量处理

import { chunk } from '@afjs/telegram-bot-sdk'

// 分批处理大量操作
const batches = chunk(largeArray, 10)
for (const batch of batches) {
  await Promise.allSettled(
    batch.map(item => processItem(item))
  )
}

🐛 故障排除

常见错误

"Unauthorized" 错误
- 检查 Bot Token 是否正确
- 确保 Token 没有过期
"Chat not found" 错误
- 确保聊天 ID 正确
- 用户可能删除了与机器人的对话
"Message is too long" 错误
- 消息长度不能超过 4096 字符
- 使用 truncateText 函数截断长文本
"Too Many Requests" 错误
- 触发了 API 限制
- 添加延迟或减少请求频率

调试技巧

// 启用详细日志
const bot = new TelegramBot({
  token: process.env.BOT_TOKEN!,
  timeout: 30000
})

// 添加错误处理
bot.sendMessage({ chat_id: chatId, text: 'test' })
  .then(result => console.log('成功:', result))
  .catch(error => console.error('失败:', error.message))

📄 许可证

本项目采用 MIT 许可证。