refactor(core): 统一代码风格并移除未使用的导入

本次提交主要进行代码风格的统一和现代化改造,具体包括:
- 使用 `|` 联合类型替代 `typing.Optional`,以符合 PEP 604 的现代语法。
- 移除多个文件中未被使用的导入语句,清理代码。
- 调整了部分日志输出的级别,使其更符合调试场景。
- 统一了部分文件的导入顺序和格式。
This commit is contained in:
minecraft1024a
2025-10-07 20:16:47 +08:00
parent b3d1c93b87
commit 4f9b31d188
14 changed files with 55 additions and 67 deletions

View File

@@ -110,4 +110,4 @@ class ScoringAPI:
# 创建全局API实例 - 系统级服务
scoring_api = ScoringAPI()
scoring_api = ScoringAPI()

View File

@@ -3,7 +3,7 @@ from __future__ import annotations
import re
from pathlib import Path
from re import Pattern
from typing import Any, Optional, Union, cast
from typing import Any, cast
from src.common.logger import get_logger
from src.plugin_system.base.base_action import BaseAction
@@ -119,7 +119,7 @@ class ComponentRegistry:
def register_component(
self, self_component_info: ComponentInfo, component_class: ComponentClassType
) -> bool: # noqa: C901 (保持原有结构, 以后可再拆)
) -> bool:
"""注册组件
Args:
@@ -533,8 +533,8 @@ class ComponentRegistry:
# === 组件查询方法 ===
def get_component_info(
self, component_name: str, component_type: Optional[ComponentType] = None
) -> Optional[ComponentInfo]:
self, component_name: str, component_type: ComponentType | None = None
) -> ComponentInfo | None:
# sourcery skip: class-extract-method
"""获取组件信息,支持自动命名空间解析
@@ -578,16 +578,9 @@ class ComponentRegistry:
def get_component_class(
self,
component_name: str,
component_type: Optional[ComponentType] = None,
component_type: ComponentType | None = None,
) -> (
type[BaseCommand]
| type[BaseAction]
| type[BaseEventHandler]
| type[BaseTool]
| type[PlusCommand]
| type[BaseChatter]
| type[BaseInterestCalculator]
| None
type[BaseCommand | BaseAction | BaseEventHandler | BaseTool | PlusCommand | BaseChatter | BaseInterestCalculator] | None
):
"""获取组件类,支持自动命名空间解析
@@ -655,7 +648,7 @@ class ComponentRegistry:
"""获取Action注册表"""
return self._action_registry.copy()
def get_registered_action_info(self, action_name: str) -> Optional[ActionInfo]:
def get_registered_action_info(self, action_name: str) -> ActionInfo | None:
"""获取Action信息"""
info = self.get_component_info(action_name, ComponentType.ACTION)
return info if isinstance(info, ActionInfo) else None
@@ -670,7 +663,7 @@ class ComponentRegistry:
"""获取Command注册表"""
return self._command_registry.copy()
def get_registered_command_info(self, command_name: str) -> Optional[CommandInfo]:
def get_registered_command_info(self, command_name: str) -> CommandInfo | None:
"""获取Command信息"""
info = self.get_component_info(command_name, ComponentType.COMMAND)
return info if isinstance(info, CommandInfo) else None
@@ -714,7 +707,7 @@ class ComponentRegistry:
"""获取LLM可用的Tool列表"""
return self._llm_available_tools.copy()
def get_registered_tool_info(self, tool_name: str) -> Optional[ToolInfo]:
def get_registered_tool_info(self, tool_name: str) -> ToolInfo | None:
"""获取Tool信息
Args:
@@ -733,7 +726,7 @@ class ComponentRegistry:
self._plus_command_registry: dict[str, type[PlusCommand]] = {}
return self._plus_command_registry.copy()
def get_registered_plus_command_info(self, command_name: str) -> Optional[PlusCommandInfo]:
def get_registered_plus_command_info(self, command_name: str) -> PlusCommandInfo | None:
"""获取PlusCommand信息
Args:
@@ -751,7 +744,7 @@ class ComponentRegistry:
"""获取事件处理器注册表"""
return self._event_handler_registry.copy()
def get_registered_event_handler_info(self, handler_name: str) -> Optional[EventHandlerInfo]:
def get_registered_event_handler_info(self, handler_name: str) -> EventHandlerInfo | None:
"""获取事件处理器信息"""
info = self.get_component_info(handler_name, ComponentType.EVENT_HANDLER)
return info if isinstance(info, EventHandlerInfo) else None
@@ -773,14 +766,14 @@ class ComponentRegistry:
self._enabled_chatter_registry: dict[str, type[BaseChatter]] = {}
return self._enabled_chatter_registry.copy()
def get_registered_chatter_info(self, chatter_name: str) -> Optional[ChatterInfo]:
def get_registered_chatter_info(self, chatter_name: str) -> ChatterInfo | None:
"""获取Chatter信息"""
info = self.get_component_info(chatter_name, ComponentType.CHATTER)
return info if isinstance(info, ChatterInfo) else None
# === 插件查询方法 ===
def get_plugin_info(self, plugin_name: str) -> Optional[PluginInfo]:
def get_plugin_info(self, plugin_name: str) -> PluginInfo | None:
"""获取插件信息"""
return self._plugins.get(plugin_name)

View File

@@ -1,5 +1,4 @@
import asyncio
import importlib
import os
from importlib.util import module_from_spec, spec_from_file_location
from pathlib import Path

View File

@@ -3,7 +3,6 @@
提供独立的兴趣管理功能,不依赖任何插件
"""
from typing import Optional
from src.chat.interest_system import bot_interest_manager
from src.common.logger import get_logger
@@ -41,7 +40,7 @@ class InterestService:
logger.error(f"初始化智能兴趣系统失败: {e}")
self.is_initialized = False
async def calculate_interest_match(self, content: str, keywords: Optional[list[str]] = None):
async def calculate_interest_match(self, content: str, keywords: list[str] | None = None):
"""
计算内容与兴趣的匹配度
@@ -105,4 +104,4 @@ class InterestService:
# 创建全局实例
interest_service = InterestService()
interest_service = InterestService()

View File

@@ -4,7 +4,6 @@
"""
import time
from typing import Optional
from src.common.database.sqlalchemy_models import UserRelationships, get_db_session
from src.common.logger import get_logger
@@ -110,7 +109,7 @@ class RelationshipService:
"user_name": ""
}
async def update_user_relationship(self, user_id: str, relationship_score: float, relationship_text: Optional[str] = None, user_name: Optional[str] = None):
async def update_user_relationship(self, user_id: str, relationship_score: float, relationship_text: str | None = None, user_name: str | None = None):
"""
更新用户关系数据
@@ -160,7 +159,7 @@ class RelationshipService:
except Exception as e:
logger.error(f"更新用户关系失败: {user_id}, 错误: {e}")
def _get_from_cache(self, user_id: str) -> Optional[dict]:
def _get_from_cache(self, user_id: str) -> dict | None:
"""从缓存获取数据"""
if user_id in self._cache:
cached_data = self._cache[user_id]
@@ -179,7 +178,7 @@ class RelationshipService:
"last_updated": time.time()
}
async def _fetch_from_database(self, user_id: str) -> Optional[UserRelationships]:
async def _fetch_from_database(self, user_id: str) -> UserRelationships | None:
"""从数据库获取关系数据"""
try:
async with get_db_session() as session:
@@ -217,7 +216,7 @@ class RelationshipService:
"cache_keys": list(self._cache.keys())
}
def clear_cache(self, user_id: Optional[str] = None):
def clear_cache(self, user_id: str | None = None):
"""清理缓存"""
if user_id:
if user_id in self._cache:
@@ -229,4 +228,4 @@ class RelationshipService:
# 创建全局实例
relationship_service = RelationshipService()
relationship_service = RelationshipService()