x11_template_vue2/backend/app/services/machine_store.py

"""机型配置管理 - 从JSON文件读写"""
import json
from pathlib import Path
from typing import Dict, Any, Optional


class MachineStore:
    def __init__(self, data_dir: Path):
        self.config_path = data_dir / "machine_configs.json"
        self.reuse_path = data_dir / "reuse_map.json"
        self._configs: Dict[str, Any] = {}
        self._reuse_map: Dict[str, Any] = {}
        self._load()

    def _load(self):
        if self.config_path.exists():
            self._configs = json.loads(self.config_path.read_text(encoding="utf-8"))
        if self.reuse_path.exists():
            self._reuse_map = json.loads(self.reuse_path.read_text(encoding="utf-8"))

    def _save(self):
        self.config_path.write_text(json.dumps(self._configs, ensure_ascii=False, indent=2), encoding="utf-8")

    def _save_reuse(self):
        self.reuse_path.write_text(json.dumps(self._reuse_map, ensure_ascii=False, indent=2), encoding="utf-8")

    @property
    def configs(self) -> Dict[str, Any]:
        return self._configs

    @property
    def reuse_map(self) -> Dict[str, Any]:
        return self._reuse_map

    def get(self, model: str) -> Optional[Dict[str, Any]]:
        return self._configs.get(model)

    def create(self, model: str, config: Dict[str, Any]) -> Dict[str, Any]:
        if model in self._configs:
            raise ValueError(f"机型 '{model}' 已存在")
        self._configs[model] = config
        self._save()
        return config

    def update(self, model: str, config: Dict[str, Any]) -> Dict[str, Any]:
        if model not in self._configs:
            raise ValueError(f"机型 '{model}' 不存在")
        self._configs[model] = config
        self._save()
        return config

    def delete(self, model: str):
        if model not in self._configs:
            raise ValueError(f"机型 '{model}' 不存在")
        del self._configs[model]
        # 同时清理复用映射
        if model in self._reuse_map:
            del self._reuse_map[model]
            self._save_reuse()
        self._save()

    def set_reuse(self, model: str, reuse: Optional[Dict[str, str]]):
        if reuse:
            self._reuse_map[model] = reuse
        elif model in self._reuse_map:
            del self._reuse_map[model]
        self._save_reuse()


def load_machine_store(data_dir: Path) -> MachineStore:
    return MachineStore(data_dir)