import os import json from typing import Dict, Any, Optional, Callable from .schema_handler import SchemaHandler # Import SchemaHandler class ConfigHandler: def __init__( self, data_path: str, script_groups_path: str, get_workdir_func: Callable[[], Optional[str]], schema_handler: SchemaHandler, ): self.data_path = data_path self.script_groups_path = script_groups_path self._get_working_directory = ( get_workdir_func # Function to get current workdir ) self.schema_handler = schema_handler # Instance of SchemaHandler def get_config(self, level: str, group: str = None) -> Dict[str, Any]: """ Get configuration for specified level. Applies default values from the corresponding schema if the config file doesn't exist or is missing keys with defaults. """ config_data = {} needs_save = False schema = None data_path = self._get_config_path(level, group) schema_path_for_debug = "N/A" # For logging if not data_path: if level == "3": # Level 3 depends on working directory return {} # Return empty if working dir not set for L3 else: return { "error": f"Could not determine config path for level {level}, group {group}" } # Determine schema path for logging purposes (actual loading done by schema_handler) if level == "1": schema_path_for_debug = os.path.join(self.data_path, "esquema_general.json") elif level == "2" and group: schema_path_for_debug = os.path.join( self.script_groups_path, group, "esquema_group.json" ) elif level == "3" and group: schema_path_for_debug = os.path.join( self.script_groups_path, group, "esquema_work.json" ) elif level == "3": schema_path_for_debug = "N/A (Level 3 without group)" # Get schema using SchemaHandler try: schema = self.schema_handler.get_schema(level, group) except Exception as e: print( f"Warning: Could not load schema for level {level}, group {group}. Defaults will not be applied. Error: {e}" ) schema = None # Try to load existing data data_file_exists = os.path.exists(data_path) if data_file_exists: try: with open(data_path, "r", encoding="utf-8") as f_data: content = f_data.read() if content.strip(): config_data = json.loads(content) else: print( f"Warning: Data file {data_path} is empty. Will initialize with defaults." ) needs_save = True except json.JSONDecodeError: print( f"Warning: Could not decode JSON from {data_path}. Will initialize with defaults." ) config_data = {} needs_save = True except Exception as e: print( f"Error reading data from {data_path}: {e}. Will attempt to initialize with defaults." ) config_data = {} needs_save = True else: # File doesn't exist print( f"Info: Data file not found at {data_path}. Will initialize with defaults." ) needs_save = True # Apply defaults from schema if schema and isinstance(schema, dict) and "properties" in schema: schema_properties = schema.get("properties", {}) if isinstance(schema_properties, dict): for key, prop_definition in schema_properties.items(): if ( isinstance(prop_definition, dict) and key not in config_data and "default" in prop_definition ): print( f"Info: Applying default for '{key}' from schema {schema_path_for_debug}" ) config_data[key] = prop_definition["default"] needs_save = True else: print( f"Warning: 'properties' in schema {schema_path_for_debug} is not a dictionary. Cannot apply defaults." ) # Save if needed if needs_save: try: print(f"Info: Saving updated config data to: {data_path}") os.makedirs(os.path.dirname(data_path), exist_ok=True) with open(data_path, "w", encoding="utf-8") as f_data: json.dump(config_data, f_data, indent=2, ensure_ascii=False) except IOError as e: print(f"Error: Could not write data file to {data_path}: {e}") except Exception as e: print(f"Unexpected error saving data to {data_path}: {e}") return config_data def update_config( self, level: str, data: Dict[str, Any], group: str = None ) -> Dict[str, str]: """Update configuration for specified level.""" path = self._get_config_path(level, group) if not path: return { "status": "error", "message": f"Could not determine config path for level {level}, group {group}", } try: os.makedirs(os.path.dirname(path), exist_ok=True) with open(path, "w", encoding="utf-8") as f: json.dump(data, f, indent=2, ensure_ascii=False) print(f"Info: Config successfully updated at {path}") return {"status": "success"} except Exception as e: print(f"Error updating config at {path}: {str(e)}") return {"status": "error", "message": str(e)} def _get_config_path( self, level: str, group: Optional[str] = None ) -> Optional[str]: """Helper to determine the config file path.""" if level == "1": return os.path.join(self.data_path, "data.json") elif level == "2": if not group: return None return os.path.join(self.script_groups_path, group, "data.json") elif level == "3": working_directory = self._get_working_directory() if working_directory and os.path.isdir(working_directory): return os.path.join(working_directory, "data.json") else: return None # Cannot determine L3 path without valid workdir else: return None