mirror of
https://github.com/nonebot/nonebot2.git
synced 2025-07-28 00:31:14 +00:00
clean up files
This commit is contained in:
yanyongyu
@ -1,3 +0,0 @@
|
||||
# 概览
|
||||
|
||||
在 [指南](../guide/) 中，我们已经介绍了 NoneBot 的基础用法，利用这些基础用法已经可以编写很多有趣的功能。而在进阶这一部分，我们将介绍 NoneBot 的高级特性、特殊用法和最佳实践，以帮助你编写更加复杂、功能更丰富的机器人。
|
||||
@ -1,110 +0,0 @@
|
||||
# 命令参数
|
||||
|
||||
## `session.get()` 和参数解析器
|
||||
|
||||
## 类 Shell 参数解析
|
||||
|
||||
`nonebot.argparse` 模块主要继承自 Python 内置的同名模块（`argparse`），用于解析命令的参数。在需要编写类 shell 语法的命令的时候，使用此模块可以大大提高开发效率。
|
||||
|
||||
「类 shell 语法」指的是形如 `some-command --verbose -n 3 --name=some-name argument1 argument2` 的类似于 shell 命令的语法。
|
||||
|
||||
下面给出一个使用 `argparse` 模块的实际例子：
|
||||
|
||||
```python {1-15}
|
||||
@on_command('schedule', shell_like=True)
|
||||
async def _(session: CommandSession):
|
||||
parser = ArgumentParser(session=session, usage=USAGE)
|
||||
parser.add_argument('-S', '--second')
|
||||
parser.add_argument('-M', '--minute')
|
||||
parser.add_argument('-H', '--hour')
|
||||
parser.add_argument('-d', '--day')
|
||||
parser.add_argument('-m', '--month')
|
||||
parser.add_argument('-w', '--day-of-week')
|
||||
parser.add_argument('-f', '--force', action='store_true', default=False)
|
||||
parser.add_argument('-v', '--verbose', action='store_true', default=False)
|
||||
parser.add_argument('--name', required=True)
|
||||
parser.add_argument('commands', nargs='+')
|
||||
|
||||
args = parser.parse_args(session.argv)
|
||||
|
||||
if not re.match(r'[_a-zA-Z][_a-zA-Z0-9]*', args.name):
|
||||
await session.send('计划任务名必须仅包含字母、数字、下划线，且以字母或下划线开头')
|
||||
return
|
||||
|
||||
parsed_commands: List[ScheduledCommand] = []
|
||||
invalid_commands: List[str] = []
|
||||
|
||||
if args.verbose:
|
||||
parsed_commands.append(
|
||||
ScheduledCommand(('echo',), f'开始执行计划任务 {args.name}……'))
|
||||
|
||||
for cmd_str in args.commands:
|
||||
cmd, current_arg = parse_command(session.bot, cmd_str)
|
||||
if cmd:
|
||||
tmp_session = CommandSession(session.bot, session.ctx, cmd,
|
||||
current_arg=current_arg)
|
||||
if await cmd.run(tmp_session, dry=True):
|
||||
parsed_commands.append(ScheduledCommand(cmd.name, current_arg))
|
||||
continue
|
||||
invalid_commands.append(cmd_str)
|
||||
if invalid_commands:
|
||||
invalid_commands_joined = '\n'.join(
|
||||
[f'- {c}' for c in invalid_commands])
|
||||
await session.send(f'计划任务添加失败，'
|
||||
f'因为下面的 {len(invalid_commands)} 个命令无法被运行'
|
||||
f'（命令不存在或权限不够）：\n'
|
||||
f'{invalid_commands_joined}')
|
||||
return
|
||||
|
||||
trigger_args = {k: v for k, v in args.__dict__.items()
|
||||
if k in {'second', 'minute', 'hour', 'day', 'month', 'day_of_week'}}
|
||||
try:
|
||||
job = await scheduler.add_scheduled_commands(
|
||||
parsed_commands,
|
||||
job_id=scheduler.make_job_id(PLUGIN_NAME, context_id(session.ctx), args.name),
|
||||
ctx=session.ctx,
|
||||
trigger='cron', **trigger_args,
|
||||
replace_existing=args.force
|
||||
)
|
||||
except scheduler.JobIdConflictError:
|
||||
# a job with same name exists
|
||||
await session.send(f'计划任务 {args.name} 已存在，'
|
||||
f'若要覆盖请使用 --force 参数')
|
||||
return
|
||||
|
||||
await session.send(f'计划任务 {args.name} 添加成功')
|
||||
await session.send(format_job(args.name, job))
|
||||
|
||||
|
||||
USAGE = r"""
|
||||
添加计划任务
|
||||
|
||||
使用方法：
|
||||
schedule.add [OPTIONS] --name NAME COMMAND [COMMAND ...]
|
||||
|
||||
OPTIONS：
|
||||
-h, --help 显示本使用帮助
|
||||
-S SECOND, --second SECOND 定时器的秒参数
|
||||
-M MINUTE, --minute MINUTE 定时器的分参数
|
||||
-H HOUR, --hour HOUR 定时器的时参数
|
||||
-d DAY, --day DAY 定时器 的日参数
|
||||
-m MONTH, --month MONTH 定时器的月参数
|
||||
-w DAY_OF_WEEK, --day-of-week DAY_OF_WEEK 定时器的星期参数
|
||||
-f, --force 强制覆盖已有的同名计划任务
|
||||
-v, --verbose 在执行计划任务时输出更多信息
|
||||
|
||||
NAME：
|
||||
计划任务名称
|
||||
|
||||
COMMAND：
|
||||
要计划执行的命令，如果有空格或特殊字符，需使用引号括起来
|
||||
""".strip()
|
||||
```
|
||||
|
||||
上面的例子出自 [cczu-osa/aki](https://github.com/cczu-osa/aki) 项目的计划任务插件，这里我们只关注前 15 行。
|
||||
|
||||
`on_command` 的 `shell_like=True` 参数告诉 NoneBot 这个命令需要使用类 shell 语法，NoneBot 会自动添加命令参数解析器来使用 Python 内置的 `shlex` 包分割参数。分割后的参数被放在 `session.args['argv']`，可通过 `session.argv` 属性来快速获得。
|
||||
|
||||
命令处理函数中，使用 `nonebot.argparse` 模块包装后的 `ArgumentParser` 类来解析参数，具体 `ArgumentParser` 添加参数的方法，请参考 [`argparse`](https://docs.python.org/3/library/argparse.html)。在使用 `add_argument()` 方法添加需要解析的参数后，使用 `parse_args()` 方法最终将 `argv` 解析为 `argparse.Namespace` 对象。
|
||||
|
||||
特别地，`parse_args()` 方法如果遇到需要打印帮助或报错并退出程序的情况（具体可以通过使用 Python 内置的 `argparse.ArgumentParser` 来体验），行为会更改为发送消息给当前 session 对应的上下文。注意到，`ArgumentParser` 类初始化时传入了 `session` 和 `usage` 参数，分别用于发送消息和使用帮助的内容。
|
||||
@ -1 +0,0 @@
|
||||
# 命令组
|
||||
@ -1,9 +0,0 @@
|
||||
# 命令会话
|
||||
|
||||
## 生命周期
|
||||
|
||||
## 状态数据
|
||||
|
||||
## 暂停、终止
|
||||
|
||||
## 切换上下文
|
||||
@ -1 +0,0 @@
|
||||
# 配置
|
||||
@ -1 +0,0 @@
|
||||
# 数据库
|
||||
@ -1 +0,0 @@
|
||||
# `on_*` 装饰器
|
||||
@ -1,80 +0,0 @@
|
||||
# 部署
|
||||
|
||||
## 基本部署
|
||||
|
||||
NoneBot 所基于的 python-aiocqhttp 库使用的 web 框架是 Quart，因此 NoneBot 的部署方法和 Quart 一致（[Deploying Quart](https://pgjones.gitlab.io/quart/tutorials/deployment.html)）。
|
||||
|
||||
Quart 官方建议使用 Hypercorn 来部署，这需要一个 ASGI app 对象，在 NoneBot 中，可使用 `nonebot.get_bot().asgi` 获得 ASGI app 对象。
|
||||
|
||||
具体地，通常在项目根目录下创建一个 `run.py` 文件如下：
|
||||
|
||||
```python
|
||||
import os
|
||||
import sys
|
||||
|
||||
import nonebot
|
||||
|
||||
import config
|
||||
|
||||
nonebot.init(config)
|
||||
bot = nonebot.get_bot()
|
||||
app = bot.asgi
|
||||
|
||||
if __name__ == '__main__':
|
||||
bot.run()
|
||||
```
|
||||
|
||||
然后使用下面命令部署：
|
||||
|
||||
```python
|
||||
hypercorn run:app
|
||||
```
|
||||
|
||||
另外，NoneBot 配置文件的 `DEBUG` 项默认为 `True`，在生产环境部署时请注意修改为 `False` 以提高性能。
|
||||
|
||||
## 使用 Docker Compose 与 酷Q 同时部署
|
||||
|
||||
Docker Compose 是 Docker 官方提供的一个命令行工具，用来定义和运行由多个容器组成的应用。通过建立一个名为 `docker-compose.yml` 的文件，可以将部署过程中需要的参数记录在其中，并由单个命令完成应用的创建和启动。
|
||||
|
||||
`docker-compose.yml` 文件的示例如下：
|
||||
|
||||
```yaml
|
||||
version: "3"
|
||||
services:
|
||||
|
||||
cqhttp:
|
||||
image: richardchien/cqhttp:latest
|
||||
volumes:
|
||||
- "./coolq:/home/user/coolq" # 用于保存COOLQ文件的目录
|
||||
environment:
|
||||
- COOLQ_ACCOUNT=123456 # 指定要登陆的QQ号，用于自动登录
|
||||
- FORCE_ENV=true
|
||||
- CQHTTP_USE_HTTP=false
|
||||
- CQHTTP_USE_WS=false
|
||||
- CQHTTP_USE_WS_REVERSE=true
|
||||
- CQHTTP_WS_REVERSE_API_URL=ws://nonebot:8080/ws/api/
|
||||
- CQHTTP_WS_REVERSE_EVENT_URL=ws://nonebot:8080/ws/event/
|
||||
depends_on:
|
||||
- nonebot
|
||||
ports: 9000:9000 # noVNC 端口，用于从浏览器控制 酷Q
|
||||
|
||||
nonebot:
|
||||
build: ./nonebot # 构建nonebot执行环境，Dockerfile见下面的例子
|
||||
expose:
|
||||
- "8080"
|
||||
environment:
|
||||
- TZ=Asia/Shanghai
|
||||
volumes:
|
||||
- "./qbot:/root/qbot" # 项目文件所在目录
|
||||
command: python3 /root/qbot/bot.py
|
||||
```
|
||||
|
||||
部分说明见注释。NoneBot 运行环境由文件 `./nonebot/Dockerfile` 控制构建。如果项目中使用了第三方库，可以在这一步骤进行安装。`Dockerfile` 内容例如：
|
||||
|
||||
```Dockerfile
|
||||
FROM alpine
|
||||
RUN apk add --no-cache tzdata python3 py3-multidict py3-yarl && \
|
||||
pip3 install --no-cache-dir "nonebot[scheduler]"
|
||||
```
|
||||
|
||||
上述文件编辑完成后，输入命令 `docker-compose up` 即可一次性启动酷Q和 NoneBot（可通过 `docker-compose up -d` 在后台启动。更多 Docker Compose 用法见 [官方文档](https://docs.docker.com/compose/reference/overview/)。
|
||||
@ -1,7 +0,0 @@
|
||||
# 大型应用的最佳实践
|
||||
|
||||
## 使用独立 Logger
|
||||
|
||||
## 项目结构
|
||||
|
||||
## 根据运行环境加载不同的配置
|
||||
@ -1,23 +0,0 @@
|
||||
# 日志
|
||||
|
||||
`nonebot.log` 模块提供了一个 `logger` 对象，可用于日志。
|
||||
|
||||
使用 `nonebot.init()` 配置 NoneBot 时，`logger` 对象的日志级别会随 `DEBUG` 配置项的不同而不同，如果 `DEBUG` 为 `True`，则日志级别为 `DEBUG`，否则为 `INFO`。你也可以在 `nonebot.init()` 调用之后自行设置 `logger` 的日志级别。
|
||||
|
||||
举例：
|
||||
|
||||
```python
|
||||
import logging
|
||||
|
||||
import nonebot
|
||||
from nonebot.log import logger
|
||||
|
||||
import config
|
||||
|
||||
|
||||
nonebot.init(config)
|
||||
# logger.setLevel(logging.WARNING)
|
||||
|
||||
logger.info('Starting')
|
||||
nonebot.run()
|
||||
```
|
||||
@ -1,5 +0,0 @@
|
||||
# 消息处理
|
||||
|
||||
## CQ 码和消息段
|
||||
|
||||
## Expression
|
||||
@ -1 +0,0 @@
|
||||
# 权限控制
|
||||
@ -1,161 +0,0 @@
|
||||
# 计划任务
|
||||
|
||||
NoneBot 可选地内置了计划任务功能，在指南的 [添加计划任务](../guide/scheduler.md) 已经进行了简单的介绍。这里列出更多常见的用法。
|
||||
|
||||
## 固定的计划任务
|
||||
|
||||
可以利用固定的*触发器*（trigger）来触发某些任务。
|
||||
|
||||
### 一次性任务
|
||||
|
||||
[`date` 触发器](https://apscheduler.readthedocs.io/en/stable/modules/triggers/date.html#module-apscheduler.triggers.date) 固定时间触发，仅触发一次：
|
||||
|
||||
```python
|
||||
from datetime import datetime
|
||||
|
||||
@nonebot.scheduler.scheduled_job(
|
||||
'date',
|
||||
run_date=datetime(2021, 1, 1, 0, 0),
|
||||
# timezone=None,
|
||||
)
|
||||
async def _():
|
||||
await bot.send_group_msg(group_id=123456,
|
||||
message="2021，新年快乐！")
|
||||
```
|
||||
|
||||
### 定期任务
|
||||
|
||||
[`cron` 触发器](https://apscheduler.readthedocs.io/en/stable/modules/triggers/cron.html#module-apscheduler.triggers.cron) 从 `start_date` 开始到 `end_date` 结束，根据类似 [Cron](https://wiki.archlinux.org/index.php/Cron_(%E7%AE%80%E4%BD%93%E4%B8%AD%E6%96%87)) 的规则触发任务：
|
||||
|
||||
```python
|
||||
@nonebot.scheduler.scheduled_job(
|
||||
'cron',
|
||||
# year=None,
|
||||
# month=None,
|
||||
# day=None,
|
||||
# week=None,
|
||||
day_of_week="mon,tue,wed,thu,fri",
|
||||
hour=7,
|
||||
# minute=None,
|
||||
# second=None,
|
||||
# start_date=None,
|
||||
# end_date=None,
|
||||
# timezone=None,
|
||||
)
|
||||
async def _():
|
||||
await bot.send_group_msg(group_id=123456,
|
||||
message="起床啦！")
|
||||
```
|
||||
|
||||
### 间隔任务
|
||||
|
||||
[`interval` 触发器](https://apscheduler.readthedocs.io/en/stable/modules/triggers/interval.html#module-apscheduler.triggers.interval) 从 `start_date` 开始，每间隔一段时间触发，到 `end_date` 结束：
|
||||
|
||||
```python
|
||||
@nonebot.scheduler.scheduled_job(
|
||||
'interval',
|
||||
# weeks=0,
|
||||
# days=0,
|
||||
# hours=0,
|
||||
minutes=5,
|
||||
# seconds=0,
|
||||
# start_date=time.now(),
|
||||
# end_date=None,
|
||||
)
|
||||
async def _():
|
||||
has_new_item = check_new_item()
|
||||
if has_new_item:
|
||||
await bot.send_group_msg(group_id=123456,
|
||||
message="XX有更新啦！")
|
||||
```
|
||||
|
||||
## 动态的计划任务
|
||||
|
||||
有时，我们需要机器人在运行的过程中，添加或删除计划任务，那么我们就需要 `scheduler.add_job` 来帮忙。这里，我们以*一次性任务*为例，其他类型的任务可以用相同的方法：
|
||||
|
||||
```python
|
||||
import datetime
|
||||
|
||||
from apscheduler.triggers.date import DateTrigger # 一次性触发器
|
||||
# from apscheduler.triggers.cron import CronTrigger # 定期触发器
|
||||
# from apscheduler.triggers.interval import IntervalTrigger # 间隔触发器
|
||||
from nonebot import on_command, scheduler
|
||||
|
||||
@on_command('赖床')
|
||||
async def _(session: CommandSession):
|
||||
await session.send('我会在5分钟后再喊你')
|
||||
|
||||
# 制作一个“5分钟后”触发器
|
||||
delta = datetime.timedelta(minutes=5)
|
||||
trigger = DateTrigger(
|
||||
run_date=datetime.datetime.now() + delta
|
||||
)
|
||||
|
||||
# 添加任务
|
||||
scheduler.add_job(
|
||||
func=session.send, # 要添加任务的函数，不要带参数
|
||||
trigger=trigger, # 触发器
|
||||
args=('不要再赖床啦！',), # 函数的参数列表，注意：只有一个值时，不能省略末尾的逗号
|
||||
# kwargs=None,
|
||||
misfire_grace_time=60, # 允许的误差时间，建议不要省略
|
||||
# jobstore='default', # 任务储存库，在下一小节中说明
|
||||
)
|
||||
```
|
||||
|
||||
<!-- ## 持久化存储任务
|
||||
|
||||
有时，我们动态添加的一些计划任务需要持久化储存，否则任务会在重启后丢失，这时我们就需要 `jobstore` 来帮忙。
|
||||
|
||||
APScheduler 可以将任务存储在内存中或数据库中，默认 jobstore 将所有任务储存在内存中，关闭后即丢失。这里，我们以 SQLite 为例，将任务添加到数据库中。
|
||||
|
||||
我们先创建一个数据库：
|
||||
|
||||
```python
|
||||
import asyncio
|
||||
import os
|
||||
|
||||
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
|
||||
from nonebot import on_command, scheduler, get_bot
|
||||
|
||||
database_path = os.path.join(os.path.dirname(__file__), 'job_store.db')
|
||||
|
||||
store = SQLAlchemyJobStore(url='sqlite:///'+database_path)
|
||||
|
||||
scheduler.add_jobstore(store, alias='my_job_store')
|
||||
```
|
||||
|
||||
之后，我们在添加新任务时，可以指定任务的储存库
|
||||
|
||||
```python
|
||||
async def alarm(*args, **kwargs):
|
||||
bot = get_bot()
|
||||
await bot.send(*args, **kwargs)
|
||||
|
||||
@on_command('提醒收菜')
|
||||
async def _(session: CommandSession):
|
||||
await session.send('我会在一天后提醒你收菜')
|
||||
delta = datetime.timedelta(days=1)
|
||||
trigger = DateTrigger(
|
||||
run_date=datetime.datetime.now() + delta
|
||||
)
|
||||
|
||||
# 添加任务
|
||||
scheduler.add_job(
|
||||
func=alarm,
|
||||
trigger=trigger,
|
||||
kwargs = {
|
||||
'context': session.ctx,
|
||||
'message': '起床收菜啦！',
|
||||
},
|
||||
misfire_grace_time=60,
|
||||
jobstore='my_job_store', # 任务储存库，指定为刚才创建的储存库
|
||||
)
|
||||
```
|
||||
|
||||
**踩坑预警：**
|
||||
|
||||
由于 `apscheduler` 自带的 jobstore 无法将协程任务储存进数据库，
|
||||
所以必须将任务转化为同步任务再储存。
|
||||
并且 `apscheduler` 中的 `AsyncIOScheduler` 在执行同步任务时，会新建一个执行器（executor），
|
||||
导致这个任务里无法使用 `asyncio.get_running_loop()` 来获取事件循环，
|
||||
只能使用 `asyncio.run(...)` 来运行异步函数。 -->
|
||||
@ -1,39 +0,0 @@
|
||||
# Server App
|
||||
|
||||
如果需要对 web 框架进行更详细的控制，可以通过 `bot.server_app` 访问到内部的 Quart 对象，之后可以像使用 Quart 的 app 对象一样添加路由、设置生命周期处理函数等。
|
||||
|
||||
:::tip 提示
|
||||
Quart 是一个与 Flask 具有相同 API 的异步 web 框架，其用法可以参考 [官方文档](https://pgjones.gitlab.io/quart/)。
|
||||
:::
|
||||
|
||||
## 自定义路由
|
||||
|
||||
这里以一个简单的管理页面为例：
|
||||
|
||||
```python
|
||||
import nonebot
|
||||
|
||||
bot = nonebot.get_bot() # 在此之前必须已经 init
|
||||
|
||||
@bot.server_app.route('/admin')
|
||||
async def admin():
|
||||
await bot.send_private_msg(12345678, '你的主页被访问了')
|
||||
return '欢迎来到管理页面'
|
||||
```
|
||||
|
||||
启动 NoneBot 后访问 <http://127.0.0.1:8080/admin>，你会看见管理页面的欢迎词，并收到机器人的提醒。
|
||||
|
||||
## 处理生命周期事件
|
||||
|
||||
有时可能需要在 NoneBot 启动时初始化数据库连接池，例如：
|
||||
|
||||
```python
|
||||
import nonebot
|
||||
|
||||
bot = nonebot.get_bot() # 在此之前必须已经 init
|
||||
|
||||
@bot.server_app.before_serving
|
||||
async def init_db():
|
||||
# 这会在 NoneBot 启动后立即运行
|
||||
pass
|
||||
```
|
||||
Reference in New Issue
Block a user