ConfigWizard

Mission: Ensures configuration consistency and compliance across development, staging, and production environments.

Large JSON/YAML Config Validation & Auto-Fix with Versioning

import os
import json
import yaml
import logging
from datetime import datetime

logging.basicConfig(level=logging.INFO)

class ConfigWizardAgent:
    def __init__(self, backup_dir="./config_backups"):
        self.backup_dir = backup_dir
        if not os.path.exists(backup_dir):
            os.makedirs(backup_dir)

    def scan_and_fix_config(self, config_path: str, auto_apply: bool = True):
        logging.info(f"Scanning config: {config_path}")
        ext = os.path.splitext(config_path)[1].lower()
        if ext == ".json":
            with open(config_path, 'r', encoding='utf-8') as f:
                config_data = json.load(f)
        elif ext in [".yml", ".yaml"]:
            with open(config_path, 'r', encoding='utf-8') as f:
                config_data = yaml.safe_load(f)
        else:
            logging.error(f"Unsupported config format: {ext}")
            return

        issues_found = self._validate_config(config_data)
        if issues_found and auto_apply:
            self._backup_config(config_path)
            self._auto_fix(config_data, issues_found)
            self._save_config(config_data, config_path, ext)
            logging.info(f"Auto-fixed {len(issues_found)} issues in {config_path}")
        else:
            logging.info(f"Found {len(issues_found)} issues, auto_apply={auto_apply}")

    def _validate_config(self, cfg_data: dict) -> list:
        issues = []
        # Example checks: "debug" should be false, "port" within a range, "db.user" non-empty
        if cfg_data.get('debug', True) is True:
            issues.append(('debug', 'Should be set to False in production.'))

        if 'port' in cfg_data and not (1024 < cfg_data['port'] < 65535):
            issues.append(('port', 'Port must be within 1025-65534.'))

        db_cfg = cfg_data.get('database', {})
        if not db_cfg.get('user'):
            issues.append(('database.user', 'Database user is missing or blank.'))

        return issues

    def _auto_fix(self, cfg_data: dict, issues: list):
        for issue_key, msg in issues:
            if issue_key == 'debug':
                cfg_data['debug'] = False
            elif issue_key == 'port':
                cfg_data['port'] = 8080
            elif issue_key == 'database.user':
                cfg_data.setdefault('database', {})
                cfg_data['database']['user'] = "default_user"

    def _backup_config(self, config_path: str):
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        basename = os.path.basename(config_path)
        backup_name = f"{basename}.{timestamp}.bak"
        backup_path = os.path.join(self.backup_dir, backup_name)
        logging.info(f"Backing up original config to {backup_path}")
        with open(config_path, 'rb') as src, open(backup_path, 'wb') as dst:
            dst.write(src.read())

    def _save_config(self, cfg_data: dict, config_path: str, ext: str):
        if ext == ".json":
            with open(config_path, 'w', encoding='utf-8') as f:
                json.dump(cfg_data, f, indent=2)
        else:
            with open(config_path, 'w', encoding='utf-8') as f:
                yaml.safe_dump(cfg_data, f, sort_keys=False)

# Usage:
# wizard = ConfigWizardAgent()
# wizard.scan_and_fix_config("./config_staging.yaml", auto_apply=True)