gardel/Mofox-Core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

ruff, typing, api, bug fix

Browse Source
This commit is contained in:
UnCLASPrommer
2025-07-15 16:50:29 +08:00
parent 4ebcf4e056
commit b5fd959fe1
23 changed files with 335 additions and 238 deletions
Download Patch File Download Diff File Expand all files Collapse all files

48
src/plugin_system/apis/chat_api.py
View File

@@ -39,7 +39,12 @@ class ChatManager:
Returns:
List[ChatStream]: 聊天流列表
Raises:
TypeError: 如果 platform 不是字符串或 SpecialTypes 枚举类型
"""
if not isinstance(platform, (str, SpecialTypes)):
raise TypeError("platform 必须是字符串或是 SpecialTypes 枚举")
streams = []
try:
for _, stream in get_chat_manager().streams.items():
@@ -60,6 +65,8 @@ class ChatManager:
Returns:
List[ChatStream]: 群聊聊天流列表
"""
if not isinstance(platform, (str, SpecialTypes)):
raise TypeError("platform 必须是字符串或是 SpecialTypes 枚举")
streams = []
try:
for _, stream in get_chat_manager().streams.items():
@@ -79,7 +86,12 @@ class ChatManager:
Returns:
List[ChatStream]: 私聊聊天流列表
Raises:
TypeError: 如果 platform 不是字符串或 SpecialTypes 枚举类型
"""
if not isinstance(platform, (str, SpecialTypes)):
raise TypeError("platform 必须是字符串或是 SpecialTypes 枚举")
streams = []
try:
for _, stream in get_chat_manager().streams.items():
@@ -102,7 +114,17 @@ class ChatManager:
Returns:
Optional[ChatStream]: 聊天流对象，如果未找到返回None
Raises:
ValueError: 如果 group_id 为空字符串
TypeError: 如果 group_id 不是字符串类型或 platform 不是字符串或 SpecialTypes
"""
if not isinstance(group_id, str):
raise TypeError("group_id 必须是字符串类型")
if not isinstance(platform, (str, SpecialTypes)):
raise TypeError("platform 必须是字符串或是 SpecialTypes 枚举")
if not group_id:
raise ValueError("group_id 不能为空")
try:
for _, stream in get_chat_manager().streams.items():
if (
@@ -129,7 +151,17 @@ class ChatManager:
Returns:
Optional[ChatStream]: 聊天流对象，如果未找到返回None
Raises:
ValueError: 如果 user_id 为空字符串
TypeError: 如果 user_id 不是字符串类型或 platform 不是字符串或 SpecialTypes
"""
if not isinstance(user_id, str):
raise TypeError("user_id 必须是字符串类型")
if not isinstance(platform, (str, SpecialTypes)):
raise TypeError("platform 必须是字符串或是 SpecialTypes 枚举")
if not user_id:
raise ValueError("user_id 不能为空")
try:
for _, stream in get_chat_manager().streams.items():
if (
@@ -153,9 +185,15 @@ class ChatManager:
Returns:
str: 聊天类型 ("group", "private", "unknown")
Raises:
TypeError: 如果 chat_stream 不是 ChatStream 类型
ValueError: 如果 chat_stream 为空
"""
if not isinstance(chat_stream, ChatStream):
raise TypeError("chat_stream 必须是 ChatStream 类型")
if not chat_stream:
raise ValueError("chat_stream cannot be None")
raise ValueError("chat_stream 不能为 None")
if hasattr(chat_stream, "group_info"):
return "group" if chat_stream.group_info else "private"
@@ -170,9 +208,15 @@ class ChatManager:
Returns:
Dict[str, Any]: 聊天流信息字典
Raises:
TypeError: 如果 chat_stream 不是 ChatStream 类型
ValueError: 如果 chat_stream 为空
"""
if not chat_stream:
return {}
raise ValueError("chat_stream 不能为 None")
if not isinstance(chat_stream, ChatStream):
raise TypeError("chat_stream 必须是 ChatStream 类型")
try:
info: Dict[str, Any] = {

43
src/plugin_system/apis/emoji_api.py
View File

@@ -8,6 +8,8 @@
count = emoji_api.get_count()
"""
import random
from typing import Optional, Tuple, List
from src.common.logger import get_logger
from src.chat.emoji_system.emoji_manager import get_emoji_manager
@@ -29,7 +31,15 @@ async def get_by_description(description: str) -> Optional[Tuple[str, str, str]]
Returns:
Optional[Tuple[str, str, str]]: (base64编码, 表情包描述, 匹配的情感标签) 或 None
Raises:
ValueError: 如果描述为空字符串
TypeError: 如果描述不是字符串类型
"""
if not description:
raise ValueError("描述不能为空")
if not isinstance(description, str):
raise TypeError("描述必须是字符串类型")
try:
logger.debug(f"[EmojiAPI] 根据描述获取表情包: {description}")
@@ -55,7 +65,7 @@ async def get_by_description(description: str) -> Optional[Tuple[str, str, str]]
return None
async def get_random(count: int = 1) -> Optional[List[Tuple[str, str, str]]]:
async def get_random(count: Optional[int] = 1) -> Optional[List[Tuple[str, str, str]]]:
"""随机获取指定数量的表情包
Args:
@@ -63,8 +73,17 @@ async def get_random(count: int = 1) -> Optional[List[Tuple[str, str, str]]]:
Returns:
Optional[List[Tuple[str, str, str]]]: 包含(base64编码, 表情包描述, 随机情感标签)的元组列表，如果失败则为None
Raises:
TypeError: 如果count不是整数类型
ValueError: 如果count为负数
"""
if count <= 0:
if not isinstance(count, int):
raise TypeError("count 必须是整数类型")
if count < 0:
raise ValueError("count 不能为负数")
if count == 0:
logger.warning("[EmojiAPI] count 为0，返回空列表")
return []
try:
@@ -90,8 +109,6 @@ async def get_random(count: int = 1) -> Optional[List[Tuple[str, str, str]]]:
count = len(valid_emojis)
# 随机选择
import random
selected_emojis = random.sample(valid_emojis, count)
results = []
@@ -128,7 +145,15 @@ async def get_by_emotion(emotion: str) -> Optional[Tuple[str, str, str]]:
Returns:
Optional[Tuple[str, str, str]]: (base64编码, 表情包描述, 匹配的情感标签) 或 None
Raises:
ValueError: 如果情感标签为空字符串
TypeError: 如果情感标签不是字符串类型
"""
if not emotion:
raise ValueError("情感标签不能为空")
if not isinstance(emotion, str):
raise TypeError("情感标签必须是字符串类型")
try:
logger.info(f"[EmojiAPI] 根据情感获取表情包: {emotion}")
@@ -146,8 +171,6 @@ async def get_by_emotion(emotion: str) -> Optional[Tuple[str, str, str]]:
return None
# 随机选择匹配的表情包
import random
selected_emoji = random.choice(matching_emojis)
emoji_base64 = image_path_to_base64(selected_emoji.full_path)
@@ -185,11 +208,11 @@ def get_count() -> int:
return 0
def get_info() -> dict:
def get_info():
"""获取表情包系统信息
Returns:
dict: 包含表情包数量、最大数量等信息
dict: 包含表情包数量、最大数量、可用数量信息
"""
try:
emoji_manager = get_emoji_manager()
@@ -203,7 +226,7 @@ def get_info() -> dict:
return {"current_count": 0, "max_count": 0, "available_emojis": 0}
def get_emotions() -> list:
def get_emotions() -> List[str]:
"""获取所有可用的情感标签
Returns:
@@ -223,7 +246,7 @@ def get_emotions() -> list:
return []
def get_descriptions() -> list:
def get_descriptions() -> List[str]:
"""获取所有表情包描述
Returns:

10
src/plugin_system/apis/generator_api.py
View File

@@ -5,11 +5,12 @@
使用方式：
from src.plugin_system.apis import generator_api
replyer = generator_api.get_replyer(chat_stream)
success, reply_set = await generator_api.generate_reply(chat_stream, action_data, reasoning)
success, reply_set, _ = await generator_api.generate_reply(chat_stream, action_data, reasoning)
"""
import traceback
from typing import Tuple, Any, Dict, List, Optional
from rich.traceback import install
from src.common.logger import get_logger
from src.chat.replyer.default_generator import DefaultReplyer
from src.chat.message_receive.chat_stream import ChatStream
@@ -17,6 +18,8 @@ from src.chat.utils.utils import process_llm_response
from src.chat.replyer.replyer_manager import replyer_manager
from src.plugin_system.base.component_types import ActionInfo
install(extra_lines=3)
logger = get_logger("generator_api")
@@ -44,7 +47,12 @@ def get_replyer(
Returns:
Optional[DefaultReplyer]: 回复器对象，如果获取失败则返回None
Raises:
ValueError: chat_stream 和 chat_id 均为空
"""
if not chat_id and not chat_stream:
raise ValueError("chat_stream 和 chat_id 不可均为空")
try:
logger.debug(f"[GeneratorAPI] 正在获取回复器，chat_id: {chat_id}, chat_stream: {'有' if chat_stream else '无'}")
return replyer_manager.get_replyer(

16
src/plugin_system/apis/llm_api.py
View File

@@ -14,7 +14,6 @@ from src.config.config import global_config
logger = get_logger("llm_api")
# =============================================================================
# LLM模型API函数
# =============================================================================
@@ -31,8 +30,21 @@ def get_available_models() -> Dict[str, Any]:
logger.error("[LLMAPI] 无法获取模型列表：全局配置中未找到 model 配置")
return {}
# 自动获取所有属性并转换为字典形式
rets = {}
models = global_config.model
return models
attrs = dir(models)
for attr in attrs:
if not attr.startswith("__"):
try:
value = getattr(models, attr)
if not callable(value): # 排除方法
rets[attr] = value
except Exception as e:
logger.debug(f"[LLMAPI] 获取属性 {attr} 失败: {e}")
continue
return rets
except Exception as e:
logger.error(f"[LLMAPI] 获取可用模型失败: {e}")
return {}

10
src/plugin_system/apis/send_api.py
View File

@@ -114,7 +114,11 @@ async def _send_to_target(
# 发送消息
sent_msg = await heart_fc_sender.send_message(
bot_message, typing=typing, set_reply=(anchor_message is not None), storage_message=storage_message, show_log=show_log
bot_message,
typing=typing,
set_reply=(anchor_message is not None),
storage_message=storage_message,
show_log=show_log,
)
if sent_msg:
@@ -363,7 +367,9 @@ async def custom_to_stream(
Returns:
bool: 是否发送成功
"""
return await _send_to_target(message_type, content, stream_id, display_message, typing, reply_to, storage_message, show_log)
return await _send_to_target(
message_type, content, stream_id, display_message, typing, reply_to, storage_message, show_log
)
async def text_to_group(