# backend/core/config_manager.py from pathlib import Path import json from typing import Dict, Any, Optional, List from datetime import datetime class BaseConfigManager: """Base class for all configuration managers""" def __init__(self, config_path: Path, schema_path: Optional[Path] = None): self.config_path = Path(config_path) self.schema_path = Path(schema_path) if schema_path else None self._schema = None self._config = None def _load_schema(self) -> Dict[str, Any]: """Load configuration schema""" if not self.schema_path or not self.schema_path.exists(): return {"config_schema": {}} try: with open(self.schema_path, 'r', encoding='utf-8') as f: return json.load(f) except Exception as e: print(f"Error loading schema: {e}") return {"config_schema": {}} def _load_config(self) -> Dict[str, Any]: """Load configuration data""" if not self.config_path.exists(): return {} try: with open(self.config_path, 'r', encoding='utf-8') as f: return json.load(f) except Exception as e: print(f"Error loading config: {e}") return {} def _save_config(self, config: Dict[str, Any]): """Save configuration data""" self.config_path.parent.mkdir(parents=True, exist_ok=True) try: with open(self.config_path, 'w', encoding='utf-8') as f: json.dump(config, f, indent=4) except Exception as e: print(f"Error saving config: {e}") raise def _validate_config(self, config: Dict[str, Any], schema: Dict[str, Any]) -> Dict[str, Any]: """Validate configuration against schema""" validated = {} schema_fields = schema.get("config_schema", {}) for key, field_schema in schema_fields.items(): if key in config: validated[key] = self._validate_field(key, config[key], field_schema) elif field_schema.get("required", False): raise ValueError(f"Required field '{key}' is missing") else: validated[key] = field_schema.get("default") return validated def _validate_field(self, key: str, value: Any, field_schema: Dict[str, Any]) -> Any: """Validate a single field value""" field_type = field_schema.get("type") if value is None or value == "": if field_schema.get("required", False): raise ValueError(f"Field '{key}' is required") return field_schema.get("default") try: if field_type == "string": return str(value) elif field_type == "number": return float(value) if "." in str(value) else int(value) elif field_type == "boolean": if isinstance(value, str): return value.lower() == "true" return bool(value) elif field_type == "directory": path = Path(value) if not path.is_absolute(): path = Path(self.config_path).parent / path path.mkdir(parents=True, exist_ok=True) return str(path) elif field_type == "select": if value not in field_schema.get("options", []): raise ValueError(f"Invalid option '{value}' for field '{key}'") return value else: return value except Exception as e: raise ValueError(f"Invalid value for field '{key}': {str(e)}") def get_schema(self) -> Dict[str, Any]: """Get configuration schema""" if self._schema is None: self._schema = self._load_schema() return self._schema def get_config(self) -> Dict[str, Any]: """Get current configuration""" if self._config is None: self._config = self._load_config() return self._config def update_schema(self, schema: Dict[str, Any]): """Update configuration schema""" if not self.schema_path: raise ValueError("No schema path configured") try: self.schema_path.parent.mkdir(parents=True, exist_ok=True) with open(self.schema_path, 'w', encoding='utf-8') as f: json.dump(schema, f, indent=4) self._schema = schema except Exception as e: print(f"Error saving schema: {e}") raise class ProfileManager(BaseConfigManager): """Manager for application profiles""" DEFAULT_PROFILE = { "id": "default", "name": "Default Profile", "llm_settings": { "model": "gpt-4", "temperature": 0.7, "api_key": "" } } def __init__(self, data_dir: Path): super().__init__( config_path=data_dir / "profiles.json", schema_path=data_dir / "profile_schema.json" ) self.profiles = self._load_profiles() def _load_profiles(self) -> Dict[str, Dict]: """Load all profiles""" if self.config_path.exists(): try: with open(self.config_path, 'r', encoding='utf-8') as f: profiles = json.load(f) if "default" not in profiles: profiles["default"] = self._create_default_profile() return profiles except Exception as e: print(f"Error loading profiles: {e}") return {"default": self._create_default_profile()} else: profiles = {"default": self._create_default_profile()} self._save_profiles(profiles) return profiles def _create_default_profile(self) -> Dict[str, Any]: """Create default profile""" profile = self.DEFAULT_PROFILE.copy() now = datetime.now().isoformat() profile["created_at"] = now profile["updated_at"] = now return profile def _save_profiles(self, profiles: Dict[str, Dict]): """Save all profiles""" try: self.config_path.parent.mkdir(parents=True, exist_ok=True) with open(self.config_path, 'w', encoding='utf-8') as f: json.dump(profiles, f, indent=4) except Exception as e: print(f"Error saving profiles: {e}") raise def get_all_profiles(self) -> List[Dict[str, Any]]: """Get all profiles""" return list(self.profiles.values()) def get_profile(self, profile_id: str) -> Optional[Dict[str, Any]]: """Get specific profile""" return self.profiles.get(profile_id) def create_profile(self, profile_data: Dict[str, Any]) -> Dict[str, Any]: """Create new profile""" if "id" not in profile_data: raise ValueError("Profile must have an id") profile_id = profile_data["id"] if profile_id in self.profiles: raise ValueError(f"Profile {profile_id} already exists") # Add timestamps now = datetime.now().isoformat() profile_data["created_at"] = now profile_data["updated_at"] = now # Validate against schema schema = self.get_schema() validated_data = self._validate_config(profile_data, schema) # Add to profiles self.profiles[profile_id] = validated_data self._save_profiles(self.profiles) return validated_data def update_profile(self, profile_id: str, profile_data: Dict[str, Any]) -> Dict[str, Any]: """Update existing profile""" if profile_id not in self.profiles: raise ValueError(f"Profile {profile_id} not found") if profile_id == "default" and "id" in profile_data: raise ValueError("Cannot change id of default profile") # Update timestamp profile_data["updated_at"] = datetime.now().isoformat() # Validate against schema schema = self.get_schema() validated_data = self._validate_config(profile_data, schema) # Update profile self.profiles[profile_id].update(validated_data) self._save_profiles(self.profiles) return self.profiles[profile_id] def delete_profile(self, profile_id: str): """Delete profile""" if profile_id == "default": raise ValueError("Cannot delete default profile") if profile_id not in self.profiles: raise ValueError(f"Profile {profile_id} not found") del self.profiles[profile_id] self._save_profiles(self.profiles) class ScriptGroupManager(BaseConfigManager): """Manager for script group configuration""" def __init__(self, group_dir: Path): super().__init__( config_path=group_dir / "data.json", schema_path=group_dir.parent / "config.json" ) def get_work_dir(self) -> Optional[str]: """Get work directory from configuration""" config = self.get_config() return config.get("work_dir") def update_config(self, config_data: Dict[str, Any]) -> Dict[str, Any]: """Update configuration""" # Add timestamp config_data["updated_at"] = datetime.now().isoformat() # Validate against schema schema = self.get_schema() validated_data = self._validate_config(config_data, schema) # Save configuration self._save_config(validated_data) self._config = validated_data return validated_data class WorkDirConfigManager(BaseConfigManager): """Manager for work directory configuration""" def __init__(self, work_dir: str, group_id: str): self.work_dir = Path(work_dir) self.group_id = group_id super().__init__( config_path=self.work_dir / "script_config.json", schema_path=None # Schema is loaded from group configuration ) def get_group_config(self) -> Dict[str, Any]: """Get configuration for current group""" config = self.get_config() return config.get("group_settings", {}).get(self.group_id, {}) def update_group_config(self, settings: Dict[str, Any]): """Update configuration for current group""" config = self.get_config() if "group_settings" not in config: config["group_settings"] = {} config["group_settings"][self.group_id] = settings config["updated_at"] = datetime.now().isoformat() self._save_config(config) self._config = config