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

Merge branch 'dev' of https://github.com/MoFox-Studio/MoFox-Core into dev

Browse Source
This commit is contained in:
Windpicker-owo
2025-12-08 17:19:33 +08:00
parent a1d60ab026 f9b193c86d
commit 01bcfb491a
136 changed files with 2188 additions and 2230 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -13,7 +13,7 @@
"""
from enum import Enum
from typing import Any, TYPE_CHECKING
from typing import TYPE_CHECKING, Any
from src.common.logger import get_logger

26
src/plugin_system/apis/cross_context_api.py
View File

@@ -3,13 +3,11 @@
"""
import time
from typing import Any, TYPE_CHECKING
from src.common.message_repository import find_messages
from typing import TYPE_CHECKING, Any
from src.chat.message_receive.chat_stream import get_chat_manager
from src.chat.utils.chat_message_builder import (
build_readable_messages_with_id,
get_raw_msg_before_timestamp_with_chat,
)
from src.common.logger import get_logger
from src.common.message_repository import get_user_messages_from_streams
@@ -31,20 +29,20 @@ async def build_cross_context_s4u(
"""
# 记录S4U上下文构建开始
logger.debug("[S4U] Starting S4U context build.")
# 检查全局配置是否存在且包含必要部分
if not global_config or not global_config.cross_context or not global_config.bot:
logger.error("全局配置尚未初始化或缺少关键配置，无法构建S4U上下文。")
return ""
# 获取跨上下文配置
cross_context_config = global_config.cross_context
# 检查目标用户信息和用户ID是否存在
if not target_user_info or not (user_id := target_user_info.get("user_id")):
logger.warning(f"[S4U] Failed: target_user_info ({target_user_info}) or user_id is missing.")
return ""
# 记录目标用户ID
logger.debug(f"[S4U] Target user ID: {user_id}")
@@ -56,14 +54,14 @@ async def build_cross_context_s4u(
# --- 1. 优先处理私聊上下文 ---
# 获取与目标用户的私聊流ID
private_stream_id = chat_manager.get_stream_id(chat_stream.platform, user_id, is_group=False)
# 如果存在私聊流且不是当前聊天流
if private_stream_id and private_stream_id != chat_stream.stream_id:
logger.debug(f"[S4U] Found private chat with target user: {private_stream_id}")
try:
# 定义需要获取消息的用户ID列表（目标用户和机器人自己）
user_ids_to_fetch = [str(user_id), str(global_config.bot.qq_account)]
# 从指定私聊流中获取双方的消息
messages_by_stream = await get_user_messages_from_streams(
user_ids=user_ids_to_fetch,
@@ -71,12 +69,12 @@ async def build_cross_context_s4u(
timestamp_after=time.time() - (3 * 24 * 60 * 60), # 最近3天的消息
limit_per_stream=cross_context_config.s4u_limit,
)
# 如果获取到了私聊消息
if private_messages := messages_by_stream.get(private_stream_id):
chat_name = await chat_manager.get_stream_name(private_stream_id) or "私聊"
title = f'[以下是您与"{chat_name}"的近期私聊记录]\n'
# 格式化消息为可读字符串
formatted, _ = await build_readable_messages_with_id(private_messages, timestamp_mode="relative")
private_context_block = f"{title}{formatted}"
@@ -86,7 +84,7 @@ async def build_cross_context_s4u(
# --- 2. 处理其他群聊上下文 ---
streams_to_scan = []
# 根据S4U配置模式（白名单/黑名单）确定要扫描的聊天范围
if cross_context_config.s4u_mode == "whitelist":
# 白名单模式：只扫描在白名单中的聊天
@@ -95,7 +93,7 @@ async def build_cross_context_s4u(
platform, chat_type, chat_raw_id = chat_str.split(":")
is_group = chat_type == "group"
stream_id = chat_manager.get_stream_id(platform, chat_raw_id, is_group=is_group)
# 排除当前聊和私聊
if stream_id and stream_id != chat_stream.stream_id and stream_id != private_stream_id:
streams_to_scan.append(stream_id)
@@ -113,7 +111,7 @@ async def build_cross_context_s4u(
blacklisted_streams.add(stream_id)
except ValueError:
logger.warning(f"无效的S4U黑名单格式: {chat_str}")
# 将不在黑名单中的流添加到扫描列表
streams_to_scan.extend(
stream_id for stream_id in chat_manager.streams

137
src/plugin_system/apis/permission_api.py
View File

@@ -13,9 +13,9 @@
"""
from abc import ABC, abstractmethod # ABC: 抽象基类，abstractmethod: 抽象方法装饰器
from dataclasses import dataclass # dataclass: 自动生成 __init__, __repr__ 等方法的装饰器
from enum import Enum # Enum: 枚举类型基类
from typing import Any # Any: 表示任意类型
from dataclasses import dataclass # dataclass: 自动生成 __init__, __repr__ 等方法的装饰器
from enum import Enum # Enum: 枚举类型基类
from typing import Any # Any: 表示任意类型
from src.common.logger import get_logger
@@ -25,7 +25,7 @@ logger = get_logger(__name__) # 获取当前模块的日志记录器
class PermissionLevel(Enum):
"""
权限等级枚举类。
定义了系统中的权限等级，目前只有 MASTER（管理员/主人）级别。
MASTER 用户拥有最高权限，可以执行所有操作。
"""
@@ -36,9 +36,9 @@ class PermissionLevel(Enum):
class PermissionNode:
"""
权限节点数据类。
每个权限节点代表一个具体的权限项，例如"发送消息"、"管理用户"等。
属性:
node_name: 权限节点名称，例如 "plugin.chat.send_message"
description: 权限描述，用于向用户展示这个权限的用途
@@ -55,9 +55,9 @@ class PermissionNode:
class UserInfo:
"""
用户信息数据类。
用于唯一标识一个用户，通过平台+用户ID的组合确定用户身份。
属性:
platform: 用户所在平台，例如 "qq", "telegram", "discord"
user_id: 用户在该平台上的唯一标识ID
@@ -68,7 +68,7 @@ class UserInfo:
def __post_init__(self):
"""
dataclass 的后初始化钩子。
确保 user_id 始终是字符串类型，即使传入的是数字也会被转换。
这样可以避免类型不一致导致的比较问题。
"""
@@ -78,25 +78,25 @@ class UserInfo:
class IPermissionManager(ABC):
"""
权限管理器抽象接口（Interface）。
这是一个抽象基类，定义了权限管理器必须实现的所有方法。
具体的权限管理实现类需要继承此接口并实现所有抽象方法。
使用抽象接口的好处：
1. 解耦：PermissionAPI 不需要知道具体的实现细节
2. 可测试：可以轻松创建 Mock 实现用于测试
3. 可替换：可以随时更换不同的权限管理实现
"""
@abstractmethod
async def check_permission(self, user: UserInfo, permission_node: str) -> bool:
"""
检查用户是否拥有指定权限。
Args:
user: 要检查的用户信息
permission_node: 权限节点名称
Returns:
bool: True 表示用户拥有该权限，False 表示没有
"""
@@ -106,12 +106,12 @@ class IPermissionManager(ABC):
async def is_master(self, user: UserInfo) -> bool:
"""
检查用户是否是管理员/主人。
管理员拥有最高权限，通常绕过所有权限检查。
Args:
user: 要检查的用户信息
Returns:
bool: True 表示是管理员，False 表示不是
"""
@@ -121,12 +121,12 @@ class IPermissionManager(ABC):
async def register_permission_node(self, node: PermissionNode) -> bool:
"""
注册一个新的权限节点。
插件在加载时会调用此方法注册自己需要的权限。
Args:
node: 要注册的权限节点信息
Returns:
bool: True 表示注册成功，False 表示失败（可能是重复注册）
"""
@@ -136,11 +136,11 @@ class IPermissionManager(ABC):
async def grant_permission(self, user: UserInfo, permission_node: str) -> bool:
"""
授予用户指定权限。
Args:
user: 目标用户信息
permission_node: 要授予的权限节点名称
Returns:
bool: True 表示授权成功，False 表示失败
"""
@@ -150,11 +150,11 @@ class IPermissionManager(ABC):
async def revoke_permission(self, user: UserInfo, permission_node: str) -> bool:
"""
撤销用户的指定权限。
Args:
user: 目标用户信息
permission_node: 要撤销的权限节点名称
Returns:
bool: True 表示撤销成功，False 表示失败
"""
@@ -164,10 +164,10 @@ class IPermissionManager(ABC):
async def get_user_permissions(self, user: UserInfo) -> list[str]:
"""
获取用户拥有的所有权限列表。
Args:
user: 目标用户信息
Returns:
list[str]: 用户拥有的权限节点名称列表
"""
@@ -177,7 +177,7 @@ class IPermissionManager(ABC):
async def get_all_permission_nodes(self) -> list[PermissionNode]:
"""
获取系统中所有已注册的权限节点。
Returns:
list[PermissionNode]: 所有权限节点的列表
"""
@@ -187,10 +187,10 @@ class IPermissionManager(ABC):
async def get_plugin_permission_nodes(self, plugin_name: str) -> list[PermissionNode]:
"""
获取指定插件注册的所有权限节点。
Args:
plugin_name: 插件名称
Returns:
list[PermissionNode]: 该插件注册的权限节点列表
"""
@@ -200,27 +200,27 @@ class IPermissionManager(ABC):
class PermissionAPI:
"""
权限API封装类。
这是对外暴露的权限操作接口，插件和其他模块通过这个类来进行权限相关操作。
它封装了底层的 IPermissionManager，提供更简洁的调用方式。
使用方式：
from src.plugin_system.apis.permission_api import permission_api
# 检查权限
has_perm = await permission_api.check_permission("qq", "12345", "chat.send")
# 检查是否是管理员
is_admin = await permission_api.is_master("qq", "12345")
设计模式：
这是一个单例模式的变体，模块级别的 permission_api 实例供全局使用。
"""
def __init__(self):
"""
初始化 PermissionAPI。
初始时权限管理器为 None，需要在系统启动时通过 set_permission_manager 设置。
"""
self._permission_manager: IPermissionManager | None = None # 底层权限管理器实例
@@ -228,9 +228,9 @@ class PermissionAPI:
def set_permission_manager(self, manager: IPermissionManager):
"""
设置权限管理器实例。
这个方法应该在系统启动时被调用，注入具体的权限管理器实现。
Args:
manager: 实现了 IPermissionManager 接口的权限管理器实例
"""
@@ -239,10 +239,10 @@ class PermissionAPI:
def _ensure_manager(self):
"""
确保权限管理器已设置（内部辅助方法）。
如果权限管理器未设置，抛出 RuntimeError 异常。
这是一个防御性编程措施，帮助开发者快速发现配置问题。
Raises:
RuntimeError: 当权限管理器未设置时
"""
@@ -252,17 +252,17 @@ class PermissionAPI:
async def check_permission(self, platform: str, user_id: str, permission_node: str) -> bool:
"""
检查用户是否拥有指定权限。
这是最常用的权限检查方法，在执行需要权限的操作前调用。
Args:
platform: 用户所在平台（如 "qq", "telegram"）
user_id: 用户ID
permission_node: 要检查的权限节点名称
Returns:
bool: True 表示用户拥有权限，False 表示没有
Example:
if await permission_api.check_permission("qq", "12345", "admin.ban_user"):
# 执行封禁操作
@@ -276,13 +276,13 @@ class PermissionAPI:
async def is_master(self, platform: str, user_id: str) -> bool:
"""
检查用户是否是管理员/主人。
管理员是系统的最高权限用户，通常在配置文件中指定。
Args:
platform: 用户所在平台
user_id: 用户ID
Returns:
bool: True 表示是管理员，False 表示不是
"""
@@ -302,19 +302,19 @@ class PermissionAPI:
) -> bool:
"""
注册一个新的权限节点。
插件在初始化时应调用此方法注册自己需要的权限节点。
Args:
node_name: 权限节点名称，建议使用 "插件名.功能.操作" 的格式
description: 权限描述，向用户解释这个权限的作用
plugin_name: 注册此权限的插件名称
default_granted: 是否默认授予所有用户（默认 False，需要显式授权）
allow_relative: 预留参数，是否允许相对权限名（目前未使用）
Returns:
bool: True 表示注册成功，False 表示失败
Example:
await permission_api.register_permission_node(
node_name="my_plugin.chat.send_image",
@@ -324,7 +324,6 @@ class PermissionAPI:
)
"""
self._ensure_manager()
original_name = node_name # 保存原始名称（预留给相对路径处理）
# 创建权限节点对象
node = PermissionNode(node_name, description, plugin_name, default_granted)
@@ -337,17 +336,17 @@ class PermissionAPI:
async def grant_permission(self, platform: str, user_id: str, permission_node: str) -> bool:
"""
授予用户指定权限。
通常由管理员调用，给某个用户赋予特定权限。
Args:
platform: 目标用户所在平台
user_id: 目标用户ID
permission_node: 要授予的权限节点名称
Returns:
bool: True 表示授权成功，False 表示失败
Example:
# 授予用户管理权限
await permission_api.grant_permission("qq", "12345", "admin.manage_users")
@@ -360,14 +359,14 @@ class PermissionAPI:
async def revoke_permission(self, platform: str, user_id: str, permission_node: str) -> bool:
"""
撤销用户的指定权限。
通常由管理员调用，移除某个用户的特定权限。
Args:
platform: 目标用户所在平台
user_id: 目标用户ID
permission_node: 要撤销的权限节点名称
Returns:
bool: True 表示撤销成功，False 表示失败
"""
@@ -379,16 +378,16 @@ class PermissionAPI:
async def get_user_permissions(self, platform: str, user_id: str) -> list[str]:
"""
获取用户拥有的所有权限列表。
可用于展示用户的权限信息，或进行批量权限检查。
Args:
platform: 目标用户所在平台
user_id: 目标用户ID
Returns:
list[str]: 用户拥有的所有权限节点名称列表
Example:
perms = await permission_api.get_user_permissions("qq", "12345")
print(f"用户拥有以下权限: {perms}")
@@ -401,16 +400,16 @@ class PermissionAPI:
async def get_all_permission_nodes(self) -> list[dict[str, Any]]:
"""
获取系统中所有已注册的权限节点。
返回所有插件注册的权限节点信息，可用于权限管理界面展示。
Returns:
list[dict]: 权限节点信息列表，每个字典包含：
- node_name: 权限节点名称
- description: 权限描述
- plugin_name: 所属插件名称
- default_granted: 是否默认授予
Note:
返回字典而非 PermissionNode 对象，便于序列化和API响应。
"""
@@ -432,12 +431,12 @@ class PermissionAPI:
async def get_plugin_permission_nodes(self, plugin_name: str) -> list[dict[str, Any]]:
"""
获取指定插件注册的所有权限节点。
用于查看某个特定插件定义了哪些权限。
Args:
plugin_name: 插件名称
Returns:
list[dict]: 该插件的权限节点信息列表，格式同 get_all_permission_nodes
"""

2
src/plugin_system/apis/plugin_manage_api.py
View File

@@ -217,7 +217,7 @@ async def load_plugin(plugin_name: str) -> bool:
logger.info(f"插件 '{plugin_name}' 加载成功。")
else:
logger.error(f"插件 '{plugin_name}' 加载失败。")
return success

2
src/plugin_system/apis/schedule_api.py
View File

@@ -208,7 +208,7 @@ class ScheduleAPI:
if not time_range:
continue
try:
event_start_str, event_end_str = time_range.split("-")
event_start_str, _event_end_str = time_range.split("-")
event_start = datetime.strptime(event_start_str.strip(), "%H:%M").time()
if start <= event_start < end:
activities_in_range.append(event)

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

@@ -93,7 +93,9 @@ import uuid
from typing import TYPE_CHECKING, Any
from mofox_wire import MessageEnvelope
from src.common.data_models.database_data_model import DatabaseUserInfo
if TYPE_CHECKING:
from src.common.data_models.database_data_model import DatabaseMessages

6
src/plugin_system/apis/unified_scheduler.py
View File

@@ -615,7 +615,7 @@ class UnifiedScheduler:
async def _execute_task(self, task: ScheduleTask) -> None:
"""执行单个任务（完全隔离）"""
execution = task.start_execution()
task.start_execution()
self._deadlock_detector.register_task(task.schedule_id, task.task_name)
try:
@@ -755,7 +755,7 @@ class UnifiedScheduler:
async def _execute_event_task(self, task: ScheduleTask, event_params: dict[str, Any]) -> None:
"""执行事件触发的任务"""
execution = task.start_execution()
task.start_execution()
self._deadlock_detector.register_task(task.schedule_id, task.task_name)
try:
@@ -859,7 +859,7 @@ class UnifiedScheduler:
for i, timeout in enumerate(timeouts):
try:
# 使用 asyncio.wait 代替 wait_for，避免重新抛出异常
done, pending = await asyncio.wait({task._asyncio_task}, timeout=timeout)
done, _pending = await asyncio.wait({task._asyncio_task}, timeout=timeout)
if done:
# 任务已完成（可能是正常完成或被取消）