611 lines
21 KiB
Python
611 lines
21 KiB
Python
"""
|
|
Silique (https://www.silique.fr)
|
|
Copyright (C) 2022-2025
|
|
|
|
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 <http://www.gnu.org/licenses/>.
|
|
"""
|
|
|
|
from typing import Any, List, Optional
|
|
from io import BytesIO
|
|
|
|
from rich.tree import Tree
|
|
from rich.console import Console
|
|
from rich.table import Table
|
|
from rich.panel import Panel
|
|
from ruamel.yaml import YAML
|
|
|
|
from tiramisu import owners
|
|
from tiramisu.error import PropertiesOptionError, ConfigError
|
|
from rougail.utils import undefined
|
|
|
|
from .i18n import _
|
|
from .__version__ import __version__
|
|
|
|
|
|
class RougailOutputConsole:
|
|
variable_hidden_color = "orange1"
|
|
value_unmodified_color = "gold1"
|
|
value_modified_color = "green"
|
|
value_default_color = None
|
|
|
|
def __init__(
|
|
self,
|
|
config: "Config",
|
|
metaconfig=None,
|
|
*,
|
|
rougailconfig: "RougailConfig" = None,
|
|
user_data_errors: Optional[list] = None,
|
|
user_data_warnings: Optional[list] = None,
|
|
config_owner_is_path: bool = False,
|
|
layer_datas=None,
|
|
**kwargs,
|
|
) -> None:
|
|
if rougailconfig is None:
|
|
from rougail import RougailConfig
|
|
rougailconfig = RougailConfig
|
|
self.rougailconfig = rougailconfig
|
|
self.config = config
|
|
self.metaconfig = metaconfig
|
|
self.layer_datas = layer_datas
|
|
self.config_owner_is_path = config_owner_is_path
|
|
self.is_mandatory = self.rougailconfig["console.mandatory"]
|
|
self.show_secrets = self.rougailconfig["console.show_secrets"]
|
|
self.max_width = self.rougailconfig["console.max_width"]
|
|
self.key_is_description = self.rougailconfig["console.key_is_description"]
|
|
self.variable_default_enable = False
|
|
self.variable_hidden_enable = False
|
|
self.value_modified_enable = False
|
|
self.value_unmodified_enable = False
|
|
self.value_default_enable = False
|
|
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.console = Console(force_terminal=True, width=self.max_width)
|
|
self.out = []
|
|
self.root = self.get_root()
|
|
|
|
def mandatory(self):
|
|
try:
|
|
mandatories = self.config.value.mandatory()
|
|
except (ConfigError, PropertiesOptionError) as err:
|
|
self.errors.append(_("Error in config: {0}").format(err))
|
|
return
|
|
except ValueError as err:
|
|
self.errors.append(str(err))
|
|
return
|
|
options_with_error = []
|
|
options = {}
|
|
for option in mandatories:
|
|
try:
|
|
option.value.get()
|
|
except PropertiesOptionError:
|
|
options_with_error.append(option)
|
|
else:
|
|
parent = self.config
|
|
current_options = options
|
|
for child_path in option.path().split("."):
|
|
child = parent.option(child_path)
|
|
if self.key_is_description:
|
|
description = child.description(uncalculated=True)
|
|
else:
|
|
description = child.path(uncalculated=True)
|
|
if child.isdynamic() and "{{ identifier }}" in description:
|
|
description = description.replace(
|
|
"{{ identifier }}", str(child.identifiers()[-1])
|
|
)
|
|
|
|
if child.isoptiondescription():
|
|
current_options = current_options.setdefault(description, {})
|
|
parent = child
|
|
else:
|
|
current_options.setdefault(None, []).append(description)
|
|
if options:
|
|
self.errors.append(
|
|
{_("The following variables are mandatory but have no value:"): options}
|
|
)
|
|
elif options_with_error:
|
|
if self.key_is_description:
|
|
variables = [option.description() for option in options_with_error]
|
|
else:
|
|
variables = [option.path() for option in options_with_error]
|
|
self.errors.append(
|
|
{
|
|
_(
|
|
"The following variables are inaccessible but are empty and mandatory:"
|
|
): variables
|
|
}
|
|
)
|
|
|
|
def exporter(self) -> bool:
|
|
if self.is_mandatory:
|
|
ori_properties = self.config.property.exportation()
|
|
self.config.property.read_write()
|
|
if not self.user_data_errors and not self.errors:
|
|
self.mandatory()
|
|
self.config.property.importation(ori_properties)
|
|
warnings = self.user_data_warnings + self.warnings
|
|
if warnings:
|
|
self.display_warnings(warnings)
|
|
|
|
errors = self.user_data_errors + self.errors
|
|
if errors:
|
|
self.display_errors(errors)
|
|
return False
|
|
old_path_in_description = self.config.information.get(
|
|
"path_in_description", True
|
|
)
|
|
self.config.information.set("path_in_description", False)
|
|
self.parse_options(
|
|
self.config,
|
|
self.root,
|
|
)
|
|
self.config.information.set("path_in_description", old_path_in_description)
|
|
if self.output.tree and self.output.tree.children:
|
|
self.header()
|
|
self.end()
|
|
return True
|
|
|
|
def run(self) -> str:
|
|
with self.console.capture() as capture:
|
|
ret = self.print()
|
|
return ret, capture.get()
|
|
|
|
def print(self) -> None:
|
|
ret = self.exporter()
|
|
for out in self.out:
|
|
self.console.print(out)
|
|
return ret
|
|
|
|
def parse_options(
|
|
self,
|
|
conf,
|
|
parent,
|
|
):
|
|
for option in conf:
|
|
if option.isoptiondescription():
|
|
family = parent.add_family(option)
|
|
if option.isleadership():
|
|
self.parse_leadership(
|
|
option,
|
|
family,
|
|
)
|
|
else:
|
|
self.parse_options(
|
|
option,
|
|
family,
|
|
)
|
|
else:
|
|
parent.add_variable(option)
|
|
|
|
def parse_leadership(
|
|
self,
|
|
conf,
|
|
parent,
|
|
):
|
|
leader, *followers = list(conf)
|
|
leader_values = leader.value.get()
|
|
for idx, leader_value in enumerate(leader_values):
|
|
leader_obj = parent.add_family(leader)
|
|
leader_obj.add_variable(
|
|
leader,
|
|
value=leader_value,
|
|
leader_index=idx,
|
|
)
|
|
for follower in followers:
|
|
if follower.index() != idx:
|
|
continue
|
|
leader_obj.add_variable(follower)
|
|
|
|
def header(self):
|
|
caption_line = ""
|
|
if self.variable_default_enable:
|
|
caption_line += _("Variable") + "\n"
|
|
if self.variable_hidden_enable:
|
|
caption_line += f'[{self.variable_hidden_color}]{_("Unmodifiable variable")}[/{self.variable_hidden_color}]\n'
|
|
header_value = ""
|
|
if self.value_unmodified_enable:
|
|
header_value = f'[{self.value_unmodified_color}]{_("Default value")}[/{self.value_unmodified_color}]\n'
|
|
if self.value_modified_enable:
|
|
header_value += f'[{self.value_modified_color}]{_("Modified value")}[/{self.value_modified_color}]\n'
|
|
if self.value_default_enable:
|
|
header_value += f'(:hourglass_flowing_sand: {_("Original default value")})\n'
|
|
caption = Table.grid(padding=1, collapse_padding=True)
|
|
caption.pad_edge = False
|
|
caption.add_row(caption_line[:-1], header_value[:-1])
|
|
self.out.append(Panel.fit(caption, title=_("Caption")))
|
|
#
|
|
layers = Table.grid(padding=1, collapse_padding=True)
|
|
caption.pad_edge = False
|
|
if self.layer_datas:
|
|
max_len = 0
|
|
for datas in self.layer_datas.values():
|
|
for data in datas.values():
|
|
max_len = max(max_len, len(data))
|
|
display_layers = ["" for i in range(max_len)]
|
|
for datas in self.layer_datas.values():
|
|
for data in datas.values():
|
|
last_index = len(data) - 1
|
|
for idx in range(max_len):
|
|
if last_index < idx:
|
|
display_layers[idx] += "\n"
|
|
else:
|
|
display_layers[idx] += data[idx] + "\n"
|
|
layers.add_row(*[layer[:-1] for layer in display_layers])
|
|
self.out.append(Panel.fit(layers, title=_("Layers")))
|
|
|
|
def display_errors(
|
|
self,
|
|
errors,
|
|
) -> None:
|
|
tree = Tree(
|
|
f"[bold][bright_red]:stop_sign: {_('ERRORS')}[/bright_red][/bold]",
|
|
guide_style="bold bright_red",
|
|
)
|
|
for error in errors:
|
|
self.display_error(tree, error)
|
|
self.out.append(tree)
|
|
|
|
def display_error(self, tree, error):
|
|
if isinstance(error, list):
|
|
for err in error:
|
|
self.display_error(tree, err)
|
|
return tree
|
|
elif isinstance(error, dict):
|
|
for key, value in error.items():
|
|
if key is None:
|
|
# it's variables, no more families
|
|
self.display_error(tree, value)
|
|
else:
|
|
sub_tree = self.display_error(tree, key)
|
|
self.display_error(sub_tree, value)
|
|
else:
|
|
return tree.add(error)
|
|
|
|
def display_warnings(
|
|
self,
|
|
warnings: list,
|
|
) -> None:
|
|
tree = Tree(
|
|
f"[bold][bright_yellow]:bell: {_('WARNINGS')}[/bright_yellow][/bold]",
|
|
guide_style="bold bright_yellow",
|
|
)
|
|
for warning in warnings:
|
|
tree.add(warning)
|
|
self.out.append(tree)
|
|
|
|
def get_root(self) -> None:
|
|
yaml = YAML()
|
|
yaml.indent(mapping=2, sequence=4, offset=2)
|
|
self.output = OutputFamily(
|
|
_("Variables:"),
|
|
None,
|
|
self,
|
|
yaml,
|
|
self.key_is_description,
|
|
no_icon=True,
|
|
)
|
|
return self.output
|
|
|
|
def end(self):
|
|
self.out.append(self.output.tree)
|
|
|
|
|
|
class OutputFamily:
|
|
def __init__(
|
|
self,
|
|
family,
|
|
parent,
|
|
root,
|
|
_yaml,
|
|
key_is_description,
|
|
*,
|
|
is_leader: bool = False,
|
|
no_icon: bool = False,
|
|
) -> None:
|
|
self.tree = None
|
|
self.root = root
|
|
self.parent = parent
|
|
self.family = family
|
|
self.is_leader = is_leader
|
|
self.no_icon = no_icon
|
|
self._yaml = _yaml
|
|
self.key_is_description = key_is_description
|
|
|
|
def get_tree(self):
|
|
if self.tree is None:
|
|
if self.parent is None:
|
|
tree = Tree
|
|
else:
|
|
tree = self.parent.add
|
|
if self.is_leader:
|
|
self.tree = tree(
|
|
":notebook: " + _("{0}:").format(self.family),
|
|
guide_style="bold bright_blue",
|
|
)
|
|
elif self.no_icon:
|
|
self.tree = tree(
|
|
self.family,
|
|
guide_style="bold bright_blue",
|
|
)
|
|
else:
|
|
self.tree = tree(
|
|
f":open_file_folder: {self.family}",
|
|
guide_style="bold bright_blue",
|
|
)
|
|
return self.tree
|
|
|
|
def add_family(
|
|
self,
|
|
option,
|
|
) -> 'OutputFamily':
|
|
properties = option.property.get()
|
|
if "hidden" in properties:
|
|
self.root.variable_hidden_enable = True
|
|
color = self.root.variable_hidden_color
|
|
else:
|
|
self.root.variable_default_enable = True
|
|
color = None
|
|
if self.key_is_description:
|
|
key_name = option.description()
|
|
else:
|
|
key_name = option.name()
|
|
return OutputFamily(
|
|
self.colorize(
|
|
[
|
|
{
|
|
"value": key_name,
|
|
"color": color,
|
|
"loaded_from": None,
|
|
}
|
|
]
|
|
),
|
|
self.get_tree(),
|
|
self.root,
|
|
self._yaml,
|
|
self.key_is_description,
|
|
)
|
|
|
|
def add_variable(
|
|
self, option, value: Any = undefined, leader_index: Optional[int] = None
|
|
):
|
|
# value is undefined but not for leader variable
|
|
properties = option.property.get()
|
|
variable_color = None
|
|
hidden = False
|
|
if "hidden" in properties:
|
|
self.root.variable_hidden_enable = True
|
|
variable_color = self.root.variable_hidden_color
|
|
else:
|
|
self.root.variable_default_enable = True
|
|
values = []
|
|
collect_values = []
|
|
if not option.owner.isdefault():
|
|
self.root.value_modified_enable = True
|
|
follower_index = option.index()
|
|
if follower_index is not None:
|
|
loaded_from = option.information.get(
|
|
f"loaded_from_{follower_index}", None
|
|
)
|
|
else:
|
|
loaded_from = option.information.get("loaded_from", None)
|
|
if value is undefined:
|
|
value = option.value.get()
|
|
values.append(
|
|
{
|
|
"value": value,
|
|
"color": self.root.value_modified_color,
|
|
"loaded_from": loaded_from,
|
|
}
|
|
)
|
|
suboption = option
|
|
subconfig = self.root.config
|
|
if "force_store_value" not in option.property.get() and option.information.get(
|
|
"default_value_makes_sense", True
|
|
):
|
|
while True:
|
|
default_value = suboption.value.default()
|
|
if leader_index is not None:
|
|
if len(default_value) > leader_index:
|
|
default_value = default_value[leader_index]
|
|
else:
|
|
default_value = None
|
|
is_root_metaconfig = False
|
|
if (
|
|
subconfig is None
|
|
or not subconfig.path()
|
|
or "." not in subconfig.path()
|
|
):
|
|
is_root_metaconfig = True
|
|
subconfig = self.get_subconfig_with_default_value(suboption)
|
|
if (
|
|
(values and default_value is None) or default_value == []
|
|
) and is_root_metaconfig:
|
|
break
|
|
if not values:
|
|
self.root.value_unmodified_enable = True
|
|
color = self.root.value_unmodified_color
|
|
else:
|
|
self.root.value_default_enable = True
|
|
color = self.root.value_default_color
|
|
index = option.index()
|
|
parent_option = subconfig.option(option.path(), index)
|
|
if is_root_metaconfig:
|
|
loaded_from = None
|
|
else:
|
|
parent_option = subconfig.option(option.path(), index)
|
|
if index is not None:
|
|
key = f"loaded_from_{index}"
|
|
else:
|
|
key = "loaded_from"
|
|
loaded_from = parent_option.information.get(key, None)
|
|
values.append(
|
|
{
|
|
"value": default_value,
|
|
"color": color,
|
|
"loaded_from": loaded_from,
|
|
}
|
|
)
|
|
if is_root_metaconfig:
|
|
break
|
|
elif not values:
|
|
if leader_index is None and option.index() is not None:
|
|
if not option.isfollower() or not option.issubmulti():
|
|
default_value = None
|
|
else:
|
|
default_value = []
|
|
else:
|
|
default_value = None
|
|
if not values:
|
|
self.root.value_unmodified_enable = True
|
|
color = self.root.value_unmodified_color
|
|
else:
|
|
self.root.value_default_enable = True
|
|
color = self.root.value_default_color
|
|
follower_index = option.index()
|
|
if follower_index is not None:
|
|
loaded_from = option.information.get(
|
|
f"loaded_from_{follower_index}", None
|
|
)
|
|
else:
|
|
loaded_from = option.information.get("loaded_from", None)
|
|
values.append(
|
|
{
|
|
"value": None,
|
|
"color": color,
|
|
"loaded_from": loaded_from,
|
|
}
|
|
)
|
|
if self.key_is_description:
|
|
key_name = option.description()
|
|
else:
|
|
key_name = option.name()
|
|
key = self.colorize(
|
|
[
|
|
{
|
|
"value": key_name,
|
|
"color": variable_color,
|
|
"loaded_from": None,
|
|
}
|
|
]
|
|
)
|
|
value = self.colorize(
|
|
values,
|
|
option,
|
|
)
|
|
if isinstance(value, list):
|
|
subtree = self.get_tree().add(
|
|
":notebook: " + _("{0}:").format(key),
|
|
guide_style="bold bright_blue",
|
|
)
|
|
for val in value:
|
|
subtree.add(str(val))
|
|
else:
|
|
self.get_tree().add(":notebook: " + _("{0}: {1}").format(key, value))
|
|
|
|
def get_subconfig_with_default_value(self, suboption):
|
|
default_owner = suboption.owner.default()
|
|
if default_owner == owners.default:
|
|
return self.root.config
|
|
if not self.root.config_owner_is_path:
|
|
subconfig = self.root.config
|
|
while True:
|
|
subconfig = subconfig.parent()
|
|
if not subconfig.owner.isdefault():
|
|
break
|
|
else:
|
|
subconfig = self.root.metaconfig
|
|
for child in default_owner.split(".")[1:]:
|
|
subconfig = subconfig.config(child)
|
|
return subconfig
|
|
|
|
def colorize(
|
|
self,
|
|
values,
|
|
option=None,
|
|
) -> str:
|
|
multi = False
|
|
ret = []
|
|
default = []
|
|
for idx, data in enumerate(values):
|
|
value = data["value"]
|
|
if isinstance(value, list):
|
|
multi = True
|
|
else:
|
|
value = [value]
|
|
for vidx, val in enumerate(value):
|
|
if len(ret) == vidx:
|
|
ret.append("")
|
|
default.append(False)
|
|
if idx:
|
|
if not default[vidx]:
|
|
if ret[vidx]:
|
|
ret[vidx] += " "
|
|
ret[vidx] += "("
|
|
default[vidx] = True
|
|
else:
|
|
ret[vidx] += " "
|
|
ret[vidx] += ":hourglass_flowing_sand: "
|
|
if option:
|
|
val = self.convert_value(
|
|
option,
|
|
val,
|
|
)
|
|
color = data["color"]
|
|
if color is not None:
|
|
ret[vidx] += f"[{color}]{val}[/{color}]"
|
|
else:
|
|
ret[vidx] += val
|
|
loaded_from = data["loaded_from"]
|
|
if loaded_from:
|
|
ret[vidx] += f" :arrow_backward: {loaded_from}"
|
|
for idx in range(len(ret)):
|
|
if default[idx]:
|
|
ret[idx] += ")"
|
|
if not multi:
|
|
if not ret:
|
|
return ""
|
|
return ret[0]
|
|
return ret
|
|
|
|
def convert_value(
|
|
self,
|
|
option,
|
|
value: Any,
|
|
) -> str:
|
|
"""Dump variable, means transform bool, ... to yaml string"""
|
|
if (
|
|
value is not None
|
|
and not self.root.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
|
|
|
|
|
|
RougailOutput = RougailOutputConsole
|
|
|
|
|
|
__all__ = ("RougailOutputConsole",)
|