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:
"""总通道"""
self.i_chan = Chan[Any]()
self.o_chan = Chan[Any]()
'常规操作'
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>')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 += 1func restart_process(self, name: Optional[str] = None)
Description: 停止轻雪
Arguments:
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:
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:
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