rougail-cli/src/rougail/cli/__main__.py

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

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 3 of the License, or
(at your option) any later version.

Mtools 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 Mtools.  If not, see <http://www.gnu.org/licenses/>.
"""

import os
from warnings import filterwarnings
from pathlib import Path
from sys import exit
from warnings import warn

from tiramisu_cmdline_parser import TiramisuCmdlineParser
from tiramisu.error import PropertiesOptionError
from tiramisu import MetaConfig

from rougail import Rougail
from rougail.user_data import UserData, mandatories
from rougail.config import get_rougail_config
from rougail.utils import load_modules
from rougail.error import RougailWarning

try:
    from rougail.user_data_yaml import RougailUserDataYaml
except ImportError:
    RougailUserDataYaml = None
try:
    from rougail.user_data_environment import RougailUserDataEnvironment
except ImportError:
    RougailUserDataEnvironment = None
from .rougailconfig import load as rougailconfig_load

from .i18n import _


ENV_PREFIX = "ROUGAILCLI"


def _main(arguments, do_not_print):
    global print_traceback
    rougailconfig = get_rougail_config(
        backward_compatibility=False, add_extra_options=False
    )
    cmd_config = load_cmd_user_data(rougailconfig, arguments)
    print_traceback = rougailconfig["cli.debug"]
    if rougailconfig["cli.versions"]:
        versions = display_version(cmd_config)
        if do_not_print:
            return list(versions)
        for version in versions:
            print(version)
        exit()
    layer_datas, metaconfig, config, err_warn = load_user_data(rougailconfig)
    if config and (not rougailconfig["cli.load_config"] or not rougailconfig["cli.read_write"]):
        config.property.read_only()
    if config and rougailconfig["cli.mandatory"]:
        mandatories(config, err_warn["errors"])
    output = get_output(rougailconfig,
                        metaconfig,
                        config,
                        err_warn,
                        layer_datas,
                        )
    if do_not_print:
        return output.run()
    ret = output.print()
    if ret is False:
        exit(1)


def load_cmd_user_data(rougailconfig, arguments):
    rougailconfig.generate_config()
    cmd_config = rougailconfig.config
    cmd_config.information.set("description_type", "path_and_description")
    origin_prop = cmd_config.property.default("read_write", "append")
    cmd_config.property.setdefault(
        frozenset(origin_prop | {"not_for_commandline"}), "read_write", "append"
    )
    cmd_config.property.read_write()
    config_files = None
    if RougailUserDataYaml:
        _config_files = os.environ.pop(f"{ENV_PREFIX}_CLI.CONFIG_FILE", None)
        if _config_files:
            _config_files = _config_files.split(',')
        else:
            _config_files = cmd_config.forcepermissive.option("cli.config_file").value.get()
        _config_files = [_config_file for _config_file in _config_files if Path(_config_file).is_file()]
        if _config_files:
            config_files = _config_files
    if RougailUserDataEnvironment:
        env_prefix = ENV_PREFIX
    else:
        env_prefix = None
    user_data = rougailconfig_load(rougailconfig, config_files, env_prefix, True, _arguments=arguments, _generate=False, _add_help=False)
    display_warnings = rougailconfig["cli.warnings"]
    manage_warnings(display_warnings)
    if not cmd_config.option("cli.versions").value.get():
        if display_warnings and user_data["warnings"]:
            for warning in user_data["warnings"]:
                if isinstance(warning, dict):
                    for w, var in warning.items():
                        if var:
                            warn(UserWarning(_('unable to load {0}, {1}').format(var.option.impl_get_display_name(var, with_quote=True), w)))
                        else:
                            warn(w)
                else:
                    warn(warning)
        if manage_warnings:
            # replays to display errors if needed
            parser = TiramisuCmdlineParser(
                cmd_config,
                short_name_max_len=2,
            )
            parser.parse_args(arguments)
        if user_data["errors"]:
            raise Exception(user_data["errors"][0])

    cmd_config.property.setdefault(origin_prop, "read_write", "append")
    cmd_config.property.read_only()
    cmd_config.property.remove("not_for_commandline")

    return cmd_config


def display_version(cmd_config):
    versions = []
    from tiramisu import __version__
    yield(f"tiramisu: {__version__}")

    from tiramisu_cmdline_parser import __version__
    yield(f"tiramisu-cmdline-parser: {__version__}")

    from rougail import __version__
    yield(f"rougail: {__version__}")

    from . import __version__
    yield(f"rougail-cli: {__version__}")

    for step in ["structural", "user_data", "output"]:
        display_step = step.replace("_", "-")
        for step_name in sorted(cmd_config.unrestraint.option(f"step.{step}").value.list()):
            path = (
                Path(__file__).parent.parent
                / (step + "_" + step_name)
                / "__init__.py"
            )
            if path.is_file():
                try:
                    module = load_modules(
                        "rougail." + step + "_" + step_name, str(path)
                    )
                    yield(
                        f"rougail-{display_step}-{step_name}: {module.__version__}"
                    )
                except Exception as err:
                    pass


def manage_warnings(warnings):
    if not warnings:
        filterwarnings("ignore", category=DeprecationWarning)
        filterwarnings("ignore", category=RougailWarning)
    else:
        filterwarnings("default", category=DeprecationWarning)
        filterwarnings("default", category=RougailWarning)


def load_user_data(rougailconfig):
    layer_datas = []
    if not rougailconfig["cli.load_config"]:
        return None, None, None, {"errors": [], "warnings": []}
    try:
        user_data_names = rougailconfig["step.user_data"]
    except PropertiesOptionError:
        user_data_names = []
    if rougailconfig["tiramisu_cache"]:
        load_from_tiramisu_cache = rougailconfig["cli.load_from_tiramisu_cache"]
    else:
        load_from_tiramisu_cache = False
    rougail = Rougail(rougailconfig, load_from_tiramisu_cache=load_from_tiramisu_cache)
    has_layers = rougailconfig["cli.layers"]
    last_layers = len(user_data_names) - 1
    if has_layers and user_data_names:
        layer_name = user_data_names[-1]
    else:
        layer_name = None
    subconfig = metaconfig = rougail.run(name=layer_name)
    try:
        read_write = set()
        read_only = set()
        if rougailconfig["cli.inaccessible_read_write_modes"]:
            read_write = set(rougailconfig["cli.inaccessible_read_write_modes"])
        if rougailconfig["cli.inaccessible_read_only_modes"]:
            read_only = set(rougailconfig["cli.inaccessible_read_only_modes"])
        if rougailconfig["cli.inaccessible_modes"]:
            modes = set(rougailconfig["cli.inaccessible_modes"])
            read_only |= modes
            read_write |= modes
        if read_write:
            metaconfig.property.setdefault(
                frozenset(metaconfig.property.default("read_write", "append") | read_write), "read_write", "append"
            )
            metaconfig.property.setdefault(
                frozenset(metaconfig.property.default("read_only", "remove") | (read_write - read_only)), "read_only", "remove"
            )
            for p in read_write:
                metaconfig.permissive.add(p)
        if read_only:
            metaconfig.property.setdefault(
                frozenset(metaconfig.property.default("read_only", "append") | read_only), "read_only", "append"
            )
            for p in read_only:
                metaconfig.permissive.add(p)
        if read_write or read_only:
            metaconfig.property.read_write()
    except:
        pass
    metaconfig.information.set("description_type", rougailconfig["cli.description_type"])
    rougail_user_datas = {}
    interactive_user_datas = {}
    for ud_idx, user_data_name in enumerate(reversed(user_data_names)):
        path = (
            Path(__file__).parent.parent
            / ("user_data_" + user_data_name)
            / "__init__.py"
        )
        if not path.is_file():
            raise Exception(
                _('cannot find "user_data" module "{0}"').format(user_data_name)
            )
        module = load_modules("rougail.user_data_" + user_data_name, str(path))
        rougail_user_data = module.RougailUserData
        has_several_layers = hasattr(rougail_user_data, 'has_several_layers') and rougail_user_data.has_several_layers
        if has_layers and has_several_layers:
            layers_len = rougail_user_data(None, rougailconfig=rougailconfig).count_layers()
        else:
            layers_len = 1
        for idx in range(layers_len):
            if ud_idx and has_layers:
                metaconfig = MetaConfig([metaconfig], name=user_data_name)
            if not idx:
                has_interactive_user_data = hasattr(rougail_user_data, 'interactive_user_data') and rougail_user_data.interactive_user_data
                if has_interactive_user_data:
                    interactive_user_datas[user_data_name] = rougail_user_data
                else:
                    if interactive_user_datas:
                        raise Exception(_('interactive user data "{0}" is loader before uninteractive user data "{1}"').format(list(interactive_user_datas), user_data_name))
                    rougail_user_datas[user_data_name] = rougail_user_data
    root_metaconfig = metaconfig
    err_warn = {"errors": [], "warnings": []}
    invalid_user_data_error = rougailconfig["cli.invalid_user_data_error"]
    unknown_user_data_error = rougailconfig["cli.unknown_user_data_error"]
    user_datas = []
    for idx, user_data_name in enumerate(reversed(rougail_user_datas)):
        rougail_user_data = rougail_user_datas[user_data_name]
        if has_layers and idx:
            metaconfig = metaconfig.config(user_data_name)
            metaconfig.owner.set(metaconfig.path())
        for ud_idx, user_data in enumerate(rougail_user_data(
                    subconfig,
                    rougailconfig=rougailconfig,
                ).run()):
            if has_layers and ud_idx:
                metaconfig = metaconfig.config(user_data_name)
                metaconfig.owner.set(metaconfig.path())
            if has_layers:
                layer_datas.append(user_data["source"])
                new_err_warn = UserData(metaconfig).user_data([user_data], invalid_user_data_error=invalid_user_data_error, unknown_user_data_error=unknown_user_data_error)
                for level, datas in new_err_warn.items():
                    if datas:
                        err_warn[level].extend(datas)
            else:
                user_datas.append(user_data)
    if user_datas and not has_layers:
        new_err_warn = UserData(metaconfig).user_data(user_datas, invalid_user_data_error=invalid_user_data_error, unknown_user_data_error=unknown_user_data_error)
        for level, datas in new_err_warn.items():
            if datas:
                err_warn[level].extend(datas)
    user_datas = []
    for idx, user_data_name in enumerate(reversed(interactive_user_datas)):
        rougail_user_data = interactive_user_datas[user_data_name]
        if has_layers and idx:
            metaconfig = metaconfig.config(user_data_name)
            metaconfig.owner.set(metaconfig.path())
        for ud_idx, user_data in enumerate(rougail_user_data(
                    subconfig,
                    rougailconfig=rougailconfig,
                ).run()):
            if has_layers and ud_idx:
                metaconfig = metaconfig.config(user_data_name)
                metaconfig.owner.set(metaconfig.path())
            if has_layers:
                layer_datas.append(user_data["source"])
                new_err_warn = UserData(metaconfig).user_data([user_data], invalid_user_data_error=invalid_user_data_error, unknown_user_data_error=unknown_user_data_error)
                for level, datas in new_err_warn.items():
                    if datas:
                        err_warn[level].extend(datas)
            else:
                user_datas.append(user_data)
    if user_datas and not has_layers:
        new_err_warn = UserData(metaconfig).user_data(user_datas, invalid_user_data_error=invalid_user_data_error, unknown_user_data_error=unknown_user_data_error)
        for level, datas in new_err_warn.items():
            if datas:
                err_warn[level].extend(datas)
    return layer_datas, root_metaconfig, subconfig, err_warn


def get_output(rougailconfig, metaconfig, config, err_warn, layer_datas):
    # output
    output_name = rougailconfig["step.output"]
    path = Path(__file__).parent.parent / ("output_" + output_name) / "__init__.py"
    if not path.is_file():
        raise Exception(
            _('cannot find cli file for "output_name" module "{0}"').format(output_name)
        )
    module = load_modules("rougail.output_" + output_name, str(path))
    root = rougailconfig["cli.root"]
    if root:
        subconfig = config.option(root)
    else:
        subconfig = config
    output = module.RougailOutput(
        config=subconfig,
        true_config=config,
        rougailconfig=rougailconfig,
        user_data_errors=err_warn["errors"],
        user_data_warnings=err_warn["warnings"],
        config_owner_is_path=True,
        root_config=metaconfig,
        layer_datas=layer_datas,
    )
    return output


def main(arguments=None, do_not_print=False):
    global print_traceback
    print_traceback = True
    try:
        return _main(arguments, do_not_print)
    except Exception as err:
        if print_traceback:
            import traceback
            traceback.print_exc()
        exit(_("ERROR: {0}").format(err))