from __future__ import annotations import json from pathlib import Path from typing import Dict, List, Optional from fastapi import HTTPException from ..schemas import TemplateCreate, TemplateRecord, TemplateUpdate class TemplateStore: """Simple JSON-file based store for template definitions.""" def __init__(self, data_path: Path): self.data_path = data_path self.data_path.parent.mkdir(parents=True, exist_ok=True) if not self.data_path.exists(): self.data_path.write_text(json.dumps([], ensure_ascii=False, indent=2)) self._cache: Dict[str, TemplateRecord] = {} self._load() def _load(self) -> None: raw = json.loads(self.data_path.read_text(encoding='utf-8')) self._cache = {item["id"]: TemplateRecord(**item) for item in raw} def _persist(self) -> None: # Use JSON mode so datetime fields are converted to ISO strings. serializable = [record.model_dump(mode="json") for record in self._cache.values()] self.data_path.write_text( json.dumps(serializable, ensure_ascii=False, indent=2), encoding='utf-8' ) def list_templates(self, machine_model: str = None, device_type: str = None, template_type: str = None) -> List[TemplateRecord]: templates = list(self._cache.values()) if machine_model: templates = [t for t in templates if t.machine_model == machine_model] if device_type: templates = [t for t in templates if t.device_type == device_type] if template_type: templates = [t for t in templates if t.template_type == template_type] return templates def get_template(self, template_id: str) -> TemplateRecord: try: record = self._cache[template_id] # 如果是复用模板，返回源模板的内容（但保留当前模板的元数据） if record.reusable_from: source = self._cache.get(record.reusable_from) if source: # 返回源模板，但保留当前模板的元数据 merged = source.model_copy() merged.id = record.id merged.name = record.name merged.category = record.category merged.device_type = record.device_type merged.machine_model = record.machine_model merged.template_type = record.template_type merged.type_id = record.type_id merged.zone_ids = record.zone_ids merged.card_type = record.card_type merged.reusable_from = record.reusable_from merged.reuse_targets = record.reuse_targets merged.description = record.description merged.created_at = record.created_at merged.updated_at = record.updated_at return merged return record except KeyError as exc: raise HTTPException(status_code=404, detail="模板不存在") from exc def create_template(self, payload: TemplateCreate) -> TemplateRecord: record = TemplateRecord(**payload.model_dump()) self._cache[record.id] = record self._persist() return record def update_template(self, template_id: str, payload: TemplateUpdate) -> TemplateRecord: record = self.get_template(template_id) update_data = payload.model_dump(exclude_none=True) updated = record.model_copy(update=update_data) from datetime import datetime updated.updated_at = datetime.utcnow() self._cache[template_id] = updated self._persist() return updated def delete_template(self, template_id: str) -> None: if template_id not in self._cache: raise HTTPException(status_code=404, detail="模板不存在") record = self._cache[template_id] # 如果该模板被其他模板复用，禁止删除 if record.reuse_targets: raise HTTPException(status_code=400, detail=f"该模板被 {len(record.reuse_targets)} 个模板复用，无法删除") # 如果这是复用模板，需要从源模板的 reuse_targets 中移除 if record.reusable_from and record.reusable_from in self._cache: source = self._cache[record.reusable_from] if template_id in source.reuse_targets: source.reuse_targets.remove(template_id) del self._cache[template_id] self._persist() def generate_redis_keys(self, record: TemplateRecord) -> dict: """根据模板信息自动生成 Redis 一级 key 列表""" if not record.type_id or not record.zone_ids: return {"second": record.redis_key_second or "", "first": []} # 二级 key card = record.card_type or "single_card" second_key = f"broadcast-template-jinja:{card}:{record.device_type}:{record.template_type}" # 模板类型到 card_type 和 recommend_type 的映射 template_type_map = { "common": ("common", None), "time_popular": ("time_made_popularity", "1"), "time_custom": ("time_made_popularity", "2"), "time_season": ("time_season_storage", None), # 同时有 time_season_in "time_care": ("time_care", None), "time_environmental": ("time_environmental", None), "time_weather": ("time_weather", None), "time_health": ("time_health", None), } card_type_key, recommend_type = template_type_map.get(record.template_type, (record.template_type, None)) first_keys = [] for zone_id in record.zone_ids: if record.template_type == "common": key = f"broadcast-template-jinja:{record.type_id}:{zone_id}:common" first_keys.append(key) else: # 非 common 模板，需要为每个 zone_id 和 recommend_type 生成 key if recommend_type: key = f"broadcast-template-jinja:{record.type_id}:{zone_id}:{card_type_key}:{recommend_type}" first_keys.append(key) else: # 对于 season，同时生成 storage 和 in if record.template_type == "time_season": first_keys.append(f"broadcast-template-jinja:{record.type_id}:{zone_id}:time_season_storage:1") first_keys.append(f"broadcast-template-jinja:{record.type_id}:{zone_id}:time_season_in:1") else: key = f"broadcast-template-jinja:{record.type_id}:{zone_id}:{card_type_key}:1" first_keys.append(key) return {"second": second_key, "first": first_keys} def list_categories(self, machine_model: str = None) -> List[str]: templates = self._cache.values() if machine_model: templates = [t for t in templates if t.machine_model == machine_model] categories = {record.category for record in templates} return sorted(categories) def list_machine_models(self) -> List[str]: models = {record.machine_model for record in self._cache.values()} return sorted(models) def list_device_types(self, machine_model: str = None) -> List[str]: templates = self._cache.values() if machine_model: templates = [t for t in templates if t.machine_model == machine_model] device_types = {record.device_type for record in templates} return sorted(device_types) def list_template_types(self, machine_model: str = None, device_type: str = None) -> List[str]: templates = self._cache.values() if machine_model: templates = [t for t in templates if t.machine_model == machine_model] if device_type: templates = [t for t in templates if t.device_type == device_type] template_types = {record.template_type for record in templates} return sorted(template_types) def load_store(base_dir: Path) -> TemplateStore: data_path = base_dir / "data" / "templates.json" return TemplateStore(data_path)