Skip to content

# Module `liteyuki.bot` ​

### class `LiteyukiBot` ​

#### func `__init__(self, **kwargs) -> None` ​

Description: 初始化轻雪实例

Arguments:

> - **kwargs: 配置

Source code or View on GitHub
python

    def __init__(self, **kwargs) -> None:
        """常规操作"""
        print_logo()
        global _BOT_INSTANCE
        _BOT_INSTANCE = self
        '配置'
        self.config: dict[str, Any] = kwargs
        '初始化'
        self.init(**self.config)
        logger.info('Liteyuki is initializing...')
        '生命周期管理'
        self.lifespan = Lifespan()
        self.process_manager: ProcessManager = ProcessManager(lifespan=self.lifespan)
        '事件循环'
        self.loop = asyncio.new_event_loop()
        asyncio.set_event_loop(self.loop)
        self.stop_event = threading.Event()
        self.call_restart_count = 0
        '加载插件加载器'
        load_plugin('liteyuki.plugins.plugin_loader')

#### func `run(self)` ​

Description: 外部启动接口

Source code or View on GitHub
python

    def run(self):
        self.process_manager.start_all()
        try:
            asyncio.run(self._run())
        except KeyboardInterrupt:
            logger.opt(colors=True).info('<y>Liteyuki is stopping...</y>')
            self.stop()
        logger.opt(colors=True).info('<y>Liteyuki is stopped...</y>')

#### async func `keep_alive(self)` ​

Description: 保持轻雪运行

Source code or View on GitHub
python

    async def keep_alive(self):
        logger.info('Liteyuki is keeping alive...')
        try:
            while not self.stop_event.is_set():
                await asyncio.sleep(0.1)
        except Exception:
            logger.info('Liteyuki is exiting...')
            self.stop()

#### func `restart(self, delay: int = 0)` ​

Description: 重启轻雪本体

Arguments:

> - delay (`int`, optional): 延迟重启时间. Defaults to 0.

Source code or View on GitHub
python

    def restart(self, delay: int=0):
        if self.call_restart_count < 1:
            executable = sys.executable
            args = sys.argv
            logger.info('Restarting LiteyukiBot...')
            time.sleep(delay)
            if platform.system() == 'Windows':
                cmd = 'start'
            elif platform.system() == 'Linux':
                cmd = 'nohup'
            elif platform.system() == 'Darwin':
                cmd = 'open'
            else:
                cmd = 'nohup'
            self.process_manager.terminate_all()
            threading.Thread(target=os.system, args=(f"{cmd} {executable} {' '.join(args)}",), daemon=True).start()
            sys.exit(0)
        self.call_restart_count += 1

#### func `restart_process(self, name: Optional[str] = None)` ​

Description: 停止轻雪

Arguments:

> - name (`Optional`[`str`]): 进程名. Defaults to None.

Source code or View on GitHub
python

    def restart_process(self, name: Optional[str]=None):
        if name is not None:
            chan_active = get_channel(f'{name}-active')
            chan_active.send(1)
        else:
            for process_name in self.process_manager.processes:
                chan_active = get_channel(f'{process_name}-active')
                chan_active.send(1)

#### func `init(self, *args, **kwargs)` ​

Description: 初始化轻雪, 自动调用

Arguments:

> - *args: 参数
> - **kwargs: 关键字参数

Source code or View on GitHub
python

    def init(self, *args, **kwargs):
        self.init_logger()

#### func `init_logger(self)` ​

Description: 初始化日志

Source code or View on GitHub
python

    def init_logger(self):
        init_log(config=self.config)

#### func `stop(self)` ​

Description: 停止轻雪

Source code or View on GitHub
python

    def stop(self):
        self.process_manager.terminate_all()
        self.stop_event.set()

#### func `on_before_start(self, func: LIFESPAN_FUNC) -> LIFESPAN_FUNC` ​

Description: 注册启动前的函数

Arguments:

> - func (`LIFESPAN_FUNC`): 生命周期函数

Return: `LIFESPAN_FUNC`: 生命周期函数

Source code or View on GitHub
python

    def on_before_start(self, func: LIFESPAN_FUNC) -> LIFESPAN_FUNC:
        return self.lifespan.on_before_start(func)

#### func `on_after_start(self, func: LIFESPAN_FUNC)` ​

Description: 注册启动后的函数

Arguments:

> - func (`LIFESPAN_FUNC`): 生命周期函数

Return: `LIFESPAN_FUNC`: 生命周期函数

Source code or View on GitHub
python

    def on_after_start(self, func: LIFESPAN_FUNC):
        return self.lifespan.on_after_start(func)

#### func `on_after_shutdown(self, func: LIFESPAN_FUNC)` ​

Description: 注册停止后的函数：未实现

Arguments:

> - func (`LIFESPAN_FUNC`): 生命周期函数

Return: `LIFESPAN_FUNC`: 生命周期函数

Source code or View on GitHub
python

    def on_after_shutdown(self, func: LIFESPAN_FUNC):
        return self.lifespan.on_after_shutdown(func)

#### func `on_before_process_shutdown(self, func: PROCESS_LIFESPAN_FUNC)` ​

Description: 注册进程停止前的函数，为子进程停止时调用

Arguments:

> - func (`PROCESS_LIFESPAN_FUNC`): 生命周期函数

Return: `PROCESS_LIFESPAN_FUNC`: 生命周期函数

Source code or View on GitHub
python

    def on_before_process_shutdown(self, func: PROCESS_LIFESPAN_FUNC):
        return self.lifespan.on_before_process_shutdown(func)

#### func `on_before_process_restart(self, func: PROCESS_LIFESPAN_FUNC) -> PROCESS_LIFESPAN_FUNC` ​

Description: 注册进程重启前的函数，为子进程重启时调用

Arguments:

> - func (`PROCESS_LIFESPAN_FUNC`): 生命周期函数

Return: `PROCESS_LIFESPAN_FUNC`: 生命周期函数

Source code or View on GitHub
python

    def on_before_process_restart(self, func: PROCESS_LIFESPAN_FUNC) -> PROCESS_LIFESPAN_FUNC:
        return self.lifespan.on_before_process_restart(func)

#### func `on_after_restart(self, func: LIFESPAN_FUNC)` ​

Description: 注册重启后的函数：未实现

Arguments:

> - func (`LIFESPAN_FUNC`): 生命周期函数

Return: `LIFESPAN_FUNC`: 生命周期函数

Source code or View on GitHub
python

    def on_after_restart(self, func: LIFESPAN_FUNC):
        return self.lifespan.on_after_restart(func)

### func `get_bot() -> LiteyukiBot` ​

Description: 获取轻雪实例

Return: `LiteyukiBot`: 轻雪实例

Source code or View on GitHub
python

    def get_bot() -> LiteyukiBot:
        if IS_MAIN_PROCESS:
            if _BOT_INSTANCE is None:
                raise RuntimeError('Liteyuki instance not initialized.')
            return _BOT_INSTANCE
        else:
            raise RuntimeError("Can't get bot instance in sub process.")

### func `get_config(key: str, default: Any = None) -> Any` ​

Description: 获取配置

Arguments:

> - key (`str`): 配置键
> - default (`Any`, optional): 默认值. Defaults to None.

Return: `Any`: 配置值

Source code or View on GitHub
python

    def get_config(key: str, default: Any=None) -> Any:
        return get_bot().config.get(key, default)

### func `get_config_with_compat(key: str, compat_keys: tuple[str], default: Any = None) -> Any` ​

Description: 获取配置，兼容旧版本

Arguments:

> - key (`str`): 配置键
> - compat_keys (`tuple` `str`): 兼容键
> - default (`Any`, optional): 默认值. Defaults to None.

Return: `Any`: 配置值

Source code or View on GitHub
python

    def get_config_with_compat(key: str, compat_keys: tuple[str], default: Any=None) -> Any:
        if key in get_bot().config:
            return get_bot().config[key]
        for compat_key in compat_keys:
            if compat_key in get_bot().config:
                logger.warning(f'Config key "{compat_key}" will be deprecated, use "{key}" instead.')
                return get_bot().config[compat_key]
        return default