297 lines
11 KiB
Python
297 lines
11 KiB
Python
# 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 |