使用Python写一个Telegram bot吧|Telegram bot教程

2.9k
小技术
Pythontelegrambot代码
本文最后更新于 1 年前，文中所描述的信息可能已发生改变。

TIP提示

2024-01-14 更新:

修复了一些错误, 删除了一些不必要的内容

讨论群

本文只是我最初写 bot 时的一点笔记, 更建议你直接看 Python-Telegram-Bot 的文档

前言 ​

为了激发群友们水群的积极性， 为了学习和应用 python 知识，我最近在开发一个 telegram 的 bot，所以写这篇文章以记录和分享

在这篇文章中，将使用 Python-Telegram-Bot ，基于 Python 的异步特性与 Telegram 友好开放的 API，开发一个兼顾实用性和趣味的 bot ，并使用 Docker 在任何地方部署 bot

Demo: kmua-bot

本文不是从零开始的教程，阅读本文前，你需要具有一点点(真的很少一点)的 python 编程的基础。

准备 ​

环境搭建 ​

使用你喜欢的工具创建虚拟环境，并安装 python-telegram-bot 库

shell
pip install python-telegram-bot

bot 申请 ​

私聊 @BotFather。发送 /newbot，根据提示一步步创建，记得妥善保存最后的 API Token。

获取你的 id ​

每个 tg 用户都有一串唯一标识，即为 user_id，可以私聊 @kmua 发送 /id 来获取它

编程 ​

开始:响应/start ​

在项目文件夹内，新建 bot.py，开始编写 bot 最基础的功能，让其响应 /start 命令

首先，导入包

python
from telegram import Update
from telegram.ext import ApplicationBuilder, ContextTypes, CommandHandler

Update 类是从 Telegram 服务器获取到的各种"更新", 包括而不限于用户发送的消息, 群组的信息变化等.

然后，写一个异步函数 start()，当收到 /start 命令时，要调用它。

python
async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
    text = '你好~我是一个bot'
    await context.bot.send_message(chat_id=update.effective_chat.id,text=text)

这个函数接受两个参数 update 和 context，形参冒号后是类型注解。

context.bot.send_message() 方法即为让 bot 发送消息，它能接受的参数其实很多，但是往往只需要 chat_id ，和 text 就够了，它们分别表示 要发送消息给的用户或群组 id ，要发送的文本。

而 update.effective_chat.id 即为当前有效对话的 id ，bot 收到的是哪里的消息 ，它就指向哪里的 id。

现在，实例化一个处理器(handler)

python
start_handler = CommandHandler('start', start)

CommandHandler 类可以实现当收到某个命令时，调用某个函数（命令和函数名可以不一样），我们将其实例化为了 start_handler ，并且将命令名字 'start' 和对应要回调的函数名 start 传递给它

然后，启动bot

python
# 构建 bot
TOKEN='你 bot 的 api token'
application = ApplicationBuilder().token(TOKEN).build()
# 注册 handler
application.add_handler(start_handler)
# run!
application.run_polling()

最后，就可以使用 python bot.py 启动你的 bot ，对 bot 发送 /start ，它应该就会回复你 ”你好~我是一个bot”。 使用 Ctrl + C 结束程序运行。

完整 bot.py :

python
from telegram import Update
from telegram.ext import ApplicationBuilder, ContextTypes, CommandHandler

async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
    text = '你好~我是一个bot'
    await context.bot.send_message(chat_id=update.effective_chat.id,text=text)

start_handler = CommandHandler('start', start)

TOKEN='你 bot 的 api token'
application = ApplicationBuilder().token(TOKEN).build()
application.add_handler(start_handler)
application.run_polling()

TIP提示

如果你的环境无法直接访问 Telegram 服务器，可以设置代理

全局代理, 代理整个程序.

python
os.environ['http_proxy'] = '代理地址'
os.environ['https_proxy'] = '代理地址'

若想仅为 telegram bot api 设置代理,则需要改动一下构建 bot 部分的代码:

http(s) 代理:

python
proxy_url = 'http://USERNAME:PASSWORD@PROXY_HOST:PROXY_PORT'
application = ApplicationBuilder().token("TOKEN").proxy_url(proxy_url).get_updates_proxy_url(proxy_url).build()

socks5 代理,需要安装 python-telegram-bot[socks], 然后:

python
proxy_url = "socks5://user:pass@host:port"
application = ApplicationBuilder().token("TOKEN").proxy_url(proxy_url).get_updates_proxy_url(proxy_url).build()

到这里，其实你已经了解到了最基本的 telegram bot 编写规则，即：

  1. 编写回调, 上面的例子中即为 start 这个函数.
  2. 决定调用回调函数的规则. 上述例子中，即为收到 /start 命令时, 调用 start 函数
  3. 实例化 handler , 注册给 application。上述例子中，即为 application.add_handler(start_handler)

接下来，写一些更有趣的功能，使用更复杂一些的规则来调用这些功能。

授予群成员头衔 ​

让群成员可以通过 bot 自助获得一个头衔吧，比如，群友可以在群里使用 /t@botname 好人 来给自己加上 “好人” 的头衔。

当然，前提是 bot 自己要是管理员，并且具有相应的权限。

要实现这个功能，可以这样写:

python
async def set_right(update: Update, context: ContextTypes.DEFAULT_TYPE):
    chat_id = update.effective_chat.id
    user_id = update.effective_user.id
    bot_username_len = len(update._bot.name)
    custom_title = update.effective_message.text[3+bot_username_len:]
    if not custom_title:
        custom_title = update.effective_user.username
    try:
        await context.bot.promote_chat_member(chat_id=chat_id, user_id=user_id, can_manage_chat=True)
        await context.bot.set_chat_administrator_custom_title(chat_id=chat_id, user_id=user_id, custom_title=custom_title)
        text = f'好,你现在是{custom_title}啦'
        await context.bot.send_message(chat_id=chat_id, reply_to_message_id=update.effective_message.id, text=text)
    except:
        await context.bot.send_message(chat_id=chat_id, text='不行!!')

然后，添加这个功能的handler:

python
set_right_handler = CommandHandler('t', set_right)
application.add_handler(set_right_handler)

注意handler的添加应该在 application.run_polling() 之前，也就是说 application.run_polling() 才是真正开始运行 bot

简单解释一下 set_right 这个函数:

首先，设置了调用这个函数时的 chat_id 和 user_id 为当前聊天和当前用户

然后用 len(update._bot.name) 获取 bot 自己用户名的长度，用于下面的切片选取用户想要的头衔

而 custom_title = update.effective_message.text[3+bot_username_len:] 即获取用户想要的头衔，存储在 custom_title 中，下面的 if not 的作用是，如果用户没有发送他想要的头衔，那么默认头衔为他的用户名

由于只有管理员才有头衔，所以我们使用 promote_chat_member() 方法设置成员权限，将其的 can_manage_chat 权限设置为 True. 即使这样设置了，用户也只有一个管理员的名头，实际上没有任何权限

TIP提示

要想赋予权限，你可以向 promote_chat_member() 方法中继续添加参数，如 can_pin_messages=True 可置顶消息

然后就可以使用 set_chat_administrator_custom_title() 方法设置成员头衔了，如果成功了，会继续执行接下来的语句，将结果反馈给用户。

响应未知命令 ​

如果 bot 只能回应设置好的命令就太无趣了，所以再编写一个当收到未知命令时的执行功能 比如，想要让 bot 回应未知命令说 “我不会这个哦~”，可以这样写

python
async def unknown(update: Update, context: ContextTypes.DEFAULT_TYPE):
    await context.bot.send_message(chat_id=update.effective_chat.id, text="我不会这个哦~")

然后，添加它的 handler。这时候注意到：

前面的两个功能都是回应一个特定的命令，在这里，我们希望它回应除已经设定的命令外的其他所有命令，该怎么做？

这就需要用到 filters ，即过滤需要处理器处理的更新. 并且要设置功能的优先级

首先导入它

python
from telegram.ext import filters

然后可以这样写:

python
unknown_handler = MessageHandler(filters.COMMAND, unknown)

filters.COMMAND 即为过滤出命令。而设置优先级其实很简单，在不对处理器分组的情况下, 添加 handler 的顺序越靠后，对应功能的优先级越低

也就是说，我们需要在其他命令类 handler 之后，添加 unknown_handler

python
application.add_handler(start_handler)
application.add_handler(set_right_handler)
application.add_handler(unkonw_handler)

application.run_polling()

简单的关键词回复 ​

bot 最常见的功能之一就是根据关键词回复特定内容，比如，当对 bot 说早安之类的话时，让 bot 对此做出回应。下面来实现这这一功能

由于命令和普通消息是两种不同的消息类，处理它们的方法是不一样的，针对非命令消息，要使用 MessageHandler:

python
from telegram.ext import MessageHandler

首先还是要编写这个功能执行的函数。

python
import random

async def ohayo(update: Update, context: ContextTypes.DEFAULT_TYPE):
    texts = ['早上好呀','我的小鱼你醒了，还记得清晨吗','哦哈哟~']
    await context.bot.send_message(chat_id=update.effective_chat.id, text=random.choice(text))

由于只回复一个固定的消息过于单调，所以我们使用 random 来从一个列表中随机选择一个回复

仍然是前文所提到的, 回调函数只负责发送消息，而何时调用这个函数，并不是这个函数本身要做的事，而是由接下来我们要编写的 MessageHandler 中的 filters 决定的:

python
filter_ohayo = filters.Regex('早安|早上好|哦哈哟|ohayo')
ohayo_handler = MessageHandler(filter_ohayo, ohayo)

filters.Regex() 即为使用正则表达式过滤，我们使用 '早安|早上好|哦哈哟|ohayo' 这个表达式，很容易理解何时会调用 ohayo 这个函数

记得添加它的 handler

python
application.add_handler(ohayo_handler)

所有完整 bot.py:

python
import random

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

async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
    text = '你好~我是一个bot'
    await context.bot.send_message(chat_id=update.effective_chat.id,text=text)


async def set_right(update: Update, context: ContextTypes.DEFAULT_TYPE):
    chat_id = update.effective_chat.id
    user_id = update.effective_user.id
    bot_username_len = len(update._bot.name)
    custom_title = update.effective_message.text[3+bot_username_len:]
    if not custom_title:
        custom_title = update.effective_user.username
    try:
        await context.bot.promote_chat_member(chat_id=chat_id, user_id=user_id, can_manage_chat=True)
        await context.bot.set_chat_administrator_custom_title(chat_id=chat_id, user_id=user_id, custom_title=custom_title)
        text = f'好,你现在是{custom_title}啦'
        await context.bot.send_message(chat_id=chat_id, reply_to_message_id=update.effective_message.id, text=text)
    except:
        await context.bot.send_message(chat_id=chat_id, text='不行!!')


async def unknown(update: Update, context: ContextTypes.DEFAULT_TYPE):
    await context.bot.send_message(chat_id=update.effective_chat.id, text="我不会这个哦~")


async def ohayo(update: Update, context: ContextTypes.DEFAULT_TYPE):
    texts = ['早上好呀','我的小鱼你醒了，还记得清晨吗','哦哈哟~']
    await context.bot.send_message(chat_id=update.effective_chat.id, text=random.choice(text))


start_handler = CommandHandler('start', start)
set_right_handler = CommandHandler('p', set_right)
unknown_handler = MessageHandler(filters.COMMAND, unknown)
filter_ohayo = filters.Regex('早安|早上好|哦哈哟|ohayo')
ohayo_handler = MessageHandler(filter_ohayo, ohayo)

TOKEN='你 bot 的 api token'
application = ApplicationBuilder().token(TOKEN).build()
application.add_handler(start_handler)
application.add_handler(set_right_handler)
application.add_handler(unkonw_handler)
application.add_handler(ohayo_handler)
# run!
application.run_polling()

部署 ​

python 程序的环境管理很烦人,用 docker 来跑再合适不过.

Docker 本地构建 ​

如果要在本地构建 docker 镜像,参考下面的 Dockerfile 文件

Dockerfile
FROM python:3.12-slim
COPY . /app
WORKDIR /app
RUN pip install -r requirements.txt
ENTRYPOINT [ "python","/app/bot.py" ]

将 Dockerfile 文件放到合适的目录(项目根目录即可),然后执行

docker build -t bot .

注意后面有一个 .

然后使用 docker run -d bot 启动容器,运行 bot

使用 GitHub action 自动构建 ​

github action 可以构建 docker 镜像并发布到 ghcr,方便在其他地方部署 docker 容器.你依然需要在项目中写好 Dockerfile

在项目中新建 .github/workflows/build-docker.yml,参考以下配置

yml
name: Build and publish docker container

on:
 push:
     branches:
     - main
  workflow_dispatch:
    
    

jobs:
  publish:
    name: Publish container image
    runs-on: ubuntu-20.04
    env:
      TZ: Asia/Shanghai
    steps:
      - name: Checkout
        uses: actions/checkout@v3

      - name: OCI meta
        id: meta
        uses: docker/metadata-action@v4.1.1
        with:
          images: ghcr.io/${{ github.repository }}
          tags: |
            type=edge,branch=main
            type=ref,event=branch
            type=semver,pattern={{version}}
            type=semver,pattern={{major}}.{{minor}}
            type=semver,pattern={{major}}
            type=sha
      - name: Set up Docker Buildx
        uses: docker/setup-buildx-action@v2
        
      - name: Set up QEMU
        uses: docker/setup-qemu-action@v2

      - name: Login to GHCR
        uses: docker/login-action@v2
        with:
          registry: ghcr.io
          username: ${{ github.repository_owner }}
          password: ${{ secrets.GITHUB_TOKEN }}

      - name: Build and push
        uses: docker/build-push-action@v3
        with:
          context: .
          push: true
          tags: ${{ steps.meta.outputs.tags }}
          labels: ${{ steps.meta.outputs.labels }}
          platforms: linux/amd64,linux/arm64
          cache-from: type=gha
          cache-to: type=gha,mode=max

然后可以使用 docker pull 的方法运行,或者使用 docker compose:

新建 docker-compose.yml,参考以下内容

yml
version: "3"
services:
  kmua:
    image: ghcr.io/krau/kmua-bot:main
    container_name: kmua-main
    init: true
    volumes:
      - ./data:/kmua/data
      - ./logs:/kmua/logs
      - ./config.yml:/kmua/config.yml
    environment:
      - TZ=Asia/Shanghai

注意根据自己需要更改

然后就可以使用 docker compose up -d 启动容器

将 hexo 博客迁移至 valaxy
ChatGPT简单使用体验|你的下一个Google，何必是搜索引擎