File size: 5,435 Bytes
10b8d56 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 | """
插件状态持久化模块
管理插件的启用/禁用状态，与 plugin.json 的 enabled 字段独立。
"""
import json
import logging
from pathlib import Path
from datetime import datetime
from typing import Optional
from app.plugins.models import PluginStateStore, PluginStateEntry
logger = logging.getLogger(__name__)
class PluginStateStoreManager:
"""插件状态文件管理器
事实源：data/plugin_state.json
以插件名为 key 保存用户启停状态。
"""
def __init__(self, state_file_path: Path):
"""初始化状态文件管理器
Args:
state_file_path: 状态文件路径
"""
self.state_file_path = state_file_path
self._store: Optional[PluginStateStore] = None
def _load_store(self) -> PluginStateStore:
"""加载状态文件
如果文件不存在，返回空的 store。
"""
if self._store is not None:
return self._store
if not self.state_file_path.exists():
logger.info(f"状态文件不存在，创建新的: {self.state_file_path}")
self._store = PluginStateStore()
return self._store
try:
with open(self.state_file_path, "r", encoding="utf-8") as f:
data = json.load(f)
self._store = PluginStateStore(**data)
logger.info(f"加载状态文件: {len(self._store.plugins)} 个插件")
return self._store
except Exception as e:
logger.error(f"加载状态文件失败: {e}")
self._store = PluginStateStore()
return self._store
def _save_store(self) -> bool:
"""保存状态文件"""
if self._store is None:
return False
try:
# 确保目录存在
self.state_file_path.parent.mkdir(parents=True, exist_ok=True)
# 序列化为 JSON
data = self._store.model_dump(mode="json")
with open(self.state_file_path, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=2, default=str)
logger.debug(f"保存状态文件: {self.state_file_path}")
return True
except Exception as e:
logger.error(f"保存状态文件失败: {e}")
return False
def is_enabled(self, plugin_name: str) -> Optional[bool]:
"""检查插件是否启用
Args:
plugin_name: 插件名
Returns:
True/False 表示状态，None 表示没有记录
"""
store = self._load_store()
entry = store.plugins.get(plugin_name)
if entry is None:
return None
return entry.enabled
def set_enabled(
self, plugin_name: str, enabled: bool, updated_by: str = "user"
) -> bool:
"""设置插件启用状态
Args:
plugin_name: 插件名
enabled: 是否启用
updated_by: 更新者
Returns:
是否成功
"""
store = self._load_store()
store.plugins[plugin_name] = PluginStateEntry(
enabled=enabled,
updated_at=datetime.now(),
updated_by=updated_by,
)
return self._save_store()
def remove_plugin(self, plugin_name: str) -> bool:
"""移除插件状态记录
Args:
plugin_name: 插件名
Returns:
是否成功
"""
store = self._load_store()
if plugin_name in store.plugins:
del store.plugins[plugin_name]
return self._save_store()
return True
def get_all_states(self) -> dict:
"""获取所有插件状态
Returns:
{plugin_name: enabled} 字典
"""
store = self._load_store()
return {name: entry.enabled for name, entry in store.plugins.items()}
def initialize_from_metadata(
self, plugin_name: str, default_enabled: bool = False
) -> bool:
"""从 plugin.json 初始化状态（首次扫描时）
如果状态文件中已有记录，不覆盖。
Args:
plugin_name: 插件名
default_enabled: 默认启用状态（来自 plugin.json）
Returns:
是否新增了记录
"""
store = self._load_store()
if plugin_name in store.plugins:
# 已有记录，不覆盖
return False
# 新增记录
store.plugins[plugin_name] = PluginStateEntry(
enabled=default_enabled,
updated_at=datetime.now(),
updated_by="system",
)
self._save_store()
return True
# 全局实例（延迟初始化）
_state_store_manager: Optional[PluginStateStoreManager] = None
def get_state_store_manager(state_file_path: Optional[Path] = None) -> PluginStateStoreManager:
"""获取状态文件管理器实例
Args:
state_file_path: 状态文件路径（首次调用时必须提供）
Returns:
PluginStateStoreManager 实例
"""
global _state_store_manager
if _state_store_manager is None:
if state_file_path is None:
raise ValueError("首次调用必须提供 state_file_path")
_state_store_manager = PluginStateStoreManager(state_file_path)
return _state_store_manager
|