Telegram 开发实战：从零搭建一个智能群管机器人

Telegram 作为全球最受欢迎的开源即时通讯平台之一，其开放的 Bot API 让开发者可以轻松构建各种自动化工具。这篇文章分享搭建一个智能群管机器人的完整流程，覆盖 API 接入、消息处理、权限管理和实用功能实现。

一、准备工作：创建你的第一个 Bot

第一步：找 BotFather 要一个身份

在 Telegram 搜索 @BotFather，发送 /newbot 命令，按提示给你的机器人起名字和用户名。完成后你会拿到一串 Token，形如：

1234567890:ABCdefGHIjklMNOpqrsTUVwxyz

这个 Token 是你的机器人身份证，保存好别泄露。

第二步：配置 Webhook 或 Polling

两种接收消息的方式：

Webhook：Telegram 主动把消息推送到你的服务器（需要 HTTPS）
Polling：你的程序定时去 Telegram 服务器拉取消息（开发调试更方便）

开发阶段建议用 Polling，上线后再切 Webhook。

示例代码（Python + python-telegram-bot 库）：

from telegram import Update
from telegram.ext import Application, CommandHandler, MessageHandler, filters, ContextTypes

BOT_TOKEN = “你的Token”

async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
await update.message.reply_text(“你好！我是群管机器人，发送 /help 查看功能。”)

async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
help_text = “””
可用命令：
/start – 启动机器人
/help – 显示帮助
/kick – 踢出用户（管理员专用）
/ban – 封禁用户（管理员专用）
/warn – 警告用户
/stats – 查看群统计
“””
await update.message.reply_text(help_text)

app = Application.builder().token(BOT_TOKEN).build()
app.add_handler(CommandHandler(“start”, start))
app.add_handler(CommandHandler(“help”, help_command))
app.run_polling()

运行后，你的机器人就能响应 /start 和 /help 命令了。

二、核心功能实现

1. 自动欢迎新成员

群组里有新人加入时自动打招呼，是群管机器人的标配功能。

async def welcome(update: Update, context: ContextTypes.DEFAULT_TYPE):
for member in update.message.new_chat_members:
if not member.is_bot:
welcome_msg = f”欢迎 {member.full_name} 加入本群！请阅读置顶消息了解群规。”
await update.message.reply_text(welcome_msg)

app.add_handler(MessageHandler(filters.StatusUpdate.NEW_CHAT_MEMBERS, welcome))

2. 关键词自动回复

检测群消息中的关键词，触发自动回复或删除违规内容。

async def keyword_handler(update: Update, context: ContextTypes.DEFAULT_TYPE):
text = update.message.text.lower()
banned_words = [“广告”, “加微信”, “兼职”]

for word in banned_words:
if word in text:
await update.message.delete()
await update.message.reply_text(f”检测到违规内容，已自动删除。”)
return

app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, keyword_handler))

3. 警告系统

三振出局机制：用户收到 3 次警告后自动禁言。

warning_count = {}

async def warn(update: Update, context: ContextTypes.DEFAULT_TYPE):
if not update.message.reply_to_message:
await update.message.reply_text(“请回复要警告的消息使用此命令。”)
return

user_id = update.message.reply_to_message.from_user.id
user_name = update.message.reply_to_message.from_user.full_name

warning_count[user_id] = warning_count.get(user_id, 0) + 1
count = warning_count[user_id]

await update.message.reply_text(f”已警告 {user_name}（第{count}次）”)

if count >= 3:
chat_id = update.effective_chat.id
await context.bot.restrict_chat_member(
chat_id,
user_id,
permissions={“can_send_messages”: False}
)
await update.message.reply_text(f”{user_name} 因累计3次警告已被禁言。”)

app.add_handler(CommandHandler(“warn”, warn))

三、权限管理：谁能用管理员命令

踢人、封禁等敏感操作必须限制权限，否则任何群成员都能乱来。

async def admin_only(update: Update, context: ContextTypes.DEFAULT_TYPE):
user_id = update.message.from_user.id
chat_id = update.effective_chat.id

admins = await context.bot.get_chat_administrators(chat_id)
admin_ids = [admin.user.id for admin in admins]

if user_id not in admin_ids:
await update.message.reply_text(“此命令仅限管理员使用。”)
return False
return True

async def kick(update: Update, context: ContextTypes.DEFAULT_TYPE):
if not await admin_only(update, context):
return

if update.message.reply_to_message:
user_id = update.message.reply_to_message.from_user.id
chat_id = update.effective_chat.id
await context.bot.ban_chat_member(chat_id, user_id)
await update.message.reply_text(“已踢出该用户。”)

app.add_handler(CommandHandler(“kick”, kick))

四、进阶功能：群数据统计

统计群消息数量、活跃用户、最活跃时段等，帮助管理员了解群动态。

import json
from datetime import datetime
from collections import defaultdict

stats = {
“total_messages”: 0,
“users”: defaultdict(int),
“hourly”: defaultdict(int)
}

async def track_stats(update: Update, context: ContextTypes.DEFAULT_TYPE):
stats[“total_messages”] += 1
user_id = update.message.from_user.id
stats[“users”][user_id] += 1
hour = datetime.now().hour
stats[“hourly”][hour] += 1

async def show_stats(update: Update, context: ContextTypes.DEFAULT_TYPE):
total = stats[“total_messages”]
unique_users = len(stats[“users”])
peak_hour = max(stats[“hourly”].items(), key=lambda x: x[1])[0]

msg = f”“”
群数据统计：
总消息数：{total}
活跃用户：{unique_users}
最活跃时段：{peak_hour}点
“””
await update.message.reply_text(msg)

app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, track_stats))
app.add_handler(CommandHandler(“stats”, show_stats))

五、部署上线

本地开发用 Polling，正式上线建议切 Webhook。

Webhook 部署步骤：

1. 准备一个有 HTTPS 的服务器（可用 Nginx + Let’s Encrypt）
2. 设置 Webhook URL

import requests
WEBHOOK_URL = “https://yourdomain.com/webhook”
requests.get(f”https://api.telegram.org/bot{BOT_TOKEN}/setWebhook?url={WEBHOOK_URL}“)

3. 在服务器上运行 Flask 或 FastAPI 接收推送

from flask import Flask, request
app_flask = Flask(name)

@app_flask.route(“/webhook”, methods=[“POST”])
def webhook():
update = Update.de_json(request.get_json(), app.bot)
app.update_queue.put(update)
return “ok”

app_flask.run(port=8443)

六、常见问题与踩坑

问题一：机器人没有权限操作消息

解决方案：确保机器人在群里是管理员，且拥有删除消息、封禁用户等权限。

问题二：Webhook 设置失败

解决方案：检查 HTTPS 证书是否有效，URL 必须是 443 端口。

问题三：中文乱码

解决方案：确保代码文件保存为 UTF-8 编码，Python 文件开头加：

–– coding: utf-8 ––

问题四：机器人响应慢

解决方案：用异步处理耗时操作（如调用外部 API），不要阻塞主消息循环。

七、总结

搭建一个 Telegram 群管机器人，核心流程：

1. 找 BotFather 创建机器人获取 Token
2. 用 python-telegram-bot 库接入 API
3. 实现消息处理、权限控制、功能模块
4. 本地测试用 Polling，上线切 Webhook
5. 持续迭代功能，监控运行状态

Telegram Bot API 文档完善、社区活跃，是练手机器人开发的绝佳平台。以上代码可以直接运行，你也可以在此基础上扩展更多功能——天气查询、汇率转换、AI 对话集成，想象力是唯一的边界。