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 = "" 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, } @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", {}) 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), ) 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()