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" SCENES_DIR = DATA_DIR / "scenes" @dataclass class DashboardPlacement: dashboard_id: str = "" x: float = 0.0 y: float = 0.0 width: int = 400 height: int = 300 z_index: int = 0 scale_x: float = 1.0 scale_y: float = 1.0 aspect_ratio: str = "auto" render_mode: str = "contain" custom_config: dict[str, Any] = field(default_factory=dict) @dataclass class SceneCanvas: width: int = 1920 height: int = 1080 label: str = "16:9" placements: list[DashboardPlacement] = field(default_factory=list) @dataclass class Scene: id: str = field(default_factory=lambda: str(uuid.uuid4())) name: str = "New Scene" game_id: str = "" description: str = "" canvases: list[SceneCanvas] = field(default_factory=list) created_at: str = field(default_factory=lambda: datetime.now().isoformat()) updated_at: str = field(default_factory=lambda: datetime.now().isoformat()) def to_dict(self) -> dict[str, Any]: d = asdict(self) d["canvases"] = [asdict(c) for c in self.canvases] return d @classmethod def from_dict(cls, d: dict[str, Any]) -> Scene: canvases = [] for cd in d.get("canvases", []): placements = [DashboardPlacement(**p) for p in cd.get("placements", [])] canvases.append(SceneCanvas( width=cd.get("width", 1920), height=cd.get("height", 1080), label=cd.get("label", ""), placements=placements, )) return cls( id=d.get("id", str(uuid.uuid4())), name=d.get("name", "New Scene"), game_id=d.get("game_id", ""), description=d.get("description", ""), canvases=canvases, created_at=d.get("created_at", datetime.now().isoformat()), updated_at=d.get("updated_at", datetime.now().isoformat()), ) class SceneManager: def __init__(self): SCENES_DIR.mkdir(parents=True, exist_ok=True) def list_all(self, game_id: str | None = None) -> list[dict[str, Any]]: scenes = [] for item in SCENES_DIR.iterdir(): if not item.is_dir(): continue f = item / "manifest.json" if not f.exists(): continue try: with open(f, "r", encoding="utf-8") as fp: data = json.load(fp) scene = Scene.from_dict(data) if game_id is None or scene.game_id == game_id: d = scene.to_dict() for ext in ["jpg","jpeg","png","webp","gif"]: if (item / f"preview.{ext}").exists(): d["preview_ext"] = ext break scenes.append(d) except Exception as e: logger.error("Failed to load scene %s: %s", item.name, e) return scenes def get(self, scene_id: str) -> dict[str, Any] | None: item = SCENES_DIR / scene_id f = item / "manifest.json" if not f.exists(): return None try: with open(f, "r", encoding="utf-8") as fp: data = json.load(fp) d = Scene.from_dict(data).to_dict() for ext in ["jpg","jpeg","png","webp","gif"]: if (item / f"preview.{ext}").exists(): d["preview_ext"] = ext break return d except Exception as e: logger.error("Failed to read scene %s: %s", scene_id, e) return None def save(self, scene: Scene) -> bool: scene.updated_at = datetime.now().isoformat() item = SCENES_DIR / scene.id item.mkdir(parents=True, exist_ok=True) try: with open(item / "manifest.json", "w", encoding="utf-8") as f: json.dump(scene.to_dict(), f, indent=2, ensure_ascii=False) logger.info("Scene saved: %s", scene.id) return True except Exception as e: logger.error("Failed to save scene %s: %s", scene.id, e) return False def delete(self, scene_id: str) -> bool: import shutil item = SCENES_DIR / scene_id if item.exists(): shutil.rmtree(item) logger.info("Scene deleted: %s", scene_id) return True return False def export_scene_zip(self, scene_id: str) -> bytes | None: item = SCENES_DIR / scene_id if not item.exists(): return None buf = io.BytesIO() with zipfile.ZipFile(buf, 'w', zipfile.ZIP_DEFLATED) as zf: for f in sorted(item.rglob("*")): if f.is_file(): zf.write(f, f.relative_to(item)) logger.info("Scene exported as zip: %s", scene_id) return buf.getvalue() def import_scene_zip(self, zip_data: bytes) -> bool: try: with zipfile.ZipFile(io.BytesIO(zip_data), 'r') as zf: if "manifest.json" not in zf.namelist(): logger.error("No manifest.json found in zip") return False data = json.loads(zf.read("manifest.json").decode('utf-8')) scene = Scene.from_dict(data) item = SCENES_DIR / scene.id if item.exists(): import shutil; shutil.rmtree(item) item.mkdir(parents=True) for name in zf.namelist(): if name.endswith('/'): continue dest = item / name dest.parent.mkdir(parents=True, exist_ok=True) dest.write_bytes(zf.read(name)) logger.info("Scene imported from zip: %s", scene.id) return True except Exception as e: logger.error("Failed to import scene zip: %s", e) return False scene_manager = SceneManager()