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

合并utils_api

Browse Source
This commit is contained in:
UnCLAS-Prommer
2025-07-28 12:37:37 +08:00
parent 81ef6f4897
commit c0375f5dd9
6 changed files with 57 additions and 613 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
docs/plugins/api/plugin-manage-api.md
View File

@@ -36,7 +36,18 @@ def list_registered_plugins() -> List[str]:
**Returns:**
- `List[str]` - 已注册的插件名称列表。
### 3. 卸载指定的插件
### 3. 获取插件路径
```python
def get_plugin_path(plugin_name: str) -> str:
```
获取指定插件的路径。
**Args:**
- `plugin_name` (str): 要查询的插件名称。
**Returns:**
- `str` - 插件的路径，如果插件不存在则 raise ValueError。
### 4. 卸载指定的插件
```python
async def remove_plugin(plugin_name: str) -> bool:
```
@@ -48,7 +59,7 @@ async def remove_plugin(plugin_name: str) -> bool:
**Returns:**
- `bool` - 卸载是否成功。
### 4. 重新加载指定的插件
### 5. 重新加载指定的插件
```python
async def reload_plugin(plugin_name: str) -> bool:
```
@@ -60,7 +71,7 @@ async def reload_plugin(plugin_name: str) -> bool:
**Returns:**
- `bool` - 重新加载是否成功。
### 5. 加载指定的插件
### 6. 加载指定的插件
```python
def load_plugin(plugin_name: str) -> Tuple[bool, int]:
```
@@ -72,7 +83,7 @@ def load_plugin(plugin_name: str) -> Tuple[bool, int]:
**Returns:**
- `Tuple[bool, int]` - 加载是否成功，成功或失败的个数。
### 6. 添加插件目录
### 7. 添加插件目录
```python
def add_plugin_directory(plugin_directory: str) -> bool:
```
@@ -84,7 +95,7 @@ def add_plugin_directory(plugin_directory: str) -> bool:
**Returns:**
- `bool` - 添加是否成功。
### 7. 重新扫描插件目录
### 8. 重新扫描插件目录
```python
def rescan_plugin_directory() -> Tuple[int, int]:
```

435
docs/plugins/api/utils-api.md
View File

@@ -1,435 +0,0 @@
# 工具API
工具API模块提供了各种辅助功能，包括文件操作、时间处理、唯一ID生成等常用工具函数。
## 导入方式
```python
from src.plugin_system.apis import utils_api
```
## 主要功能
### 1. 文件操作
#### `get_plugin_path(caller_frame=None) -> str`
获取调用者插件的路径
**参数：**
- `caller_frame`：调用者的栈帧，默认为None（自动获取）
**返回：**
- `str`：插件目录的绝对路径
**示例：**
```python
plugin_path = utils_api.get_plugin_path()
print(f"插件路径: {plugin_path}")
```
#### `read_json_file(file_path: str, default: Any = None) -> Any`
读取JSON文件
**参数：**
- `file_path`：文件路径，可以是相对于插件目录的路径
- `default`：如果文件不存在或读取失败时返回的默认值
**返回：**
- `Any`：JSON数据或默认值
**示例：**
```python
# 读取插件配置文件
config = utils_api.read_json_file("config.json", {})
settings = utils_api.read_json_file("data/settings.json", {"enabled": True})
```
#### `write_json_file(file_path: str, data: Any, indent: int = 2) -> bool`
写入JSON文件
**参数：**
- `file_path`：文件路径，可以是相对于插件目录的路径
- `data`：要写入的数据
- `indent`：JSON缩进
**返回：**
- `bool`：是否写入成功
**示例：**
```python
data = {"name": "test", "value": 123}
success = utils_api.write_json_file("output.json", data)
```
### 2. 时间相关
#### `get_timestamp() -> int`
获取当前时间戳
**返回：**
- `int`：当前时间戳（秒）
#### `format_time(timestamp: Optional[int] = None, format_str: str = "%Y-%m-%d %H:%M:%S") -> str`
格式化时间
**参数：**
- `timestamp`：时间戳，如果为None则使用当前时间
- `format_str`：时间格式字符串
**返回：**
- `str`：格式化后的时间字符串
#### `parse_time(time_str: str, format_str: str = "%Y-%m-%d %H:%M:%S") -> int`
解析时间字符串为时间戳
**参数：**
- `time_str`：时间字符串
- `format_str`：时间格式字符串
**返回：**
- `int`：时间戳（秒）
### 3. 其他工具
#### `generate_unique_id() -> str`
生成唯一ID
**返回：**
- `str`：唯一ID
## 使用示例
### 1. 插件数据管理
```python
from src.plugin_system.apis import utils_api
class DataPlugin(BasePlugin):
def __init__(self):
self.plugin_path = utils_api.get_plugin_path()
self.data_file = "plugin_data.json"
self.load_data()
def load_data(self):
"""加载插件数据"""
default_data = {
"users": {},
"settings": {"enabled": True},
"stats": {"message_count": 0}
}
self.data = utils_api.read_json_file(self.data_file, default_data)
def save_data(self):
"""保存插件数据"""
return utils_api.write_json_file(self.data_file, self.data)
async def handle_action(self, action_data, chat_stream):
# 更新统计信息
self.data["stats"]["message_count"] += 1
self.data["stats"]["last_update"] = utils_api.get_timestamp()
# 保存数据
if self.save_data():
return {"success": True, "message": "数据已保存"}
else:
return {"success": False, "message": "数据保存失败"}
```
### 2. 日志记录系统
```python
class PluginLogger:
def __init__(self, plugin_name: str):
self.plugin_name = plugin_name
self.log_file = f"{plugin_name}_log.json"
self.logs = utils_api.read_json_file(self.log_file, [])
def log_event(self, event_type: str, message: str, data: dict = None):
"""记录事件"""
log_entry = {
"id": utils_api.generate_unique_id(),
"timestamp": utils_api.get_timestamp(),
"formatted_time": utils_api.format_time(),
"event_type": event_type,
"message": message,
"data": data or {}
}
self.logs.append(log_entry)
# 保持最新的100条记录
if len(self.logs) > 100:
self.logs = self.logs[-100:]
# 保存到文件
utils_api.write_json_file(self.log_file, self.logs)
def get_logs_by_type(self, event_type: str) -> list:
"""获取指定类型的日志"""
return [log for log in self.logs if log["event_type"] == event_type]
def get_recent_logs(self, count: int = 10) -> list:
"""获取最近的日志"""
return self.logs[-count:]
# 使用示例
logger = PluginLogger("my_plugin")
logger.log_event("user_action", "用户发送了消息", {"user_id": "123", "message": "hello"})
```
### 3. 配置管理系统
```python
class ConfigManager:
def __init__(self, config_file: str = "plugin_config.json"):
self.config_file = config_file
self.default_config = {
"enabled": True,
"debug": False,
"max_users": 100,
"response_delay": 1.0,
"features": {
"auto_reply": True,
"logging": True
}
}
self.config = self.load_config()
def load_config(self) -> dict:
"""加载配置"""
return utils_api.read_json_file(self.config_file, self.default_config)
def save_config(self) -> bool:
"""保存配置"""
return utils_api.write_json_file(self.config_file, self.config, indent=4)
def get(self, key: str, default=None):
"""获取配置值，支持嵌套访问"""
keys = key.split('.')
value = self.config
for k in keys:
if isinstance(value, dict) and k in value:
value = value[k]
else:
return default
return value
def set(self, key: str, value):
"""设置配置值，支持嵌套设置"""
keys = key.split('.')
config = self.config
for k in keys[:-1]:
if k not in config:
config[k] = {}
config = config[k]
config[keys[-1]] = value
def update_config(self, updates: dict):
"""批量更新配置"""
def deep_update(base, updates):
for key, value in updates.items():
if isinstance(value, dict) and key in base and isinstance(base[key], dict):
deep_update(base[key], value)
else:
base[key] = value
deep_update(self.config, updates)
# 使用示例
config = ConfigManager()
print(f"调试模式: {config.get('debug', False)}")
print(f"自动回复: {config.get('features.auto_reply', True)}")
config.set('features.new_feature', True)
config.save_config()
```
### 4. 缓存系统
```python
class PluginCache:
def __init__(self, cache_file: str = "plugin_cache.json", ttl: int = 3600):
self.cache_file = cache_file
self.ttl = ttl # 缓存过期时间（秒）
self.cache = self.load_cache()
def load_cache(self) -> dict:
"""加载缓存"""
return utils_api.read_json_file(self.cache_file, {})
def save_cache(self):
"""保存缓存"""
return utils_api.write_json_file(self.cache_file, self.cache)
def get(self, key: str):
"""获取缓存值"""
if key not in self.cache:
return None
item = self.cache[key]
current_time = utils_api.get_timestamp()
# 检查是否过期
if current_time - item["timestamp"] > self.ttl:
del self.cache[key]
return None
return item["value"]
def set(self, key: str, value):
"""设置缓存值"""
self.cache[key] = {
"value": value,
"timestamp": utils_api.get_timestamp()
}
self.save_cache()
def clear_expired(self):
"""清理过期缓存"""
current_time = utils_api.get_timestamp()
expired_keys = []
for key, item in self.cache.items():
if current_time - item["timestamp"] > self.ttl:
expired_keys.append(key)
for key in expired_keys:
del self.cache[key]
if expired_keys:
self.save_cache()
return len(expired_keys)
# 使用示例
cache = PluginCache(ttl=1800) # 30分钟过期
cache.set("user_data_123", {"name": "张三", "score": 100})
user_data = cache.get("user_data_123")
```
### 5. 时间处理工具
```python
class TimeHelper:
@staticmethod
def get_time_info():
"""获取当前时间的详细信息"""
timestamp = utils_api.get_timestamp()
return {
"timestamp": timestamp,
"datetime": utils_api.format_time(timestamp),
"date": utils_api.format_time(timestamp, "%Y-%m-%d"),
"time": utils_api.format_time(timestamp, "%H:%M:%S"),
"year": utils_api.format_time(timestamp, "%Y"),
"month": utils_api.format_time(timestamp, "%m"),
"day": utils_api.format_time(timestamp, "%d"),
"weekday": utils_api.format_time(timestamp, "%A")
}
@staticmethod
def time_ago(timestamp: int) -> str:
"""计算时间差"""
current = utils_api.get_timestamp()
diff = current - timestamp
if diff < 60:
return f"{diff}秒前"
elif diff < 3600:
return f"{diff // 60}分钟前"
elif diff < 86400:
return f"{diff // 3600}小时前"
else:
return f"{diff // 86400}天前"
@staticmethod
def parse_duration(duration_str: str) -> int:
"""解析时间段字符串，返回秒数"""
import re
pattern = r'(\d+)([smhd])'
matches = re.findall(pattern, duration_str.lower())
total_seconds = 0
for value, unit in matches:
value = int(value)
if unit == 's':
total_seconds += value
elif unit == 'm':
total_seconds += value * 60
elif unit == 'h':
total_seconds += value * 3600
elif unit == 'd':
total_seconds += value * 86400
return total_seconds
# 使用示例
time_info = TimeHelper.get_time_info()
print(f"当前时间: {time_info['datetime']}")
last_seen = 1699000000
print(f"最后见面: {TimeHelper.time_ago(last_seen)}")
duration = TimeHelper.parse_duration("1h30m") # 1小时30分钟 = 5400秒
```
## 最佳实践
### 1. 错误处理
```python
def safe_file_operation(file_path: str, data: dict):
"""安全的文件操作"""
try:
success = utils_api.write_json_file(file_path, data)
if not success:
logger.warning(f"文件写入失败: {file_path}")
return success
except Exception as e:
logger.error(f"文件操作出错: {e}")
return False
```
### 2. 路径处理
```python
import os
def get_data_path(filename: str) -> str:
"""获取数据文件的完整路径"""
plugin_path = utils_api.get_plugin_path()
data_dir = os.path.join(plugin_path, "data")
# 确保数据目录存在
os.makedirs(data_dir, exist_ok=True)
return os.path.join(data_dir, filename)
```
### 3. 定期清理
```python
async def cleanup_old_files():
"""清理旧文件"""
plugin_path = utils_api.get_plugin_path()
current_time = utils_api.get_timestamp()
for filename in os.listdir(plugin_path):
if filename.endswith('.tmp'):
file_path = os.path.join(plugin_path, filename)
file_time = os.path.getmtime(file_path)
# 删除超过24小时的临时文件
if current_time - file_time > 86400:
os.remove(file_path)
```
## 注意事项
1. **相对路径**：文件路径支持相对于插件目录的路径
2. **自动创建目录**：写入文件时会自动创建必要的目录
3. **错误处理**：所有函数都有错误处理，失败时返回默认值
4. **编码格式**：文件读写使用UTF-8编码
5. **时间格式**：时间戳使用秒为单位
6. **JSON格式**：JSON文件使用可读性好的缩进格式

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

@@ -1,4 +1,6 @@
from typing import Tuple, List
def list_loaded_plugins() -> List[str]:
"""
列出所有当前加载的插件。
@@ -23,6 +25,27 @@ def list_registered_plugins() -> List[str]:
return plugin_manager.list_registered_plugins()
def get_plugin_path(plugin_name: str) -> str:
"""
获取指定插件的路径。
Args:
plugin_name (str): 插件名称。
Returns:
str: 插件目录的绝对路径。
Raises:
ValueError: 如果插件不存在。
"""
from src.plugin_system.core.plugin_manager import plugin_manager
if plugin_path := plugin_manager.get_plugin_path(plugin_name):
return plugin_path
else:
raise ValueError(f"插件 '{plugin_name}' 不存在。")
async def remove_plugin(plugin_name: str) -> bool:
"""
卸载指定的插件。
@@ -71,6 +94,7 @@ def load_plugin(plugin_name: str) -> Tuple[bool, int]:
return plugin_manager.load_registered_plugin_classes(plugin_name)
def add_plugin_directory(plugin_directory: str) -> bool:
"""
添加插件目录。
@@ -84,6 +108,7 @@ def add_plugin_directory(plugin_directory: str) -> bool:
return plugin_manager.add_plugin_directory(plugin_directory)
def rescan_plugin_directory() -> Tuple[int, int]:
"""
重新扫描插件目录，加载新插件。

168
src/plugin_system/apis/utils_api.py
View File

@@ -1,168 +0,0 @@
"""工具类API模块
提供了各种辅助功能
使用方式：
from src.plugin_system.apis import utils_api
plugin_path = utils_api.get_plugin_path()
data = utils_api.read_json_file("data.json")
timestamp = utils_api.get_timestamp()
"""
import os
import json
import time
import inspect
import datetime
import uuid
from typing import Any, Optional
from src.common.logger import get_logger
logger = get_logger("utils_api")
# =============================================================================
# 文件操作API函数
# =============================================================================
def get_plugin_path(caller_frame=None) -> str:
"""获取调用者插件的路径
Args:
caller_frame: 调用者的栈帧，默认为None（自动获取）
Returns:
str: 插件目录的绝对路径
"""
try:
if caller_frame is None:
caller_frame = inspect.currentframe().f_back # type: ignore
plugin_module_path = inspect.getfile(caller_frame) # type: ignore
plugin_dir = os.path.dirname(plugin_module_path)
return plugin_dir
except Exception as e:
logger.error(f"[UtilsAPI] 获取插件路径失败: {e}")
return ""
def read_json_file(file_path: str, default: Any = None) -> Any:
"""读取JSON文件
Args:
file_path: 文件路径，可以是相对于插件目录的路径
default: 如果文件不存在或读取失败时返回的默认值
Returns:
Any: JSON数据或默认值
"""
try:
# 如果是相对路径，则相对于调用者的插件目录
if not os.path.isabs(file_path):
caller_frame = inspect.currentframe().f_back # type: ignore
plugin_dir = get_plugin_path(caller_frame)
file_path = os.path.join(plugin_dir, file_path)
if not os.path.exists(file_path):
logger.warning(f"[UtilsAPI] 文件不存在: {file_path}")
return default
with open(file_path, "r", encoding="utf-8") as f:
return json.load(f)
except Exception as e:
logger.error(f"[UtilsAPI] 读取JSON文件出错: {e}")
return default
def write_json_file(file_path: str, data: Any, indent: int = 2) -> bool:
"""写入JSON文件
Args:
file_path: 文件路径，可以是相对于插件目录的路径
data: 要写入的数据
indent: JSON缩进
Returns:
bool: 是否写入成功
"""
try:
# 如果是相对路径，则相对于调用者的插件目录
if not os.path.isabs(file_path):
caller_frame = inspect.currentframe().f_back # type: ignore
plugin_dir = get_plugin_path(caller_frame)
file_path = os.path.join(plugin_dir, file_path)
# 确保目录存在
os.makedirs(os.path.dirname(file_path), exist_ok=True)
with open(file_path, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=indent)
return True
except Exception as e:
logger.error(f"[UtilsAPI] 写入JSON文件出错: {e}")
return False
# =============================================================================
# 时间相关API函数
# =============================================================================
def get_timestamp() -> int:
"""获取当前时间戳
Returns:
int: 当前时间戳（秒）
"""
return int(time.time())
def format_time(timestamp: Optional[int | float] = None, format_str: str = "%Y-%m-%d %H:%M:%S") -> str:
"""格式化时间
Args:
timestamp: 时间戳，如果为None则使用当前时间
format_str: 时间格式字符串
Returns:
str: 格式化后的时间字符串
"""
try:
if timestamp is None:
timestamp = time.time()
return datetime.datetime.fromtimestamp(timestamp).strftime(format_str)
except Exception as e:
logger.error(f"[UtilsAPI] 格式化时间失败: {e}")
return ""
def parse_time(time_str: str, format_str: str = "%Y-%m-%d %H:%M:%S") -> int:
"""解析时间字符串为时间戳
Args:
time_str: 时间字符串
format_str: 时间格式字符串
Returns:
int: 时间戳（秒）
"""
try:
dt = datetime.datetime.strptime(time_str, format_str)
return int(dt.timestamp())
except Exception as e:
logger.error(f"[UtilsAPI] 解析时间失败: {e}")
return 0
# =============================================================================
# 其他工具函数
# =============================================================================
def generate_unique_id() -> str:
"""生成唯一ID
Returns:
str: 唯一ID
"""
return str(uuid.uuid4())

3
src/plugin_system/base/base_action.py
View File

@@ -208,7 +208,7 @@ class BaseAction(ABC):
return False, f"等待新消息失败: {str(e)}"
async def send_text(
self, content: str, reply_to: str = "", reply_to_platform_id: str = "", typing: bool = False
self, content: str, reply_to: str = "", typing: bool = False
) -> bool:
"""发送文本消息
@@ -227,7 +227,6 @@ class BaseAction(ABC):
text=content,
stream_id=self.chat_id,
reply_to=reply_to,
reply_to_platform_id=reply_to_platform_id,
typing=typing,
)

12
src/plugin_system/core/plugin_manager.py
View File

@@ -225,6 +225,18 @@ class PluginManager:
"""
return list(self.plugin_classes.keys())
def get_plugin_path(self, plugin_name: str) -> Optional[str]:
"""
获取指定插件的路径。
Args:
plugin_name: 插件名称
Returns:
Optional[str]: 插件目录的绝对路径，如果插件不存在则返回None。
"""
return self.plugin_paths.get(plugin_name)
# === 私有方法 ===
# == 目录管理 ==
def _ensure_plugin_directories(self) -> None: