250 lines
9.3 KiB
Python
250 lines
9.3 KiB
Python
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
|
|
from typing import Any
|
|
|
|
from utils.logger import get_logger
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
DATA_DIR = Path(__file__).resolve().parent.parent / "data"
|
|
USER_DASHBOARDS_DIR = DATA_DIR / "dashboards"
|
|
BUILTIN_DASHBOARDS_DIR = Path(__file__).resolve().parent.parent / "dashboards"
|
|
|
|
|
|
@dataclass
|
|
class DashboardTheme:
|
|
id: str = ""
|
|
name: str = ""
|
|
category: str = "basic"
|
|
description: str = ""
|
|
author: str = ""
|
|
version: str = "1.0.0"
|
|
icon: str = "📊"
|
|
aspect_ratio: str = "auto"
|
|
render_mode: str = "contain"
|
|
supported_games: str = "all"
|
|
created_at: str = ""
|
|
updated_at: str = ""
|
|
is_builtin: bool = False
|
|
dir_path: str = ""
|
|
preview_ext: str = ""
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
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,
|
|
"supported_games": self.supported_games,
|
|
"created_at": self.created_at,
|
|
"updated_at": self.updated_at,
|
|
"is_builtin": self.is_builtin,
|
|
"preview_ext": self.preview_ext,
|
|
}
|
|
|
|
@classmethod
|
|
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", {})
|
|
dir_path = manifest_path.parent
|
|
preview_ext = ""
|
|
for ext in ["jpg", "jpeg", "png", "webp", "gif"]:
|
|
if (dir_path / f"preview.{ext}").exists():
|
|
preview_ext = ext
|
|
break
|
|
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"),
|
|
supported_games=data.get("supported_games", "all"),
|
|
created_at=data.get("created_at", ""),
|
|
updated_at=data.get("updated_at", ""),
|
|
is_builtin=is_builtin,
|
|
dir_path=str(manifest_path.parent),
|
|
preview_ext=preview_ext,
|
|
)
|
|
except Exception as e:
|
|
logger.error("Failed to load dashboard manifest %s: %s", manifest_path, e)
|
|
return None
|
|
|
|
|
|
class DashboardManager:
|
|
def __init__(self):
|
|
USER_DASHBOARDS_DIR.mkdir(parents=True, exist_ok=True)
|
|
|
|
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]]:
|
|
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:
|
|
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 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,
|
|
"supported_games": theme.supported_games,
|
|
"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(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:
|
|
logger.error("Failed to save dashboard %s: %s", theme.id, e)
|
|
return False
|
|
|
|
def delete(self, theme_id: str) -> bool:
|
|
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 False
|
|
|
|
def export_theme_zip(self, theme_id: str) -> bytes | None:
|
|
theme_dir = self.get_dir(theme_id)
|
|
if not theme_dir:
|
|
return None
|
|
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_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
|
|
|
|
def get_categories(self) -> list[str]:
|
|
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()
|