开发一个自己的 Telegram Bot

虽然每天都在用 Telegram Bot，但一直没深入了解过，最近折腾了一下，发现 TG 机器人能实现的功能太多了，非常实用，API 也很完善，然而还是遇到了不少问题，特别是用 Nginx 配合 Webhook 的时候，搞了一晚上都没成功，因此这里简单的记录一下，排一下坑。

准备工作

这里跳过 Bot 的申请过程。网上有多种语言的集成 API，PHP、Node.、Rust、Python 等等，但人生苦短，我选 Python。API 则选择了 pyTelegramBotAPI。

以下所有演示为在 Debian 11+ 系统中使用 root 用户进行。

配置虚拟环境

Debian 11 下配置

apt install python3-pip
pip3 install virtualenv

创建一个工程目录：

mkdir /home/mybot && cd /home/mybot

创建并激活虚拟环境：

virtualenv env
source env/bin/activate

接下来的依赖就会安装到虚拟环境中，要退出环境的话使用 deactivate 命令。

如果以后要将项目转移到别的地方，先进入虚拟环境中导出所有包到 requirements.txt：

pip3 freeze > requirements.txt

然后在新的环境中安装：

pip3 install -r requirements.txt

Debian 12 下配置

今天（2023-10-08）将系统升级到 Debian 12 (bookworm) 后发现 uWSGI 启动失败，提示错误：

uwsgi: error while loading shared libraries: libpython3.9.so.1.0: cannot open shared

显然是因为 Debian 12 升级了系统 Python 版本，而我升级系统后执行了 autoclean，于是准备重新安装初始化一个虚拟环境，执行 pip3 install virtualenv 后提示：

error: externally-managed-environment

× This environment is externally managed
╰─> To install Python packages system-widetry apt install
    python3-xyzwhere xyz is the package you are trying to
    install.
  
    If you wish to install a non-Debian-packaged Python package,
    create a virtual environment using python3 -m venv path/to/venv.
    Then use path/to/venv/bin/python and path/to/venv/bin/pip. Make
    sure you have python3-full installed.
  
    If you wish to install a non-Debian packaged Python application,
    it may be easiest to use pipx install xyzwhich will manage a
    virtual environment for you. Make sure you have pipx installed.
  
    See /usr/share/doc/python3.11/README.venv for more information.

发现 Python 3.11+ 会默认要求用户在 venv 虚拟环境中安装 Package，于是按照它建议的来：

apt install python3-venv

创建一个工程目录：

mkdir /home/mybot && cd /home/mybot

创建并激活虚拟环境：

python3 -m venv env
source env/bin/activate

安装所需依赖

pip install pyTelegramBotAPI
pip install flask
pip install uwsgi 
# 或者不用 uwsgi，使用更简单的 gunicorn
# pip install gunicorn

设置 Webhook

获取 Bot 的更新可以通过 Polling 或者 Webhook，这里选择更适合生产环境的 Webhook，设置 Webhook 很简单，只需要在浏览器中访问一下下面的链接就行了：

https://api.telegram.org/bot{bot_token}/setWebhook?url={your_link}

例如：

https://api.telegram.org/bot123456:ABC-DEF1234ghIkl-zyx57W2v1u123ew11/setWebhook?url=https://www.example.com

之后可以通过下面的链接查看设置的 Webhook 信息：

https://api.telegram.org/bot{bot_token}/getWebhookInfo

配置 WSGI 服务器

开发测试环境中直接使用 Flask 接收 Webhook 即可，如果是部署生产环境，建议使用 WSGI 服务器网关接口提高性能。常见的 WSGI 服务器可以选择 uWSGI 或者 Gunicorn，uWSGI 是由 C 语言编写，通过编译安装，Gunicorn 由 Python 编写。一般来说，uWSGI 的性能更好，尤其是在高并发环境下，而 Gunicorn 更加简单，容易上手，这里你可以根据自己的需求选择。

使用 uWSGI

运行 uWSGI 可以通过命令行或者文件配置，这里使用 .ini 文件配置，创建一个 uwsgi.ini 配置文件：

[uwsgi]
project = mybot
path = /home/mybot
wsgi-file = %(path)/mybot.py
socket = %(path)/run/%(project).sock
pidfile = %(path)/run/uwsgi.pid
stats = %(path)/run/uwsgi.status
chmod-socket = 666
callable = app
master = true
vacuum = true
processes = 4
threads = 2

启动服务：

uwsgi --ini uwsgi.ini

使用 Gunicorn

在项目路径下新建一个 gunicorn_config.py 配置文件：

bind = '127.0.0.1:8000'  # 绑定的主机和端口
workers = 3  # worker 进程的数量，建议的数量是 (2*CPU)+1
worker_class = 'sync'  # worker 进程的类型
timeout = 30  # 超时时间，单位为秒
keepalive = 2  # keep-alive 连接的最大数量
errorlog = '-'  # 错误日志文件，默认输出到 stdout
accesslog = '-'  # 访问日志文件，默认输出到 stdout
loglevel = 'info'  # 日志级别：debuginfowarningerrorcritical
# access_log_format = '%({X-Forwarded-For}i)s %(l)s %(u)s %(t)s "%(r)s" %(s)s %(b)s "%(f)s" "%(a)s"' # 获取真实 IP

启动服务：

env/bin/gunicorn -c gunicorn_config.py mybot:app

创建守护进程

投入生产环境运行时，可以使用 systemd 控制进程。这里简单写一个使用 systemd 来创建进程并保持后台运行的示例：

vim /etc/systemd/system/mybot.service

使用 uWSGI：

[Unit]
Description=uWSGI - Telegram Bot by atpx.com
After=network.target

[Service]
WorkingDirectory=/home/mybot
ExecStart=/home/mybot/env/bin/uwsgi --ini /home/mybot/uwsgi.ini
ExecStop=/home/mybot/env/bin/uwsgi --stop /home/mybot/run/uwsgi.pid
ExecReload=/home/mybot/env/bin/uwsgi --reload /home/mybot/run/uwsgi.pid
Restart=on-failure
RestartSec=20s

[Install]
WantedBy=multi-user.target

使用 Gunicorn：

[Unit]
Description=Gunicorn - Telegram Bot by atpx.com
After=network.target

[Service]
WorkingDirectory=/home/mybot
ExecStartPre=/bin/bash -c 'source /home/mybot/env/bin/activate'
ExecStart=/home/mybot/env/bin/gunicorn -c gunicorn_config.py mybot:app
ExecStopPost=/bin/bash -c 'deactivate'
Restart=on-failure
RestartSec=20s

[Install]
WantedBy=multi-user.target

配置 Nginx

通常建议再加一层 Nginx 反代到 WSGI 服务器以提高安全性和性能。同时 Webhook 只能使用 https，因此你需要签发 SSL 证书，可以申请免费的 Let’s Encrypt 证书，具体步骤可查看这篇文章 使用acme.sh自动签发和更新证书，Nginx 配置文件中的反代部分配置如下：

# 使用 uWSGI
location /SECRET_PATH {
    include uwsgi_params;
    uwsgi_pass unix:/home/mybot/run/mybot.sock;
}

# 或者使用 Gunicorn：
location /SECRET_PATH {
    proxy_redirect off;
    proxy_set_header Host $http_host;
    proxy_set_header X-Real-IP $remote_addr;
    proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
    proxy_set_header X-Forwarded-Proto $scheme;
    proxy_pass http://127.0.0.1:8000;
}

其中的 SECRET_PATH 可以是任意的字符串，推荐使用足够长的随机字符串，后面编写 Bot 代码时需要和这里一致。

编写第一个 Bot

这里的 /home/mybot/mybot.py 是一个简单的 Bot 示例，来源于 pyTelegramBotAPI 的 webhook_examples，并由 atpX 稍作修改。

#!/usr/bin/python3
# -*- coding: utf-8 -*-

import time
import telebot
from flask import Flaskrequestabort

# Bot Token / 你申请到的 Bot API Token
API_TOKEN = ''
# Your domain / 前面 Webhook 中设置的域名，例如 https://www.example.com
WEBHOOK_HOST = ''
# Nginx reverse proxy path / Nginx 反代中设置的自定义路径
SECRET_PATH = ''

WEBHOOK_URL = "%s/%s" % (WEBHOOK_HOSTSECRET_PATH)

bot = telebot.TeleBot(API_TOKENthreaded=False)

app = Flask(__name__)

@app.route('/'methods=['GET''HEAD'])
def index():
    return ''

@app.route(WEBHOOK_URL_PATHmethods=['POST'])
def webhook():
    if request.headers.get('content-type') == 'application/on':
        on_string = request.get_data().decode('utf-8')
        update = telebot.types.Update.de_on(on_string)
        bot.process_new_updates([update])
        return 'success'
    else:
        abort(403)

# Handle '/start'
@bot.message_handler(commands=['start'])
def send_welcome(message):
    bot.send_message(message.chat.id'Nyanpasu~~')

# Handle all other messages
@bot.message_handler(func=lambda message: Truecontent_types=['text'])
def echo_message(message):
    bot.reply_to(messagemessage.text)

# 使用 Gunicorn 时注释掉下方三行
bot.remove_webhook()
time.sleep(0.1)
bot.set_webhook(url=WEBHOOK_URL)

# 使用 Gunicorn 时取消注释下方两行
# if __name__ == '__main__':
#     app.run(host='127.0.0.1'port=8000)

到这里，一个简单的 Bot 就完成了，接下来运行脚本，访问你的 Bot 发送 /start 就可以看到自动回复，输入其他文本信息就会变成复读机(￣、￣)。

接下来，你可以使用 Python 配合 API 实现任何你想实现的功能。

赏阅。悦赏！