""" Repair engine for fixing detected issues in generated configurations. Implements intelligent repairs rather than blind retries. """ import json from typing import Any, Dict, List, Optional, Tuple from schemas import FieldType class RepairEngine: """Repairs detected issues in generated configurations.""" def __init__(self): self.repairs_made = [] def repair_json(self, json_str: str) -> Tuple[bool, Optional[Dict[str, Any]]]: """Attempt to repair malformed JSON.""" # Try obvious fixes first attempts = [ json_str, json_str.rstrip(",") + "}", # Remove trailing comma before } json_str.rstrip(",") + "]", # Remove trailing comma before ] self._fix_unquoted_keys(json_str), self._fix_missing_quotes(json_str), ] for attempt in attempts: try: data = json.loads(attempt) self.repairs_made.append(f"Fixed malformed JSON") return True, data except json.JSONDecodeError: continue return False, None def _fix_unquoted_keys(self, json_str: str) -> str: """Fix unquoted keys in JSON.""" import re # Pattern to match unquoted keys return re.sub(r'(\w+):', r'"\1":', json_str) def _fix_missing_quotes(self, json_str: str) -> str: """Attempt to fix missing quotes around values.""" # This is risky, so be conservative return json_str def repair_missing_fields(self, config: Dict[str, Any], required_fields: List[str]) -> Dict[str, Any]: """Add missing required fields with sensible defaults.""" for field in required_fields: if field not in config or config[field] is None: default = self._get_default_for_field(field) config[field] = default self.repairs_made.append(f"Added missing field '{field}' with default: {default}") return config def _get_default_for_field(self, field_name: str) -> Any: """Provide sensible default for a field.""" defaults = { "app_name": "GeneratedApp", "app_description": "Auto-generated application", "database_schema": [], "api_schema": [], "ui_schema": [], "auth_config": {"type": "basic"}, "roles": [{"name": "user", "permissions": []}], "business_logic": {}, "validation_metadata": {}, } return defaults.get(field_name, "") def repair_database_schema(self, db_schema: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """Repair issues in database schema.""" repaired = [] for table in db_schema: if not isinstance(table, dict): continue # Ensure required fields if "name" not in table: table["name"] = f"table_{len(repaired)}" self.repairs_made.append(f"Added missing table name") if "fields" not in table: table["fields"] = [] self.repairs_made.append(f"Table '{table['name']}': added empty fields list") if "primary_key" not in table: table["primary_key"] = "id" self.repairs_made.append(f"Table '{table['name']}': added default primary_key") # Repair fields if isinstance(table["fields"], list): table["fields"] = self._repair_fields(table["fields"], table["name"]) repaired.append(table) return repaired def _repair_fields(self, fields: List[Dict[str, Any]], table_name: str) -> List[Dict[str, Any]]: """Repair field definitions.""" repaired = [] for field in fields: if not isinstance(field, dict): continue # Ensure name if "name" not in field: field["name"] = f"field_{len(repaired)}" self.repairs_made.append(f"Table '{table_name}': added field name") # Ensure valid type if "type" not in field: field["type"] = FieldType.STRING.value self.repairs_made.append(f"Table '{table_name}', field '{field['name']}': set default type to string") else: valid_types = [t.value for t in FieldType] if field["type"] not in valid_types: field["type"] = FieldType.STRING.value self.repairs_made.append( f"Table '{table_name}', field '{field['name']}': fixed invalid type" ) # Ensure required field if "required" not in field: field["required"] = True repaired.append(field) return repaired def repair_api_schema(self, api_schema: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """Repair issues in API schema.""" repaired = [] valid_methods = ["GET", "POST", "PUT", "DELETE", "PATCH"] for endpoint in api_schema: if not isinstance(endpoint, dict): continue # Ensure path if "path" not in endpoint: endpoint["path"] = f"/endpoint_{len(repaired)}" self.repairs_made.append(f"API: added missing endpoint path") # Ensure valid method if "method" not in endpoint: endpoint["method"] = "GET" self.repairs_made.append(f"API {endpoint['path']}: added default method GET") elif endpoint["method"] not in valid_methods: endpoint["method"] = "POST" self.repairs_made.append(f"API {endpoint['path']}: fixed invalid HTTP method") # Ensure description if "description" not in endpoint: endpoint["description"] = f"Endpoint for {endpoint['path']}" self.repairs_made.append(f"API {endpoint['path']}: added auto-description") repaired.append(endpoint) return repaired def repair_ui_schema(self, ui_schema: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """Repair issues in UI schema.""" repaired = [] for page in ui_schema: if not isinstance(page, dict): continue # Ensure path if "path" not in page: page["path"] = f"/page_{len(repaired)}" self.repairs_made.append(f"UI: added missing page path") # Ensure title if "title" not in page: page["title"] = page["path"].replace("/", " ").title() self.repairs_made.append(f"UI page {page['path']}: added auto-generated title") # Ensure components if "components" not in page: page["components"] = [] self.repairs_made.append(f"UI page {page['path']}: added empty components list") repaired.append(page) return repaired def repair_roles_config(self, roles: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """Repair roles configuration.""" repaired = [] for role in roles: if not isinstance(role, dict): continue if "name" not in role: role["name"] = f"role_{len(repaired)}" self.repairs_made.append(f"Added missing role name") if "permissions" not in role: role["permissions"] = [] self.repairs_made.append(f"Role '{role['name']}': added empty permissions list") if "description" not in role: role["description"] = f"Role: {role['name']}" self.repairs_made.append(f"Role '{role['name']}': added auto-description") repaired.append(role) return repaired def repair_config(self, config: Dict[str, Any]) -> Tuple[Dict[str, Any], List[str]]: """Run complete repair on configuration.""" self.repairs_made = [] # Repair missing top-level fields required_fields = [ "app_name", "app_description", "database_schema", "api_schema", "ui_schema", "auth_config", "roles", "business_logic" ] config = self.repair_missing_fields(config, required_fields) # Repair each layer if isinstance(config["database_schema"], list): config["database_schema"] = self.repair_database_schema(config["database_schema"]) if isinstance(config["api_schema"], list): config["api_schema"] = self.repair_api_schema(config["api_schema"]) if isinstance(config["ui_schema"], list): config["ui_schema"] = self.repair_ui_schema(config["ui_schema"]) if isinstance(config["roles"], list): config["roles"] = self.repair_roles_config(config["roles"]) return config, self.repairs_made