Skip to content

Commit

Permalink
✨ 分离magicoca和croterline
Browse files Browse the repository at this point in the history
  • Loading branch information
@snowykami
snowykami committed Oct 12, 2024
1 parent a77f97f commit 98a9d64
Show file tree
Hide file tree
Showing 91 changed files with 3,676 additions and 3,213 deletions.
5 changes: 3 additions & 2 deletions .github/workflows/pypi-publish.yml
View file
Original file line number Diff line number Diff line change
@@ -1,8 +1,9 @@
name: Publish

on:
release:
types: [published]
push:
tags:
- 'v*'

jobs:
pypi-publish:
Expand Down
42 changes: 30 additions & 12 deletions liteyuki/bot/__init__.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -8,18 +8,21 @@
import time
from typing import Any, Optional

from liteyuki.bot.lifespan import (LIFESPAN_FUNC, Lifespan, PROCESS_LIFESPAN_FUNC)
from liteyuki.bot.lifespan import LIFESPAN_FUNC, Lifespan, PROCESS_LIFESPAN_FUNC
from liteyuki.comm.channel import get_channel
from liteyuki.core.manager import ProcessManager
from liteyuki.log import init_log, logger
from liteyuki.plugin import load_plugin
from liteyuki.utils import IS_MAIN_PROCESS

# new version
from liteyuki.core.manager import sub_process_manager

__all__ = [
"LiteyukiBot",
"get_bot",
"get_config",
"get_config_with_compat",
"LiteyukiBot",
"get_bot",
"get_config",
"get_config_with_compat",
]


Expand Down Expand Up @@ -60,6 +63,7 @@ async def _run(self):
启动逻辑
"""
await self.lifespan.before_start() # 启动前钩子
sub_process_manager.start_all()
await self.lifespan.after_start() # 启动后钩子
await self.keep_alive()

Expand Down Expand Up @@ -108,7 +112,11 @@ def restart(self, delay: int = 0):
cmd = "nohup"
self.process_manager.terminate_all()
# 进程退出后重启
threading.Thread(target=os.system, args=(f"{cmd} {executable} {' '.join(args)}",), daemon=True).start()
threading.Thread(
target=os.system,
args=(f"{cmd} {executable} {' '.join(args)}",),
daemon=True,
).start()
sys.exit(0)
self.call_restart_count += 1

Expand Down Expand Up @@ -189,7 +197,9 @@ def on_before_process_shutdown(self, func: PROCESS_LIFESPAN_FUNC):
"""
return self.lifespan.on_before_process_shutdown(func)

def on_before_process_restart(self, func: PROCESS_LIFESPAN_FUNC) -> PROCESS_LIFESPAN_FUNC:
def on_before_process_restart(
self, func: PROCESS_LIFESPAN_FUNC
) -> PROCESS_LIFESPAN_FUNC:
"""
注册进程重启前的函数，为子进程重启时调用
Args:
Expand All @@ -211,7 +221,7 @@ def on_after_restart(self, func: LIFESPAN_FUNC):
return self.lifespan.on_after_restart(func)


_BOT_INSTANCE: LiteyukiBot
_BOT_INSTANCE: LiteyukiBot | None = None


def get_bot() -> LiteyukiBot:
Expand Down Expand Up @@ -241,7 +251,9 @@ def get_config(key: str, default: Any = None) -> Any:
return get_bot().config.get(key, default)


def get_config_with_compat(key: str, compat_keys: tuple[str], default: Any = None) -> Any:
def get_config_with_compat(
key: str, compat_keys: tuple[str], default: Any = None
) -> Any:
"""
获取配置，兼容旧版本
Args:
Expand All @@ -256,14 +268,18 @@ def get_config_with_compat(key: str, compat_keys: tuple[str], default: Any = Non
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.")
logger.warning(
f'Config key "{compat_key}" will be deprecated, use "{key}" instead.'
)
return get_bot().config[compat_key]
return default


def print_logo():
"""@litedoc-hide"""
print("\033[34m" + r"""
print(
"\033[34m"
+ r"""
__ ______ ________ ________ __ __ __ __ __ __ ______
/ | / |/ |/ |/ \ / |/ | / |/ | / |/ |
$$ | $$$$$$/ $$$$$$$$/ $$$$$$$$/ $$ \ /$$/ $$ | $$ |$$ | /$$/ $$$$$$/
Expand All @@ -273,4 +289,6 @@ def print_logo():
$$ |_____ _$$ |_ $$ | $$ |_____ $$ | $$ \__$$ |$$ |$$ \ _$$ |_
$$ |/ $$ | $$ | $$ | $$ | $$ $$/ $$ | $$ |/ $$ |
$$$$$$$$/ $$$$$$/ $$/ $$$$$$$$/ $$/ $$$$$$/ $$/ $$/ $$$$$$/
""" + "\033[0m")
"""
+ "\033[0m"
)
127 changes: 77 additions & 50 deletions liteyuki/comm/channel.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4,20 +4,31 @@
"""
import asyncio
from multiprocessing import Pipe
from typing import Any, Callable, Coroutine, Generic, Optional, TypeAlias, TypeVar, get_args
from typing import (
Any,
Callable,
Coroutine,
Generic,
Optional,
TypeAlias,
TypeVar,
get_args,
)

from liteyuki.log import logger
from liteyuki.utils import IS_MAIN_PROCESS, is_coroutine_callable

T = TypeVar("T")

SYNC_ON_RECEIVE_FUNC: TypeAlias = Callable[[T], Any] # 同步接收函数
ASYNC_ON_RECEIVE_FUNC: TypeAlias = Callable[[T], Coroutine[Any, Any, Any]] # 异步接收函数
ON_RECEIVE_FUNC: TypeAlias = SYNC_ON_RECEIVE_FUNC | ASYNC_ON_RECEIVE_FUNC # 接收函数
SYNC_ON_RECEIVE_FUNC: TypeAlias = Callable[[T], Any] # 同步接收函数
ASYNC_ON_RECEIVE_FUNC: TypeAlias = Callable[
[T], Coroutine[Any, Any, Any]
] # 异步接收函数
ON_RECEIVE_FUNC: TypeAlias = SYNC_ON_RECEIVE_FUNC | ASYNC_ON_RECEIVE_FUNC # 接收函数

SYNC_FILTER_FUNC: TypeAlias = Callable[[T], bool] # 同步过滤函数
ASYNC_FILTER_FUNC: TypeAlias = Callable[[T], Coroutine[Any, Any, bool]] # 异步过滤函数
FILTER_FUNC: TypeAlias = SYNC_FILTER_FUNC | ASYNC_FILTER_FUNC # 过滤函数
SYNC_FILTER_FUNC: TypeAlias = Callable[[T], bool] # 同步过滤函数
ASYNC_FILTER_FUNC: TypeAlias = Callable[[T], Coroutine[Any, Any, bool]] # 异步过滤函数
FILTER_FUNC: TypeAlias = SYNC_FILTER_FUNC | ASYNC_FILTER_FUNC # 过滤函数

_func_id: int = 0
_channel: dict[str, "Channel"] = {}
Expand All @@ -39,7 +50,9 @@ def __init__(self, name: str, type_check: Optional[bool] = None):
"""

self.conn_send, self.conn_recv = Pipe()
self._conn_send_inner, self._conn_recv_inner = Pipe() # 内部通道，用于子进程通信
self._conn_send_inner, self._conn_recv_inner = (
Pipe()
) # 内部通道，用于子进程通信
self._closed = False
self._on_main_receive_func_ids: list[int] = []
self._on_sub_receive_func_ids: list[int] = []
Expand All @@ -64,15 +77,17 @@ def __init__(self, name: str, type_check: Optional[bool] = None):
_channel[name] = self
logger.debug(f"Channel {name} initialized in main process")
else:
logger.debug(f"Channel {name} initialized in sub process, should manually set in main process")
logger.debug(
f"Channel {name} initialized in sub process, should manually set in main process"
)

def _get_generic_type(self) -> Optional[type]:
"""
获取通道传递泛型类型
Returns:
Optional[type]: 泛型类型
"""
if hasattr(self, '__orig_class__'):
if hasattr(self, "__orig_class__"):
return get_args(self.__orig_class__)[0]
return None

Expand All @@ -98,7 +113,10 @@ def _validate_structure(self, data: Any, structure: type) -> bool:
elif isinstance(structure, dict):
if not isinstance(data, dict):
return False
return all(k in data and self._validate_structure(data[k], structure[k]) for k in structure)
return all(
k in data and self._validate_structure(data[k], structure[k])
for k in structure
)
return False

def __str__(self):
Expand All @@ -113,10 +131,12 @@ def send(self, data: T):
if self.type_check:
_type = self._get_generic_type()
if _type is not None and not self._validate_structure(data, _type):
raise TypeError(f"Data must be an instance of {_type}, {type(data)} found")
raise TypeError(
f"Data must be an instance of {_type}, {type(data)} found"
)

if self._closed:
raise RuntimeError("Cannot send to a closed channel_")
raise RuntimeError("Cannot send to a closed channel")
self.conn_send.send(data)

def receive(self) -> T:
Expand All @@ -126,7 +146,7 @@ def receive(self) -> T:
T: 数据
"""
if self._closed:
raise RuntimeError("Cannot receive from a closed channel_")
raise RuntimeError("Cannot receive from a closed channel")

while True:
data = self.conn_recv.recv()
Expand All @@ -142,7 +162,9 @@ async def async_receive(self) -> T:
data = await loop.run_in_executor(None, self.receive)
return data

def on_receive(self, filter_func: Optional[FILTER_FUNC] = None) -> Callable[[Callable[[T], Any]], Callable[[T], Any]]:
def on_receive(
self, filter_func: Optional[FILTER_FUNC] = None
) -> Callable[[Callable[[T], Any]], Callable[[T], Any]]:
"""
接收数据并执行函数
Args:
Expand Down Expand Up @@ -187,37 +209,52 @@ async def _run_on_receive_funcs(self, data: Any):
data: 数据
"""
if IS_MAIN_PROCESS:
[asyncio.create_task(_callback_funcs[func_id](data)) for func_id in self._on_main_receive_func_ids]
[
asyncio.create_task(_callback_funcs[func_id](data))
for func_id in self._on_main_receive_func_ids
]
else:
[asyncio.create_task(_callback_funcs[func_id](data)) for func_id in self._on_sub_receive_func_ids]
[
asyncio.create_task(_callback_funcs[func_id](data))
for func_id in self._on_sub_receive_func_ids
]


"""子进程可用的主动和被动通道"""
active_channel: Channel = Channel(name="active_channel") # 主动通道
active_channel: Channel = Channel(name="active_channel") # 主动通道
passive_channel: Channel = Channel(name="passive_channel") # 被动通道
publish_channel: Channel[tuple[str, dict[str, Any]]] = Channel(name="publish_channel") # 发布通道
publish_channel: Channel[tuple[str, dict[str, Any]]] = Channel(
name="publish_channel"
) # 发布通道
"""通道传递通道，主进程创建单例，子进程初始化时实例化"""
channel_deliver_active_channel: Channel[Channel[Any]] # 主动通道传递通道
channel_deliver_passive_channel: Channel[tuple[str, dict[str, Any]]] # 被动通道传递通道
channel_deliver_active_channel: Channel[Channel[Any]] # 主动通道传递通道
channel_deliver_passive_channel: Channel[tuple[str, dict[str, Any]]] # 被动通道传递通道

if IS_MAIN_PROCESS:
channel_deliver_active_channel = Channel(name="channel_deliver_active_channel") # 主动通道传递通道
channel_deliver_passive_channel = Channel(name="channel_deliver_passive_channel") # 被动通道传递通道


@channel_deliver_passive_channel.on_receive(filter_func=lambda data: data[0] == "set_channel")
channel_deliver_active_channel = Channel(
name="channel_deliver_active_channel"
) # 主动通道传递通道
channel_deliver_passive_channel = Channel(
name="channel_deliver_passive_channel"
) # 被动通道传递通道

@channel_deliver_passive_channel.on_receive(
filter_func=lambda data: data[0] == "set_channel"
)
def on_set_channel(data: tuple[str, dict[str, Any]]):
name, channel = data[1]["name"], data[1]["channel_"]
set_channel(name, channel)


@channel_deliver_passive_channel.on_receive(filter_func=lambda data: data[0] == "get_channel")
@channel_deliver_passive_channel.on_receive(
filter_func=lambda data: data[0] == "get_channel"
)
def on_get_channel(data: tuple[str, dict[str, Any]]):
name, recv_chan = data[1]["name"], data[1]["recv_chan"]
recv_chan.send(get_channel(name))


@channel_deliver_passive_channel.on_receive(filter_func=lambda data: data[0] == "get_channels")
@channel_deliver_passive_channel.on_receive(
filter_func=lambda data: data[0] == "get_channels"
)
def on_get_channels(data: tuple[str, dict[str, Any]]):
recv_chan = data[1]["recv_chan"]
recv_chan.send(get_channels())
Expand All @@ -231,7 +268,9 @@ def set_channel(name: str, channel: "Channel"):
channel ([`Channel`](#class-channel-generic-t)): 通道实例
"""
if not isinstance(channel, Channel):
raise TypeError(f"channel_ must be an instance of Channel, {type(channel)} found")
raise TypeError(
f"channel_ must be an instance of Channel, {type(channel)} found"
)

if IS_MAIN_PROCESS:
if name in _channel:
Expand All @@ -241,10 +280,11 @@ def set_channel(name: str, channel: "Channel"):
# 请求主进程设置通道
channel_deliver_passive_channel.send(
(
"set_channel", {
"name" : name,
"channel_": channel,
}
"set_channel",
{
"name": name,
"channel_": channel,
},
)
)

Expand Down Expand Up @@ -273,13 +313,7 @@ def get_channel(name: str) -> "Channel":
else:
recv_chan = Channel[Channel[Any]]("recv_chan")
channel_deliver_passive_channel.send(
(
"get_channel",
{
"name" : name,
"recv_chan": recv_chan
}
)
("get_channel", {"name": name, "recv_chan": recv_chan})
)
return recv_chan.receive()

Expand All @@ -294,12 +328,5 @@ def get_channels() -> dict[str, "Channel"]:
return _channel
else:
recv_chan = Channel[dict[str, Channel[Any]]]("recv_chan")
channel_deliver_passive_channel.send(
(
"get_channels",
{
"recv_chan": recv_chan
}
)
)
channel_deliver_passive_channel.send(("get_channels", {"recv_chan": recv_chan}))
return recv_chan.receive()
26 changes: 0 additions & 26 deletions liteyuki/comm/rpc.py
View file

This file was deleted.

Loading

0 comments on commit 98a9d64

Please sign in to comment.