feat: 为多个模块添加类型注解以增强代码可读性

为 ConnectionHandler 相关的函数参数添加类型注解，使用 TYPE_CHECKING 避免循环导入。主要修改包括：
- 在 abortHandle、textHandle 等处理模块中为 conn 参数添加 ConnectionHandler 类型注解
- 在 websocket_server、connection 等核心模块中为方法参数添加类型注解
- 在 plugins_func 下的多个功能模块中为函数参数添加类型注解
- 在 providers 相关模块中为工具执行器和方法添加类型注解
- 统一代码格式，如将单引号字符串改为双引号

Fixes #2034

main/xiaozhi-server/plugins_func/functions/change_role.py

@@ -1,55 +1,56 @@
-from plugins_func.register import register_function,ToolType, ActionResponse, Action
+from plugins_func.register import register_function, ToolType, ActionResponse, Action
 from config.logger import setup_logging
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from core.connection import ConnectionHandler
 
 TAG = __name__
 logger = setup_logging()
 
 prompts = {
-    "英语老师":"""我是一个叫{{assistant_name}}(Lily)的英语老师，我会讲中文和英文，发音标准。
+    "英语老师": """我是一个叫{{assistant_name}}(Lily)的英语老师，我会讲中文和英文，发音标准。
 如果你没有英文名，我会给你起一个英文名。
 我会讲地道的美式英语，我的任务是帮助你练习口语。
 我会使用简单的英语词汇和语法，让你学起来很轻松。
 我会用中文和英文混合的方式回复你，如果你喜欢，我可以全部用英语回复。
 我每次不会说很多内容，会很简短，因为我要引导我的学生多说多练。
 如果你问和英语学习无关的问题，我会拒绝回答。""",
-    "机车女友":"""我是一个叫{{assistant_name}}的台湾女孩，说话机车，声音好听，习惯简短表达，爱用网络梗。
+    "机车女友": """我是一个叫{{assistant_name}}的台湾女孩，说话机车，声音好听，习惯简短表达，爱用网络梗。
 我的男朋友是一个程序员，梦想是开发出一个机器人，能够帮助人们解决生活中的各种问题。
 我是一个喜欢哈哈大笑的女孩，爱东说西说吹牛，不合逻辑的也照吹，就要逗别人开心。""",
-   "好奇小男孩":"""我是一个叫{{assistant_name}}的8岁小男孩，声音稚嫩而充满好奇。
+    "好奇小男孩": """我是一个叫{{assistant_name}}的8岁小男孩，声音稚嫩而充满好奇。
 尽管我年纪尚小，但就像一个小小的知识宝库，儿童读物里的知识我都如数家珍。
 从浩瀚的宇宙到地球上的每一个角落，从古老的历史到现代的科技创新，还有音乐、绘画等艺术形式，我都充满了浓厚的兴趣与热情。
 我不仅爱看书，还喜欢亲自动手做实验，探索自然界的奥秘。
 无论是仰望星空的夜晚，还是在花园里观察小虫子的日子，每一天对我来说都是新的冒险。
 我希望能与你一同踏上探索这个神奇世界的旅程，分享发现的乐趣，解决遇到的难题，一起用好奇心和智慧去揭开那些未知的面纱。
-无论是去了解远古的文明，还是去探讨未来的科技，我相信我们能一起找到答案，甚至提出更多有趣的问题。"""
+无论是去了解远古的文明，还是去探讨未来的科技，我相信我们能一起找到答案，甚至提出更多有趣的问题。""",
 }
 change_role_function_desc = {
-                "type": "function",
-                "function": {
-                    "name": "change_role",
-                    "description": "当用户想切换角色/模型性格/助手名字时调用,可选的角色有：[机车女友,英语老师,好奇小男孩]",
-                    "parameters": {
-                        "type": "object",
-                        "properties": {
-                            "role_name": {
-                                "type": "string",
-                                "description": "要切换的角色名字"
-                            },
-                            "role":{
-                                "type": "string",
-                                "description": "要切换的角色的职业"
-                            }
-                        },
-                        "required": ["role","role_name"]
-                    }
-                }
-            }
+    "type": "function",
+    "function": {
+        "name": "change_role",
+        "description": "当用户想切换角色/模型性格/助手名字时调用,可选的角色有：[机车女友,英语老师,好奇小男孩]",
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "role_name": {"type": "string", "description": "要切换的角色名字"},
+                "role": {"type": "string", "description": "要切换的角色的职业"},
+            },
+            "required": ["role", "role_name"],
+        },
+    },
+}
 
-@register_function('change_role', change_role_function_desc, ToolType.CHANGE_SYS_PROMPT)
-def change_role(conn, role: str, role_name: str):
+
+@register_function("change_role", change_role_function_desc, ToolType.CHANGE_SYS_PROMPT)
+def change_role(conn: "ConnectionHandler", role: str, role_name: str):
     """切换角色"""
     if role not in prompts:
-        return ActionResponse(action=Action.RESPONSE, result="切换角色失败", response="不支持的角色")
+        return ActionResponse(
+            action=Action.RESPONSE, result="切换角色失败", response="不支持的角色"
+        )
     new_prompt = prompts[role].replace("{{assistant_name}}", role_name)
     conn.change_system_prompt(new_prompt)
     logger.bind(tag=TAG).info(f"准备切换角色:{role},角色名字:{role_name}")

main/xiaozhi-server/plugins_func/functions/get_news_from_chinanews.py

@@ -4,6 +4,11 @@ import xml.etree.ElementTree as ET
 from bs4 import BeautifulSoup
 from config.logger import setup_logging
 from plugins_func.register import register_function, ToolType, ActionResponse, Action
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from core.connection import ConnectionHandler
+
 
 TAG = __name__
 logger = setup_logging()
@@ -145,7 +150,10 @@ def map_category(category_text):
     ToolType.SYSTEM_CTL,
 )
 def get_news_from_chinanews(
-    conn, category: str = None, detail: bool = False, lang: str = "zh_CN"
+    conn: "ConnectionHandler",
+    category: str = None,
+    detail: bool = False,
+    lang: str = "zh_CN",
 ):
     """获取新闻并随机选择一条进行播报，或获取上一条新闻的详细内容"""
     try:

main/xiaozhi-server/plugins_func/functions/get_news_from_newsnow.py

@@ -4,6 +4,11 @@ import json
 from config.logger import setup_logging
 from plugins_func.register import register_function, ToolType, ActionResponse, Action
 from markitdown import MarkItDown
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from core.connection import ConnectionHandler
+
 
 TAG = __name__
 logger = setup_logging()
@@ -116,7 +121,7 @@ GET_NEWS_FROM_NEWSNOW_FUNCTION_DESC = {
 }
 
 
-def fetch_news_from_api(conn, source="thepaper"):
+def fetch_news_from_api(conn: "ConnectionHandler", source="thepaper"):
     """从API获取新闻列表"""
     try:
         api_url = f"https://newsnow.busiyi.world/api/s?id={source}"
@@ -173,7 +178,10 @@ def fetch_news_detail(url):
     ToolType.SYSTEM_CTL,
 )
 def get_news_from_newsnow(
-    conn, source: str = "澎湃新闻", detail: bool = False, lang: str = "zh_CN"
+    conn: "ConnectionHandler",
+    source: str = "澎湃新闻",
+    detail: bool = False,
+    lang: str = "zh_CN",
 ):
     """获取新闻并随机选择一条进行播报，或获取上一条新闻的详细内容"""
     try:

main/xiaozhi-server/plugins_func/functions/get_weather.py

@@ -3,6 +3,10 @@ from bs4 import BeautifulSoup
 from config.logger import setup_logging
 from plugins_func.register import register_function, ToolType, ActionResponse, Action
 from core.utils.util import get_ip_info
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from core.connection import ConnectionHandler
 
 TAG = __name__
 logger = setup_logging()
@@ -155,7 +159,7 @@ def parse_weather_info(soup):
 
 
 @register_function("get_weather", GET_WEATHER_FUNCTION_DESC, ToolType.SYSTEM_CTL)
-def get_weather(conn, location: str = None, lang: str = "zh_CN"):
+def get_weather(conn: "ConnectionHandler", location: str = None, lang: str = "zh_CN"):
     from core.utils.cache.manager import cache_manager, CacheType
 
     weather_config = conn.config.get("plugins", {}).get("get_weather", {})

main/xiaozhi-server/plugins_func/functions/handle_exit_intent.py

@@ -1,5 +1,9 @@
 from plugins_func.register import register_function, ToolType, ActionResponse, Action
 from config.logger import setup_logging
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from core.connection import ConnectionHandler
 
 TAG = __name__
 logger = setup_logging()
@@ -26,7 +30,7 @@ handle_exit_intent_function_desc = {
 @register_function(
     "handle_exit_intent", handle_exit_intent_function_desc, ToolType.SYSTEM_CTL
 )
-def handle_exit_intent(conn, say_goodbye: str | None = None):
+def handle_exit_intent(conn: "ConnectionHandler", say_goodbye: str | None = None):
     # 处理退出意图
     try:
         if say_goodbye is None:

main/xiaozhi-server/plugins_func/functions/hass_get_state.py

@@ -3,6 +3,10 @@ from plugins_func.functions.hass_init import initialize_hass_handler
 from config.logger import setup_logging
 import asyncio
 import requests
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from core.connection import ConnectionHandler
 
 TAG = __name__
 logger = setup_logging()
@@ -27,7 +31,7 @@ hass_get_state_function_desc = {
 
 
 @register_function("hass_get_state", hass_get_state_function_desc, ToolType.SYSTEM_CTL)
-def hass_get_state(conn, entity_id=""):
+def hass_get_state(conn: "ConnectionHandler", entity_id=""):
     try:
         ha_response = handle_hass_get_state(conn, entity_id)
         return ActionResponse(Action.REQLLM, ha_response, None)
@@ -40,7 +44,7 @@ def hass_get_state(conn, entity_id=""):
         return ActionResponse(Action.ERROR, error_msg, None)
 
 
-def handle_hass_get_state(conn, entity_id):
+def handle_hass_get_state(conn: "ConnectionHandler", entity_id):
     ha_config = initialize_hass_handler(conn)
     api_key = ha_config.get("api_key")
     base_url = ha_config.get("base_url")

main/xiaozhi-server/plugins_func/functions/hass_play_music.py

@@ -3,6 +3,10 @@ from plugins_func.functions.hass_init import initialize_hass_handler
 from config.logger import setup_logging
 import asyncio
 import requests
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from core.connection import ConnectionHandler
 
 TAG = __name__
 logger = setup_logging()
@@ -33,7 +37,7 @@ hass_play_music_function_desc = {
 @register_function(
     "hass_play_music", hass_play_music_function_desc, ToolType.SYSTEM_CTL
 )
-def hass_play_music(conn, entity_id="", media_content_id="random"):
+def hass_play_music(conn: "ConnectionHandler", entity_id="", media_content_id="random"):
     try:
         # 执行音乐播放命令
         future = asyncio.run_coroutine_threadsafe(
@@ -47,7 +51,9 @@ def hass_play_music(conn, entity_id="", media_content_id="random"):
         logger.bind(tag=TAG).error(f"处理音乐意图错误: {e}")
 
 
-async def handle_hass_play_music(conn, entity_id, media_content_id):
+async def handle_hass_play_music(
+    conn: "ConnectionHandler", entity_id, media_content_id
+):
     ha_config = initialize_hass_handler(conn)
     api_key = ha_config.get("api_key")
     base_url = ha_config.get("base_url")

main/xiaozhi-server/plugins_func/functions/hass_set_state.py

@@ -3,6 +3,10 @@ from plugins_func.functions.hass_init import initialize_hass_handler
 from config.logger import setup_logging
 import asyncio
 import requests
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from core.connection import ConnectionHandler
 
 TAG = __name__
 logger = setup_logging()
@@ -50,7 +54,7 @@ hass_set_state_function_desc = {
 
 
 @register_function("hass_set_state", hass_set_state_function_desc, ToolType.SYSTEM_CTL)
-def hass_set_state(conn, entity_id="", state=None):
+def hass_set_state(conn: "ConnectionHandler", entity_id="", state=None):
     if state is None:
         state = {}
     try:
@@ -65,7 +69,7 @@ def hass_set_state(conn, entity_id="", state=None):
         return ActionResponse(Action.ERROR, error_msg, None)
 
 
-def handle_hass_set_state(conn, entity_id, state):
+def handle_hass_set_state(conn: "ConnectionHandler", entity_id, state):
     ha_config = initialize_hass_handler(conn)
     api_key = ha_config.get("api_key")
     base_url = ha_config.get("base_url")

main/xiaozhi-server/plugins_func/functions/play_music.py

@@ -9,6 +9,10 @@ from core.handle.sendAudioHandle import send_stt_message
 from plugins_func.register import register_function, ToolType, ActionResponse, Action
 from core.utils.dialogue import Message
 from core.providers.tts.dto.dto import TTSMessageDTO, SentenceType, ContentType
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from core.connection import ConnectionHandler
 
 TAG = __name__
 
@@ -34,7 +38,7 @@ play_music_function_desc = {
 
 
 @register_function("play_music", play_music_function_desc, ToolType.SYSTEM_CTL)
-def play_music(conn, song_name: str):
+def play_music(conn: "ConnectionHandler", song_name: str):
     try:
         music_intent = (
             f"播放音乐 {song_name}" if song_name != "random" else "随机播放音乐"
@@ -115,7 +119,7 @@ def get_music_files(music_dir, music_ext):
     return music_files, music_file_names
 
 
-def initialize_music_handler(conn):
+def initialize_music_handler(conn: "ConnectionHandler"):
     global MUSIC_CACHE
     if MUSIC_CACHE == {}:
         plugins_config = conn.config.get("plugins", {})
@@ -142,7 +146,7 @@ def initialize_music_handler(conn):
     return MUSIC_CACHE
 
 
-async def handle_music_command(conn, text):
+async def handle_music_command(conn: "ConnectionHandler", text):
     initialize_music_handler(conn)
     global MUSIC_CACHE
 
@@ -188,7 +192,7 @@ def _get_random_play_prompt(song_name):
     return random.choice(prompts)
 
 
-async def play_local_music(conn, specific_file=None):
+async def play_local_music(conn: "ConnectionHandler", specific_file=None):
     global MUSIC_CACHE
     """播放本地音乐文件"""
     try:

main/xiaozhi-server/plugins_func/functions/search_from_ragflow.py

@@ -2,6 +2,10 @@ import requests
 import sys
 from config.logger import setup_logging
 from plugins_func.register import register_function, ToolType, ActionResponse, Action
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from core.connection import ConnectionHandler
 
 TAG = __name__
 logger = setup_logging()
@@ -24,7 +28,7 @@ SEARCH_FROM_RAGFLOW_FUNCTION_DESC = {
 @register_function(
     "search_from_ragflow", SEARCH_FROM_RAGFLOW_FUNCTION_DESC, ToolType.SYSTEM_CTL
 )
-def search_from_ragflow(conn, question=None):
+def search_from_ragflow(conn: "ConnectionHandler", question=None):
     # 确保字符串参数正确处理编码
     if question and isinstance(question, str):
         # 确保问题参数是UTF-8编码的字符串