t0nyx1ang/nonebot-plugin-scheduled-broadcast

An event-based scheduled broadcaster aiming for all nonebot2 adapters.

Nonebot Plugin Scheduled Broadcast

✨ 一款可配置的, 不依赖具体适配器的, 基于事件的定时广播插件. ✨

特性

• 基于一般事件 nonebot.adapters.Event 构造, 从而支持全部协议的信息发送.
• 定时广播由文件配置, 基于 apscheduler 提供类 crontab 的配置参数.

安装

使用 nb-cli 安装

• 在 nonebot2 项目的根目录下打开命令行, 输入以下指令即可安装:

    nb-cli plugin install nonebot-plugin-scheduled-broadcast

使用包管理器安装

• 使用 pdm 安装:

    pdm add nonebot-plugin-scheduled-broadcast

• 使用 poetry 安装:

    poetry add nonebot-plugin-scheduled-broadcast

• 使用 pip 安装: (虽然不太推荐但大概率没事)

    pip install nonebot-plugin-scheduled-broadcast

使用

启动广播

• 保证发送消息的 id 位于 SUPERUSER 用户组中.
• 在需要启动广播的地方发送:

    启动广播/enablebc [-bid 广播ID]

• 广播ID 是可选项, 如果用户没有输入 广播ID, 则会使用对应事件的 session_id 去掉 user_id 的结果作为 广播ID, 如果此时 广播ID 为空, 则不支持自动生成 广播ID.
• 如果是第一次使用, 机器人在执行上述命令后, 在 nonebot2 项目的根目录下生成一个名为 broadcast_policy.json 的配置文件.
• 如果该机器人下已经存在相同的 广播ID, 配置文件将会保留, 同时广播作业将被 恢复, 如果不需要手动修改配置文件, 不需要重启.

待触发指令的编写

• 由于本插件提供了一个装饰器, 可以按照如下方式编写待触发指令:

from nonebot import require
from nonebot.adapters import Event
from nonebot.log import logger

require("nonebot_plugin_scheduled_broadcast")

from nonebot_plugin_scheduled_broadcast import broadcast

@broadcast('example')
async def _(self_id: str, event: Event):
    """Scheduled example broadcast."""
    message = generate_your_message()
    try:
        bot = nonebot.get_bots()[self_id]  # select the target bot
        await bot.send(event=event, message=message)  # send message
    except Exception:
        logger.error(traceback.format_exc())  # print logs

• 触发指令编写完成以后, 启动机器人即可.

配置待触发指令的广播时间

• 保证发送消息的 id 位于 SUPERUSER 用户组中.
• 在需要启动广播的地方发送:

    设置广播/setbc 待触发指令 [-bid 广播ID] [-s 秒] [-m 分] [-h 时] [-w 周数] [-d 星期几] [-D 日] [-M 月] [-Y 年]

• 广播ID 是可选项, 如果用户没有输入 广播ID, 则会使用对应事件的 session_id 去掉 user_id 的结果作为 广播ID, 如果此时 广播ID 为空, 则不支持自动生成 广播ID.
• 配置项支持不含空格的 apscheduler 语法, 当配置项留空时, 将会使得该指令不被触发.

关闭广播

• 保证发送消息的 id 位于 SUPERUSER 用户组中.
• 在需要关闭广播的地方发送:

    关闭广播/disablebc [-bid 广播ID]

• 广播ID 是可选项, 如果用户没有输入 广播ID, 则会使用对应事件的 session_id 去掉 user_id 的结果作为 广播ID, 如果此时 广播ID 为空, 则不支持自动生成 广播ID.
• 停止广播之后, 对应 广播ID 中的 enable 将被置为 false, 为保持可复用性, 其余部分将不会改变, 同时广播作业将被 暂停, 如果不需要修改 config 中的内容, 不需要重启.

配置

• broadcast_policy_location: 代表配置文件的存放位置, 默认值为 ./broadcast_policy.json

注意事项

• 由于每一个 Event 几乎不会相同, 建议不要在同一个地方多次执行 启动广播 命令, 可能会刷屏的.
• 由于用到了 pickle 的序列化和反序列化功能, 而该功能具有潜在的安全风险, 请谨慎对待来源不明的配置文件. 如果不确定配置文件的安全性, 建议重新启动广播以生成新的 广播ID, 然后替换配置文件中的 config 键里面的内容.
• 目前版本使用了 nonebot.driver.Driver 中的 on_bot_connect 和 on_bot_disconnect 钩子函数用于检测 Bot 的连接与断开, 但是适配器 adapter-console 中检测断开的时机过晚, 导致终端退出时 scheduler 仍然在后台运行, 不建议在生产环境引入该适配器.

配置文件的手动填写

• 一般而言, 配置文件不需要手动修改, 但是本指南仍然会介绍配置文件的结构:

{
    "self_id": {
        "broadcast_id": {
            "config": {
                "example": {"second": "*/30"}
            },
            "data": "event data (b64encode)",
            "hash": "event hash (sha256)",
            "enable": true
        },
        "another broadcast_id": {
            // ...
        }
    },
    "another self_id": {
        // ...
    }
}

• config 的可用配置与 apscheduler 一致, 以下给出一个示例, 该示例表示, example 命令会在每天 10:24, 每隔 10 秒被触发:

{
    "config": {
        "example": {
            "hour": 10,
            "minute": 24,
            "second": "*/10"
        }
    }
}

协议

• 本项目使用 MIT 协议

nonebot-plugin-scheduled-broadcast's Issues

[Feature] 自动发布更新机制

当推送 tag 时，发布至 pypi 并推送更新

[Bug] 如果配置文件有多余的键，会导致定时任务无法暂停或者恢复

pause和resume函数需要修改，只把创建了装饰器的命令名称记录下来

[Feature] 仅当 Bot 连接时恢复 scheduler，在 Bot 断开连接时暂停 scheduler

nonebot2.driver 中存在与 Bot 连接相关的钩子函数，可以将其用于检测 Bot 的连接情况

[Feature] 不重启机器人，对已有指令进行在线配置

由于引入了 valid_command 项，对已有指令进行在线配置成为可能。
在某个 broadcast_id 下，配置完成之后或许可以立即刷新 scheduler 的内容。
配置可以采用 apscheduler 自带的 crontab-like 格式。

[Feature] 使用nb2提供的session_id代替用户输入广播id

session_id一般来说包括了事件发生的位置和触发事件的人物，因此将session_id中的user_id信息清除后，应该能够得到一个合理的地点信息

[Bug] event 中可能没有 self_id

建议改成在事件的bot中获取self_id

[Breaking Feature] event 不使用 pickle 进行序列化

警告：这会引起破坏性更新，需要给出兼容的转换工具。

由于 pickle 的反序列化存在安全问题，且手动修改不够方便，使用 json 格式来进行序列化是比较好的方案。但是由于 adapters.event 类是一个抽象基类，实际发送信息需要得知真实的类名是什么，否则是无法反序列化的。但是我们现在拥有了 inspect 和 importlib 工具，能够得到 bot.send 方法中用到的对应 adapter 的基类，这个基类就不是抽象基类了，这样就能够反序列化了。

代码有两种实现方案：

1. 基于send函数中的类型注解，在send函数不存在类型注解的时候就会报错，层级在bot，获得event的方法必须在用户端实现（因为不确定具体的bot支持怎样的event）：

import inspect

async def _(bot: Bot, event: Event):
    dump = event.dict()
    sig = inspect.signature(bot.send)
    event1 = sig.parameters["event"].annotation.parse_obj(dump)

2. 基于原event的具体类型，好处是不需要依赖类型注解，层级在event，获得event的方法可以在broadcast装饰器内实现，坏处就是json文件里面需要多两个键来记__module__和__name__：

import importlib

async def _(bot: Bot, event: Event):
    dump = event.dict()
    c_event = importlib.import_module(event.__module__)
    m_event = getattr(c_event, event.__class__.__name__)
    event1 = m_event.parse_obj(dump)

[Feature] 建议将 broadcast_id 改为自定义

目前 broadcast_id 非常长，很难记。建议将其改为用户自定义 broadcast_id，仅去校验id是否重复即可