114 lines
3.7 KiB
Python
114 lines
3.7 KiB
Python
import json
|
|
import os
|
|
import shutil
|
|
from typing import Dict, Any, Optional
|
|
from datetime import datetime
|
|
|
|
CONFIG_FILENAME = "crumbforest_config.json"
|
|
|
|
class ConfigService:
|
|
_config: Optional[Dict[str, Any]] = None
|
|
_config_path: Optional[str] = None
|
|
|
|
@classmethod
|
|
def _resolve_path(cls) -> str:
|
|
"""Resolve the absolute path to the config file."""
|
|
if cls._config_path:
|
|
return cls._config_path
|
|
|
|
# Try finding it
|
|
paths_to_try = [
|
|
# 1. In root (for docker -v mapped volume)
|
|
os.path.abspath(CONFIG_FILENAME),
|
|
# 2. One level up (if running from app/)
|
|
os.path.abspath(os.path.join("..", CONFIG_FILENAME)),
|
|
# 3. Explicit /config dir
|
|
"/config/crumbforest_config.json"
|
|
]
|
|
|
|
for p in paths_to_try:
|
|
if os.path.exists(p):
|
|
cls._config_path = p
|
|
return p
|
|
|
|
# Default fallback (might create new file later)
|
|
return os.path.abspath(CONFIG_FILENAME)
|
|
|
|
@classmethod
|
|
def load_config(cls, force_reload: bool = False) -> Dict[str, Any]:
|
|
if cls._config is None or force_reload:
|
|
path = cls._resolve_path()
|
|
try:
|
|
with open(path, 'r', encoding='utf-8') as f:
|
|
cls._config = json.load(f)
|
|
# print(f"Loaded config from {path}")
|
|
except Exception as e:
|
|
print(f"Error loading config from {path}: {e}")
|
|
cls._config = {"roles": {}, "groups": {}, "theme_variants": {}}
|
|
|
|
return cls._config
|
|
|
|
@classmethod
|
|
def save_config(cls, new_config: Dict[str, Any]) -> bool:
|
|
"""
|
|
Save new configuration atomically with backup.
|
|
"""
|
|
path = cls._resolve_path()
|
|
backup_path = f"{path}.bak"
|
|
temp_path = f"{path}.tmp"
|
|
|
|
try:
|
|
# 1. Validate JSON first (implicit by type typing, but meaningful check: must have 'roles')
|
|
if 'roles' not in new_config:
|
|
raise ValueError("Config must contain 'roles' key")
|
|
|
|
# 2. Create Backup
|
|
if os.path.exists(path):
|
|
shutil.copy2(path, backup_path)
|
|
|
|
# 3. Write to Temp File
|
|
with open(temp_path, 'w', encoding='utf-8') as f:
|
|
json.dump(new_config, f, indent=4, ensure_ascii=False)
|
|
|
|
# 4. Atomic Rename
|
|
os.replace(temp_path, path)
|
|
|
|
# 5. Update Memory
|
|
cls._config = new_config
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"FAILED to save config: {e}")
|
|
if os.path.exists(temp_path):
|
|
os.remove(temp_path)
|
|
return False
|
|
|
|
@classmethod
|
|
def get_raw_config(cls) -> str:
|
|
"""Get config as raw JSON string."""
|
|
path = cls._resolve_path()
|
|
try:
|
|
with open(path, 'r', encoding='utf-8') as f:
|
|
return f.read()
|
|
except:
|
|
return "{}"
|
|
|
|
# --- Accessors ---
|
|
@classmethod
|
|
def get_role(cls, role_id: str) -> Optional[Dict[str, Any]]:
|
|
config = cls.load_config()
|
|
return config.get('roles', {}).get(role_id)
|
|
|
|
@classmethod
|
|
def get_group(cls, group_id: str) -> Optional[Dict[str, Any]]:
|
|
config = cls.load_config()
|
|
return config.get('groups', {}).get(group_id)
|
|
|
|
@classmethod
|
|
def get_theme(cls, theme_id: str) -> Optional[Dict[str, Any]]:
|
|
config = cls.load_config()
|
|
return config.get('theme_variants', {}).get(theme_id)
|
|
|
|
# Alias for backward compatibility if needed, though we should update imports
|
|
ConfigLoader = ConfigService
|