nonebot 源碼閱讀筆記

前言

nonebot 是一個 QQ 消息機器人框架，它的一些實現機制，值得參考。

• nonebot

NoneBot

初始化（配置加載）

閱讀 nonebot 文檔，第一個示例如下：

import nonebot

if __name__ == '__main__':
    nonebot.init()
    nonebot.load_builtin_plugins()
    nonebot.run(host='127.0.0.1', port=8080)

首先思考一下，要運行幾個 QQ 機器人，肯定是要保存一些動態的數據的。但是從上面的示例看，我們並沒有創建什么對象來保存動態數據，很簡單的就直接調用 nontbot.run() 了。這說明動態的數據被隱藏在了 nonebot 內部。

接下來詳細分析這幾行代碼：

第一步是 nonebot.init()，該方法源碼如下：

#  這個全局變量用於保存 NoneBot 對象
_bot: Optional[NoneBot] = None

def init(config_object: Optional[Any] = None) -> None:
    global _bot
    _bot = NoneBot(config_object)  # 通過傳入的配置對象，構造 NoneBot 實例。

    if _bot.config.DEBUG:   # 根據是否 debug 模式，來配置日志級別
        logger.setLevel(logging.DEBUG)
    else:
        logger.setLevel(logging.INFO)

    # 在 websocket 啟動前，先啟動 scheduler（通過調用 quart 的 before_serving 裝飾器）
    # 這實際上是將 _start_scheduler 包裝成一個 coroutine，然后丟到 quart 的 before_serving_funcs 隊列中去。
    _bot.server_app.before_serving(_start_scheduler)

def _start_scheduler():
    if scheduler and not scheduler.running:  # 這個 scheduler 是使用的 apscheduler.schedulers.asyncio.AsyncIOScheduler
        scheduler.configure(_bot.config.APSCHEDULER_CONFIG)  # 配置 scheduler 參數，該參數可通過 `nonebot.init()` 配置
        scheduler.start()  # 啟動 scheduler，用於定時任務（如定時發送消息、每隔一定時間執行某任務）
        logger.info('Scheduler started')

可以看到，nonebot.init() 做了三件事：

1. 通過傳入的配置對象，構造 NoneBot 實例。該實例對用戶不可見
2. 配置日志級別
3. 讓 quart 在服務啟動前，先啟動 AsyncIOScheduler。
   • AsyncIOScheduler 是一個異步 scheduler，這意味着它本身也會由 asyncio 的 eventloop 調度。它和 quart 是並發執行的。

1. plugins 加載機制

第二步是 nonebot.load_builtin_plugins()，直接加載了 nonebot 內置的插件。該函數來自 plugin.py：

class Plugin:
    __slots__ = ('module', 'name', 'usage')

    def __init__(self, module: Any,
                 name: Optional[str] = None,
                 usage: Optional[Any] = None):
        self.module = module  # 插件對象本身
        self.name = name   # 插件名稱
        self.usage = usage  # 插件的 help 字符串

# 和 `_bot` 類似的設計，用全局變量保存狀態
_plugins: Set[Plugin] = set()

def load_plugin(module_name: str) -> bool:
    try:
        module = importlib.import_module(module_name)   #  通過模塊名，動態 import 該模塊
        name = getattr(module, '__plugin_name__', None)
        usage = getattr(module, '__plugin_usage__', None)  # 模塊的全局變量
        _plugins.add(Plugin(module, name, usage))              # 將加載好的模塊放入 _plugins
        logger.info(f'Succeeded to import "{module_name}"')
        return True
    except Exception as e:
        logger.error(f'Failed to import "{module_name}", error: {e}')
        logger.exception(e)
        return False


def load_plugins(plugin_dir: str, module_prefix: str) -> int:
    count = 0
    for name in os.listdir(plugin_dir):  # 遍歷指定的文件夾
        path = os.path.join(plugin_dir, name)
        if os.path.isfile(path) and \
                (name.startswith('_') or not name.endswith('.py')):
            continue
        if os.path.isdir(path) and \
                (name.startswith('_') or not os.path.exists(
                    os.path.join(path, '__init__.py'))):
            continue

        m = re.match(r'([_A-Z0-9a-z]+)(.py)?', name)
        if not m:
            continue

        if load_plugin(f'{module_prefix}.{m.group(1)}'):  # 嘗試加載該模塊
            count += 1
    return count

def load_builtin_plugins() -> int:
    plugin_dir = os.path.join(os.path.dirname(__file__), 'plugins')  # 得到內部 plugins 目錄的路徑
    return load_plugins(plugin_dir, 'nonebot.plugins')  # 直接加載該目錄下的所有插件

def get_loaded_plugins() -> Set[Plugin]:
    """
    獲取所有加載好的插件，一般用於提供命令幫助。
    比如在收到 "幫助 拆字" 時，就從這里查詢到 “拆字” 插件的 usage，返回給用戶。

    :return: a set of Plugin objects
    """
    return _plugins

這就是插件的動態加載機制，可以看到獲取已加載插件的唯一方法，就是 get_loaded_plugins()，而且 plugins 是用集合來保存的。

1. 優化：仔細想想，我覺得用字典（Dict）來代替 Set 會更好一些，用“插件名”索引，這樣可以防止出現同名的插件，而且查詢插件時也不需要遍歷整個 Set。

2. 思考：插件是 python 模塊，但是這里加載好了，卻沒有手動將它注冊到別的地方，那加載它還有什么用？

   • 插件中的“命令解析器”、“消息處理器”等，都是使用的是 nonebot 的裝飾器裝飾了的。
   • 該裝飾器會直接將命令處理函數，連同命令解析參數等直接注冊到 nonebot 的命令集合中。（這個后面會看到。）因此不需要在 load_plugin() 中手動注冊。

這兩行之后，就直接 nonebot.run() 啟動 quart 服務器了。

QQ消息的處理

從第一個例子中，只能看到上面這些。接下來考慮寫一個自定義插件，看看 nonebot 的消息處理機制。項目結構如下：

awesome-bot
├── awesome
│   └── plugins
│       └── usage.py
├── bot.py
└── config.py  # 配置文件，寫法參考 nonebot.default_config，建議使用類方式保存配置

bot.py:

from os import path

import nonebot
import config

if __name__ == '__main__':
    nonebot.init(config)  # 使用自定義配置
    nonebot.load_plugins(  # 加載 awesome/plugins 下的自定義插件
        path.join(path.dirname(__file__), 'awesome', 'plugins'),
        'awesome.plugins'
    )
    nonebot.run()

usage.py:

import nonebot
from nonebot import on_command, CommandSession


@on_command('usage', aliases=['使用幫助', '幫助', '使用方法'])
async def _(session: CommandSession):
    """之前說過的“幫助”命令"""
    plugins = list(filter(lambda p: p.name, nonebot.get_loaded_plugins()))
    arg = session.current_arg_text.strip().lower()
    if not arg:
        session.finish(
            '我現在支持的功能有：\n\n' + '\n'.join(p.name for p in plugins))
    for p in plugins:  # 如果 plugins 換成 dict 類型，就不需要循環遍歷了
        if p.name.lower() == arg:
            await session.send(p.usage)

查看裝飾器 on_command 的內容，有 command/__init__.py：

# key: one segment of command name
# value: subtree or a leaf Command object
_registry = {}  # type: Dict[str, Union[Dict, Command]]  # 保存命令與命令處理器

# key: alias
# value: real command name
_aliases = {}  # type: Dict[str, CommandName_T]  # 保存命令的別名（利用別名，從這里查找真正的命令名稱，再用該名稱查找命令處理器）

# key: context id
# value: CommandSession object
_sessions = {}  # type: Dict[str, CommandSession]  # 保存與用戶的會話，這樣就能支持一些需要關聯上下文的命令。比如賽文續傳，或者需要花一定時間執行的命令，Session 有個 is_running。


def on_command(name: Union[str, CommandName_T], *,
               aliases: Iterable[str] = (),
               permission: int = perm.EVERYBODY,
               only_to_me: bool = True,
               privileged: bool = False,
               shell_like: bool = False) -> Callable:
    """
    用於注冊命令處理器

    :param name: 命令名稱 (e.g. 'echo' or ('random', 'number'))
    :param aliases: 命令別名，建議用元組
    :param permission: 該命令的默認權限
    :param only_to_me: 是否只處理發送給“我”的消息
    :param privileged: 已經存在此 session 時，是否仍然能被運行
    :param shell_like: 使用類似 shell 的語法傳遞參數
    """

    def deco(func: CommandHandler_T) -> CommandHandler_T:
        if not isinstance(name, (str, tuple)):
            raise TypeError('the name of a command must be a str or tuple')
        if not name:
            raise ValueError('the name of a command must not be empty')

        cmd_name = (name,) if isinstance(name, str) else name

        cmd = Command(name=cmd_name, func=func, permission=permission,
                      only_to_me=only_to_me, privileged=privileged)  # 構造命令處理器
        if shell_like:
            async def shell_like_args_parser(session):
                session.args['argv'] = shlex.split(session.current_arg)

            cmd.args_parser_func = shell_like_args_parser

        current_parent = _registry
        for parent_key in cmd_name[:-1]:  # 循環將命令樹添加到 _registry
            current_parent[parent_key] = current_parent.get(parent_key) or {}
            current_parent = current_parent[parent_key]
        current_parent[cmd_name[-1]] = cmd

        for alias in aliases:  # 保存命令別名
            _aliases[alias] = cmd_name

        return CommandFunc(cmd, func)

    return deco

該裝飾器將命令處理器注冊到模塊的全局變量中，然后 quart 在收到消息時，會調用該模塊的如下方法，查找對應的命令處理器，並使用它處理該命令：

async def handle_command(bot: NoneBot, ctx: Context_T) -> bool:
    """
    嘗試將消息解析為命令，如果解析成功，而且用戶擁有權限，就執行該命令。否則忽略。

    此函數會被 "handle_message" 調用
    """
    cmd, current_arg = parse_command(bot, str(ctx['message']).lstrip())  # 嘗試解析該命令
    is_privileged_cmd = cmd and cmd.privileged
    if is_privileged_cmd and cmd.only_to_me and not ctx['to_me']:
        is_privileged_cmd = False
    disable_interaction = is_privileged_cmd

    if is_privileged_cmd:
        logger.debug(f'Command {cmd.name} is a privileged command')

    ctx_id = context_id(ctx)

    if not is_privileged_cmd:
        # wait for 1.5 seconds (at most) if the current session is running
        retry = 5
        while retry > 0 and \
                _sessions.get(ctx_id) and _sessions[ctx_id].running:
            retry -= 1
            await asyncio.sleep(0.3)

    check_perm = True
    session = _sessions.get(ctx_id) if not is_privileged_cmd else None
    if session:
        if session.running:
            logger.warning(f'There is a session of command '
                           f'{session.cmd.name} running, notify the user')
            asyncio.ensure_future(send(
                bot, ctx,
                render_expression(bot.config.SESSION_RUNNING_EXPRESSION)
            ))
            # pretend we are successful, so that NLP won't handle it
            return True

        if session.is_valid:
            logger.debug(f'Session of command {session.cmd.name} exists')
            # since it's in a session, the user must be talking to me
            ctx['to_me'] = True
            session.refresh(ctx, current_arg=str(ctx['message']))
            # there is no need to check permission for existing session
            check_perm = False
        else:
            # the session is expired, remove it
            logger.debug(f'Session of command {session.cmd.name} is expired')
            if ctx_id in _sessions:
                del _sessions[ctx_id]
            session = None

    if not session:
        if not cmd:
            logger.debug('Not a known command, ignored')
            return False
        if cmd.only_to_me and not ctx['to_me']:
            logger.debug('Not to me, ignored')
            return False
        session = CommandSession(bot, ctx, cmd, current_arg=current_arg)  # 構造命令 Session，某些上下文相關的命令需要用到。
        logger.debug(f'New session of command {session.cmd.name} created')

    return await _real_run_command(session, ctx_id, check_perm=check_perm,  # 這個函數將命令處理函數包裝成 task，然后等待該 task 完成，再返回結果。
                                   disable_interaction=disable_interaction)

Web 中的 Session 一般是用於保存登錄狀態，而聊天程序的 session，則主要是保存上下文。

如果要做賽文續傳與成績統計，Session 和 Command 肯定是需要的，但是不能像 nonebot 這樣做。
NoneBot 的命令格式限制得比較嚴，沒法用來解析跟打器自動發送的成績消息。也許命令應該更寬松：

1. 命令前綴仍然通過全局配置來做，但是用 dict 來存，給每個前綴一個名字，默認使用 default。
   • @command 應該給一個參數用於指定前綴：None 為不需要前綴，默認為 config.Prefix.DEFAULT.
2. 添加一個正則消息匹配的命令注冊器，要匹配多個正則，則多次使用該裝飾器。正則匹配到的 groupdict 會被傳到命令處理器中。

其他

還有就是 NoneBot 作者提到的一些問題：

1. 基於 python-aiocqhttp（跟 酷Q 強耦合），無法支持其它機器人平台：我寫 xhup-bot 時，也需要把這一點考慮進去。機器人核心不應該依賴任何平台相關的東西。
2. 過於以命令為核心：這也是我體會到的。這導致很多功能無法使用 nonebot 實現。只能借助底層的 on_message。
3. 沒有全局黑名單機制，無法簡單地屏蔽其它 bot 的消息。全局黑名單感覺還算比較容易做。
4. 權限控制功能不夠強大，無法進行單用戶和群組粒度的控制：我這邊也有考慮這個。
   • 細粒度權限控制的話，可以將 on_command 的 permission 當成該命令的默認權限。然后可以在 config 里針對不同的群/用戶，添加不同的權限。
   • 但是這可能會導致配置變復雜。最好還是通過后端提供的 Web 網頁來配置。每個群管理都可以自己配置自己群的一些權限。然后 bot 在啟動時通過 http 從后端獲取配置信息。
   • 會話只針對單用戶，無法簡單地實現多用戶游戲功能：這個我暫時不需要。。而且我的 xhup-bot 是有后端的，我覺得這個可以放到后端做。

本文為個人雜談，不保證正確。如有錯誤，還請指正。