from __future__ import annotations import json import os import shutil import uuid 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" 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())) 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()) is_builtin: bool = False def to_dict(self) -> dict[str, Any]: return asdict(self) @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__}) class DashboardManager: def __init__(self): DASHBOARDS_DIR.mkdir(parents=True, exist_ok=True) self._cache: dict[str, DashboardTheme] = {} self._load_builtins() 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 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()] 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 def save(self, theme: DashboardTheme) -> bool: theme.updated_at = datetime.now().isoformat() theme.is_builtin = False filepath = DASHBOARDS_DIR / f"{theme.id}.json" try: with open(filepath, "w", encoding="utf-8") as f: json.dump(theme.to_dict(), f, indent=2, ensure_ascii=False) 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: 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() logger.info("Dashboard deleted: %s", theme_id) return True def export_theme(self, theme_id: str) -> dict[str, Any] | None: theme = self.get(theme_id) if not theme: return None result = { "type": "dashboard_theme", "version": "1.0", "data": theme, "html": self._read_template_file(theme["id"]), } return result def import_theme(self, data: dict[str, Any]) -> bool: if data.get("type") != "dashboard_theme": 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) dashboard_manager = DashboardManager()