API configurazione GUI

app.py

@@ -154,6 +154,7 @@ async def local_then_core(request: Request, call_next):
     internal_core_proxy_paths = {
         "/reslevis/updateAlarm",
         "/reslevis/updateCoreSettings",
+        "/reslevis/updateUserPreferences",
     }
     # only proxy CRUD for Reslevis (change prefix or methods if needed)
     if (

logica_reslevis/config.py

@@ -27,4 +27,5 @@ TRACK_JSON_PATH = DATA_DIR / "tracks.json"
 TRACKER_ZONE_JSON_PATH = DATA_DIR / "tracker_zone.json"
 SETTING_JSON_PATH = DATA_DIR / "settings.json"
 GUI_CONFIG_JSON_PATH = DATA_DIR / "gui_config.json"
+USER_PREFERENCES_JSON_PATH = DATA_DIR / "user_preferences.json"
 

logica_reslevis/user_preferences.py

@@ -0,0 +1,88 @@
+import json
+from datetime import datetime, timezone
+from typing import Any, Dict, List, Optional
+from uuid import NAMESPACE_URL, uuid5
+
+from fastapi.encoders import jsonable_encoder
+
+from schemas.reslevis import UserPreferencesUpdateItem
+from .config import USER_PREFERENCES_JSON_PATH
+from .gateway import _LockedFile, _atomic_write, _norm_str
+
+
+def _model_to_dict(model: Any) -> Dict[str, Any]:
+    if hasattr(model, "model_dump"):
+        return model.model_dump(exclude_unset=True, exclude_none=True)
+    return model.dict(exclude_unset=True, exclude_none=True)
+
+
+class UserPreferencesJsonRepository:
+    def __init__(self, json_path: str = USER_PREFERENCES_JSON_PATH):
+        self.path = json_path
+
+    def _read_all(self) -> List[Dict[str, Any]]:
+        with _LockedFile(self.path, "r") as fp:
+            try:
+                fp.seek(0)
+                data = fp.read().strip()
+                return json.loads(data) if data else []
+            except json.JSONDecodeError:
+                return []
+
+    def _write_all(self, rows: List[Dict[str, Any]]) -> None:
+        payload = json.dumps(rows, ensure_ascii=False, indent=2)
+        _atomic_write(self.path, payload)
+
+    def _index_by_uid(self, rows: List[Dict[str, Any]], uid: str) -> Optional[int]:
+        target = _norm_str(uid)
+        for i, row in enumerate(rows):
+            if _norm_str(row.get("uid")) == target:
+                return i
+        return None
+
+    def _default_for_uid(self, uid: str) -> Dict[str, Any]:
+        return {
+            "id": str(uuid5(NAMESPACE_URL, f"reslevis:user-preferences:{uid}")),
+            "uid": uid,
+            "language": "it",
+            "tables": {},
+            "updated_at": None,
+        }
+
+    def get(self, uid: str) -> Dict[str, Any]:
+        rows = self._read_all()
+        idx = self._index_by_uid(rows, uid)
+        if idx is None:
+            return self._default_for_uid(uid)
+
+        obj = self._default_for_uid(uid)
+        obj.update(rows[idx])
+        obj["uid"] = uid
+        obj["tables"] = obj.get("tables") or {}
+        obj["language"] = obj.get("language") or "it"
+        return obj
+
+    def update(self, uid: str, item: UserPreferencesUpdateItem) -> Dict[str, Any]:
+        rows = self._read_all()
+        idx = self._index_by_uid(rows, uid)
+        current = self._default_for_uid(uid) if idx is None else self.get(uid)
+        payload = jsonable_encoder(_model_to_dict(item))
+
+        if "language" in payload and payload["language"] is not None:
+            current["language"] = payload["language"]
+
+        if "tables" in payload and payload["tables"] is not None:
+            tables = current.get("tables") or {}
+            for table_name, table_preferences in payload["tables"].items():
+                tables[table_name] = table_preferences or {"hiddenColumns": {}}
+            current["tables"] = tables
+
+        current["uid"] = uid
+        current["updated_at"] = datetime.now(timezone.utc).isoformat()
+
+        if idx is None:
+            rows.append(current)
+        else:
+            rows[idx] = current
+        self._write_all(rows)
+        return current

routes/reslevis.py

@@ -24,6 +24,8 @@ from schemas.reslevis import (
     TrackerZoneItem,
     SettingItem,
     GuiConfigItem,
+    UserPreferencesItem,
+    UserPreferencesUpdateItem,
     CoreSettingsItem,
     CoreSettingsUpdateItem,
 )
@@ -37,6 +39,7 @@ from logica_reslevis.tracker import TrackerJsonRepository
 from logica_reslevis.operator import OperatorJsonRepository
 from logica_reslevis.setting import SettingJsonRepository
 from logica_reslevis.gui_config import GuiConfigJsonRepository
+from logica_reslevis.user_preferences import UserPreferencesJsonRepository
 from logica_reslevis.subject import SubjectJsonRepository
 from logica_reslevis.alarm import AlarmJsonRepository
 from logica_reslevis.track import TrackJsonRepository
@@ -94,6 +97,7 @@ track_repo = TrackJsonRepository()
 tracker_zone_repo = TrackerZoneJsonRepository()
 setting_repo = SettingJsonRepository()
 gui_config_repo = GuiConfigJsonRepository()
+user_preferences_repo = UserPreferencesJsonRepository()
 
 def _none_if_empty(v):
     return None if v in ("", None, 0, "0") else v
@@ -155,6 +159,18 @@ def _normalize_zone(row: dict) -> dict:
     row["groups"] = _uuid_list(row.get("groups"))
     return row
 
+
+def _uid_from_claims(claims: dict) -> str:
+    uid = (
+        claims.get("preferred_username")
+        or claims.get("username")
+        or claims.get("email")
+        or claims.get("sub")
+    )
+    if not uid:
+        raise HTTPException(status_code=401, detail="User identity not found in token")
+    return str(uid)
+
 CORE_GET_SYNC = {
     "/reslevis/getGateways": (gateway_repo, _normalize_gateway),
     "/reslevis/getZones": (zone_repo, _normalize_zone),
@@ -604,6 +620,29 @@ def getGuiConfigs():
     return gui_config_repo.list()
 
 
+@router.get(
+    "/getUserPreferences",
+    response_model=UserPreferencesItem,
+    tags=["Reslevis"],
+)
+def getUserPreferences(current_user: dict = Depends(get_current_user)):
+    uid = _uid_from_claims(current_user)
+    return user_preferences_repo.get(uid)
+
+
+@router.put(
+    "/updateUserPreferences",
+    response_model=UserPreferencesItem,
+    tags=["Reslevis"],
+)
+def updateUserPreferences(
+    item: UserPreferencesUpdateItem,
+    current_user: dict = Depends(get_current_user),
+):
+    uid = _uid_from_claims(current_user)
+    return user_preferences_repo.update(uid, item)
+
+
 @router.post("/postGuiConfig", tags=["Reslevis"], dependencies=[Depends(get_current_user)])
 def postGuiConfig(item: GuiConfigItem):
     gui_config_repo.add(item)

schemas/reslevis.py

@@ -1,9 +1,10 @@
-from pydantic import BaseModel
+from pydantic import BaseModel, Field
 from typing import Optional
 from uuid import UUID
 from typing import Optional, Union, Literal
 from typing import List
 from typing import Tuple
+from typing import Dict
 
 class BuildingItem(BaseModel):
     id: UUID
@@ -113,6 +114,24 @@ class GuiConfigItem(BaseModel):
     role: Optional[Literal["developer", "administrator", "user"]] = None
     debug: Optional[bool] = None
 
+
+class TablePreferenceItem(BaseModel):
+    hiddenColumns: Dict[str, bool] = Field(default_factory=dict)
+
+
+class UserPreferencesItem(BaseModel):
+    id: Optional[UUID] = None
+    uid: str
+    language: Literal["it", "en"] = "it"
+    tables: Dict[str, TablePreferenceItem] = Field(default_factory=dict)
+    updated_at: Optional[str] = None
+
+
+class UserPreferencesUpdateItem(BaseModel):
+    language: Optional[Literal["it", "en"]] = None
+    tables: Optional[Dict[str, TablePreferenceItem]] = None
+
+
 class OperatorItem(BaseModel):
     id: UUID
     name: str