rougail/src/rougail/user_data.py

"""
Silique (https://www.silique.fr)
Copyright (C) 2022-2026

distribued with GPL-2 or later license

This program is free software; you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation; either version 2 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 General Public License for more details.

You should have received a copy of the GNU General Public License
along with this program; if not, write to the Free Software
Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
"""

from typing import List

from tiramisu import Calculation, Params, ParamValue, owners
from tiramisu.error import (
    PropertiesOptionError,
    AttributeOptionError,
    LeadershipError,
    ConfigError,
    CancelParam,
    display_list,
)
from .utils import undefined, get_properties_to_string
from .tiramisu import (
    normalize_family,
    CONVERT_OPTION,
    get_identifier_from_dynamic_family,
)
from .error import DictConsistencyError

from .i18n import _


class UserData:
    def __init__(self, config) -> None:
        self.config = config

    def user_data(
        self,
        user_data: List[dict],
        *,
        invalid_user_data_error: bool = False,
        unknown_user_data_error: bool = False,
    ):
        self.values = {}
        self.errors = []
        self.warnings = []
        self.invalid_user_data_error = invalid_user_data_error
        self.unknown_user_data_error = unknown_user_data_error
        if self.invalid_user_data_error:
            self.invalids = self.errors
        else:
            self.invalids = self.warnings
        if self.unknown_user_data_error:
            self.unknowns = self.errors
        else:
            self.unknowns = self.warnings
        self.show_secrets = False
        self._populate_values(user_data)
        self._auto_configure_dynamics()
        self._populate_config()
        self.properties_to_string = get_properties_to_string()
        self._populate_error_warnings()
        return {
            "errors": self.errors,
            "warnings": self.warnings,
        }

    def _populate_values(self, user_data):
        for datas in user_data:
            options = datas.get("options", {})
            source = datas["source"]
            for name, data in datas.get("values", {}).items():
                self.values[name] = {
                    "source": source,
                    "values": data,
                    "options": options.copy(),
                }
            self.errors.extend(datas.get("errors", []))
            self.warnings.extend(datas.get("warnings", []))

    def _get_variable(self, config):
        try:
            for subconfig in config:
                if subconfig.isoptiondescription():
                    yield from self._get_variable(subconfig)
                else:
                    yield subconfig
        except (ConfigError, ValueError) as err:
            err.prefix = ""
            if self.invalid_user_data_error:
                msg = str(err)
            else:
                msg = _('{0}, it will be ignored').format(err)
            self.invalids.append({msg: err.subconfig})

    def _auto_configure_dynamics(self):
        cache = {}
        added = []
        for path, data in list(self.values.items()):
            try:
                option = self.config.option(path)
                option.name()
            except (ConfigError, PropertiesOptionError):
                pass
            except AttributeError:
                self._not_found_is_dynamic(self.config, path, cache, added, data)

    def _not_found_is_dynamic(self, config, path, cache, added, data):
        """if path is not found, check if parent is a dynamic family"""
        current_path = ""
        identifiers = []
        # get parent
        for name in path.split(".")[:-1]:
            if current_path:
                current_path += "."
            current_path += name
            if current_path in cache:
                config, identifier = cache[current_path]
                identifiers.append(identifier)
                continue
            tconfig = config.option(name)
            try:
                tconfig.group_type()
                # object exists, so current config is the temporary config
                config = tconfig
                if config.isdynamic(only_self=True):
                    identifiers.append(config.identifiers()[-1])
            except AttributeError:
                # try to found the good dynamic family
                try:
                    lists = config.list(uncalculated=True)
                except PropertiesOptionError:
                    lists = []
                for tconfig in lists:
                    if not tconfig.isdynamic(only_self=True):
                        # it's not a dynamic variable
                        continue
                    identifier = get_identifier_from_dynamic_family(
                        tconfig.name(uncalculated=True), name
                    )
                    if identifier == "{{ identifier }}":
                        continue
                    if identifier != normalize_family(identifier):
                        msg = _(
                            'cannot load variable path "{0}", the identifier "{1}" is not valid in {2}'
                        ).format(path, identifier, data["source"])
                        self.invalids.append(msg)
                        continue
                    if identifier is None:
                        # it's a dynamic variable but doesn't match the current name
                        continue
                    # get the variable associate to this dynamic family
                    dynamic_variable = tconfig.information.get(
                        "dynamic_variable",
                        None,
                    )
                    if not dynamic_variable:
                        # it's the good dynamic variable but it's not linked to a variable
                        # so cannot change the variable
                        continue
                    option_type = self.config.option(dynamic_variable).information.get(
                        "type"
                    )
                    dyn_options_values = (
                        self.config.option(dynamic_variable).get().impl_getdefault()
                    )
                    if "{{ identifier }}" in dynamic_variable:
                        for s in identifiers:
                            dynamic_variable = dynamic_variable.replace(
                                "{{ identifier }}", str(s), 1
                            )
                    # do not add values in variable if has already a value
                    if dynamic_variable not in self.values and not dyn_options_values:
                        self.values[dynamic_variable] = {"values": []}
                        added.append(dynamic_variable)
                    elif dynamic_variable not in added:
                        continue
                    config = tconfig
                    identifiers.append(identifier)
                    typ = CONVERT_OPTION.get(option_type, {}).get("func")
                    if typ:
                        identifier = typ(identifier)
                    if identifier not in self.values[dynamic_variable]["values"]:
                        self.values[dynamic_variable]["values"].append(identifier)
                    cache[current_path] = config, identifier
                    break

    def convert_value(self, path, option, options, value):
        # converted value
        needs_convert = options.get("needs_convert", False)
        if option.ismulti():
            if options.get("multi_separator") and not isinstance(value, list):
                value = value.split(options["multi_separator"])
                self.values[path]["values"] = value
                if option.issubmulti():
                    value = [[val] for val in value]
            if option.issubmulti():
                for idx, val in enumerate(value):
                    if isinstance(val, list):
                        value[idx] = [convert_value(option, v, needs_convert) for v in val]
            elif isinstance(value, list):
                value = [convert_value(option, val, needs_convert) for val in value]
            if needs_convert:
                self.values[path]["values"] = value
                self.values[path]["options"]["needs_convert"] = needs_convert
        elif not isinstance(value, list):
            value = convert_value(option, value, needs_convert)
        return value

    def _populate_config(self):
        dynamics_variable = []
        while self.values:
            value_is_set = False
            for option in self._get_variable(self.config):
                values_path = path = option.path()
                if path not in self.values:
                    if path in dynamics_variable or not option.isdynamic():
                        continue
                    values_path = option.path(uncalculated=True)
                    if values_path not in self.values:
                        continue
                options = self.values[values_path].get("options", {})
                if (
                    options.get("allow_secrets_variables", True) is False
                    and option.type() == "password"
                ):
                    self.errors.append({
                        _(
                            'the variable contains secrets and should not be defined in {0}'
                            ).format(self.values[values_path]["source"]): option._subconfig}
                    )
                    self.values.pop(path)
                    continue
                value = self.convert_value(
                    path, option, options, self.values[values_path]["values"]
                )
                index = option.index()
                if index is not None:
                    if isinstance(value, tuple):
                        self.values[values_path]["values"] = []
                        #                        for i in range(len(option.parent().leader().value.get())):
                        for i in range(option.value.len()):
                            self.values[values_path]["values"].append(value)
                        value = self.values[values_path]["values"]
                    if not isinstance(value, list) or index >= len(value):
                        continue
                    value = value[index]
                    option_without_index = option.index(None)
                else:
                    option_without_index = option
                if option.isleader():
                    # set value for a leader, it began to remove all values!
                    len_leader = option.value.len()
                    if len_leader:
                        for idx in range(len_leader - 1, -1, -1):
                            option.value.pop(idx)
                try:
                    self.set_value(option, value, options)
                    value_is_set = True
                except Exception as err:
                    if path != option.path():
                        self.values[option.path()] = self.values.pop(values_path)
                else:
                    if "source" in self.values[values_path]:
                        option_without_index.information.set(
                            "loaded_from",
                            _("loaded from {0}").format(
                                self.values[values_path]["source"]
                            ),
                        )
                    # value is correctly set, remove variable to the set
                    if index is not None:
                        # if it's a follower waiting for all followers are sets
                        self.values[values_path]["values"][index] = undefined
                        for tmp_value in self.values[values_path]["values"]:
                            if tmp_value != undefined:
                                break
                        else:
                            if path in self.values:
                                self.values.pop(path)
                            else:
                                dynamics_variable.append(path)
                    elif path in self.values:
                        self.values.pop(path)
                    else:
                        dynamics_variable.append(path)
            if not value_is_set:
                break

    def _display_value(self, option, value):
        if not self.show_secrets and option.type() == "password":
            if not isinstance(value, list):
                value = "*" * 10
            else:
                value = ["*" * 10 for val in value]
        if isinstance(value, list):
            value = display_list(value, add_quote=True)
        else:
            value = '"' + value + '"'
        return value

    def _populate_error_warnings(self):
        # we don't find variable, apply value just to get error or warning messages
        for path, options in self.values.items():
            value = options["values"]
            if options.get("secret_manager"):
                option = self.config.forcepermissive.option(path)
            else:
                option = self.config.option(path)
            if option.issymlinkoption():
                err = _('it\'s a symlink option so we cannot set the value {0}').format(self._display_value(option, value))
                if self.invalid_user_data_error:
                    msg = _('{0}, it has been loading from {1}').format(err, options["source"])
                else:
                    msg = _('{0}, it will be ignored when loading from {1}').format(err, options["source"])
                self.unknowns.append({msg: option._subconfig})
                continue
            try:
                if option.isoptiondescription():
                    if value:
                        if self.invalid_user_data_error:
                            msg = _(
                                    'it\'s a family so we cannot set the value {0}, it has been loading from {1}'
                                    )
                        else:
                            msg = _(
                                    'it\'s a family so we cannot set the value {0}, it will be ignored when loading from {1}'
                                    )
                        self.invalids.append({msg.format(
                                self._display_value(option, value),
                                options["source"],
                                ): option._subconfig}
                        )
                    continue
            except ConfigError as err:
                self.invalids.append({
                    _("{0}, it has been loaded from {1}").format(err, options["source"]): option._subconfig}
                )
                continue
            except PropertiesOptionError as err:
                self.unknowns.append({
                    _("{0}, it has been loaded from {1}").format(err, options["source"]): option._subconfig}
                )
                continue
            except AttributeOptionError as err:
                if err.code == "option-not-found":
                    if self.unknown_user_data_error:
                        msg = _(
                            'variable or family "{0}" does not exist, it has been loading from {1}'
                            )
                    else:
                        msg = _(
                            'variable or family "{0}" does not exist, it will be ignored when loading from {1}'
                            )
                    self.unknowns.append(msg.format(err.path, options["source"]))
                elif err.code == "option-dynamic":
                    if self.invalid_user_data_error:
                        msg = _(
                        '"{0}" is the name of a dynamic family, it has been loading from {1}'
                        )
                    else:
                        msg = _(
                        '"{0}" is the name of a dynamic family, it will be ignored when loading from {1}'
                        )
                    self.invalids.append(
                            {msg.format(option.description(with_quote=True), options["source"]): option._subconfig}
                    )
                else:
                    self.invalids.append(
                        _("{0} loaded from {1}").format(err, options["source"])
                    )
                continue
            value = self.convert_value(
                path, option, self.values[path].get("options", {}), value
            )
            if option.isfollower():
                indexes = range(len(value))
                values = value
            else:
                indexes = [None]
            for index in indexes:
                try:
                    if option.isfollower():
                        value = values[index]
                        if value is undefined or isinstance(value, CancelParam):
                            continue
                        option.index(index).value.set(value)
                    else:
                        self.set_value(option, value, options.get("options", {}))
                except PropertiesOptionError as err:
                    if err.code == "property-error":
                        properties = display_list(
                            [_(prop) for prop in err.proptype], add_quote=False
                        )
                        err_path = err.subconfig.path
                        err_description = err.subconfig.option.impl_get_display_name(err.subconfig, with_quote=True)
                        display_name = option.description(with_quote=True)
                        if index is not None:
                            if path == err_path:
                                if self.unknown_user_data_error:
                                    msg = _(
                                        'variable {0} at index "{1}" is {2}, it has been loading from {3}'
                                    )
                                else:
                                    msg = _(
                                        'variable {0} at index "{1}" is {2}, it will be ignored when loading from {3}'
                                    )
                                self.unknowns.append({
                                    msg.format(
                                        display_name,
                                        index,
                                        properties,
                                        options["source"],
                                        ): option._subconfig}
                                )
                            else:
                                if self.unknown_user_data_error:
                                    msg = _(
                                        'family {0} is {1}, {2} at index "{3}", it has been loading from {4}'
                                    )
                                else:
                                    msg = _(
                                        'family {0} is {1}, {2} at index "{3}", it will be ignored when loading from {4}'
                                    )
                                self.unknowns.append({
                                    msg.format(
                                        err_description,
                                        properties,
                                        display_name,
                                        index,
                                        options["source"],
                                        ): option._subconfig}
                                )
                        else:
                            if path == err_path:
                                if self.unknown_user_data_error:
                                    msg = _(
                                        "variable has propery {0}, it has been loading from {1}"
                                    )
                                else:
                                    msg = _(
                                        "variable has property {0}, it will be ignored when loading from {1}"
                                    )
                                self.unknowns.append({
                                    msg.format(
                                        properties, options["source"]
                                        ): option._subconfig}
                                )
                            else:
                                if self.unknown_user_data_error:
                                    msg = _(
                                        "family {0} has property {1}, so cannot access to {2}, it has been loading from {3}"
                                    )
                                else:
                                    msg = _(
                                        "family {0} has property {1}, so cannot access to {2}, it will be ignored when loading from {3}"
                                    )
                                self.unknowns.append({
                                    msg.format(
                                        err_description,
                                        properties,
                                        display_name,
                                        options["source"],
                                        ): option._subconfig}
                                )
                    else:
                        if self.unknown_user_data_error:
                            msg = _(
                                "{0}, it has been loading from {1}"
                            )
                        else:
                            msg = _(
                                "{0}, it will be ignored when loading from {1}"
                            )
                        self.unknowns.append({
                            msg.format(err, options["source"]): option._subconfig}
                        )
                except LeadershipError as err:
                    if self.unknown_user_data_error:
                        msg = _(
                            "{0}, it has been loading from {1}"
                        )
                    else:
                        msg = _(
                            "{0}, it will be ignored when loading from {1}"
                        )
                    self.unknowns.append({
                        msg.format(err, options["source"]): option._subconfig}
                    )
                except ConfigError as err:
                    err.prefix = ""
                    if self.invalid_user_data_error:
                        msg = _('{0}, it has been loading from {1}').format(err, options["source"])
                    else:
                        msg = _('{0}, it will be ignored when loading from {1}').format(err, options["source"])
                    self.invalids.append({msg: option._subconfig})
                except ValueError as err:
                    err.prefix = ""
                    type_ = option.type(translation=True)
                    msg = _('the value {0} is an invalid {1}, {2}').format(
                            self._display_value(option, value),
                            type_,
                            err,
                            )
                    if self.invalid_user_data_error:
                        msg += _(', it has been loading from {0}').format(options["source"])
                    else:
                        msg += _(', it will be ignored when loading from {0}').format(options["source"])
                    self.invalids.append({msg: option._subconfig})
                except AttributeOptionError as err:
                    err.prefix = ""
                    if err.code == "option-dynamic":
                        continue
                    raise err from err

    def set_value(self, option, value, options):
        is_secret_manager = options.get("secret_manager", False)
        if is_secret_manager and isinstance(value, tuple):
            # it's a function
            params = tuple([ParamValue(val) for val in value[1:]])
            value = Calculation(value[0], Params(params, kwargs={"option": ParamValue(option)}))
            option = option.forcepermissive
            add_validation = True
        else:
            add_validation = False
        option.value.set(value)
        if add_validation:
            option.property.add("validator")


def convert_value(option, value, needs_convert):
    if value == "":
        return None
    option_type = option.information.get("type")
    if option_type == "port":
        func = CONVERT_OPTION[option_type]["func"]
        try:
            return func(value)
        except:
            return value
    if option_type == "choice":
        try:
            choices = option.value.list()
            if value not in choices and isinstance(value, str):
                # FIXME add other tests (boolean, float, ...)
                if value.isnumeric() and int(value) in choices:
                    value = int(value)
        except:
            pass
    func = CONVERT_OPTION.get(option_type, {}).get("func")
    if func:
        try:
            return func(value)
        except:
            pass
    return value