diff --git a/pkg/core/app.py b/pkg/core/app.py index 1ed53042..25fc49f6 100644 --- a/pkg/core/app.py +++ b/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 diff --git a/pkg/core/boot.py b/pkg/core/boot.py index 22af7ff8..5663c2ef 100644 --- a/pkg/core/boot.py +++ b/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" ] diff --git a/pkg/core/note.py b/pkg/core/note.py new file mode 100644 index 00000000..6ffbff51 --- /dev/null +++ b/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 diff --git a/pkg/core/notes/__init__.py b/pkg/core/notes/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/pkg/core/notes/n001_classic_msgs.py b/pkg/core/notes/n001_classic_msgs.py new file mode 100644 index 00000000..bdc5c44e --- /dev/null +++ b/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() \ No newline at end of file diff --git a/pkg/core/notes/n002_selection_mode_on_windows.py b/pkg/core/notes/n002_selection_mode_on_windows.py new file mode 100644 index 00000000..961d697d --- /dev/null +++ b/pkg/core/notes/n002_selection_mode_on_windows.py @@ -0,0 +1,21 @@ +from __future__ import annotations + +import typing +import os +import sys +import logging + +from .. import note, app + + +@note.note_class("SelectionModeOnWindows", 2) +class SelectionModeOnWindows(note.LaunchNote): + """Windows 上的选择模式提示信息 + """ + + async def need_show(self) -> bool: + return os.name == 'nt' + + async def yield_note(self) -> typing.AsyncGenerator[typing.Tuple[str, int], None]: + + yield """您正在使用 Windows 系统,若窗口左上角显示处于”选择“模式,程序将被暂停运行,此时请右键窗口中空白区域退出选择模式。""", logging.INFO diff --git a/pkg/core/stages/build_app.py b/pkg/core/stages/build_app.py index fbdb5e8f..85bd48a3 100644 --- a/pkg/core/stages/build_app.py +++ b/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 diff --git a/pkg/core/stages/show_notes.py b/pkg/core/stages/show_notes.py new file mode 100644 index 00000000..91cdbf01 --- /dev/null +++ b/pkg/core/stages/show_notes.py @@ -0,0 +1,28 @@ +from __future__ import annotations + +from .. import stage, app, note +from ..notes import n001_classic_msgs, n002_selection_mode_on_windows + + +@stage.stage_class("ShowNotesStage") +class ShowNotesStage(stage.BootingStage): + """显示启动信息阶段 + """ + + async def run(self, ap: app.Application): + + # 排序 + note.preregistered_notes.sort(key=lambda x: x.number) + + for note_cls in note.preregistered_notes: + try: + 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) + except Exception as e: + continue diff --git a/pkg/utils/announce.py b/pkg/utils/announce.py index d17ac62a..ff1e363a 100644 --- a/pkg/utils/announce.py +++ b/pkg/utils/announce.py @@ -4,6 +4,7 @@ import json import typing import os import base64 +import logging import pydantic import requests @@ -107,17 +108,20 @@ class AnnouncementManager: async def show_announcements( self - ): + ) -> typing.Tuple[str, int]: """显示公告""" try: announcements = await self.fetch_new() + ann_text = "" for ann in announcements: - self.ap.logger.info(f'[公告] {ann.time}: {ann.content}') + ann_text += f"[公告] {ann.time}: {ann.content}\n" if announcements: await self.ap.ctr_mgr.main.post_announcement_showed( ids=[item.id for item in announcements] ) + + return ann_text, logging.INFO except Exception as e: - self.ap.logger.warning(f'获取公告时出错: {e}') + return f'获取公告时出错: {e}', logging.WARNING diff --git a/pkg/utils/version.py b/pkg/utils/version.py index cba575b6..aa98be26 100644 --- a/pkg/utils/version.py +++ b/pkg/utils/version.py @@ -1,6 +1,8 @@ from __future__ import annotations import os +import typing +import logging import time import requests @@ -213,11 +215,11 @@ class VersionManager: async def show_version_update( self - ): + ) -> typing.Tuple[str, int]: try: if await self.ap.ver_mgr.is_new_version_available(): - self.ap.logger.info("有新版本可用,请使用 !update 命令更新") + return "有新版本可用,请使用管理员账号发送 !update 命令更新", logging.INFO except Exception as e: - self.ap.logger.warning(f"检查版本更新时出错: {e}") + return f"检查版本更新时出错: {e}", logging.WARNING