""" Silique (https://www.silique.fr) Copyright (C) 2022-2026 This program is free software: you can redistribute it and/or modify it under the terms of the GNU Lesser General Public License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any later version. This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more details. You should have received a copy of the GNU Lesser General Public License along with this program. If not, see . """ from typing import Optional, Any from ruamel.yaml import YAML from io import BytesIO from tiramisu.error import ( PropertiesOptionError, LeadershipError, ConfigError, AttributeOptionError, ) from tiramisu import owners, groups from .config import OutPuts from .i18n import _ class RougailOutputDisplay: def __init__( self, config: "Config", *, rougailconfig: "RougailConfig" = None, root_config=None, user_data_errors: Optional[list] = None, user_data_warnings: Optional[list] = None, config_owner_is_path: bool = False, layer_datas=None, true_config=None, **kwargs, ) -> None: if rougailconfig is None: from rougail import RougailConfig rougailconfig = RougailConfig self.rougailconfig = rougailconfig self.config = config if true_config is None: self.true_config = config else: self.true_config = true_config self.root_config = root_config self.layer_datas = layer_datas self.config_owner_is_path = config_owner_is_path self.errors = [] self.warnings = [] if user_data_errors is None: user_data_errors = [] self.user_data_errors = user_data_errors if user_data_warnings is None: user_data_warnings = [] self.user_data_warnings = user_data_warnings # self.out = [] self.nodes = None self.yaml = YAML() self.yaml.indent(mapping=2, sequence=4, offset=2) def run(self) -> str: output_format = self.rougailconfig["display.output_format"] output = OutPuts().get()[output_format](self.rougailconfig) warnings = self.user_data_warnings + self.warnings errors = self.user_data_errors + self.errors level = None errors_warnings_dict = {} if not errors and self.nodes is None: show_secrets = self.rougailconfig["display.show_secrets"] self.nodes = Node( self.yaml, show_secrets, self.config, self.root_config, self.config_owner_is_path, errors, ) if warnings: level = "warnings" # output.display_warnings(errors_warnings_dict, warnings) for warning in warnings: output.error_warn_to_dict(warning, errors_warnings_dict, "warning") if errors: level = "errors" for error in errors: output.error_warn_to_dict(error, errors_warnings_dict, "error") if level: if level == "errors": tree = output.error_header() else: tree = output.warning_header() ret = output.parse_error_warning( tree, errors_warnings_dict, output.display_error, level ) if errors: return False, ret ret += "\n" else: ret = "" code, run = output.run(self.nodes, self.layer_datas) return code, ret + run def print(self) -> None: status, ret = self.run() print(ret) return status class Node: def __init__( self, yaml, show_secrets, config, root_config, config_owner_is_path, errors, *, node=None, values: Optional[dict] = None, level: int = 0, leader_index: Optional[int] = None, ) -> None: self.yaml = yaml self.show_secrets = show_secrets self.config = config self.root_config = root_config self.config_owner_is_path = config_owner_is_path self.errors = errors if node is None: values = self.config.value.get() self.level = level if node: self.hidden = "hidden" in node.property.get() self.description = node.description() else: self.hidden = None self.description = _("Variables:") self.level = level self.children = [] if node and node.isoptiondescription() and node.isleadership(): values_iter = iter(values.items()) leader, leader_values = next(values_iter) followers_values = { idx: {leader: value} for idx, value in enumerate(leader_values) } for follower, follower_value in values_iter: followers_values[follower.index()][follower] = follower_value for idx, fvalues in enumerate(followers_values.values()): self.add_node(leader, fvalues, leader_index=idx) else: for option, value in values.items(): if option.isoptiondescription(): if option.group_type() == groups.namespace and not value: continue if option.isleadership(): leader, leader_values = next(iter(value.items())) if not leader_values: self.add_leaf(leader, [], description=option.description()) else: self.add_node(option, value) else: if value == {}: self.add_leaf(option, "{}") else: self.add_node(option, value) else: if option.isleader(): index = leader_index else: index = None self.add_leaf(option, value, leader_index=index) def add_node( self, option, values, *, leader_index: Optional[int] = None, ) -> "Node": self.children.append( { "type": "node", "node": Node( self.yaml, self.show_secrets, self.config, self.root_config, self.config_owner_is_path, self.errors, node=option, values=values, level=self.level + 1, leader_index=leader_index, ), } ) def add_leaf( self, option, value: Any, *, leader_index: Optional[int] = None, description: Optional[str] = None, ): properties = option.property.get() if description is None: description = option.description() icon = "leaf" else: icon = "node" self.children.append( { "type": "leaf", "description": description, "values": self.get_values(option, value, leader_index, properties), "icon": icon, "hidden": "hidden" in properties, }, ) def get_values( self, option: "Option", value: Any, leader_index: Optional[int], properties: list[str], ) -> None: if option.isoptiondescription(): return [{"is_default": True, "value": value, "loaded_from": None}] values = [] meta_config = self.config meta_option = option index = option.index() if index is not None: loaded_from_key = f"loaded_from_{index}" else: loaded_from_key = "loaded_from" # default = option.owner.isdefault() or not option.information.get("default_value_makes_sense", True) force_store_value = "force_store_value" in properties is_default = option.owner.isdefault() option_path = option.path() default_owner = owners.default if self.root_config == meta_config: added = True true_default = is_default else: true_default = is_default and meta_option.owner.get() == default_owner added = not is_default or true_default secret_manager = option.information.get("secret_manager", False) while True: if values and true_default and (value in [None, []] or secret_manager): break if isinstance(value, list): value = [self.convert_value(meta_option, val) for val in value] else: value = self.convert_value(meta_option, value) if true_default: loaded_from = None else: loaded_from = meta_option.information.get(loaded_from_key, None) if added or force_store_value: values.append( { "value": value, "is_default": true_default, "loaded_from": loaded_from, } ) if true_default or force_store_value: break meta_config_path = meta_config.path() if is_default and (not meta_config_path or "." not in meta_config_path): # we are in root metaconfig and we have default value break new_meta_config = self.get_metaconfig_with_default_value( meta_config, meta_option ) if not new_meta_config: break meta_option = new_meta_config.option(option_path, index) if new_meta_config == meta_config: # we already have current value, so we search default's one is_default = True true_default = True value = meta_option.value.default() else: try: is_default = meta_option.owner.isdefault() except LeadershipError: break except AttributeOptionError: break try: value = meta_option.value.get() except ValueError as err: if not is_default: meta_option._set_subconfig() self.errors.append({str(err): meta_option._subconfig}) break except Exception as err: break true_default = is_default and meta_option.owner.get() == default_owner added = True meta_config = new_meta_config if leader_index is not None: if len(value) > leader_index: value = value[leader_index] else: value = None return values def convert_value( self, option, value: Any, ) -> str: """Dump variable, means transform bool, ... to yaml string""" if value is not None and not self.show_secrets and option.type() == "password": return "*" * 10 if isinstance(value, str): return value with BytesIO() as ymlfh: self.yaml.dump(value, ymlfh) ret = ymlfh.getvalue().decode("utf-8").strip() if ret.endswith("..."): ret = ret[:-3].strip() return ret def get_metaconfig_with_default_value(self, meta_config, option): default_owner = option.owner.default() if default_owner == owners.default: if option.owner.get() == default_owner: return None if self.root_config: return self.root_config return self.config if not self.config_owner_is_path: while True: if meta_config.type() != "metaconfig": return None meta_config = meta_config.parent() if not meta_config.owner.isdefault(): break else: meta_config = self.root_config for child in default_owner.split(".")[1:]: meta_config = meta_config.config(child) return meta_config