nonebot 源碼閱讀筆記
阿新 • • 發佈：2019-03-02
ont 第一步 狀態 ror 集合 分析 check 函數 slot
前言
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.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() 做了三件事：
- 通過傳入的配置對象，構造 NoneBot 實例。該實例對用戶不可見
- 配置日誌級別
- 讓 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 是用集合來保存的。
優化：仔細想想，我覺得用字典（Dict）來代替 Set 會更好一些，用“插件名”索引，這樣可以防止出現同名的插件，而且查詢插件時也不需要遍歷整個 Set。
- 思考：插件是 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 的命令格式限制得比較嚴，沒法用來解析跟打器自動發送的成績消息。也許命令應該更寬松：
- 命令前綴仍然通過全局配置來做，但是用 dict 來存，給每個前綴一個名字，默認使用 default。
- @command 應該給一個參數用於指定前綴：None 為不需要前綴，默認為 config.Prefix.DEFAULT.
- 添加一個正則消息匹配的命令註冊器，要匹配多個正則，則多次使用該裝飾器。正則匹配到的 groupdict 會被傳到命令處理器中。
其他
還有就是 NoneBot 作者提到的一些問題：
- 基於 python-aiocqhttp（跟 酷Q 強耦合），無法支持其它機器人平臺：我寫 xhup-bot 時，也需要把這一點考慮進去。機器人核心不應該依賴任何平臺相關的東西。
- 過於以命令為核心：這也是我體會到的。這導致很多功能無法使用 nonebot 實現。只能借助底層的 on_message。
- 沒有全局黑名單機制，無法簡單地屏蔽其它 bot 的消息。全局黑名單感覺還算比較容易做。
- 權限控制功能不夠強大，無法進行單用戶和群組粒度的控制：我這邊也有考慮這個。
- 細粒度權限控制的話，可以將 on_command 的 permission 當成該命令的默認權限。然後可以在 config 裏針對不同的群/用戶，添加不同的權限。
- 但是這可能會導致配置變復雜。最好還是通過後端提供的 Web 網頁來配置。每個群管理都可以自己配置自己群的一些權限。然後 bot 在啟動時通過 http 從後端獲取配置信息。
- 會話只針對單用戶，無法簡單地實現多用戶遊戲功能：這個我暫時不需要。。而且我的 xhup-bot 是有後端的，我覺得這個可以放到後端做。
nonebot 源碼閱讀筆記