refactor: 添加类型注解以增强代码可读性
This commit is contained in:
@@ -31,7 +31,7 @@ EMOJI_REGISTERED_DIR = DATA_DIR / "emoji_registered" # 已注册的表情包注
|
|||||||
MAX_EMOJI_FOR_PROMPT = 20 # 最大允许的表情包描述数量于图片替换的 prompt 中
|
MAX_EMOJI_FOR_PROMPT = 20 # 最大允许的表情包描述数量于图片替换的 prompt 中
|
||||||
|
|
||||||
|
|
||||||
def _ensure_directories():
|
def _ensure_directories() -> None:
|
||||||
"""确保表情包相关目录存在"""
|
"""确保表情包相关目录存在"""
|
||||||
EMOJI_DIR.mkdir(parents=True, exist_ok=True)
|
EMOJI_DIR.mkdir(parents=True, exist_ok=True)
|
||||||
EMOJI_REGISTERED_DIR.mkdir(parents=True, exist_ok=True)
|
EMOJI_REGISTERED_DIR.mkdir(parents=True, exist_ok=True)
|
||||||
@@ -47,13 +47,13 @@ class EmojiManager:
|
|||||||
表情包管理器
|
表情包管理器
|
||||||
"""
|
"""
|
||||||
|
|
||||||
def __init__(self):
|
def __init__(self) -> None:
|
||||||
_ensure_directories()
|
_ensure_directories()
|
||||||
|
|
||||||
self._emoji_num: int = 0
|
self._emoji_num: int = 0
|
||||||
self.emojis: list[MaiEmoji] = []
|
self.emojis: list[MaiEmoji] = []
|
||||||
self._maintenance_wakeup_event = asyncio.Event()
|
self._maintenance_wakeup_event: asyncio.Event = asyncio.Event()
|
||||||
self._reload_callback_registered = False
|
self._reload_callback_registered: bool = False
|
||||||
|
|
||||||
config_manager.register_reload_callback(self.reload_runtime_config)
|
config_manager.register_reload_callback(self.reload_runtime_config)
|
||||||
self._reload_callback_registered = True
|
self._reload_callback_registered = True
|
||||||
|
|||||||
19
src/main.py
19
src/main.py
@@ -1,5 +1,6 @@
|
|||||||
from maim_message import MessageServer
|
from maim_message import MessageServer
|
||||||
from rich.traceback import install
|
from rich.traceback import install
|
||||||
|
from typing import TYPE_CHECKING
|
||||||
|
|
||||||
import asyncio
|
import asyncio
|
||||||
import time
|
import time
|
||||||
@@ -31,17 +32,21 @@ install(extra_lines=3)
|
|||||||
logger = get_logger("main")
|
logger = get_logger("main")
|
||||||
|
|
||||||
|
|
||||||
|
if TYPE_CHECKING:
|
||||||
|
from src.webui.webui_server import WebUIServer
|
||||||
|
|
||||||
|
|
||||||
class MainSystem:
|
class MainSystem:
|
||||||
def __init__(self):
|
def __init__(self) -> None:
|
||||||
# 使用消息API替代直接的FastAPI实例
|
# 使用消息API替代直接的FastAPI实例
|
||||||
self.app: MessageServer = get_global_api()
|
self.app: MessageServer = get_global_api()
|
||||||
self.server: Server = get_global_server()
|
self.server: Server = get_global_server()
|
||||||
self.webui_server = None # 独立的 WebUI 服务器
|
self.webui_server: WebUIServer | None = None # 独立的 WebUI 服务器
|
||||||
|
|
||||||
# 设置独立的 WebUI 服务器
|
# 设置独立的 WebUI 服务器
|
||||||
self._setup_webui_server()
|
self._setup_webui_server()
|
||||||
|
|
||||||
def _setup_webui_server(self):
|
def _setup_webui_server(self) -> None:
|
||||||
"""设置独立的 WebUI 服务器"""
|
"""设置独立的 WebUI 服务器"""
|
||||||
from src.config.config import global_config
|
from src.config.config import global_config
|
||||||
|
|
||||||
@@ -57,7 +62,7 @@ class MainSystem:
|
|||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.error(t("startup.webui_server_init_failed", error=e))
|
logger.error(t("startup.webui_server_init_failed", error=e))
|
||||||
|
|
||||||
async def initialize(self):
|
async def initialize(self) -> None:
|
||||||
"""初始化系统组件"""
|
"""初始化系统组件"""
|
||||||
logger.info(t("startup.waking_up", nickname=global_config.bot.nickname))
|
logger.info(t("startup.waking_up", nickname=global_config.bot.nickname))
|
||||||
|
|
||||||
@@ -66,7 +71,7 @@ class MainSystem:
|
|||||||
|
|
||||||
logger.info(t("startup.initialization_completed_banner", nickname=global_config.bot.nickname))
|
logger.info(t("startup.initialization_completed_banner", nickname=global_config.bot.nickname))
|
||||||
|
|
||||||
async def _init_components(self):
|
async def _init_components(self) -> None:
|
||||||
"""初始化其他组件"""
|
"""初始化其他组件"""
|
||||||
init_start_time = time.time()
|
init_start_time = time.time()
|
||||||
|
|
||||||
@@ -128,7 +133,7 @@ class MainSystem:
|
|||||||
logger.error(t("startup.brain_external_world_failed", error=e))
|
logger.error(t("startup.brain_external_world_failed", error=e))
|
||||||
raise
|
raise
|
||||||
|
|
||||||
async def schedule_tasks(self):
|
async def schedule_tasks(self) -> None:
|
||||||
"""调度定时任务"""
|
"""调度定时任务"""
|
||||||
try:
|
try:
|
||||||
tasks = [
|
tasks = [
|
||||||
@@ -155,7 +160,7 @@ class MainSystem:
|
|||||||
# logger.info("[记忆遗忘] 记忆遗忘完成")
|
# logger.info("[记忆遗忘] 记忆遗忘完成")
|
||||||
|
|
||||||
|
|
||||||
async def main():
|
async def main() -> None:
|
||||||
"""主函数"""
|
"""主函数"""
|
||||||
system = MainSystem()
|
system = MainSystem()
|
||||||
try:
|
try:
|
||||||
|
|||||||
Reference in New Issue
Block a user