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 是有後端的，我以爲這個能夠放到後端作。

本文爲我的雜談，不保證正確。若有錯誤，還請指正。