"""Vault-specific settings management. Provides persistent storage for per-vault configuration like hidden files settings. Settings are stored in /app/data/vault_settings.json """ import json import logging from pathlib import Path from typing import Dict, Any, Optional import threading logger = logging.getLogger("obsigate.vault_settings") _SETTINGS_PATH = Path("/app/data/vault_settings.json") _settings_lock = threading.Lock() # In-memory cache of vault settings _vault_settings: Dict[str, Dict[str, Any]] = {} def load_vault_settings() -> Dict[str, Dict[str, Any]]: """Load vault settings from disk. Returns: Dict mapping vault names to their settings. """ global _vault_settings with _settings_lock: if _SETTINGS_PATH.exists(): try: content = _SETTINGS_PATH.read_text(encoding="utf-8") _vault_settings = json.loads(content) logger.info(f"Loaded settings for {len(_vault_settings)} vaults") except Exception as e: logger.warning(f"Failed to load vault settings: {e}") _vault_settings = {} else: _vault_settings = {} return dict(_vault_settings) def save_vault_settings() -> None: """Persist vault settings to disk.""" with _settings_lock: try: _SETTINGS_PATH.parent.mkdir(parents=True, exist_ok=True) _SETTINGS_PATH.write_text( json.dumps(_vault_settings, indent=2, ensure_ascii=False), encoding="utf-8" ) logger.info(f"Saved settings for {len(_vault_settings)} vaults") except Exception as e: logger.error(f"Failed to save vault settings: {e}") raise def get_vault_setting(vault_name: str) -> Optional[Dict[str, Any]]: """Get settings for a specific vault. Args: vault_name: Name of the vault. Returns: Dict with vault settings or None if not found. """ with _settings_lock: return _vault_settings.get(vault_name) def update_vault_setting(vault_name: str, settings: Dict[str, Any]) -> Dict[str, Any]: """Update settings for a specific vault. Args: vault_name: Name of the vault. settings: Dict with settings to update (partial update supported). Returns: Updated settings dict for the vault. """ with _settings_lock: if vault_name not in _vault_settings: _vault_settings[vault_name] = {} _vault_settings[vault_name].update(settings) save_vault_settings() return dict(_vault_settings[vault_name]) def delete_vault_setting(vault_name: str) -> bool: """Delete settings for a specific vault. Args: vault_name: Name of the vault. Returns: True if settings were deleted, False if vault not found. """ with _settings_lock: if vault_name in _vault_settings: del _vault_settings[vault_name] save_vault_settings() return True return False def get_all_vault_settings() -> Dict[str, Dict[str, Any]]: """Get all vault settings. Returns: Dict mapping vault names to their settings. """ with _settings_lock: return dict(_vault_settings) # Initialize settings on module load load_vault_settings()