"""UI-настройки, сохраняемые в .env.json (раздел "ui"). Хранятся на сервере → значение общее для всех клиентов и переживает перезапуск. """ from __future__ import annotations import json import os from fastapi import APIRouter, HTTPException, Request from .auth import COOKIE_NAME, auth_manager, require_role DEFAULT_UI = {"live_transcode": False, "archive_transcode": False} DEFAULT_OPT = {"single_main_stream": True, "mode_16plus": False} # серверная политика Live view def _path() -> str: return os.environ.get("NVR_SETTINGS", ".env.json") def load_ui() -> dict: try: with open(_path(), "r", encoding="utf-8") as fh: data = json.load(fh) ui = data.get("ui") or {} except (OSError, json.JSONDecodeError): ui = {} return {**DEFAULT_UI, **{k: ui[k] for k in DEFAULT_UI if k in ui}} def save_ui(patch: dict) -> dict: path = _path() try: with open(path, "r", encoding="utf-8") as fh: data = json.load(fh) except (OSError, json.JSONDecodeError): data = {} merged = load_ui() for k in DEFAULT_UI: if k in patch: merged[k] = bool(patch[k]) data["ui"] = merged with open(path, "w", encoding="utf-8") as fh: json.dump(data, fh, ensure_ascii=False, indent=2) return merged def load_optimization() -> dict: try: with open(_path(), "r", encoding="utf-8") as fh: data = json.load(fh) opt = data.get("optimization") or {} except (OSError, json.JSONDecodeError): opt = {} return {**DEFAULT_OPT, **{k: bool(opt[k]) for k in DEFAULT_OPT if k in opt}} def save_optimization(patch: dict) -> dict: path = _path() try: with open(path, "r", encoding="utf-8") as fh: data = json.load(fh) except (OSError, json.JSONDecodeError): data = {} merged = load_optimization() for k in DEFAULT_OPT: if k in patch: merged[k] = bool(patch[k]) data["optimization"] = merged with open(path, "w", encoding="utf-8") as fh: json.dump(data, fh, ensure_ascii=False, indent=2) return merged router = APIRouter(prefix="/api", tags=["settings"]) @router.get("/settings") async def get_settings() -> dict: return load_ui() @router.post("/settings") async def update_settings(request: Request) -> dict: body = await request.json() return save_ui(body) @router.get("/optimization") async def get_optimization() -> dict: """Политика Live view; читают все авторизованные клиенты.""" return load_optimization() @router.post("/optimization") async def update_optimization(request: Request) -> dict: """Менять может только суперадмин и только с подтверждением пароля.""" require_role(request, "superadmin") body = await request.json() if not auth_manager.verify_token_password(request.cookies.get(COOKIE_NAME), body.get("password", "")): raise HTTPException(403, "неверный пароль") return save_optimization(body)