本文最后更新于 11 months ago,文中所描述的信息可能已发生改变。
前言
为了激发群友们水群的积极性, 为了学习和应用 python 知识,我最近在开发一个 telegram 的 bot,所以写这篇文章以记录和分享
在这篇文章中,将使用 Python-Telegram-Bot ,基于 Python 的异步特性与 Telegram 友好开放的 API,开发一个兼顾实用性和趣味的 bot ,并使用 Docker 在任何地方部署 bot
Demo: kmua-bot
本文不是从零开始的教程,阅读本文前,你需要具有一点点(真的很少一点)的 python 编程的基础。
准备
环境搭建
使用你喜欢的工具创建虚拟环境,并安装 python-telegram-bot
库
pip install python-telegram-bot
bot 申请
私聊 @BotFather。发送 /newbot
,根据提示一步步创建,记得妥善保存最后的 API Token。
获取你的 id
每个 tg 用户都有一串唯一标识,即为 user_id,可以私聊 @kmua 发送 /id
来获取它
编程
开始:响应/start
在项目文件夹内,新建 bot.py
,开始编写 bot 最基础的功能,让其响应 /start 命令
首先,导入包
from telegram import Update
from telegram.ext import ApplicationBuilder, ContextTypes, CommandHandler
Update 类是从 Telegram 服务器获取到的各种"更新", 包括而不限于用户发送的消息, 群组的信息变化等.
然后,写一个异步函数 start(),当收到 /start 命令时,要调用它。
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)
start_handler = CommandHandler('start', start)
CommandHandler
类可以实现当收到某个命令时,调用某个函数(命令和函数名可以不一样),我们将其实例化为了 start_handler
,并且将命令名字 'start' 和对应要回调的函数名 start
传递给它
然后,启动bot
# 构建 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 :
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 服务器,可以设置代理
全局代理, 代理整个程序.
os.environ['http_proxy'] = '代理地址'
os.environ['https_proxy'] = '代理地址'
若想仅为 telegram bot api 设置代理,则需要改动一下构建 bot 部分的代码:
http(s) 代理:
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]
, 然后:
proxy_url = "socks5://user:pass@host:port"
application = ApplicationBuilder().token("TOKEN").proxy_url(proxy_url).get_updates_proxy_url(proxy_url).build()
到这里,其实你已经了解到了最基本的 telegram bot 编写规则,即:
- 编写回调, 上面的例子中即为
start
这个函数. - 决定调用回调函数的规则. 上述例子中,即为收到
/start
命令时, 调用start
函数 - 实例化
handler
, 注册给application
。上述例子中,即为application.add_handler(start_handler)
接下来,写一些更有趣的功能,使用更复杂一些的规则来调用这些功能。
授予群成员头衔
让群成员可以通过 bot 自助获得一个头衔吧,比如,群友可以在群里使用 /t@botname 好人
来给自己加上 “好人” 的头衔。
当然,前提是 bot 自己要是管理员,并且具有相应的权限。
要实现这个功能,可以这样写:
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:
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 回应未知命令说 “我不会这个哦~”,可以这样写
async def unknown(update: Update, context: ContextTypes.DEFAULT_TYPE):
await context.bot.send_message(chat_id=update.effective_chat.id, text="我不会这个哦~")
然后,添加它的 handler
。这时候注意到:
前面的两个功能都是回应一个特定的命令,在这里,我们希望它回应除已经设定的命令外的其他所有命令,该怎么做?
这就需要用到 filters
,即过滤需要处理器处理的更新. 并且要设置功能的优先级
首先导入它
from telegram.ext import filters
然后可以这样写:
unknown_handler = MessageHandler(filters.COMMAND, unknown)
filters.COMMAND
即为过滤出命令。而设置优先级其实很简单,在不对处理器分组的情况下, 添加 handler
的顺序越靠后,对应功能的优先级越低
也就是说,我们需要在其他命令类 handler
之后,添加 unknown_handler
application.add_handler(start_handler)
application.add_handler(set_right_handler)
application.add_handler(unkonw_handler)
application.run_polling()
简单的关键词回复
bot 最常见的功能之一就是根据关键词回复特定内容,比如,当对 bot 说早安之类的话时,让 bot 对此做出回应。下面来实现这这一功能
由于命令和普通消息是两种不同的消息类,处理它们的方法是不一样的,针对非命令消息,要使用 MessageHandler
:
from telegram.ext import MessageHandler
首先还是要编写这个功能执行的函数。
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
决定的:
filter_ohayo = filters.Regex('早安|早上好|哦哈哟|ohayo')
ohayo_handler = MessageHandler(filter_ohayo, ohayo)
filters.Regex()
即为使用正则表达式过滤,我们使用 '早安|早上好|哦哈哟|ohayo'
这个表达式,很容易理解何时会调用 ohayo
这个函数
记得添加它的 handler
application.add_handler(ohayo_handler)
所有完整 bot.py:
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
文件
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
,参考以下配置
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
,参考以下内容
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
启动容器