From ca34f9c2a120ea05f7eb512cf6d997ee6ccd0939 Mon Sep 17 00:00:00 2001 From: snowy Date: Wed, 31 Jul 2024 02:28:25 +0800 Subject: [PATCH] =?UTF-8?q?:bug:=20fix:=20Channel=E7=9A=84=E6=8E=A5?= =?UTF-8?q?=E6=94=B6=E8=80=85=E8=BF=87=E6=BB=A4=E5=99=A8=E7=9A=84=E9=97=AE?= =?UTF-8?q?=E9=A2=98=EF=BC=8C=E4=BC=98=E5=8C=96=E9=87=8D=E5=90=AF=E9=83=A8?= =?UTF-8?q?=E5=88=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../plugin_loader/__init__.py | 34 ++--- .../plugin_loader/data_source.py | 0 .../process_manager/__init__.py | 0 .../resource_loader/__init__.py | 0 .../what_litaco/__init__.py | 0 liteyuki/bot/__init__.py | 123 ++++++------------ liteyuki/comm/channel.py | 108 ++++++--------- liteyuki/core/__init__.py | 8 ++ liteyuki/core/manager.py | 93 +++++++++++++ .../core/nb}/adapter_manager/__init__.py | 28 ++-- .../core/nb}/adapter_manager/onebot.py | 24 ++-- .../core/nb}/adapter_manager/satori.py | 52 ++++---- .../core/nb}/driver_manager/__init__.py | 12 +- .../core/nb}/driver_manager/auto_set_env.py | 40 +++--- .../core/nb}/driver_manager/defines.py | 34 ++--- liteyuki/core/spawn_process.py | 65 +++++---- main.py | 2 +- requirements.txt | 1 + src/liteyuki_main/core.py | 68 +++++----- src/liteyuki_main/dev.py | 4 +- src/utils/base/__init__.py | 19 +++ 21 files changed, 386 insertions(+), 329 deletions(-) rename liteyuki/{plugins => _plugins}/plugin_loader/__init__.py (62%) rename liteyuki/{plugins => _plugins}/plugin_loader/data_source.py (100%) rename liteyuki/{plugins => _plugins}/process_manager/__init__.py (100%) rename liteyuki/{plugins => _plugins}/resource_loader/__init__.py (100%) rename liteyuki/{plugins => _plugins}/what_litaco/__init__.py (100%) create mode 100644 liteyuki/core/manager.py rename {src/utils => liteyuki/core/nb}/adapter_manager/__init__.py (92%) rename {src/utils => liteyuki/core/nb}/adapter_manager/onebot.py (94%) rename {src/utils => liteyuki/core/nb}/adapter_manager/satori.py (97%) rename {src/utils => liteyuki/core/nb}/driver_manager/__init__.py (94%) rename {src/utils => liteyuki/core/nb}/driver_manager/auto_set_env.py (97%) rename {src/utils => liteyuki/core/nb}/driver_manager/defines.py (96%) diff --git a/liteyuki/plugins/plugin_loader/__init__.py b/liteyuki/_plugins/plugin_loader/__init__.py similarity index 62% rename from liteyuki/plugins/plugin_loader/__init__.py rename to liteyuki/_plugins/plugin_loader/__init__.py index 1da9a131..6763b57f 100644 --- a/liteyuki/plugins/plugin_loader/__init__.py +++ b/liteyuki/_plugins/plugin_loader/__init__.py @@ -2,8 +2,6 @@ import asyncio import multiprocessing import time -from apscheduler.schedulers.asyncio import AsyncIOScheduler - from liteyuki.plugin import PluginMetadata from liteyuki import get_bot, chan @@ -30,24 +28,14 @@ def _(): common_db.save(temp_data) # 更新数据 -@liteyuki.on_before_start -def _(): - print("轻雪启动中") - - -@liteyuki.on_after_start -async def _(): - print("轻雪启动完成") - chan.send("轻雪启动完成") - - -@liteyuki.on_after_nonebot_init -async def _(): - print("NoneBot初始化完成") - - -@chan.on_receive(receiver="main") -async def _(data): - print("收到消息", data) - await asyncio.sleep(5) - +print("轻雪实例", liteyuki) +chan.send(liteyuki, "instance") +# @liteyuki.on_before_start +# def _(): +# print("轻雪启动中") +# +# +# @liteyuki.on_after_start +# async def _(): +# print("轻雪启动完成") +# chan.send("轻雪启动完成") diff --git a/liteyuki/plugins/plugin_loader/data_source.py b/liteyuki/_plugins/plugin_loader/data_source.py similarity index 100% rename from liteyuki/plugins/plugin_loader/data_source.py rename to liteyuki/_plugins/plugin_loader/data_source.py diff --git a/liteyuki/plugins/process_manager/__init__.py b/liteyuki/_plugins/process_manager/__init__.py similarity index 100% rename from liteyuki/plugins/process_manager/__init__.py rename to liteyuki/_plugins/process_manager/__init__.py diff --git a/liteyuki/plugins/resource_loader/__init__.py b/liteyuki/_plugins/resource_loader/__init__.py similarity index 100% rename from liteyuki/plugins/resource_loader/__init__.py rename to liteyuki/_plugins/resource_loader/__init__.py diff --git a/liteyuki/plugins/what_litaco/__init__.py b/liteyuki/_plugins/what_litaco/__init__.py similarity index 100% rename from liteyuki/plugins/what_litaco/__init__.py rename to liteyuki/_plugins/what_litaco/__init__.py diff --git a/liteyuki/bot/__init__.py b/liteyuki/bot/__init__.py index 43074963..8c3cac85 100644 --- a/liteyuki/bot/__init__.py +++ b/liteyuki/bot/__init__.py @@ -1,26 +1,17 @@ -import asyncio -import multiprocessing import time -from typing import Any, Coroutine, Optional +import asyncio +from typing import Any, Optional +from multiprocessing import freeze_support -import nonebot -import liteyuki -from liteyuki.plugin.load import load_plugin, load_plugins +from liteyuki.bot.lifespan import (LIFESPAN_FUNC, Lifespan) +from liteyuki.comm.channel import Channel +from liteyuki.core import IS_MAIN_PROCESS +from liteyuki.core.manager import ProcessManager +from liteyuki.core.spawn_process import mb_run, nb_run +from liteyuki.log import init_log, logger +from liteyuki.plugin import load_plugins from liteyuki.utils import run_coroutine -from liteyuki.log import logger, init_log - -from src.utils import ( - adapter_manager, - driver_manager, -) - -from liteyuki.bot.lifespan import ( - Lifespan, - LIFESPAN_FUNC, -) - -from liteyuki.core.spawn_process import nb_run, ProcessingManager __all__ = [ "LiteyukiBot", @@ -28,19 +19,21 @@ __all__ = [ ] """是否为主进程""" -IS_MAIN_PROCESS = multiprocessing.current_process().name == "MainProcess" class LiteyukiBot: def __init__(self, *args, **kwargs): global _BOT_INSTANCE _BOT_INSTANCE = self # 引用 - if not IS_MAIN_PROCESS: - self.config: dict[str, Any] = kwargs - self.lifespan: Lifespan = Lifespan() - self.init(**self.config) # 初始化 - else: - print("\033[34m" + r""" + self.config: dict[str, Any] = kwargs + self.init(**self.config) # 初始化 + + self.lifespan: Lifespan = Lifespan() + self.chan = Channel() # 进程通信通道 + self.pm: Optional[ProcessManager] = None # 启动时实例化 + + + print("\033[34m" + r""" __ ______ ________ ________ __ __ __ __ __ __ ______ / | / |/ |/ |/ \ / |/ | / |/ | / |/ | $$ | $$$$$$/ $$$$$$$$/ $$$$$$$$/ $$ \ /$$/ $$ | $$ |$$ | /$$/ $$$$$$/ @@ -52,62 +45,23 @@ $$ |/ $$ | $$ | $$ | $$ | $$ $$/ $$ | $$ |/ $$ | $$$$$$$$/ $$$$$$/ $$/ $$$$$$$$/ $$/ $$$$$$/ $$/ $$/ $$$$$$/ """ + "\033[0m") - def run(self, *args, **kwargs): - if IS_MAIN_PROCESS: - self._run_nb_in_spawn_process(*args, **kwargs) - else: - # 子进程启动 - load_plugins("liteyuki/plugins") # 加载轻雪插件 - driver_manager.init(config=self.config) - adapter_manager.init(self.config) - adapter_manager.register() - nonebot.load_plugin("src.liteyuki_main") - run_coroutine(self.lifespan.after_start()) # 启动前 + def run(self): + # load_plugins("liteyuki/plugins") # 加载轻雪插件 + self.pm = ProcessManager(bot=self, chan=self.chan) - def _run_nb_in_spawn_process(self, *args, **kwargs): - """ - 在新的进程中运行nonebot.run方法，该函数在主进程中被调用 - Args: - *args: - **kwargs: + self.pm.add_target("melobot", mb_run, **self.config) + self.pm.start("melobot") - Returns: - """ - if IS_MAIN_PROCESS: - timeout_limit: int = 20 - should_exit = False + self.pm.add_target("nonebot", nb_run, **self.config) + self.pm.start("nonebot") - while not should_exit: - ctx = multiprocessing.get_context("spawn") - event = ctx.Event() - ProcessingManager.event = event - process = ctx.Process( - target=nb_run, - args=(event,) + args, - kwargs=kwargs, - ) - process.start() # 启动进程 + run_coroutine(self.lifespan.after_start()) # 启动前 - while not should_exit: - if ProcessingManager.event.wait(1): - logger.info("Receive reboot event") - process.terminate() - process.join(timeout_limit) - if process.is_alive(): - logger.warning( - f"Process {process.pid} is still alive after {timeout_limit} seconds, force kill it." - ) - process.kill() - break - elif process.is_alive(): - liteyuki.chan.send("轻雪进程正常运行", "sub") - continue - else: - should_exit = True - - def restart(self): + def restart(self, name: Optional[str] = None): """ 停止轻雪 + Args: + name: 进程名称, 默认为None, 所有进程 Returns: """ @@ -116,8 +70,11 @@ $$$$$$$$/ $$$$$$/ $$/ $$$$$$$$/ $$/ $$$$$$/ $$/ $$/ $$$$$$/ run_coroutine(self.lifespan.before_restart()) logger.debug("Running before_shutdown functions...") run_coroutine(self.lifespan.before_shutdown()) - - ProcessingManager.restart() + if name: + self.chan.send(1, name) + else: + for name in self.pm.processes: + self.chan.send(1, name) def init(self, *args, **kwargs): """ @@ -127,11 +84,9 @@ $$$$$$$$/ $$$$$$/ $$/ $$$$$$$$/ $$/ $$$$$$/ $$/ $$/ $$$$$$/ """ self.init_config() self.init_logger() - if not IS_MAIN_PROCESS: - nonebot.init(**kwargs) - asyncio.run(self.lifespan.after_nonebot_init()) def init_logger(self): + # 修改nonebot的日志配置 init_log(config=self.config) def init_config(self): @@ -225,4 +180,8 @@ def get_bot() -> Optional[LiteyukiBot]: Returns: LiteyukiBot: 当前的轻雪实例 """ - return _BOT_INSTANCE + if IS_MAIN_PROCESS: + return _BOT_INSTANCE + else: + # 从多进程上下文中获取 + pass diff --git a/liteyuki/comm/channel.py b/liteyuki/comm/channel.py index 49d6b7e9..4474faf3 100644 --- a/liteyuki/comm/channel.py +++ b/liteyuki/comm/channel.py @@ -10,12 +10,8 @@ Copyright (C) 2020-2024 LiteyukiStudio. All Rights Reserved 本模块定义了一个通用的通道类，用于进程间通信 """ -import threading -from multiprocessing import Queue -from queue import Empty, Full -from typing import Any, Awaitable, Callable, List, Optional, TypeAlias - -from nonebot import logger +from multiprocessing import Pipe +from typing import Any, Optional, Callable, Awaitable, List, TypeAlias from liteyuki.utils import is_coroutine_callable, run_coroutine @@ -29,85 +25,66 @@ FILTER_FUNC: TypeAlias = SYNC_FILTER_FUNC | ASYNC_FILTER_FUNC class Channel: - def __init__(self, buffer_size: int = 0): - self._queue = Queue(buffer_size) + """ + 通道类，用于进程间通信 + 有两种接收工作方式，但是只能选择一种，主动接收和被动接收，主动接收使用 `receive` 方法，被动接收使用 `on_receive` 装饰器 + """ + + def __init__(self): + self.parent_conn, self.child_conn = Pipe() self._closed = False self._on_receive_funcs: List[ON_RECEIVE_FUNC] = [] self._on_receive_funcs_with_receiver: dict[str, List[ON_RECEIVE_FUNC]] = {} - self._receiving_thread = threading.Thread(target=self._start_receiver, daemon=True) - self._receiving_thread.start() - - def send( - self, - data: Any, - receiver: Optional[str] = None, - block: bool = True, - timeout: Optional[float] = None - ): + def send(self, data: Any, receiver: Optional[str] = None): """ 发送数据 Args: data: 数据 receiver: 接收者，如果为None则广播 - block: 是否阻塞 - timeout: 超时时间 - - Returns: - """ - print(f"send {data} -> {receiver}") if self._closed: raise RuntimeError("Cannot send to a closed channel") - try: - self._queue.put((data, receiver), block, timeout) - except Full: - logger.warning("Channel buffer is full, send operation is blocked") + self.child_conn.send((data, receiver)) - def receive( - self, - receiver: str = None, - block: bool = True, - timeout: Optional[float] = None - ) -> Any: + def receive(self, receiver: str = None) -> Any: """ 接收数据 Args: receiver: 接收者，如果为None则接收任意数据 - block: 是否阻塞 - timeout: 超时时间 - - Returns: - """ if self._closed: raise RuntimeError("Cannot receive from a closed channel") - try: - while True: - data, data_receiver = self._queue.get(block, timeout) - if receiver is None or receiver == data_receiver: - return data - except Empty: - if not block: - return None - raise + while True: + # 判断receiver是否为None或者receiver是否等于接收者，是则接收数据，否则不动数据 + if self.parent_conn.poll(): + data, receiver = self.parent_conn.recv() + self.parent_conn.send((data, receiver)) + self._run_on_receive_funcs(data, receiver) + return data + + def peek(self) -> Optional[Any]: + """ + 查看管道中的数据，不移除 + Returns: + """ + if self._closed: + raise RuntimeError("Cannot peek from a closed channel") + if self.parent_conn.poll(): + data, receiver = self.parent_conn.recv() + self.parent_conn.send((data, receiver)) + return data + return None def close(self): """ 关闭通道 - Returns: - """ self._closed = True - self._queue.close() - while not self._queue.empty(): - self._queue.get() + self.parent_conn.close() + self.child_conn.close() - def on_receive( - self, - filter_func: Optional[FILTER_FUNC] = None, - receiver: Optional[str] = None, - ) -> Callable[[ON_RECEIVE_FUNC], ON_RECEIVE_FUNC]: + def on_receive(self, filter_func: Optional[FILTER_FUNC] = None, receiver: Optional[str] = None) -> Callable[[ON_RECEIVE_FUNC], ON_RECEIVE_FUNC]: """ 接收数据并执行函数 Args: @@ -138,22 +115,11 @@ class Channel: return decorator - def _start_receiver(self): - """ - 使用多线程启动接收循环，在通道实例化时自动启动 - Returns: - """ - while True: - data, receiver = self._queue.get(block=True, timeout=None) - self._run_on_receive_funcs(data, receiver) - def _run_on_receive_funcs(self, data: Any, receiver: Optional[str] = None): """ 运行接收函数 Args: data: 数据 - Returns: - """ if receiver is None: for func in self._on_receive_funcs: @@ -165,8 +131,8 @@ class Channel: def __iter__(self): return self - def __next__(self, timeout: Optional[float] = None) -> Any: - return self.receive(block=True, timeout=timeout) + def __next__(self) -> Any: + return self.receive() """默认通道实例，可直接从模块导入使用""" diff --git a/liteyuki/core/__init__.py b/liteyuki/core/__init__.py index e8f45b2a..f95f8146 100644 --- a/liteyuki/core/__init__.py +++ b/liteyuki/core/__init__.py @@ -1,3 +1,11 @@ +import multiprocessing + from .spawn_process import * +from .manager import * +__all__ = [ + "IS_MAIN_PROCESS" +] + +IS_MAIN_PROCESS = multiprocessing.current_process().name == "MainProcess" diff --git a/liteyuki/core/manager.py b/liteyuki/core/manager.py new file mode 100644 index 00000000..b7f73ad2 --- /dev/null +++ b/liteyuki/core/manager.py @@ -0,0 +1,93 @@ +# -*- coding: utf-8 -*- +""" +Copyright (C) 2020-2024 LiteyukiStudio. All Rights Reserved + +@Time : 2024/7/27 上午11:12 +@Author : snowykami +@Email : snowykami@outlook.com +@File : manager.py +@Software: PyCharm +""" +import threading +from multiprocessing import Process + +from liteyuki.comm import Channel +from liteyuki.log import logger + +TIMEOUT = 10 + +__all__ = [ + "ProcessManager" +] + + +class ProcessManager: + """ + 在主进程中被调用 + """ + + def __init__(self, bot, chan: Channel): + self.bot = bot + self.chan = chan + self.processes: dict[str, tuple[callable, tuple, dict]] = {} + + def start(self, name: str, delay: int = 0): + """ + 开启后自动监控进程 + Args: + name: + delay: + + Returns: + + """ + + if name not in self.processes: + raise KeyError(f"Process {name} not found.") + + def _start(): + should_exit = False + while not should_exit: + process = Process(target=self.processes[name][0], args=(self.chan, *self.processes[name][1]), kwargs=self.processes[name][2]) + process.start() + while not should_exit: + # 0退出 1重启 + data = self.chan.receive(name) + print("Received data: ", data) + if data == 1: + logger.info("Restarting LiteyukiBot...") + process.terminate() + process.join(TIMEOUT) + if process.is_alive(): + process.kill() + break + + elif data == 0: + logger.info("Stopping LiteyukiBot...") + should_exit = True + process.terminate() + process.join(TIMEOUT) + if process.is_alive(): + process.kill() + else: + logger.warning("Unknown data received, ignored.") + + if delay: + threading.Timer(delay, _start).start() + else: + threading.Thread(target=_start).start() + + def add_target(self, name: str, target, *args, **kwargs): + self.processes[name] = (target, args, kwargs) + + def join(self): + for name, process in self.processes: + process.join() + + def terminate(self): + for name, process in self.processes: + process.terminate() + process.join(TIMEOUT) + if process.is_alive(): + process.kill() + self.processes = [] diff --git a/src/utils/adapter_manager/__init__.py b/liteyuki/core/nb/adapter_manager/__init__.py similarity index 92% rename from src/utils/adapter_manager/__init__.py rename to liteyuki/core/nb/adapter_manager/__init__.py index 2a0e73e9..daf48023 100644 --- a/src/utils/adapter_manager/__init__.py +++ b/liteyuki/core/nb/adapter_manager/__init__.py @@ -1,14 +1,14 @@ -from . import ( - satori, - onebot -) - - -def init(config: dict): - onebot.init() - satori.init(config) - - -def register(): - onebot.register() - satori.register() +from . import ( + satori, + onebot +) + + +def init(config: dict): + onebot.init() + satori.init(config) + + +def register(): + onebot.register() + satori.register() diff --git a/src/utils/adapter_manager/onebot.py b/liteyuki/core/nb/adapter_manager/onebot.py similarity index 94% rename from src/utils/adapter_manager/onebot.py rename to liteyuki/core/nb/adapter_manager/onebot.py index db19b4c9..23d2a59c 100644 --- a/src/utils/adapter_manager/onebot.py +++ b/liteyuki/core/nb/adapter_manager/onebot.py @@ -1,12 +1,12 @@ -import nonebot -from nonebot.adapters.onebot import v11, v12 - - -def init(): - pass - - -def register(): - driver = nonebot.get_driver() - driver.register_adapter(v11.Adapter) - driver.register_adapter(v12.Adapter) +import nonebot +from nonebot.adapters.onebot import v11, v12 + + +def init(): + pass + + +def register(): + driver = nonebot.get_driver() + driver.register_adapter(v11.Adapter) + driver.register_adapter(v12.Adapter) diff --git a/src/utils/adapter_manager/satori.py b/liteyuki/core/nb/adapter_manager/satori.py similarity index 97% rename from src/utils/adapter_manager/satori.py rename to liteyuki/core/nb/adapter_manager/satori.py index c5edc809..7778296d 100644 --- a/src/utils/adapter_manager/satori.py +++ b/liteyuki/core/nb/adapter_manager/satori.py @@ -1,26 +1,26 @@ -import json -import os - -import nonebot -from nonebot.adapters import satori - - -def init(config: dict): - if config.get("satori", None) is None: - nonebot.logger.info("Satori config not found, skip Satori init.") - return None - satori_config = config.get("satori") - if not satori_config.get("enable", False): - nonebot.logger.info("Satori not enabled, skip Satori init.") - return None - if os.getenv("SATORI_CLIENTS", None) is not None: - nonebot.logger.info("Satori clients already set in environment variable, skip.") - os.environ["SATORI_CLIENTS"] = json.dumps(satori_config.get("hosts", []), ensure_ascii=False) - config['satori_clients'] = satori_config.get("hosts", []) - return - - -def register(): - if os.getenv("SATORI_CLIENTS", None) is not None: - driver = nonebot.get_driver() - driver.register_adapter(satori.Adapter) +import json +import os + +import nonebot +from nonebot.adapters import satori + + +def init(config: dict): + if config.get("satori", None) is None: + nonebot.logger.info("Satori config not found, skip Satori init.") + return None + satori_config = config.get("satori") + if not satori_config.get("enable", False): + nonebot.logger.info("Satori not enabled, skip Satori init.") + return None + if os.getenv("SATORI_CLIENTS", None) is not None: + nonebot.logger.info("Satori clients already set in environment variable, skip.") + os.environ["SATORI_CLIENTS"] = json.dumps(satori_config.get("hosts", []), ensure_ascii=False) + config['satori_clients'] = satori_config.get("hosts", []) + return + + +def register(): + if os.getenv("SATORI_CLIENTS", None) is not None: + driver = nonebot.get_driver() + driver.register_adapter(satori.Adapter) diff --git a/src/utils/driver_manager/__init__.py b/liteyuki/core/nb/driver_manager/__init__.py similarity index 94% rename from src/utils/driver_manager/__init__.py rename to liteyuki/core/nb/driver_manager/__init__.py index d5cfd6a7..89d90097 100644 --- a/src/utils/driver_manager/__init__.py +++ b/liteyuki/core/nb/driver_manager/__init__.py @@ -1,6 +1,6 @@ -from .auto_set_env import auto_set_env - - -def init(config: dict): - auto_set_env(config) - return +from .auto_set_env import auto_set_env + + +def init(config: dict): + auto_set_env(config) + return diff --git a/src/utils/driver_manager/auto_set_env.py b/liteyuki/core/nb/driver_manager/auto_set_env.py similarity index 97% rename from src/utils/driver_manager/auto_set_env.py rename to liteyuki/core/nb/driver_manager/auto_set_env.py index 0492713d..506ca074 100644 --- a/src/utils/driver_manager/auto_set_env.py +++ b/liteyuki/core/nb/driver_manager/auto_set_env.py @@ -1,20 +1,20 @@ -import os - -import dotenv -import nonebot - -from .defines import * - - -def auto_set_env(config: dict): - dotenv.load_dotenv(".env") - if os.getenv("DRIVER", None) is not None: - nonebot.logger.info("Driver already set in environment variable, skip auto configure.") - return - if config.get("satori", {'enable': False}).get("enable", False): - os.environ["DRIVER"] = get_driver_string(ASGI_DRIVER, HTTPX_DRIVER, WEBSOCKETS_DRIVER) - nonebot.logger.info("Enable Satori, set driver to ASGI+HTTPX+WEBSOCKETS") - else: - os.environ["DRIVER"] = get_driver_string(ASGI_DRIVER) - nonebot.logger.info("Disable Satori, set driver to ASGI") - return +import os + +import dotenv +import nonebot + +from .defines import * + + +def auto_set_env(config: dict): + dotenv.load_dotenv(".env") + if os.getenv("DRIVER", None) is not None: + nonebot.logger.info("Driver already set in environment variable, skip auto configure.") + return + if config.get("satori", {'enable': False}).get("enable", False): + os.environ["DRIVER"] = get_driver_string(ASGI_DRIVER, HTTPX_DRIVER, WEBSOCKETS_DRIVER) + nonebot.logger.info("Enable Satori, set driver to ASGI+HTTPX+WEBSOCKETS") + else: + os.environ["DRIVER"] = get_driver_string(ASGI_DRIVER) + nonebot.logger.info("Disable Satori, set driver to ASGI") + return diff --git a/src/utils/driver_manager/defines.py b/liteyuki/core/nb/driver_manager/defines.py similarity index 96% rename from src/utils/driver_manager/defines.py rename to liteyuki/core/nb/driver_manager/defines.py index 98a250a6..b4756bf5 100644 --- a/src/utils/driver_manager/defines.py +++ b/liteyuki/core/nb/driver_manager/defines.py @@ -1,17 +1,17 @@ -ASGI_DRIVER = "~fastapi" -HTTPX_DRIVER = "~httpx" -WEBSOCKETS_DRIVER = "~websockets" - - -def get_driver_string(*argv): - output_string = "" - if ASGI_DRIVER in argv: - output_string += ASGI_DRIVER - for arg in argv: - if arg != ASGI_DRIVER: - output_string = f"{output_string}+{arg}" - return output_string - - -def get_driver_full_string(*argv): - return f"DRIVER={get_driver_string(argv)}" +ASGI_DRIVER = "~fastapi" +HTTPX_DRIVER = "~httpx" +WEBSOCKETS_DRIVER = "~websockets" + + +def get_driver_string(*argv): + output_string = "" + if ASGI_DRIVER in argv: + output_string += ASGI_DRIVER + for arg in argv: + if arg != ASGI_DRIVER: + output_string = f"{output_string}+{arg}" + return output_string + + +def get_driver_full_string(*argv): + return f"DRIVER={get_driver_string(argv)}" diff --git a/liteyuki/core/spawn_process.py b/liteyuki/core/spawn_process.py index fb56e0fe..6ea91425 100644 --- a/liteyuki/core/spawn_process.py +++ b/liteyuki/core/spawn_process.py @@ -1,37 +1,50 @@ import threading -from multiprocessing import get_context, Event +from multiprocessing import Event, Queue +from typing import Optional import nonebot -from nonebot import logger -from liteyuki.plugin.load import load_plugins +import liteyuki +from liteyuki.core.nb import adapter_manager, driver_manager timeout_limit: int = 20 -__all__ = [ - "ProcessingManager", - "nb_run", -] + +"""导出对象，用于进程通信""" +chan_in_spawn: Optional["liteyuki.Channel"] = None -class ProcessingManager: - event: Event = None +def nb_run(chan, *args, **kwargs): + """ + 初始化NoneBot并运行在子进程 + Args: - @classmethod - def restart(cls, delay: int = 0): - """ - 发送终止信号 - Args: - delay: 延迟时间，默认为0，单位秒 - Returns: - """ - if cls.event is None: - raise RuntimeError("ProcessingManager has not been initialized.") - if delay > 0: - threading.Timer(delay, function=cls.event.set).start() - return - cls.event.set() + *args: + **kwargs: + + Returns: + + """ + global chan_in_spawn + chan_in_spawn = chan + nonebot.init(**kwargs) + driver_manager.init(config=kwargs) + adapter_manager.init(kwargs) + adapter_manager.register() + nonebot.load_plugin("src.liteyuki_main") + nonebot.run() -def nb_run(event, *args, **kwargs): - ProcessingManager.event = event - nonebot.run(*args, **kwargs) +def mb_run(chan, *args, **kwargs): + """ + 初始化MeloBot并运行在子进程 + Args: + chan + *args: + **kwargs: + + Returns: + + """ + # bot = MeloBot(__name__) + # bot.init(AbstractConnector(cd_time=0)) + # bot.run() diff --git a/main.py b/main.py index 44eb1a29..c2a551c1 100644 --- a/main.py +++ b/main.py @@ -1,6 +1,6 @@ from liteyuki import LiteyukiBot from src.utils import load_from_yaml -if __name__ in ("__main__", "__mp_main__"): +if __name__ == "__main__": bot = LiteyukiBot(**load_from_yaml("config.yml")) bot.run() diff --git a/requirements.txt b/requirements.txt index 9c95cf22..a75d78b5 100644 --- a/requirements.txt +++ b/requirements.txt @@ -3,6 +3,7 @@ aiofiles~=23.2.1 colored~=2.2.4 GitPython~=3.1.42 httpx~=0.27.0 +melobot~=2.6.5 nb-cli~=1.4.1 nonebot2[fastapi,httpx,websockets]~=2.3.0 nonebot-plugin-htmlrender~=0.3.1 diff --git a/src/liteyuki_main/core.py b/src/liteyuki_main/core.py index e751fb3c..39d904e3 100644 --- a/src/liteyuki_main/core.py +++ b/src/liteyuki_main/core.py @@ -5,12 +5,13 @@ from typing import Any, AnyStr import nonebot import pip from nonebot import Bot, get_driver, require -from nonebot.adapters import satori +from nonebot.adapters import onebot, satori from nonebot.adapters.onebot.v11 import Message, escape, unescape from nonebot.exception import MockApiException from nonebot.internal.matcher import Matcher from nonebot.permission import SUPERUSER +from liteyuki import Channel from src.utils.base.config import get_config, load_from_yaml from src.utils.base.data_manager import StoredConfig, TempConfig, common_db from src.utils.base.language import get_user_lang @@ -18,9 +19,11 @@ from src.utils.base.ly_typing import T_Bot, T_MessageEvent from src.utils.message.message import MarkdownMessage as md, broadcast_to_superusers # from src.liteyuki.core import Reloader from src.utils import event as event_utils, satori_utils -from liteyuki.core import ProcessingManager +from liteyuki.core.spawn_process import chan_in_spawn + from .api import update_liteyuki from liteyuki.bot import get_bot +from ..utils.base import reload from ..utils.base.ly_function import get_function require("nonebot_plugin_alconna") @@ -78,6 +81,7 @@ async def _(bot: T_Bot, event: T_MessageEvent): ).handle() # Satori OK async def _(matcher: Matcher, bot: T_Bot, event: T_MessageEvent): + global channel_in_spawn_process await matcher.send("Liteyuki reloading") temp_data = common_db.where_one(TempConfig(), default=TempConfig()) @@ -94,9 +98,7 @@ async def _(matcher: Matcher, bot: T_Bot, event: T_MessageEvent): ) common_db.save(temp_data) - # Reloader.reload(0) - bot = get_bot() - bot.restart() + reload() @on_alconna( @@ -322,20 +324,18 @@ async def test_for_md_image(bot: T_Bot, api: str, data: dict): @driver.on_startup async def on_startup(): - # temp_data = common_db.where_one(TempConfig(), default=TempConfig()) - # # 储存重启信息 - # if temp_data.data.get("reload", False): - # delta_time = time.time() - temp_data.data.get("reload_time", 0) - # temp_data.data["delta_time"] = delta_time - # common_db.save(temp_data) # 更新数据 + temp_data = common_db.where_one(TempConfig(), default=TempConfig()) + # 储存重启信息 + if temp_data.data.get("reload", False): + delta_time = time.time() - temp_data.data.get("reload_time", 0) + temp_data.data["delta_time"] = delta_time + common_db.save(temp_data) # 更新数据 """ - 该部分迁移至轻雪生命周期 + 该部分将迁移至轻雪生命周期 Returns: """ - pass - @driver.on_shutdown async def on_shutdown(): @@ -357,19 +357,29 @@ async def _(bot: T_Bot): reload_session_id = temp_data.data.get("reload_session_id", 0) delta_time = temp_data.data.get("delta_time", 0) common_db.save(temp_data) # 更新数据 - if isinstance(bot, satori.Bot): - await bot.send_message( - channel_id=reload_session_id, - message="Liteyuki reloaded in %.2f s" % delta_time - ) - else: - await bot.call_api( - "send_msg", - message_type=reload_session_type, - user_id=reload_session_id, - group_id=reload_session_id, - message="Liteyuki reloaded in %.2f s" % delta_time - ) + + if delta_time <= 20.0: # 启动时间太长就别发了，丢人 + if isinstance(bot, satori.Bot): + await bot.send_message( + channel_id=reload_session_id, + message="Liteyuki reloaded in %.2f s" % delta_time + ) + elif isinstance(bot, onebot.v11.Bot): + await bot.send_msg( + message_type=reload_session_type, + user_id=reload_session_id, + group_id=reload_session_id, + message="Liteyuki reloaded in %.2f s" % delta_time + ) + + elif isinstance(bot, onebot.v12.Bot): + await bot.send_message( + message_type=reload_session_type, + user_id=reload_session_id, + group_id=reload_session_id, + message="Liteyuki reloaded in %.2f s" % delta_time, + detail_type="group" + ) # 每天4点更新 @@ -381,7 +391,7 @@ async def every_day_update(): if result: await broadcast_to_superusers(f"Liteyuki updated: ```\n{logs}\n```") nonebot.logger.info(f"Liteyuki updated: {logs}") - ProcessingManager.restart() + reload() else: nonebot.logger.info(logs) @@ -406,4 +416,4 @@ need_group_id = ( "get_group_member_info", "get_group_member_list", "get_group_honor_info" -) \ No newline at end of file +) diff --git a/src/liteyuki_main/dev.py b/src/liteyuki_main/dev.py index 4ed79b43..172af36a 100644 --- a/src/liteyuki_main/dev.py +++ b/src/liteyuki_main/dev.py @@ -3,8 +3,8 @@ from watchdog.observers import Observer from watchdog.events import FileSystemEventHandler from liteyuki.bot import get_bot +from src.utils.base import reload from src.utils.base.config import get_config -from liteyuki.core import ProcessingManager from src.utils.base.resource import load_resources if get_config("debug", False): @@ -38,7 +38,7 @@ if get_config("debug", False): src_excludes_extensions) or event.is_directory or "__pycache__" in event.src_path: return nonebot.logger.info(f"{event.src_path} modified, reloading bot...") - liteyuki_bot.restart() + reload() class ResourceModifiedHandler(FileSystemEventHandler): diff --git a/src/utils/base/__init__.py b/src/utils/base/__init__.py index e69de29b..9800f856 100644 --- a/src/utils/base/__init__.py +++ b/src/utils/base/__init__.py @@ -0,0 +1,19 @@ +import threading + +from nonebot import logger +from liteyuki.core.spawn_process import chan_in_spawn + + +def reload(delay: float = 0.0, receiver: str = "nonebot"): + """ + 重载LiteyukiBot(nonebot) + Args: + receiver: 指定重载的进程 + delay: + + Returns: + + """ + + chan_in_spawn.send(1, receiver) + logger.info(f"Reloading LiteyukiBot({receiver})...")