Starlight-apk
e67d2d8ef6
- crypto.py: 8个_imp_*方法改为_ModuleCache类缓存导入 - crypto.py: outer/inner加解密合并为_layer_encrypt/decrypt - crypto.py: 提取公共摘要计算方法，拆分长方法 - compiler.py: 删除_obfuscate_code中未使用的死代码 - loader.py: 3次ZIP扫描合并为1次缓存读取 - format.py: 更新为使用_ModuleCache - 合计减少205行代码（1707→1502）
93 lines
3.6 KiB
Python
93 lines
3.6 KiB
Python
import json
|
|
import os
|
|
import threading
|
|
from pathlib import Path
|
|
from typing import Any, Optional
|
|
|
|
from oss.config import get_config
|
|
from oss.logger.logger import Log
|
|
|
|
|
|
class DataStore:
|
|
"""数据存储抽象接口
|
|
|
|
默认实现使用 JSON 文件存储到 ~/.nebula/data/
|
|
后续可由 data-store 插件替换为更完善的实现
|
|
"""
|
|
|
|
def __init__(self):
|
|
config = get_config()
|
|
data_dir_env = os.environ.get("NEBULA_DATA_DIR", "")
|
|
default_dir = Path(data_dir_env) if data_dir_env else Path.home() / ".nebula" / "data"
|
|
self._base_dir = Path(config.get("DATA_DIR", str(default_dir)))
|
|
self._base_dir.mkdir(parents=True, exist_ok=True)
|
|
self._lock = threading.Lock()
|
|
|
|
def _plugin_dir(self, plugin_name: str) -> Path:
|
|
"""获取插件专属数据目录"""
|
|
pd = self._base_dir / plugin_name
|
|
pd.mkdir(parents=True, exist_ok=True)
|
|
return pd
|
|
|
|
def save(self, plugin_name: str, key: str, data: Any) -> bool:
|
|
"""保存数据"""
|
|
with self._lock:
|
|
try:
|
|
file_path = self._plugin_dir(plugin_name) / f"{key}.json"
|
|
file_path.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8")
|
|
return True
|
|
except Exception as e:
|
|
Log.error("Core", f"数据存储保存失败 [{plugin_name}/{key}]: {e}")
|
|
return False
|
|
|
|
def load(self, plugin_name: str, key: str, default: Any = None) -> Any:
|
|
"""加载数据"""
|
|
with self._lock:
|
|
try:
|
|
file_path = self._plugin_dir(plugin_name) / f"{key}.json"
|
|
if file_path.exists():
|
|
return json.loads(file_path.read_text(encoding="utf-8"))
|
|
return default
|
|
except Exception as e:
|
|
Log.error("Core", f"数据存储加载失败 [{plugin_name}/{key}]: {e}")
|
|
return default
|
|
|
|
def delete(self, plugin_name: str, key: str) -> bool:
|
|
"""删除数据"""
|
|
with self._lock:
|
|
try:
|
|
file_path = self._plugin_dir(plugin_name) / f"{key}.json"
|
|
if file_path.exists():
|
|
file_path.unlink()
|
|
return True
|
|
except Exception as e:
|
|
Log.error("Core", f"数据存储删除失败 [{plugin_name}/{key}]: {e}")
|
|
return False
|
|
|
|
def list_keys(self, plugin_name: str) -> list[str]:
|
|
"""列出插件所有数据键"""
|
|
pd = self._plugin_dir(plugin_name)
|
|
if not pd.exists():
|
|
return []
|
|
return [f.stem for f in pd.glob("*.json")]
|
|
|
|
def set_custom_path(self, plugin_name: str, custom_path: str) -> bool:
|
|
"""插件自定义存储路径（不能修改到项目目录内）"""
|
|
path = Path(custom_path).expanduser().resolve()
|
|
project_dir = Path.cwd().resolve()
|
|
if str(path).startswith(str(project_dir)):
|
|
Log.error("Core", f"插件 '{plugin_name}' 试图将数据存储到项目目录: {custom_path}")
|
|
return False
|
|
path.mkdir(parents=True, exist_ok=True)
|
|
with self._lock:
|
|
mapping_file = self._base_dir / "_custom_paths.json"
|
|
mappings = {}
|
|
if mapping_file.exists():
|
|
try:
|
|
mappings = json.loads(mapping_file.read_text())
|
|
except (json.JSONDecodeError, OSError):
|
|
pass # 映射文件不存在或损坏是正常情况
|
|
mappings[plugin_name] = str(path)
|
|
mapping_file.write_text(json.dumps(mappings, indent=2))
|
|
return True
|