127 lines
5.9 KiB
Python
127 lines
5.9 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from flask import session
|
|
|
|
from app.core_settings import AppSettings, get_settings
|
|
from app.storage.kiosk_settings import SQLiteKioskSettingsRepository
|
|
|
|
|
|
VALID_MODES = {"public", "private"}
|
|
USER_MODE_PREFIX = "user:"
|
|
DEFAULT_WIDGETS = ["hero", "history", "strings", "status", "production", "comparison", "importStatus"]
|
|
VALID_WIDGETS = {"hero", "quickMetrics", "history", "status", "strings", "production", "comparison", "distribution", "importStatus"}
|
|
VALID_REALTIME_RANGES = {"today", "yesterday", "6h", "12h", "24h", "48h", "7d"}
|
|
VALID_ANALYTICS_RANGES = {"today", "yesterday", "7d", "30d", "90d", "365d", "custom"}
|
|
|
|
|
|
class KioskSettingsService:
|
|
def __init__(self, settings: AppSettings | None = None) -> None:
|
|
self.settings = settings or get_settings()
|
|
self.repository = SQLiteKioskSettingsRepository(self.settings.storage["sqlite_path"])
|
|
|
|
def get(self, mode: str) -> dict[str, Any]:
|
|
normalized_mode = self._normalize_mode(mode)
|
|
stored = self.repository.get(normalized_mode)
|
|
if stored is None:
|
|
return self._default_payload(normalized_mode)
|
|
return self._sanitize_payload(normalized_mode, stored, persist_if_changed=False)
|
|
|
|
def update(self, mode: str, payload: dict[str, Any], updated_by: str | None = None) -> dict[str, Any]:
|
|
normalized_mode = self._normalize_mode(mode)
|
|
merged = {**self.get(normalized_mode), **(payload or {})}
|
|
cleaned = self._sanitize_payload(normalized_mode, merged, persist_if_changed=False)
|
|
return self.repository.upsert(normalized_mode, cleaned, updated_by=updated_by)
|
|
|
|
def update_from_session(self, mode: str, payload: dict[str, Any]) -> dict[str, Any]:
|
|
updated_by = session.get("auth_user")
|
|
return self.update(mode, payload, updated_by=updated_by)
|
|
|
|
def _default_payload(self, mode: str) -> dict[str, Any]:
|
|
return {
|
|
"mode": mode,
|
|
"widgets": list(DEFAULT_WIDGETS),
|
|
"realtime_range": self._default_realtime_range(),
|
|
"analytics_range": self._default_analytics_range(),
|
|
"analytics_bucket": self._default_analytics_bucket(),
|
|
"compare_mode": self._default_compare_mode(),
|
|
"updated_at": None,
|
|
"updated_by": None,
|
|
}
|
|
|
|
def _sanitize_payload(self, mode: str, payload: dict[str, Any], persist_if_changed: bool = False) -> dict[str, Any]:
|
|
cleaned = {
|
|
"mode": mode,
|
|
"widgets": self._normalize_widgets(payload.get("widgets")),
|
|
"realtime_range": self._normalize_realtime_range(payload.get("realtime_range")),
|
|
"analytics_range": self._normalize_analytics_range(payload.get("analytics_range")),
|
|
"analytics_bucket": self._normalize_bucket(payload.get("analytics_bucket")),
|
|
"compare_mode": self._normalize_compare_mode(payload.get("compare_mode")),
|
|
"updated_at": payload.get("updated_at"),
|
|
"updated_by": payload.get("updated_by"),
|
|
}
|
|
if persist_if_changed:
|
|
return self.repository.upsert(mode, cleaned, updated_by=cleaned.get("updated_by"))
|
|
return cleaned
|
|
|
|
def _normalize_mode(self, mode: str) -> str:
|
|
normalized = (mode or "").strip().lower()
|
|
if normalized in VALID_MODES:
|
|
return normalized
|
|
if normalized.startswith(USER_MODE_PREFIX) and len(normalized) > len(USER_MODE_PREFIX):
|
|
return normalized
|
|
raise ValueError("Mode musi byc jednym z: public, private")
|
|
|
|
def _normalize_widgets(self, widgets: Any) -> list[str]:
|
|
if not isinstance(widgets, list):
|
|
return list(DEFAULT_WIDGETS)
|
|
normalized: list[str] = []
|
|
for item in widgets:
|
|
widget = str(item or "").strip()
|
|
if widget in VALID_WIDGETS and widget not in normalized:
|
|
normalized.append(widget)
|
|
return normalized or list(DEFAULT_WIDGETS)
|
|
|
|
def _normalize_realtime_range(self, value: Any) -> str:
|
|
normalized = str(value or self._default_realtime_range()).strip()
|
|
return normalized if normalized in VALID_REALTIME_RANGES else self._default_realtime_range()
|
|
|
|
def _normalize_analytics_range(self, value: Any) -> str:
|
|
normalized = str(value or self._default_analytics_range()).strip()
|
|
return normalized if normalized in VALID_ANALYTICS_RANGES else self._default_analytics_range()
|
|
|
|
def _normalize_bucket(self, value: Any) -> str:
|
|
normalized = str(value or self._default_analytics_bucket()).strip()
|
|
return normalized if normalized in self.settings.analytics["bucket_labels"] else self._default_analytics_bucket()
|
|
|
|
def _normalize_compare_mode(self, value: Any) -> str:
|
|
normalized = str(value or self._default_compare_mode()).strip()
|
|
return normalized if normalized in self.settings.analytics["compare_modes"] else self._default_compare_mode()
|
|
|
|
def _default_realtime_range(self) -> str:
|
|
raw = str(self.settings.realtime.get("history_default_range", "12h"))
|
|
return raw if raw in VALID_REALTIME_RANGES else "12h"
|
|
|
|
def _default_analytics_range(self) -> str:
|
|
raw = str(self.settings.analytics.get("default_range", "30d"))
|
|
return raw if raw in VALID_ANALYTICS_RANGES else "30d"
|
|
|
|
def _default_analytics_bucket(self) -> str:
|
|
raw = str(self.settings.analytics.get("default_bucket", "day"))
|
|
return raw if raw in self.settings.analytics["bucket_labels"] else "day"
|
|
|
|
def _default_compare_mode(self) -> str:
|
|
raw = str(self.settings.analytics.get("default_compare", "none"))
|
|
return raw if raw in self.settings.analytics["compare_modes"] else "none"
|
|
|
|
|
|
_kiosk_settings_service: KioskSettingsService | None = None
|
|
|
|
|
|
def get_kiosk_settings_service() -> KioskSettingsService:
|
|
global _kiosk_settings_service
|
|
if _kiosk_settings_service is None:
|
|
_kiosk_settings_service = KioskSettingsService()
|
|
return _kiosk_settings_service
|