refactor: standalone dashboard folders with index.html, zip-based import/export (.tsd/.tss/.tsp)
- Each dashboard is now a self-contained folder: manifest.json + index.html - Auto-scan dashboards/ and data/dashboards/ on startup - Export: .tsd (dashboards), .tss (scenes), .tsp (game plugins) - all zip format - Dashboard index.html is complete standalone page (WS + data binding) - Aspect ratio constraints handled in each dashboard's own JS - Removed template-based dashboard rendering in favor of static serve - Import via file upload endpoints, export via direct file download
This commit is contained in:
+175
-109
@@ -1,9 +1,10 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import uuid
|
||||
import zipfile
|
||||
from dataclasses import dataclass, field, asdict
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
@@ -14,87 +15,142 @@ from utils.logger import get_logger
|
||||
logger = get_logger(__name__)
|
||||
|
||||
DATA_DIR = Path(__file__).resolve().parent.parent / "data"
|
||||
DASHBOARDS_DIR = DATA_DIR / "dashboards"
|
||||
USER_DASHBOARDS_DIR = DATA_DIR / "dashboards"
|
||||
BUILTIN_DASHBOARDS_DIR = Path(__file__).resolve().parent.parent / "dashboards"
|
||||
|
||||
|
||||
@dataclass
|
||||
class DashboardTheme:
|
||||
id: str = field(default_factory=lambda: str(uuid.uuid4()))
|
||||
id: str = ""
|
||||
name: str = ""
|
||||
category: str = "basic"
|
||||
description: str = ""
|
||||
author: str = ""
|
||||
version: str = "1.0.0"
|
||||
preview: str = ""
|
||||
config: dict[str, Any] = field(default_factory=dict)
|
||||
created_at: str = field(default_factory=lambda: datetime.now().isoformat())
|
||||
updated_at: str = field(default_factory=lambda: datetime.now().isoformat())
|
||||
icon: str = "📊"
|
||||
aspect_ratio: str = "auto"
|
||||
render_mode: str = "contain"
|
||||
created_at: str = ""
|
||||
updated_at: str = ""
|
||||
is_builtin: bool = False
|
||||
dir_path: str = ""
|
||||
|
||||
def to_dict(self) -> dict[str, Any]:
|
||||
return asdict(self)
|
||||
return {
|
||||
"id": self.id,
|
||||
"name": self.name,
|
||||
"category": self.category,
|
||||
"description": self.description,
|
||||
"author": self.author,
|
||||
"version": self.version,
|
||||
"icon": self.icon,
|
||||
"aspect_ratio": self.aspect_ratio,
|
||||
"render_mode": self.render_mode,
|
||||
"created_at": self.created_at,
|
||||
"updated_at": self.updated_at,
|
||||
"is_builtin": self.is_builtin,
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, d: dict[str, Any]) -> DashboardTheme:
|
||||
return cls(**{k: v for k, v in d.items() if k in cls.__dataclass_fields__})
|
||||
def from_manifest(cls, manifest_path: Path, is_builtin: bool = False) -> DashboardTheme | None:
|
||||
try:
|
||||
with open(manifest_path, "r", encoding="utf-8") as f:
|
||||
data = json.load(f)
|
||||
config = data.get("config", {})
|
||||
return cls(
|
||||
id=data.get("id", manifest_path.parent.name),
|
||||
name=data.get("name", manifest_path.parent.name),
|
||||
category=data.get("category", "basic"),
|
||||
description=data.get("description", ""),
|
||||
author=data.get("author", ""),
|
||||
version=data.get("version", "1.0.0"),
|
||||
icon=config.get("icon", "📊"),
|
||||
aspect_ratio=config.get("aspect_ratio", "auto"),
|
||||
render_mode=config.get("render_mode", "contain"),
|
||||
created_at=data.get("created_at", ""),
|
||||
updated_at=data.get("updated_at", ""),
|
||||
is_builtin=is_builtin,
|
||||
dir_path=str(manifest_path.parent),
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error("Failed to load dashboard manifest %s: %s", manifest_path, e)
|
||||
return None
|
||||
|
||||
|
||||
class DashboardManager:
|
||||
def __init__(self):
|
||||
DASHBOARDS_DIR.mkdir(parents=True, exist_ok=True)
|
||||
self._cache: dict[str, DashboardTheme] = {}
|
||||
self._load_builtins()
|
||||
USER_DASHBOARDS_DIR.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
def _load_builtins(self):
|
||||
if not BUILTIN_DASHBOARDS_DIR.exists():
|
||||
return
|
||||
for item in BUILTIN_DASHBOARDS_DIR.iterdir():
|
||||
if item.is_dir():
|
||||
cfg_file = item / "config.json"
|
||||
if cfg_file.exists():
|
||||
try:
|
||||
with open(cfg_file, "r", encoding="utf-8") as f:
|
||||
data = json.load(f)
|
||||
theme = DashboardTheme.from_dict(data)
|
||||
theme.is_builtin = True
|
||||
self._cache[theme.id] = theme
|
||||
except Exception as e:
|
||||
logger.error("Failed to load builtin dashboard %s: %s", item.name, e)
|
||||
|
||||
def _load_user_dashboards(self):
|
||||
for f in DASHBOARDS_DIR.glob("*.json"):
|
||||
try:
|
||||
with open(f, "r", encoding="utf-8") as fp:
|
||||
data = json.load(fp)
|
||||
theme = DashboardTheme.from_dict(data)
|
||||
self._cache[theme.id] = theme
|
||||
except Exception as e:
|
||||
logger.error("Failed to load dashboard %s: %s", f.name, e)
|
||||
def _scan_dirs(self, base_dir: Path, is_builtin: bool) -> list[DashboardTheme]:
|
||||
themes = []
|
||||
if not base_dir.exists():
|
||||
return themes
|
||||
for item in sorted(base_dir.iterdir()):
|
||||
if not item.is_dir():
|
||||
continue
|
||||
manifest = item / "manifest.json"
|
||||
if manifest.exists():
|
||||
theme = DashboardTheme.from_manifest(manifest, is_builtin)
|
||||
if theme:
|
||||
themes.append(theme)
|
||||
return themes
|
||||
|
||||
def list_all(self, category: str | None = None) -> list[dict[str, Any]]:
|
||||
self._cache.clear()
|
||||
self._load_builtins()
|
||||
self._load_user_dashboards()
|
||||
result = [t.to_dict() for t in self._cache.values()]
|
||||
themes = self._scan_dirs(BUILTIN_DASHBOARDS_DIR, True)
|
||||
themes += self._scan_dirs(USER_DASHBOARDS_DIR, False)
|
||||
result = [t.to_dict() for t in themes]
|
||||
if category and category != "all":
|
||||
result = [r for r in result if r.get("category") == category]
|
||||
return result
|
||||
|
||||
def get(self, theme_id: str) -> dict[str, Any] | None:
|
||||
self._cache.clear()
|
||||
self._load_builtins()
|
||||
self._load_user_dashboards()
|
||||
theme = self._cache.get(theme_id)
|
||||
return theme.to_dict() if theme else None
|
||||
for base, is_builtin in [(BUILTIN_DASHBOARDS_DIR, True), (USER_DASHBOARDS_DIR, False)]:
|
||||
theme_dir = base / theme_id
|
||||
manifest = theme_dir / "manifest.json"
|
||||
if manifest.exists():
|
||||
theme = DashboardTheme.from_manifest(manifest, is_builtin)
|
||||
if theme:
|
||||
return theme.to_dict()
|
||||
return None
|
||||
|
||||
def save(self, theme: DashboardTheme) -> bool:
|
||||
theme.updated_at = datetime.now().isoformat()
|
||||
theme.is_builtin = False
|
||||
filepath = DASHBOARDS_DIR / f"{theme.id}.json"
|
||||
def get_dir(self, theme_id: str) -> Path | None:
|
||||
for base in [BUILTIN_DASHBOARDS_DIR, USER_DASHBOARDS_DIR]:
|
||||
theme_dir = base / theme_id
|
||||
if (theme_dir / "manifest.json").exists():
|
||||
return theme_dir
|
||||
return None
|
||||
|
||||
def get_index_html(self, theme_id: str) -> str | None:
|
||||
for base in [BUILTIN_DASHBOARDS_DIR, USER_DASHBOARDS_DIR]:
|
||||
idx = base / theme_id / "index.html"
|
||||
if idx.exists():
|
||||
return idx.read_text(encoding="utf-8")
|
||||
return None
|
||||
|
||||
def save(self, theme: DashboardTheme, index_html: str = "") -> bool:
|
||||
theme_dir = USER_DASHBOARDS_DIR / theme.id
|
||||
theme_dir.mkdir(parents=True, exist_ok=True)
|
||||
manifest = {
|
||||
"id": theme.id,
|
||||
"name": theme.name,
|
||||
"category": theme.category,
|
||||
"description": theme.description,
|
||||
"author": theme.author,
|
||||
"version": theme.version,
|
||||
"config": {
|
||||
"icon": theme.icon,
|
||||
"aspect_ratio": theme.aspect_ratio,
|
||||
"render_mode": theme.render_mode,
|
||||
},
|
||||
"created_at": theme.created_at or datetime.now().isoformat(),
|
||||
"updated_at": datetime.now().isoformat(),
|
||||
}
|
||||
try:
|
||||
with open(filepath, "w", encoding="utf-8") as f:
|
||||
json.dump(theme.to_dict(), f, indent=2, ensure_ascii=False)
|
||||
with open(theme_dir / "manifest.json", "w", encoding="utf-8") as f:
|
||||
json.dump(manifest, f, indent=2, ensure_ascii=False)
|
||||
if index_html:
|
||||
with open(theme_dir / "index.html", "w", encoding="utf-8") as f:
|
||||
f.write(index_html)
|
||||
logger.info("Dashboard saved: %s", theme.id)
|
||||
return True
|
||||
except Exception as e:
|
||||
@@ -102,69 +158,79 @@ class DashboardManager:
|
||||
return False
|
||||
|
||||
def delete(self, theme_id: str) -> bool:
|
||||
self._cache.clear()
|
||||
self._load_builtins()
|
||||
self._load_user_dashboards()
|
||||
theme = self._cache.get(theme_id)
|
||||
if theme and theme.is_builtin:
|
||||
logger.warning("Cannot delete builtin dashboard: %s", theme_id)
|
||||
return False
|
||||
filepath = DASHBOARDS_DIR / f"{theme_id}.json"
|
||||
if filepath.exists():
|
||||
filepath.unlink()
|
||||
theme_dir = USER_DASHBOARDS_DIR / theme_id
|
||||
if theme_dir.exists():
|
||||
shutil.rmtree(theme_dir)
|
||||
logger.info("Dashboard deleted: %s", theme_id)
|
||||
return True
|
||||
return True
|
||||
return False
|
||||
|
||||
def export_theme(self, theme_id: str) -> dict[str, Any] | None:
|
||||
theme = self.get(theme_id)
|
||||
if not theme:
|
||||
def export_theme_zip(self, theme_id: str) -> bytes | None:
|
||||
theme_dir = self.get_dir(theme_id)
|
||||
if not theme_dir:
|
||||
return None
|
||||
result = {
|
||||
"type": "dashboard_theme",
|
||||
"version": "1.0",
|
||||
"data": theme,
|
||||
"html": self._read_template_file(theme["id"]),
|
||||
}
|
||||
return result
|
||||
buf = io.BytesIO()
|
||||
with zipfile.ZipFile(buf, 'w', zipfile.ZIP_DEFLATED) as zf:
|
||||
for f in sorted(theme_dir.rglob('*')):
|
||||
if f.is_file():
|
||||
arcname = str(f.relative_to(theme_dir))
|
||||
zf.write(f, arcname)
|
||||
logger.info("Dashboard exported as zip: %s (%d bytes)", theme_id, buf.tell())
|
||||
return buf.getvalue()
|
||||
|
||||
def import_theme(self, data: dict[str, Any]) -> bool:
|
||||
if data.get("type") != "dashboard_theme":
|
||||
def import_theme_zip(self, zip_data: bytes) -> bool:
|
||||
try:
|
||||
with zipfile.ZipFile(io.BytesIO(zip_data), 'r') as zf:
|
||||
names = zf.namelist()
|
||||
manifest_name = None
|
||||
for n in names:
|
||||
if n.endswith('manifest.json'):
|
||||
manifest_name = n
|
||||
break
|
||||
if not manifest_name:
|
||||
logger.error("No manifest.json found in zip")
|
||||
return False
|
||||
|
||||
manifest_data = json.loads(zf.read(manifest_name).decode('utf-8'))
|
||||
theme_id = manifest_data.get('id', str(uuid.uuid4()))
|
||||
|
||||
existing_dir = self.get_dir(theme_id)
|
||||
if existing_dir and existing_dir.parent == BUILTIN_DASHBOARDS_DIR:
|
||||
theme_id = str(uuid.uuid4())
|
||||
manifest_data['id'] = theme_id
|
||||
|
||||
dest = USER_DASHBOARDS_DIR / theme_id
|
||||
if dest.exists():
|
||||
shutil.rmtree(dest)
|
||||
dest.mkdir(parents=True)
|
||||
|
||||
prefix = manifest_name.rsplit('manifest.json', 1)[0]
|
||||
for name in names:
|
||||
if name == manifest_name:
|
||||
continue
|
||||
rel = name
|
||||
if prefix and name.startswith(prefix):
|
||||
rel = name[len(prefix):]
|
||||
if not rel or rel.endswith('/'):
|
||||
continue
|
||||
out_path = dest / rel
|
||||
out_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
out_path.write_bytes(zf.read(name))
|
||||
|
||||
with open(dest / 'manifest.json', 'w', encoding='utf-8') as f:
|
||||
json.dump(manifest_data, f, indent=2, ensure_ascii=False)
|
||||
|
||||
logger.info("Dashboard imported from zip: %s", theme_id)
|
||||
return True
|
||||
except Exception as e:
|
||||
logger.error("Failed to import dashboard zip: %s", e)
|
||||
return False
|
||||
theme_data = data.get("data", {})
|
||||
theme = DashboardTheme.from_dict(theme_data)
|
||||
html_content = data.get("html", "")
|
||||
if theme.id in self._cache:
|
||||
theme.id = str(uuid.uuid4())
|
||||
success = self.save(theme)
|
||||
if success and html_content:
|
||||
self._save_template_file(theme.id, html_content)
|
||||
return success
|
||||
|
||||
def _read_template_file(self, theme_id: str) -> str:
|
||||
for base in [DASHBOARDS_DIR, BUILTIN_DASHBOARDS_DIR]:
|
||||
tmpl = base / theme_id / "template.html"
|
||||
if tmpl.exists():
|
||||
return tmpl.read_text(encoding="utf-8")
|
||||
return ""
|
||||
|
||||
def _save_template_file(self, theme_id: str, content: str):
|
||||
theme_dir = DASHBOARDS_DIR / theme_id
|
||||
theme_dir.mkdir(parents=True, exist_ok=True)
|
||||
tmpl = theme_dir / "template.html"
|
||||
tmpl.write_text(content, encoding="utf-8")
|
||||
|
||||
def get_template(self, theme_id: str) -> str:
|
||||
return self._read_template_file(theme_id)
|
||||
|
||||
def get_categories(self) -> list[str]:
|
||||
self._cache.clear()
|
||||
self._load_builtins()
|
||||
self._load_user_dashboards()
|
||||
cats = set()
|
||||
for t in self._cache.values():
|
||||
if t.category:
|
||||
cats.add(t.category)
|
||||
return sorted(cats)
|
||||
themes = self._scan_dirs(BUILTIN_DASHBOARDS_DIR, True)
|
||||
themes += self._scan_dirs(USER_DASHBOARDS_DIR, False)
|
||||
cats = sorted(set(t.category for t in themes if t.category))
|
||||
return cats
|
||||
|
||||
|
||||
dashboard_manager = DashboardManager()
|
||||
|
||||
+22
-15
@@ -1,7 +1,10 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import io
|
||||
import json
|
||||
import shutil
|
||||
import uuid
|
||||
import zipfile
|
||||
from dataclasses import dataclass, field, asdict
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
@@ -124,24 +127,28 @@ class SceneManager:
|
||||
return True
|
||||
return False
|
||||
|
||||
def export_scene(self, scene_id: str) -> dict[str, Any] | None:
|
||||
scene = self.get(scene_id)
|
||||
if not scene:
|
||||
def export_scene_zip(self, scene_id: str) -> bytes | None:
|
||||
filepath = SCENES_DIR / f"{scene_id}.json"
|
||||
if not filepath.exists():
|
||||
return None
|
||||
return {
|
||||
"type": "scene",
|
||||
"version": "1.0",
|
||||
"data": scene,
|
||||
}
|
||||
buf = io.BytesIO()
|
||||
with zipfile.ZipFile(buf, 'w', zipfile.ZIP_DEFLATED) as zf:
|
||||
zf.write(filepath, "scene.json")
|
||||
logger.info("Scene exported as zip: %s", scene_id)
|
||||
return buf.getvalue()
|
||||
|
||||
def import_scene(self, data: dict[str, Any]) -> bool:
|
||||
if data.get("type") != "scene":
|
||||
def import_scene_zip(self, zip_data: bytes) -> bool:
|
||||
try:
|
||||
with zipfile.ZipFile(io.BytesIO(zip_data), 'r') as zf:
|
||||
if "scene.json" not in zf.namelist():
|
||||
logger.error("No scene.json found in zip")
|
||||
return False
|
||||
data = json.loads(zf.read("scene.json").decode('utf-8'))
|
||||
scene = Scene.from_dict(data)
|
||||
return self.save(scene)
|
||||
except Exception as e:
|
||||
logger.error("Failed to import scene zip: %s", e)
|
||||
return False
|
||||
scene_data = data.get("data", {})
|
||||
if "id" not in scene_data:
|
||||
scene_data["id"] = str(uuid.uuid4())
|
||||
scene = Scene.from_dict(scene_data)
|
||||
return self.save(scene)
|
||||
|
||||
|
||||
scene_manager = SceneManager()
|
||||
|
||||
Reference in New Issue
Block a user