File size: 9,440 Bytes
d0fdbcd | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 | """
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
|