import os import json import re from .utils import get_dict_value, set_dict_value, dict_has_key, load_json_file def get_config_value(key): return get_dict_value(RGTHREE_CONFIG, key) def extend_config(default_config, user_config): """ Returns a new config dict combining user_config into defined keys for default_config.""" cfg = {} for key, value in default_config.items(): if key not in user_config: cfg[key] = value elif isinstance(value, dict): cfg[key] = extend_config(value, user_config[key]) else: cfg[key] = user_config[key] if key in user_config else value return cfg def set_user_config(data: dict): """ Sets the user configuration.""" count = 0 for key, value in data.items(): if dict_has_key(DEFAULT_CONFIG, key): set_dict_value(USER_CONFIG, key, value) set_dict_value(RGTHREE_CONFIG, key, value) count += 1 if count > 0: write_user_config() def get_rgthree_default_config(): """ Gets the default configuration.""" return load_json_file(DEFAULT_CONFIG_FILE, default={}) def get_rgthree_user_config(): """ Gets the user configuration.""" return load_json_file(USER_CONFIG_FILE, default={}) def write_user_config(): """ Writes the user configuration.""" with open(USER_CONFIG_FILE, 'w+', encoding='UTF-8') as file: json.dump(USER_CONFIG, file, sort_keys=True, indent=2, separators=(",", ": ")) THIS_DIR = os.path.dirname(os.path.abspath(__file__)) DEFAULT_CONFIG_FILE = os.path.join(THIS_DIR, '..', 'rgthree_config.json.default') USER_CONFIG_FILE = os.path.join(THIS_DIR, '..', 'rgthree_config.json') DEFAULT_CONFIG = {} USER_CONFIG = {} RGTHREE_CONFIG = {} def refresh_config(): """Refreshes the config.""" global DEFAULT_CONFIG, USER_CONFIG, RGTHREE_CONFIG DEFAULT_CONFIG = get_rgthree_default_config() USER_CONFIG = get_rgthree_user_config() # Migrate old config options into "features" needs_to_write_user_config = False if 'patch_recursive_execution' in USER_CONFIG: del USER_CONFIG['patch_recursive_execution'] needs_to_write_user_config = True if 'features' in USER_CONFIG and 'patch_recursive_execution' in USER_CONFIG['features']: del USER_CONFIG['features']['patch_recursive_execution'] needs_to_write_user_config = True if 'show_alerts_for_corrupt_workflows' in USER_CONFIG: if 'features' not in USER_CONFIG: USER_CONFIG['features'] = {} USER_CONFIG['features']['show_alerts_for_corrupt_workflows'] = USER_CONFIG[ 'show_alerts_for_corrupt_workflows'] del USER_CONFIG['show_alerts_for_corrupt_workflows'] needs_to_write_user_config = True if 'monitor_for_corrupt_links' in USER_CONFIG: if 'features' not in USER_CONFIG: USER_CONFIG['features'] = {} USER_CONFIG['features']['monitor_for_corrupt_links'] = USER_CONFIG['monitor_for_corrupt_links'] del USER_CONFIG['monitor_for_corrupt_links'] needs_to_write_user_config = True if needs_to_write_user_config is True: print('writing new user config.') write_user_config() RGTHREE_CONFIG = extend_config(DEFAULT_CONFIG, USER_CONFIG) if "unreleased" in USER_CONFIG and "unreleased" not in RGTHREE_CONFIG: RGTHREE_CONFIG["unreleased"] = USER_CONFIG["unreleased"] def get_config(): """Returns the congfig.""" return RGTHREE_CONFIG refresh_config()