feat: 添加启动信息阶段

pkg/core/app.py

@@ -15,7 +15,7 @@ from ..command import cmdmgr
 from ..plugin import manager as plugin_mgr
 from ..pipeline import pool
 from ..pipeline import controller, stagemgr
-from ..utils import version as version_mgr, proxy as proxy_mgr
+from ..utils import version as version_mgr, proxy as proxy_mgr, announce as announce_mgr
 
 
 class Application:
@@ -69,6 +69,8 @@ class Application:
 
     ver_mgr: version_mgr.VersionManager = None
 
+    ann_mgr: announce_mgr.AnnouncementManager = None
+
     proxy_mgr: proxy_mgr.ProxyManager = None
 
     logger: logging.Logger = None

pkg/core/boot.py

@@ -7,14 +7,15 @@ from ..audit import identifier
 from . import stage
 
 # 引入启动阶段实现以便注册
-from .stages import load_config, setup_logger, build_app, migrate
+from .stages import load_config, setup_logger, build_app, migrate, show_notes
 
 
 stage_order = [
     "LoadConfigStage",
     "MigrationStage",
     "SetupLoggerStage",
-    "BuildAppStage"
+    "BuildAppStage",
+    "ShowNotesStage"
 ]
 
 

pkg/core/note.py

@@ -0,0 +1,44 @@
+from __future__ import annotations
+
+import abc
+import typing
+
+from . import app
+
+preregistered_notes: list[typing.Type[LaunchNote]] = []
+
+def note_class(name: str, number: int):
+    """注册一个启动信息
+    """
+    def decorator(cls: typing.Type[LaunchNote]) -> typing.Type[LaunchNote]:
+        cls.name = name
+        cls.number = number
+        preregistered_notes.append(cls)
+        return cls
+
+    return decorator
+
+
+class LaunchNote(abc.ABC):
+    """启动信息
+    """
+    name: str
+
+    number: int
+
+    ap: app.Application
+
+    def __init__(self, ap: app.Application):
+        self.ap = ap
+
+    @abc.abstractmethod
+    async def need_show(self) -> bool:
+        """判断当前环境是否需要显示此启动信息
+        """
+        pass
+
+    @abc.abstractmethod
+    async def yield_note(self) -> typing.AsyncGenerator[typing.Tuple[str, int], None]:
+        """生成启动信息
+        """
+        pass

pkg/core/notes/n001_classic_msgs.py

@@ -0,0 +1,20 @@
+from __future__ import annotations
+
+import typing
+
+from .. import note, app
+
+
+@note.note_class("ClassicNotes", 1)
+class ClassicNotes(note.LaunchNote):
+    """经典启动信息
+    """
+
+    async def need_show(self) -> bool:
+        return True
+
+    async def yield_note(self) -> typing.AsyncGenerator[typing.Tuple[str, int], None]:
+        
+        yield await self.ap.ann_mgr.show_announcements()
+
+        yield await self.ap.ver_mgr.show_version_update()

pkg/core/stages/build_app.py

@@ -53,12 +53,10 @@ class BuildAppStage(stage.BootingStage):
 
         # 发送公告
         ann_mgr = announce.AnnouncementManager(ap)
-        await ann_mgr.show_announcements()
+        ap.ann_mgr = ann_mgr
 
         ap.query_pool = pool.QueryPool()
 
-        await ap.ver_mgr.show_version_update()
-
         plugin_mgr_inst = plugin_mgr.PluginManager(ap)
         await plugin_mgr_inst.initialize()
         ap.plugin_mgr = plugin_mgr_inst

pkg/core/stages/show_notes.py

@@ -0,0 +1,22 @@
+from __future__ import annotations
+
+from .. import stage, app, note
+from ..notes import n001_classic_msgs
+
+
+@stage.stage_class("ShowNotesStage")
+class ShowNotesStage(stage.BootingStage):
+    """显示启动信息阶段
+    """
+
+    async def run(self, ap: app.Application):
+
+        for note_cls in note.preregistered_notes:
+            note_inst = note_cls(ap)
+            if await note_inst.need_show():
+                async for ret in note_inst.yield_note():
+                    if not ret:
+                        continue
+                    msg, level = ret
+                    if msg:
+                        ap.logger.log(level, msg)