Python中如何实现电报机器人功能?
在国内的服务器上用 python 写了一个电报机器人的脚本,使用 clash 代理来与电报通讯,但是隔几个小时就会出现一次网络问题,而且一出现机器人就会挂掉,即使后来网络恢复,机器人也响应不了电报的消息了。
我目前的解决方法是捕获到异常就退出程序,然后 docker 再把它拉起来。。
请问有没有更好的解决方法?
使用的 python 库是 python-telegram-bot
用 run_polling 的方法启动的程序。
Python中如何实现电报机器人功能?
长时间工作应当使用 webhook 来获取消息
在Python中实现电报机器人,主要依赖python-telegram-bot库。以下是完整实现步骤和代码示例:
1. 安装必要库
pip install python-telegram-bot
2. 获取Bot Token
- 在Telegram中搜索
@BotFather - 发送
/newbot创建新机器人 - 保存生成的API Token
3. 基础机器人实现代码
from telegram import Update
from telegram.ext import Application, CommandHandler, MessageHandler, filters, ContextTypes
# 替换为你的Bot Token
BOT_TOKEN = "YOUR_BOT_TOKEN_HERE"
# 处理/start命令
async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
await update.message.reply_text("你好!我是Python电报机器人。")
# 处理/help命令
async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
help_text = """
可用命令:
/start - 开始使用
/help - 显示帮助信息
/echo <文本> - 回显消息
"""
await update.message.reply_text(help_text)
# 处理/echo命令
async def echo(update: Update, context: ContextTypes.DEFAULT_TYPE):
if context.args:
text = ' '.join(context.args)
await update.message.reply_text(f"你说了:{text}")
else:
await update.message.reply_text("请在/echo后输入文本")
# 处理普通文本消息
async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
text = update.message.text
await update.message.reply_text(f"收到消息:{text}")
# 错误处理
async def error(update: Update, context: ContextTypes.DEFAULT_TYPE):
print(f"更新 {update} 导致错误 {context.error}")
def main():
# 创建应用
application = Application.builder().token(BOT_TOKEN).build()
# 注册命令处理器
application.add_handler(CommandHandler("start", start))
application.add_handler(CommandHandler("help", help_command))
application.add_handler(CommandHandler("echo", echo))
# 注册消息处理器(处理非命令文本)
application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message))
# 注册错误处理器
application.add_error_handler(error)
# 启动机器人
print("机器人启动中...")
application.run_polling(allowed_updates=Update.ALL_TYPES)
if __name__ == "__main__":
main()
4. 高级功能扩展
键盘按钮示例:
from telegram import ReplyKeyboardMarkup
async def show_keyboard(update: Update, context: ContextTypes.DEFAULT_TYPE):
keyboard = [["选项1", "选项2"], ["选项3", "选项4"]]
reply_markup = ReplyKeyboardMarkup(keyboard, resize_keyboard=True)
await update.message.reply_text("请选择:", reply_markup=reply_markup)
定时任务示例:
from telegram.ext import JobQueue
async def alarm(context: ContextTypes.DEFAULT_TYPE):
await context.bot.send_message(chat_id=context.job.chat_id, text="定时提醒!")
async def set_timer(update: Update, context: ContextTypes.DEFAULT_TYPE):
chat_id = update.message.chat_id
context.job_queue.run_once(alarm, 10, chat_id=chat_id)
await update.message.reply_text("10秒后将收到提醒")
5. 部署建议
- 本地测试:直接运行上述代码
- 服务器部署:使用
nohup或systemd保持后台运行 - 云部署:考虑使用AWS Lambda、Heroku或PythonAnywhere
- Webhook模式(生产环境推荐):
# 设置Webhook(需要HTTPS域名)
application.run_webhook(
listen="0.0.0.0",
port=8443,
webhook_url="https://yourdomain.com/webhook"
)
关键注意事项:
- Token安全:不要将Token提交到版本控制系统
- 错误处理:确保所有异步函数都有异常处理
- 速率限制:Telegram API有调用频率限制
- 日志记录:建议添加日志记录功能
调试建议:
- 使用
print()语句输出调试信息 - 查看
python-telegram-bot官方文档的示例 - 使用
@BotFather的/setdescription设置机器人描述
总结建议:从基础命令机器人开始,逐步添加键盘、定时任务等高级功能。
国外便宜年付的服务器多如牛毛,你都要用电报了,国外 vps 不标配吗?
用 webhook ,挂到 cloudflare 上可以.
用的 telethon 库,没遇到这个问题
用的库或者编写的连接没有加入断连重连的逻辑吧。
也可以写一个 systemd unit 自动重启
我觉得你现在这个做法挺好的
该死就死,别挣扎,没用的
诈尸不是自己的责任,得依赖外部法师(容器/systemd/k8s )
以我的经验:即使放在墙外,偶尔也会遇到网络不通的情况、返回 json 不 ok 的情况等等
其实可以考虑 webhook 模式,处理单个 update 之后就退出,即使失败不会影响处理其他 update
要部署在国内,用 webhook 得有一个备案好的域名+ssl 证书吧
还是搞个外面的 vps 感觉简单些
阿里的服务器折腾一天,各种连接拒绝,换成外面的 vps 一会就搞好了:)
框架选的比较有品位 ~~~ 赞一个
ptb 的轮询模式是不适合生产环境的,还是 webhook 吧,配合 fastapi 美滋滋。
至于主机,国外便宜小鸡多的是,随便一个都可以,推荐荷兰的主机。
一定用国内阿里云,那就用 cloudflare tunnel + webhook 好了。当然,自己部署 tdlib 的 telegram bot api server + clash 也可以。不过这玩意翻墙环境下不太好折腾。
这玩意你部署在国内就是个错误,你还想着在错误上做备案之类的准备?

