ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

用Python开发Telegram Bot:从入门到实战完整指南

用Python开发Telegram Bot:从入门到实战完整指南 你有没有被群里重复的问题烦到过我动手写的第一个 Telegram Bot就是用 Python 做的需求特别简单别人发一句话机器人自动回一段固定话术。后来我给同事的团队群也接了一个用来收集每日站会链接、定时推送待办清单确实省了不少事。今天就把这套“用 Python 写一个简单的 Telegram Bot电报机器人”的完整过程写下来从选方案、建机器人、写第一行代码到日志排错和长期运行一次说清楚。这个项目适合三类人刚学完 Python 基础想找个“看得见反应”的练手题的初学者需要给团队或自己搞通知机器人的开发者以及对 Bot API 不熟、想快速了解整体开发链路的人。读完你可以得到一个能自动回复消息、响应命令、定时推送、带简单按钮菜单的机器人并且理解它背后是怎么运作的。1. 先搞清楚 Telegram Bot 能做什么再选对技术方案1.1 这个机器人项目到底解决了什么问题Telegram Bot 本质上不是“一个账号”而是一段跑在你自己服务器上的程序。它通过官方提供的 Bot API 来收发消息做的事情无非三类接收用户消息、处理逻辑、回复消息。你可以把它理解为“一个挂在消息流里的自动应答服务员”它不睡觉、不摸鱼、看到消息就按规定动作干活。我最初做这个项目是想解决一个非常琐碎的场景群里总有人问“日报模板在哪”“测试环境地址是什么”“新同事入职要看哪些文档”。这些问题的答案几乎是固定的纯手工回复又无聊又容易漏。于是我用 Python 写了个机器人把常见问答做成命令和关键词回复谁触发它就回答谁。这个思路可以扩展到很多场景自动欢迎新人、定时发送每日任务、监控服务器异常后推送到群、接收外部 webhook 然后格式化成消息发出。换句话说Telegram Bot 可以当作你的消息中枢把散落的通知和指令统一进来。1.2 三种实现方式怎么选很多人第一次接触 Bot会先搜到各种各样的库容易看花眼。我按自己的经验把主流方式分成三类。第一种直接用 requests 库调 HTTP 接口。Bot API 的底层就是一组 HTTP 接口你只要发 GET/POST 请求传入 token 和参数就能收发消息。优点是没依赖、灵活缺点是所有细节都要自己处理长轮询的 offset 要自己维护、断线重连要自己写、消息解析要自己拼 JSON。适合学习原理不适合快速产出稳定程序。第二种用 python-telegram-bot 库。这是我推荐新手入门的方案。它是 Telegram Bot API 最流行的 Python 封装之一文档成熟、社区活跃、示例多。官方长期维护消息处理、命令注册、按钮回调、会话状态这些高频功能都已经封装好了你用的时候不需要重复造轮子。第三种用 aiogram 库。它也是异步框架性能不错扩展机制比较灵活在需要高并发、复杂交互的项目里表现很好。缺点是文档和示例相对概念化新手理解起来需要一点时间。我这次踩过的坑是一开始图省事直接用 requests 写代码倒是简短但一遇到网络抖动长轮询就断重启以后还会重复消费消息。后来切到 python-telegram-bot断线重连和 offset 管理都由库内部处理了程序稳定了很多。对大多数个人项目和中小型工具来说python-telegram-bot 是性价比最高的选择。1.3 同步还是异步选 python-telegram-bot v20 的理由python-telegram-bot 从 v20 开始核心代码改成了纯异步实现基于 asyncio这是很多人刚接触时会困惑的地方“我明明只是写个自动回复为什么到处是 async/await”原因在于一个机器人可能要同时面对多个用户如果一个用户触发的操作比较慢比如调一个第三方接口等 5 秒而你的代码是同步阻塞的那其他用户的消息就会卡住。异步模型让程序在等待网络响应的时候可以去处理别的请求一个进程就能撑起大量并发消息。对于新手你不需要深入理解事件循环的每个细节只需要记住规则处理函数定义成 async def耗时的调用前面加 await然后交给库的事件循环去跑就行。如果你写的函数里有 time.sleep一定要换成 await asyncio.sleep否则会阻塞整个 bot。我第一次写的时候用 time.sleep 做了个“延迟 3 秒再回复”的演示结果所有消息都跟着卡 3 秒后来才意识到问题出在这。2. 搭建环境与创建机器人的完整准备2.1 Python 环境和依赖安装建议使用 Python 3.9 以上版本python-telegram-bot v20 的最新版本对 3.9 以下的兼容性已经变差。检查自己的版本python3 --version如果还没有装 Python或者环境变量没配好先去把 Python 装好、加到 PATH 里再回来继续。项目依赖只需要一个库安装命令如下pip install python-telegram-bot如果你在安装时网速比较慢可以用国内 PyPI 镜像加速比如清华源pip install python-telegram-bot -i https://pypi.tuna.tsinghua.edu.cn/simple这里有一个经验建议在项目目录里建一个虚拟环境不要直接往全局环境里装依赖。虚拟环境的好处是隔离不同项目的依赖版本后续部署到服务器时也可以直接打包复现。创建一个虚拟环境python3 -m venv venv source venv/bin/activate # Windows 用户执行 venv\Scripts\activate激活之后再执行 pip install库就装进当前项目的虚拟环境里了。2.2 用 BotFather 创建机器人拿到 token写代码之前必须先拿到一个 bot token这是机器人的唯一身份凭证。流程很简单但因为是英文界面新手容易卡住我一步步说。在 Telegram 里搜索 BotFather打开对话框输入/newbot它会让你回答两个问题第一个是机器人展示名称display name比如 MyTestBot第二个是机器人用户名username必须以bot结尾比如my_test_bot。两个都填好后BotFather 会返回一条消息里面有一行类似123456789:AAHxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx的字符串这就是 token。拿到 token 后有两条铁律不要把它写进公开仓库不要把它发给任何人。token 相当于你家的钥匙任何人拿到它都能控制你的机器人。我用过一次教训是把 token 临时写在代码里后忘记删除结果仓库被扫描工具抓到机器人被别人接管最后只能找 BotFather 重新生成 token。所以从第一次写代码开始建议就用环境变量或本地配置文件来保存 token。bot 的原理我也顺手解释一下所有消息收发都是通过 Bot API 服务器中转的你的程序启动后会主动向 Bot API 服务器发起请求询问“有没有新的更新给我”服务器返回消息 JSON。如果你的程序没启动消息会暂时存在服务器侧但最多保留一段时间重启后再拉起就行。2.3 理解消息对象的结构与 chat_id有了 token 只是第一步你还得知道程序和用户是怎么“认出来”彼此的。每次用户给你的 bot 发消息Bot API 会返回一个 JSON 对象核心字段包括message.message_id消息唯一 ID。message.chat.id会话 ID用于定位“这条消息来自哪个对话”。message.from.id发送者的用户 ID。message.text消息文本内容。message.chat.type会话类型可能是 private私聊、group群组或 supergroup超级群组。这里最关键的是chat.id。它决定了你要把消息发到哪个会话。私聊时chat.id就是用户的 ID群聊时chat.id是群组的 ID。你要主动向用户推送消息必须提前拿到对方所在会话的chat.id机器人才能定位到“该发给谁”。我在刚开始做定时推送时就是卡在这个环节机器人能收到消息、能回复但我想“每天主动给特定的人发消息”却不知道要把消息发到哪。后来我在代码里写了一个/get_chat_id命令把update.message.chat.id打印到日志里让用户先来私聊一次拿到 chat_id再写进配置。之后所有主动推送都用这个 ID。3. 第一个能跑的机器人从 echo bot 到交互功能3.1 echo bot 完整代码与逐行拆解学习任何 Bot 项目最经典的第一课都是 echo bot用户发来什么机器人就回什么。这个程序麻雀虽小五脏俱全它包含了连接、注册处理器、启动轮询三个必要环节。先看完整代码import logging from telegram import Update from telegram.ext import Application, CommandHandler, MessageHandler, filters TOKEN 你的token logging.basicConfig( format%(asctime)s - %(name)s - %(levelname)s - %(message)s, levellogging.INFO ) async def start(update: Update, context): await update.message.reply_text(你好我是你的机器人。直接给我发消息试试。) async def echo(update: Update, context): user_text update.message.text await update.message.reply_text(f你刚才说的是{user_text}) def main(): app Application.builder().token(TOKEN).build() app.add_handler(CommandHandler(start, start)) app.add_handler(MessageHandler(filters.TEXT ~filters.COMMAND, echo)) print(bot 已启动按 CtrlC 停止) app.run_polling() if __name__ __main__: main()逐行来看Application.builder().token(TOKEN).build()这行是构建一个应用实例把我们的 token 绑定进去。它是整个机器人运行的核心容器负责管理所有处理器和网络请求。CommandHandler(start, start)的作用是注册命令处理器意思是当用户发送/start时调用start函数。Telegram 客户端通常会在用户第一次打开 bot 时自动发送/start所以这个命令适合做欢迎语。MessageHandler(filters.TEXT ~filters.COMMAND, echo)则是注册消息处理器它的条件是只处理文本消息并且排除/开头的命令消息。这样普通聊天内容都会进入echo函数。注意到两个处理函数都是async def说明它们是异步函数。await update.message.reply_text(...)表示“发送消息”这个操作耗时可能较长所以等待结果的同时让出控制权。运行这个脚本看到控制台输出“bot 已启动”后你就可以在 Telegram 里找到你的 bot私聊它试试。发一句“你好”它会回你“你刚才说的是你好”。3.2 让 bot 听懂命令CommandHandler 与参数传递echo bot 只是热身。实际项目里更多时候我们想让它执行特定指令比如/help、/weather、/tips。CommandHandler可以注册一个命令并绑定一个函数。比如加一个打招呼并带上名字的功能用户输入/greet 小明机器人回复“你好小明”async def greet(update: Update, context): args context.args if args: name .join(args) await update.message.reply_text(f你好{name}) else: await update.message.reply_text(用法/greet 你的名字) app.add_handler(CommandHandler(greet, greet))这里的context.args是一个列表Telegram 客户端会把/greet后面的所有内容按空格切分作为参数传进来。比如输入/greet 小明 同学args就是[小明, 同学]我用 .join(args)拼回完整的问候对象。有一点值得注意命令处理函数里的参数没有固定数量要求完全取决于你的业务逻辑。如果你想做一个查询类机器人完全可以定义成/query city北京 date2025-01-01然后在函数里解析context.args。我在实际开发中还有一个很实用的习惯给每个命令都加上context.args的前置校验。不要假设用户一定会按正确的格式输入。用户是懒惰的也是富有创造力的他们可能只输入/greet不按任何参数如果你的代码没有做判空程序就可能在if args这行直接报错。虽然 Telegram 机器人出错会自动捕获异常不至于让进程崩溃但错误日志一多真正的问题就被淹没了。3.3 主动推送与定时通知的写法消息回复是被动响应但在很多场景里我们需要机器人主动发起消息每天早上推送待办清单每分钟检查一次服务状态有异常时立刻告警。这时就要用到主动发送 API。在你的处理函数内部可以通过context.bot.send_message(chat_id..., text...)主动发送消息。但如果是定时任务它不一定在某个 Handler 里你需要单独拿到 bot 实例才能发消息。一个最简单可行的定时任务写法是用 asyncio 创建后台任务循环等待一定时间然后发送消息。示例import asyncio from datetime import datetime async def scheduled_task(app: Application): while True: now datetime.now() if now.hour 9 and now.minute 0: await app.bot.send_message( chat_id你的chat_id, text早上好记得看一眼今天的任务列表。 ) await asyncio.sleep(60) await asyncio.sleep(30)然后把scheduled_task作为后台任务挂到事件循环里在main()中启动import asyncio def main(): app Application.builder().token(TOKEN).build() app.add_handler(CommandHandler(start, start)) # ... 其他 handler asyncio.get_event_loop().create_task(scheduled_task(app)) app.run_polling()这里有个细节要特别注意避免在整点这个判断条件上重复触发。我在上面加了await asyncio.sleep(60)意思是整点发送一次后这一分钟内不再重复执行。当然这只是一种朴素的定时方案生产级更推荐用apscheduler这类专门的调度库来管理 cron 任务。如果你只是自用上面的写法已经够用。3.4 用内联键盘做简单的菜单交互文本回复能覆盖大多数场景但如果想让用户“点按钮”而不是“敲命令”就需要内联键盘InlineKeyboardMarkup。它是在消息下方渲染的一排按钮点击后不需要在输入框里输入任何内容体验比命令更友好。实现分两步第一步在回复消息时附加键盘第二步注册一个回调处理器接收按钮点击产生的事件。from telegram import InlineKeyboardButton, InlineKeyboardMarkup, Update from telegram.ext import CallbackQueryHandler async def show_menu(update: Update, context): keyboard [ [InlineKeyboardButton(查看今天的任务, callback_datatodo)], [InlineKeyboardButton(查看系统状态, callback_datastatus)], ] reply_markup InlineKeyboardMarkup(keyboard) await update.message.reply_text(请选择一个功能, reply_markupreply_markup) async def button_handler(update: Update, context): query update.callback_query await query.answer() if query.data todo: await query.edit_message_text(今天要完成写技术博客、跑一遍脚本、整理日志。) elif query.data status: await query.edit_message_text(系统状态正常最近一次监控时间刚刚。) app.add_handler(CommandHandler(menu, show_menu)) app.add_handler(CallbackQueryHandler(button_handler))这个交互流程的解释用户发送/menu后show_menu返回一条带有两个按钮的消息用户点击按钮时Telegram 客户端不会像普通消息那样发送一条文本而是发送一个callback_query事件button_handler收到事件后根据query.data判断用户点了哪个按钮再用query.edit_message_text把原消息改成对应的结果文字。await query.answer()也很重要它用来响应 Telegram 的回调让客户端停止按钮上的“加载中”状态。忘记调用它按钮点击可能看起来没有反应。内联键盘的嵌套列表非常灵活每个内层列表代表一行按钮。按钮数量多的时候可以做成多行多列甚至用callback_data传递复杂的参数比如order_123然后在回调函数里解析出订单号。4. 日志、异常与部署让 bot 真正稳定长期运行4.1 日志是排查问题的第一生产力写玩具 bot 时很多人不重视日志真出了问题就只能瞎猜。等你开始做定时推送、接入外部 API日志就是定位问题的第一手段。在程序开头加上logging.basicConfig让运行过程输出有时间戳、模块名、日志级别的内容。这样处理消息出错、网络请求失败、命令触发成功都能在终端看到痕迹。import logging logging.basicConfig( format%(asctime)s - %(name)s - %(levelname)s - %(message)s, levellogging.INFO )python-telegram-bot 内部本身也会输出日志所以我建议至少把日志级别设为INFO。调试时调到DEBUG可以查看更多细节比如每次请求的 payload。我自己的习惯是在关键 Handler 开头加一行logging.info(收到命令 /start来自 %s, update.effective_user.id)这样以后追溯“某条消息为什么没处理成功”能先判断是不是消息压根没进来。4.2 常见异常与重试策略网络环境再稳定也有抖动的时候。python-telegram-bot 底层会自动处理一部分超时重试但你自己的业务代码里如果调用了外部接口还是会出现异常。最常见的几个异常类型包括telegram.error.TimedOut请求超时。telegram.error.NetworkError网络不可达。telegram.error.RetryAfter被限流需要等待指定秒数。telegram.error.ChatMigrated群组升级为超级群组chat_id 变化。我的建议是在每个处理函数内部用 try/except 包住可能出错的部分尤其是调用第三方 API 的地方。一个参考结构async def get_weather(update: Update, context): try: # 请求天气 API data await fetch_weather() await update.message.reply_text(f当前温度{data[temp]}) except Exception as e: logging.error(获取天气失败%s, e) await update.message.reply_text(抱歉天气服务暂时不可用。)这里有一个取舍只在必要的地方捕获异常不要在最外层把所有异常无声吞掉。否则程序看似正常运行但具体错误连日志都没有排错会非常痛苦。4.3 两种常用的常驻运行方式本地写完、测试通过以后你不可能一直开着终端让它跑。个人电脑一关机bot 就下线了。要让它长期稳定地运行通常放在一台长期开机的服务器或开发板上。最简单的启动方式是nohup把进程放到后台即使你注销登录它也不会立刻退出nohup python3 bot.py bot.log 21 bot.log 21是把输出写进日志文件方便你后续观察运行情况。缺点是这样启动的进程没有自动重启机制进程挂了就挂了。更规范的方式是用 systemd 来托管。写一个服务文件比如/etc/systemd/system/my-telegram-bot.service[Unit] DescriptionMy Telegram Bot Afternetwork.target [Service] Useryour_user WorkingDirectory/path/to/your/project ExecStart/path/to/your/venv/bin/python bot.py Restartalways RestartSec5 EnvironmentBOT_TOKEN你的token [Install] WantedBymulti-user.target然后执行systemctl daemon-reload systemctl enable my-telegram-bot systemctl start my-telegram-bot用 systemd 的好处是进程意外退出后会自动重启开机后会自启日志可以通过journalctl -u my-telegram-bot查看。EnvironmentBOT_TOKEN...这一行很关键它让 token 不硬编码在代码里而是通过环境变量注入。我自己的项目最后就是这样的部署结构代码里用os.getenv(BOT_TOKEN)读 token服务器上通过 systemd 注入环境变量。这样即使代码传到公开仓库token 也不会泄露。4.4 安全与规范token 保护、用户隐私和频率控制机器人上线之后除了功能正确还要注意几个容易被忽视的点。token 的安全是第一位的。建议做到三件事不要把 token 写进代码仓库使用环境变量或配置文件单独管理如果发现 token 泄露立刻去 BotFather 执行/revoke重新生成同时更新代码里的 token。用户隐私方面要注意不要随意记录或转发用户的完整消息内容尤其涉及到私人对话。如果你的功能需要记录用户输入只保存必要的字段并设置合理的保存周期。频率控制也是保命的设计。如果 bot 被拉进一个很活跃的大群用户频繁发送消息而你的每次回复都调用外部 API很容易把自己搞到限流甚至封号。我一般会在关键 Handler 里做一个简单的限速判断比如同一个 chat_id 在 1 秒内最多触发一次回复。简单实现可以用字典记录chat_id - 上次时间戳超过频率则直接忽略。5. 常见问题速查与经验总结5.1 最容易遇到的 4 个错误与排查我在做这个项目的过程中从零开始踩了不少坑。把这些高频问题整理成表格方便对照排查。错误信息可能原因处理办法401 Unauthorizedtoken 写错或者 token 失效检查 token 是否完整、是否有多余空格到 BotFather 重新获取409 Conflict有多个进程同时在执行run_polling()抢同一个 bot 的更新流杀掉旧进程确保只有一个 bot 实例在运行400 Bad Request: chat not found尝试向一个不存在的chat_id发消息确认 chat_id 是否正确如果是群组确认 bot 还在群里429 Too Many Requests请求过于频繁被限流降低发送频率按响应里的retry_after等待后再发其中409 Conflict是我自己遇到最多的情况。原因往往是你 CtrlC 退出终端后旧进程没有完全结束又启动了一个新进程两个进程同时在请求同一个 bot 的更新流Telegram API 就会拒绝。排查方法很简单ps aux | grep python看看是否有残留进程有就杀掉再重新启动。5.2 被误认为“bot 失效”的几个坑有时候 bot 看起来“没反应”其实是程序在运行但没有正确处理对应情况。这类问题最迷惑人。一个典型场景是用户给 bot 发图片、表情包或贴纸而你只注册了文本消息处理器导致 bot 对非文本消息一律没反应。解决方案是需要处理非文本消息时在MessageHandler里改用filters.ALL或者分别注册filters.PHOTO、filters.STICKER等具体类型。另一个场景是群组里 bot 不响应命令。Telegram 默认群组中的 bot 需要被设置成能够读取消息如果 bot 在群组里被设为只读或者群主限制了 bot 权限它就无法收到普通消息和命令。建议在群组管理界面确认 bot 有“读取消息”的权限。还有一个很隐蔽的坑如果你同时配置了 webhook 和 polling机器人会一直报错。这两种模式是互斥的运行 polling 之前要确保没有遗留的 webhook 配置。查当前状态可以用官方 API 的deleteWebhook方法或者在 BotFather 里确认。我在切换开发模式时踩过这个坑后来写了一个小工具来统一管理 webhook 的注册和删除避免每次手工操作。5.3 踩过坑之后的小建议写 bot 项目最重要的是“先跑起来再优化”。不要一上来就想着把所有功能都塞进去。从 echo bot 开始逐步加上命令、键盘、定时任务每一步都是可以独立验证的出问题也容易定位。测试时不要只在私聊里测尽早拉一个只有你和 bot 的小群来测。群聊场景下chat.id类型、权限、命令解码这些坑会提前暴露比上线后被用户发现要好得多。日志级别设置成 INFO 之后千万不要无脑开 DEBUG 就丢在生产里。DEBUG 日志量很大日志文件会快速增长反而淹没有用的报错。生产环境用 INFO调试时才开 DEBUG。我给这么多机器人项目做过迭代最后总结出一个经验把 bot 当成一个“输入-处理-输出”的管道来设计不要让它承载过于复杂的业务逻辑。复杂的业务放到后端的普通服务里bot 只负责接收消息、调接口、回结果。这样 bot 代码简洁、稳定后续替换框架或者迁移到别的平台成本也低得多。我个人在实际操作中的体会是写一个简单的 bot 不难真正花时间的往往是想清楚“它到底要解决什么问题”。你把需求明确成“收到 X 就回复 Y定时执行 Z”代码其实就是把这些规则一条条翻译成 handler。先把最小闭环跑通再根据真实使用反馈去加功能比一开始设计一个大而全的系统靠谱得多。
返回列表