chore: perform widespread code cleanup and formatting

Perform a comprehensive code cleanup across multiple modules to improve code quality, consistency, and maintainability.

Key changes include:
- Removing numerous unused imports.
- Standardizing import order.
- Eliminating trailing whitespace and inconsistent newlines.
- Updating legacy type hints to modern syntax (e.g., `List` -> `list`).
- Making minor improvements for code robustness and style.

src/chat/chatter_manager.py

@@ -1,6 +1,6 @@
 import asyncio
 import time
-from typing import Any, TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
 
 from src.chat.planner_actions.action_manager import ChatterActionManager
 from src.common.logger import get_logger

src/chat/message_manager/context_manager.py

@@ -6,7 +6,7 @@
 
 import asyncio
 import time
-from typing import Any, TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
 
 from src.chat.energy_system import energy_manager
 from src.common.data_models.database_data_model import DatabaseMessages

src/chat/message_manager/distribution_manager.py

@@ -5,7 +5,7 @@
 
 import asyncio
 import time
-from typing import Any, TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
 
 from src.chat.chatter_manager import ChatterManager
 from src.chat.energy_system import energy_manager

src/chat/message_receive/storage.py

@@ -12,7 +12,6 @@ from src.common.data_models.database_data_model import DatabaseMessages
 from src.common.database.core import get_db_session
 from src.common.database.core.models import Images, Messages
 from src.common.logger import get_logger
-from src.config.config import global_config
 
 from .chat_stream import ChatStream
 from .message import MessageSending

src/chat/planner_actions/action_manager.py

@@ -242,9 +242,9 @@ class ChatterActionManager:
                 }
             else:
                 # 检查目标消息是否为表情包消息以及配置是否允许回复表情包
-                if target_message and getattr(target_message, 'is_emoji', False):
+                if target_message and getattr(target_message, "is_emoji", False):
                     # 如果是表情包消息且配置不允许回复表情包，则跳过回复
-                    if not getattr(global_config.chat, 'allow_reply_to_emoji', True):
+                    if not getattr(global_config.chat, "allow_reply_to_emoji", True):
                         logger.info(f"{log_prefix} 目标消息为表情包且配置不允许回复表情包，跳过回复")
                         return {"action_type": action_name, "success": True, "reply_text": "", "skip_reason": "emoji_not_allowed"}
 

src/chat/replyer/default_generator.py

@@ -1878,7 +1878,7 @@ class DefaultReplyer:
     async def build_relation_info(self, sender: str, target: str):
         # 获取用户ID
         if sender == f"{global_config.bot.nickname}(你)":
-            return f"你将要回复的是你自己发送的消息。"
+            return "你将要回复的是你自己发送的消息。"
 
         person_info_manager = get_person_info_manager()
         person_id = await person_info_manager.get_person_id_by_person_name(sender)

src/chat/utils/attention_optimizer.py

@@ -58,7 +58,7 @@ class BlockShuffler:
             for group in BlockShuffler.SWAPPABLE_BLOCK_GROUPS:
                 # 过滤出在当前上下文中实际存在的、非空的block
                 existing_blocks = [
-                    block for block in group if block in context_data and context_data[block]
+                    block for block in group if context_data.get(block)
                 ]
 
                 if len(existing_blocks) > 1:

src/chat/utils/prompt_component_manager.py

@@ -2,7 +2,6 @@ import asyncio
 import copy
 import re
 from collections.abc import Awaitable, Callable
-from typing import List
 
 from src.chat.utils.prompt_params import PromptParameters
 from src.common.logger import get_logger
@@ -119,7 +118,7 @@ class PromptComponentManager:
     async def add_injection_rule(
         self,
         prompt_name: str,
-        rules: List[InjectionRule],
+        rules: list[InjectionRule],
         content_provider: Callable[..., Awaitable[str]],
         source: str = "runtime",
     ) -> bool:

src/chat/utils/report_generator.py

@@ -228,9 +228,9 @@ class HTMLReportGenerator:
 
         # 渲染模板
         # 读取CSS和JS文件内容
-        async with aiofiles.open(os.path.join(self.jinja_env.loader.searchpath[0], "report.css"), "r", encoding="utf-8") as f:
+        async with aiofiles.open(os.path.join(self.jinja_env.loader.searchpath[0], "report.css"), encoding="utf-8") as f:
             report_css = await f.read()
-        async with aiofiles.open(os.path.join(self.jinja_env.loader.searchpath[0], "report.js"), "r", encoding="utf-8") as f:
+        async with aiofiles.open(os.path.join(self.jinja_env.loader.searchpath[0], "report.js"), encoding="utf-8") as f:
             report_js = await f.read()
         # 渲染模板
         template = self.jinja_env.get_template("report.html")

src/chat/utils/statistic.py

@@ -3,8 +3,6 @@ from collections import defaultdict
 from datetime import datetime, timedelta
 from typing import Any
 
-import aiofiles
-
 from src.common.database.compatibility import db_get, db_query
 from src.common.database.core.models import LLMUsage, Messages, OnlineTime
 from src.common.logger import get_logger
@@ -16,7 +14,7 @@ logger = get_logger("maibot_statistic")
 # 彻底异步化：删除原同步包装器 _sync_db_get，所有数据库访问统一使用 await db_get。
 
 
-from .report_generator import HTMLReportGenerator, format_online_time
+from .report_generator import HTMLReportGenerator
 from .statistic_keys import *
 
 

src/chat/utils/statistic_keys.py

@@ -1,4 +1,3 @@
-# -*- coding: utf-8 -*-
 """
 该模块用于存放统计数据相关的常量键名。
 """

src/main.py

@@ -1,4 +1,5 @@
 # 再用这个就写一行注释来混提交的我直接全部🌿飞😡
+# 🌿🌿need
 import asyncio
 import signal
 import sys
@@ -21,7 +22,6 @@ from src.common.message import get_global_api
 
 # 全局背景任务集合
 _background_tasks = set()
-from src.common.remote import TelemetryHeartBeatTask
 from src.common.server import Server, get_global_server
 from src.config.config import global_config
 from src.individuality.individuality import Individuality, get_individuality

src/memory_graph/storage/persistence.py

@@ -507,7 +507,7 @@ class PersistenceManager:
             GraphStore 对象
         """
         try:
-            async with aiofiles.open(input_file, "r", encoding="utf-8") as f:
+            async with aiofiles.open(input_file, encoding="utf-8") as f:
                 content = await f.read()
                 data = json.loads(content)
 

src/memory_graph/utils/__init__.py

@@ -9,10 +9,10 @@ from src.memory_graph.utils.time_parser import TimeParser
 
 __all__ = [
     "EmbeddingGenerator",
+    "Path",
+    "PathExpansionConfig",
+    "PathScoreExpansion",
     "TimeParser",
     "cosine_similarity",
     "get_embedding_generator",
-    "PathScoreExpansion",
-    "PathExpansionConfig",
-    "Path",
 ]

src/memory_graph/utils/memory_deduplication.py

@@ -12,7 +12,7 @@ from src.common.logger import get_logger
 from src.memory_graph.utils.similarity import cosine_similarity
 
 if TYPE_CHECKING:
-    from src.memory_graph.models import Memory
+    pass
 
 logger = get_logger(__name__)
 

src/memory_graph/utils/path_expansion.py

@@ -26,7 +26,6 @@ from src.memory_graph.utils.similarity import cosine_similarity
 if TYPE_CHECKING:
     import numpy as np
 
-    from src.memory_graph.models import Memory
     from src.memory_graph.storage.graph_store import GraphStore
     from src.memory_graph.storage.vector_store import VectorStore
 
@@ -496,7 +495,7 @@ class PathScoreExpansion:
 
         if query_embedding is None:
             # 无查询向量时，返回默认分数
-            return {nid: 0.5 for nid in node_ids}
+            return dict.fromkeys(node_ids, 0.5)
 
         # 批量获取节点数据
         node_data_list = await asyncio.gather(
@@ -870,4 +869,4 @@ class PathScoreExpansion:
         return recency_score
 
 
-__all__ = ["PathScoreExpansion", "PathExpansionConfig", "Path"]
+__all__ = ["Path", "PathExpansionConfig", "PathScoreExpansion"]

src/person_info/relationship_fetcher.py

@@ -269,7 +269,7 @@ class RelationshipFetcher:
             platform = "unknown"
             if existing_stream:
                 # 从现有记录获取platform
-                platform = getattr(existing_stream, 'platform', 'unknown') or "unknown"
+                platform = getattr(existing_stream, "platform", "unknown") or "unknown"
                 logger.debug(f"从现有ChatStream获取到platform: {platform}, stream_id: {stream_id}")
             else:
                 logger.debug(f"未找到现有ChatStream记录，使用默认platform: unknown, stream_id: {stream_id}")

src/plugin_system/base/base_action.py

@@ -742,7 +742,7 @@ class BaseAction(ABC):
         if not case_sensitive:
             search_text = search_text.lower()
 
-        matched_keywords: ClassVar = []
+        matched_keywords  = []
         for keyword in keywords:
             check_keyword = keyword if case_sensitive else keyword.lower()
             if check_keyword in search_text:

src/plugins/built_in/affinity_flow_chatter/planner/plan_filter.py

@@ -9,6 +9,7 @@ from datetime import datetime
 from typing import Any
 
 import orjson
+from json_repair import repair_json
 
 from src.chat.utils.chat_message_builder import (
     build_readable_messages_with_id,
@@ -19,7 +20,6 @@ from src.common.logger import get_logger
 from src.config.config import global_config, model_config
 from src.llm_models.utils_model import LLMRequest
 from src.mood.mood_manager import mood_manager
-from json_repair import repair_json
 from src.plugin_system.base.component_types import ActionInfo, ChatType
 from src.schedule.schedule_manager import schedule_manager
 
@@ -631,7 +631,6 @@ class ChatterPlanFilter:
             candidate_ids.add(normalized_id[1:])
 
         # 处理包含在文本中的ID格式 (如 "消息m123" -> 提取 m123)
-        import re
 
         # 尝试提取各种格式的ID
         id_patterns = [

src/plugins/built_in/affinity_flow_chatter/planner/plan_generator.py

@@ -10,7 +10,6 @@ from src.common.data_models.database_data_model import DatabaseMessages
 from src.common.data_models.info_data_model import Plan, TargetPersonInfo
 from src.config.config import global_config
 from src.plugin_system.base.component_types import ActionInfo, ChatMode, ChatType
-from src.plugin_system.core.component_registry import component_registry
 
 
 class ChatterPlanGenerator:

src/plugins/built_in/system_management/plugin.py

@@ -295,7 +295,7 @@ class SystemCommand(PlusCommand):
             if injections:
                 response_parts.append(f"🎯 **{target}** (注入源):")
                 for inj in injections:
-                    source_tag = f"({inj['source']})" if inj['source'] != 'static_default' else ''
+                    source_tag = f"({inj['source']})" if inj["source"] != "static_default" else ""
                     response_parts.append(f"  ⎿ `{inj['name']}` (优先级: {inj['priority']}) {source_tag}")
             else:
                 response_parts.append(f"🎯 **{target}** (无注入)")