Python中如何实现电报机器人功能?

在国内的服务器上用 python 写了一个电报机器人的脚本,使用 clash 代理来与电报通讯,但是隔几个小时就会出现一次网络问题,而且一出现机器人就会挂掉,即使后来网络恢复,机器人也响应不了电报的消息了。
我目前的解决方法是捕获到异常就退出程序,然后 docker 再把它拉起来。。

请问有没有更好的解决方法?

使用的 python 库是 python-telegram-bot
用 run_polling 的方法启动的程序。
Python中如何实现电报机器人功能?

13 回复

长时间工作应当使用 webhook 来获取消息


在Python中实现电报机器人,主要依赖python-telegram-bot库。以下是完整实现步骤和代码示例:

1. 安装必要库

pip install python-telegram-bot

2. 获取Bot Token

  1. 在Telegram中搜索@BotFather
  2. 发送/newbot创建新机器人
  3. 保存生成的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. 部署建议

  • 本地测试:直接运行上述代码
  • 服务器部署:使用nohupsystemd保持后台运行
  • 云部署:考虑使用AWS Lambda、Heroku或PythonAnywhere
  • Webhook模式(生产环境推荐):
# 设置Webhook(需要HTTPS域名)
application.run_webhook(
    listen="0.0.0.0",
    port=8443,
    webhook_url="https://yourdomain.com/webhook"
)

关键注意事项:

  1. Token安全:不要将Token提交到版本控制系统
  2. 错误处理:确保所有异步函数都有异常处理
  3. 速率限制:Telegram API有调用频率限制
  4. 日志记录:建议添加日志记录功能

调试建议:

  • 使用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 也可以。不过这玩意翻墙环境下不太好折腾。

这玩意你部署在国内就是个错误,你还想着在错误上做备案之类的准备?

回到顶部