nonebot2/nonebot/__init__.py

"""本模块主要定义了 NoneBot 启动所需函数，供 bot 入口文件调用。

## 快捷导入

为方便使用，本模块从子模块导入了部分内容，以下内容可以直接通过本模块导入:

- `on` => {ref}``on` <nonebot.plugin.on.on>`
- `on_metaevent` => {ref}``on_metaevent` <nonebot.plugin.on.on_metaevent>`
- `on_message` => {ref}``on_message` <nonebot.plugin.on.on_message>`
- `on_notice` => {ref}``on_notice` <nonebot.plugin.on.on_notice>`
- `on_request` => {ref}``on_request` <nonebot.plugin.on.on_request>`
- `on_startswith` => {ref}``on_startswith` <nonebot.plugin.on.on_startswith>`
- `on_endswith` => {ref}``on_endswith` <nonebot.plugin.on.on_endswith>`
- `on_fullmatch` => {ref}``on_fullmatch` <nonebot.plugin.on.on_fullmatch>`
- `on_keyword` => {ref}``on_keyword` <nonebot.plugin.on.on_keyword>`
- `on_command` => {ref}``on_command` <nonebot.plugin.on.on_command>`
- `on_shell_command` => {ref}``on_shell_command` <nonebot.plugin.on.on_shell_command>`
- `on_regex` => {ref}``on_regex` <nonebot.plugin.on.on_regex>`
- `on_type` => {ref}``on_type` <nonebot.plugin.on.on_type>`
- `CommandGroup` => {ref}``CommandGroup` <nonebot.plugin.on.CommandGroup>`
- `Matchergroup` => {ref}``MatcherGroup` <nonebot.plugin.on.MatcherGroup>`
- `load_plugin` => {ref}``load_plugin` <nonebot.plugin.load.load_plugin>`
- `load_plugins` => {ref}``load_plugins` <nonebot.plugin.load.load_plugins>`
- `load_all_plugins` => {ref}``load_all_plugins` <nonebot.plugin.load.load_all_plugins>`
- `load_from_json` => {ref}``load_from_json` <nonebot.plugin.load.load_from_json>`
- `load_from_toml` => {ref}``load_from_toml` <nonebot.plugin.load.load_from_toml>`
- `load_builtin_plugin` =>
  {ref}``load_builtin_plugin` <nonebot.plugin.load.load_builtin_plugin>`
- `load_builtin_plugins` =>
  {ref}``load_builtin_plugins` <nonebot.plugin.load.load_builtin_plugins>`
- `get_plugin` => {ref}``get_plugin` <nonebot.plugin.get_plugin>`
- `get_plugin_by_module_name` =>
  {ref}``get_plugin_by_module_name` <nonebot.plugin.get_plugin_by_module_name>`
- `get_loaded_plugins` =>
  {ref}``get_loaded_plugins` <nonebot.plugin.get_loaded_plugins>`
- `get_available_plugin_names` =>
  {ref}``get_available_plugin_names` <nonebot.plugin.get_available_plugin_names>`
- `require` => {ref}``require` <nonebot.plugin.load.require>`

FrontMatter:
    sidebar_position: 0
    description: nonebot 模块
"""

import os
from importlib.metadata import version
from typing import Any, Dict, Type, Union, TypeVar, Optional, overload

import loguru

from nonebot.log import logger as logger
from nonebot.adapters import Bot, Adapter
from nonebot.config import DOTENV_TYPE, Env, Config
from nonebot.utils import escape_tag, resolve_dot_notation
from nonebot.drivers import Driver, ASGIMixin, combine_driver

try:
    __version__ = version("nonebot2")
except Exception:  # pragma: no cover
    __version__ = None

A = TypeVar("A", bound=Adapter)

_driver: Optional[Driver] = None


def get_driver() -> Driver:
    """获取全局 {ref}`nonebot.drivers.Driver` 实例。

    可用于在计划任务的回调等情形中获取当前 {ref}`nonebot.drivers.Driver` 实例。

    返回:
        全局 {ref}`nonebot.drivers.Driver` 对象

    异常:
        ValueError: 全局 {ref}`nonebot.drivers.Driver` 对象尚未初始化
            ({ref}`nonebot.init <nonebot.init>` 尚未调用)

    用法:
        ```python
        driver = nonebot.get_driver()
        ```
    """
    if _driver is None:
        raise ValueError("NoneBot has not been initialized.")
    return _driver


@overload
def get_adapter(name: str) -> Adapter:
    """
    参数:
        name: 适配器名称

    返回:
        指定名称的 {ref}`nonebot.adapters.Adapter` 对象
    """


@overload
def get_adapter(name: Type[A]) -> A:
    """
    参数:
        name: 适配器类型

    返回:
        指定类型的 {ref}`nonebot.adapters.Adapter` 对象
    """


def get_adapter(name: Union[str, Type[Adapter]]) -> Adapter:
    """获取已注册的 {ref}`nonebot.adapters.Adapter` 实例。

    异常:
        ValueError: 指定的 {ref}`nonebot.adapters.Adapter` 未注册
        ValueError: 全局 {ref}`nonebot.drivers.Driver` 对象尚未初始化
            ({ref}`nonebot.init <nonebot.init>` 尚未调用)

    用法:
        ```python
        from nonebot.adapters.console import Adapter
        adapter = nonebot.get_adapter(Adapter)
        ```
    """
    adapters = get_adapters()
    target = name if isinstance(name, str) else name.get_name()
    if target not in adapters:
        raise ValueError(f"Adapter {target} not registered.")
    return adapters[target]


def get_adapters() -> Dict[str, Adapter]:
    """获取所有已注册的 {ref}`nonebot.adapters.Adapter` 实例。

    返回:
        所有 {ref}`nonebot.adapters.Adapter` 实例字典

    异常:
        ValueError: 全局 {ref}`nonebot.drivers.Driver` 对象尚未初始化
            ({ref}`nonebot.init <nonebot.init>` 尚未调用)

    用法:
        ```python
        adapters = nonebot.get_adapters()
        ```
    """
    return get_driver()._adapters.copy()


def get_app() -> Any:
    """获取全局 {ref}`nonebot.drivers.ASGIMixin` 对应的 Server App 对象。

    返回:
        Server App 对象

    异常:
        AssertionError: 全局 Driver 对象不是 {ref}`nonebot.drivers.ASGIMixin` 类型
        ValueError: 全局 {ref}`nonebot.drivers.Driver` 对象尚未初始化
            ({ref}`nonebot.init <nonebot.init>` 尚未调用)

    用法:
        ```python
        app = nonebot.get_app()
        ```
    """
    driver = get_driver()
    assert isinstance(driver, ASGIMixin), "app object is only available for asgi driver"
    return driver.server_app


def get_asgi() -> Any:
    """获取全局 {ref}`nonebot.drivers.ASGIMixin` 对应的
    [ASGI](https://asgi.readthedocs.io/) 对象。

    返回:
        ASGI 对象

    异常:
        AssertionError: 全局 Driver 对象不是 {ref}`nonebot.drivers.ASGIMixin` 类型
        ValueError: 全局 {ref}`nonebot.drivers.Driver` 对象尚未初始化
            ({ref}`nonebot.init <nonebot.init>` 尚未调用)

    用法:
        ```python
        asgi = nonebot.get_asgi()
        ```
    """
    driver = get_driver()
    assert isinstance(
        driver, ASGIMixin
    ), "asgi object is only available for asgi driver"
    return driver.asgi


def get_bot(self_id: Optional[str] = None) -> Bot:
    """获取一个连接到 NoneBot 的 {ref}`nonebot.adapters.Bot` 对象。

    当提供 `self_id` 时，此函数是 `get_bots()[self_id]` 的简写；
    当不提供时，返回一个 {ref}`nonebot.adapters.Bot`。

    参数:
        self_id: 用来识别 {ref}`nonebot.adapters.Bot` 的
            {ref}`nonebot.adapters.Bot.self_id` 属性

    返回:
        {ref}`nonebot.adapters.Bot` 对象

    异常:
        KeyError: 对应 self_id 的 Bot 不存在
        ValueError: 没有传入 self_id 且没有 Bot 可用
        ValueError: 全局 {ref}`nonebot.drivers.Driver` 对象尚未初始化
            ({ref}`nonebot.init <nonebot.init>` 尚未调用)

    用法:
        ```python
        assert nonebot.get_bot("12345") == nonebot.get_bots()["12345"]

        another_unspecified_bot = nonebot.get_bot()
        ```
    """
    bots = get_bots()
    if self_id is not None:
        return bots[self_id]

    for bot in bots.values():
        return bot

    raise ValueError("There are no bots to get.")


def get_bots() -> Dict[str, Bot]:
    """获取所有连接到 NoneBot 的 {ref}`nonebot.adapters.Bot` 对象。

    返回:
        一个以 {ref}`nonebot.adapters.Bot.self_id` 为键
        {ref}`nonebot.adapters.Bot` 对象为值的字典

    异常:
        ValueError: 全局 {ref}`nonebot.drivers.Driver` 对象尚未初始化
            ({ref}`nonebot.init <nonebot.init>` 尚未调用)

    用法:
        ```python
        bots = nonebot.get_bots()
        ```
    """
    return get_driver().bots


def _resolve_combine_expr(obj_str: str) -> Type[Driver]:
    drivers = obj_str.split("+")
    DriverClass = resolve_dot_notation(
        drivers[0], "Driver", default_prefix="nonebot.drivers."
    )
    if len(drivers) == 1:
        logger.trace(f"Detected driver {DriverClass} with no mixins.")
        return DriverClass
    mixins = [
        resolve_dot_notation(mixin, "Mixin", default_prefix="nonebot.drivers.")
        for mixin in drivers[1:]
    ]
    logger.trace(f"Detected driver {DriverClass} with mixins {mixins}.")
    return combine_driver(DriverClass, *mixins)


def _log_patcher(record: "loguru.Record"):
    record["name"] = (
        plugin.name
        if (module_name := record["name"])
        and (plugin := get_plugin_by_module_name(module_name))
        else (module_name and module_name.split(".")[0])
    )


def init(*, _env_file: Optional[DOTENV_TYPE] = None, **kwargs: Any) -> None:
    """初始化 NoneBot 以及 全局 {ref}`nonebot.drivers.Driver` 对象。

    NoneBot 将会从 .env 文件中读取环境信息，并使用相应的 env 文件配置。

    也可以传入自定义的 `_env_file` 来指定 NoneBot 从该文件读取配置。

    参数:
        _env_file: 配置文件名，默认从 `.env.{env_name}` 中读取配置
        kwargs: 任意变量，将会存储到 {ref}`nonebot.drivers.Driver.config` 对象里

    用法:
        ```python
        nonebot.init(database=Database(...))
        ```
    """
    global _driver
    if not _driver:
        logger.success("NoneBot is initializing...")
        env = Env()
        _env_file = _env_file or f".env.{env.environment}"
        config = Config(
            **kwargs,
            _env_file=(".env", _env_file)
            if isinstance(_env_file, (str, os.PathLike))
            else _env_file,
        )

        logger.configure(
            extra={"nonebot_log_level": config.log_level}, patcher=_log_patcher
        )
        logger.opt(colors=True).info(
            f"Current <y><b>Env: {escape_tag(env.environment)}</b></y>"
        )
        logger.opt(colors=True).debug(
            f"Loaded <y><b>Config</b></y>: {escape_tag(str(config.dict()))}"
        )

        DriverClass = _resolve_combine_expr(config.driver)
        _driver = DriverClass(env, config)


def run(*args: Any, **kwargs: Any) -> None:
    """启动 NoneBot，即运行全局 {ref}`nonebot.drivers.Driver` 对象。

    参数:
        args: 传入 {ref}`nonebot.drivers.Driver.run` 的位置参数
        kwargs: 传入 {ref}`nonebot.drivers.Driver.run` 的命名参数

    用法:
        ```python
        nonebot.run(host="127.0.0.1", port=8080)
        ```
    """
    logger.success("Running NoneBot...")
    get_driver().run(*args, **kwargs)


from nonebot.plugin import on as on
from nonebot.plugin import on_type as on_type
from nonebot.plugin import require as require
from nonebot.plugin import on_regex as on_regex
from nonebot.plugin import on_notice as on_notice
from nonebot.plugin import get_plugin as get_plugin
from nonebot.plugin import on_command as on_command
from nonebot.plugin import on_keyword as on_keyword
from nonebot.plugin import on_message as on_message
from nonebot.plugin import on_request as on_request
from nonebot.plugin import load_plugin as load_plugin
from nonebot.plugin import on_endswith as on_endswith
from nonebot.plugin import CommandGroup as CommandGroup
from nonebot.plugin import MatcherGroup as MatcherGroup
from nonebot.plugin import load_plugins as load_plugins
from nonebot.plugin import on_fullmatch as on_fullmatch
from nonebot.plugin import on_metaevent as on_metaevent
from nonebot.plugin import on_startswith as on_startswith
from nonebot.plugin import load_from_json as load_from_json
from nonebot.plugin import load_from_toml as load_from_toml
from nonebot.plugin import load_all_plugins as load_all_plugins
from nonebot.plugin import on_shell_command as on_shell_command
from nonebot.plugin import get_loaded_plugins as get_loaded_plugins
from nonebot.plugin import load_builtin_plugin as load_builtin_plugin
from nonebot.plugin import load_builtin_plugins as load_builtin_plugins
from nonebot.plugin import get_plugin_by_module_name as get_plugin_by_module_name
from nonebot.plugin import get_available_plugin_names as get_available_plugin_names