- Global connection badge in app header (persistent dashboard SSE via ConnectionProvider) - LogProvider keeps log data and SSE stream alive across page switches (no reload spinner) - Config drawer: password reveal toggle, array-type row editor for list fields - Plugin management: optimistic updates, race-free loading, reset all group overrides - Logs: load-earlier pagination; backend tail-read optimization (before/end offsets) - web_auth: Web password changes persist (sync_admin seeding only when .env has explicit key) - Dashboard: collect in thread pool + 1s result cache - ErrorBoundary + local Suspense: plugin chunk loading stays inside content area - UI polish: table row hover, mobile header/breadcrumb fixes, wider config drawer - config_standard: array type inference for list fields; array env skip - random_jm_code / steam_info: array schema migration; README web frontend note Co-Authored-By: Claude Code <noreply@anthropic.com>
38 lines
946 B
Python
38 lines
946 B
Python
"""Model 层的数据访问:唯一读写入口(此处用 JSON 文件示例)。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
|
|
# Runtime data belongs under hexi/data, not beside importable plugin code.
|
|
_DATA_DIR = Path(__file__).resolve().parents[2] / "data" / "nonebot_plugin_template"
|
|
_STORE = _DATA_DIR / "store.json"
|
|
|
|
|
|
def _load() -> dict:
|
|
if _STORE.exists():
|
|
try:
|
|
return json.loads(_STORE.read_text("utf-8"))
|
|
except json.JSONDecodeError:
|
|
return {}
|
|
return {}
|
|
|
|
|
|
def _save(data: dict) -> None:
|
|
_DATA_DIR.mkdir(parents=True, exist_ok=True)
|
|
tmp = _STORE.with_suffix(".tmp")
|
|
tmp.write_text(json.dumps(data, ensure_ascii=False, indent=2), "utf-8")
|
|
os.replace(tmp, _STORE) # 原子写
|
|
|
|
|
|
def get_record(key: str):
|
|
return _load().get(key)
|
|
|
|
|
|
def save_record(key: str, value) -> None:
|
|
data = _load()
|
|
data[key] = value
|
|
_save(data)
|