erplibre/script/execute/execute.py

#!/usr/bin/env python3
# © 2021-2026 TechnoLibre (http://www.technolibre.ca)
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl)

import datetime
import logging
import os
import re
import shutil
import subprocess
import sys
import time

try:
    import humanize
except ModuleNotFoundError as e:
    humanize = None

VENV_ERPLIBRE = ".venv.erplibre"

# Une commande construite ailleurs peut porter un secret en clair : todo.py et
# kdbx_manager.py y mettent « --default_password_auth '<mot de passe KeePass>' »,
# db_restore.py « --master_password=… ». Cette commande est affichée avant et
# après l'exécution, et journalisée en erreur : le secret finissait donc dans le
# terminal, dans les journaux et dans toute sortie CI qui les capture.
#
# On caviarde la VALEUR, jamais le nom de l'option : la commande reste lisible et
# reproductible, il ne manque que ce qui ne doit pas être lu.
_SECRET_OPTION = re.compile(
    r"(?P<opt>--?[\w-]*"
    r"(?:password|passwd|pwd|secret|token|api[-_]?key)[\w-]*"
    r"(?:\s+|=))"
    r"(?P<val>'[^']*'|\"[^\"]*\"|\S+)",
    re.IGNORECASE,
)
_SECRET_ENV = re.compile(
    r"(?P<var>\b\w*(?:PASSWORD|PASSWD|SECRET|TOKEN)\w*=)"
    r"(?P<val>'[^']*'|\"[^\"]*\"|\S+)"
)


def redact_secrets(text):
    """Remplace la valeur des options et variables porteuses de secret.

    Appliqué à CHAQUE affichage d'une commande. Filtrer au point d'affichage
    plutôt qu'à la construction est ce qui rend la garantie tenable : il n'y a
    qu'une poignée de sorties ici, alors que les commandes se construisent
    partout dans le dépôt.
    """
    if not text:
        return text
    text = _SECRET_OPTION.sub(lambda m: m.group("opt") + "'***'", text)
    return _SECRET_ENV.sub(lambda m: m.group("var") + "'***'", text)


new_path = os.path.normpath(
    os.path.join(os.path.dirname(__file__), "..", "..")
)
sys.path.append(new_path)


logging.basicConfig(
    format=(
        "%(asctime)s,%(msecs)d %(levelname)-8s [%(filename)s:%(lineno)d]"
        " %(message)s"
    ),
    datefmt="%Y-%m-%d:%H:%M:%S",
    level=logging.INFO,
)
_logger = logging.getLogger(__name__)


class Execute:
    def __init__(self) -> None:
        self.cmd_source_erplibre: str = ""
        self.cmd_source_default: str = ""
        exec_path_gnome_terminal = shutil.which("gnome-terminal")
        if exec_path_gnome_terminal:
            self.cmd_source_erplibre = (
                f"gnome-terminal -- bash -c 'source"
                f" ./{VENV_ERPLIBRE}/bin/activate;%s'"
            )
            self.cmd_source_default = "gnome-terminal -- bash -c '" f"%s'"
        else:
            exec_path_tell = shutil.which("osascript")
            if exec_path_tell:
                self.cmd_source_erplibre = (
                    "osascript -e 'tell application \"Terminal\"'"
                )
                self.cmd_source_erplibre += " -e 'tell application \"System Events\" to keystroke \"t\" using {command down}' -e 'delay 0.1' -e 'do script \""
                self.cmd_source_erplibre += f"cd {os.getcwd()}; source ./{VENV_ERPLIBRE}/bin/activate; %s\" in front window'"
                self.cmd_source_erplibre += " -e 'end tell'"
            else:
                self.cmd_source_erplibre = (
                    f"source ./{VENV_ERPLIBRE}/bin/activate;%s"
                )

    def exec_command_live(
        self,
        command: str,
        source_erplibre: bool = True,
        quiet: bool = False,
        single_source_erplibre: bool = False,
        new_window: bool = False,
        single_source_odoo: bool = False,
        source_odoo: str = "",
        new_env: dict | None = None,
        return_status_and_command: bool = False,
        return_status_and_output: bool = False,
        return_status_and_output_and_command: bool = False,
    ) -> (
        int
        | tuple[int, str]
        | tuple[int, list[str]]
        | tuple[int, str, list[str]]
    ):
        """
        Execute a command and display its output live.

        Args:
            command (str): The command to execute.
        """

        my_env = os.environ.copy()
        if new_env:
            my_env.update(new_env)

        process_start_time = time.time()
        exit_code = None
        if source_erplibre:
            # command = f"source ./{VENV_ERPLIBRE}/bin/activate && " + command
            # cmd = (
            #     f"gnome-terminal --tab -- bash -c 'source"
            #     f" ./{VENV_ERPLIBRE}/bin/activate;{command}'"
            # )
            command = self.cmd_source_erplibre % command
            # os.system(f"./script/terminal/open_terminal.sh {command}")
        elif single_source_erplibre:
            command = f"source ./{VENV_ERPLIBRE}/bin/activate && %s" % command
        elif single_source_odoo:
            if not source_odoo and os.path.exists("./.erplibre-version"):
                with open("./.erplibre-version") as f:
                    source_odoo = f.read()
            if not source_odoo:
                _logger.error(
                    "You cannot execute Odoo command if no version is"
                    f" installed. Command : {redact_secrets(command)}"
                )
                return -1
            command = f"source ./.venv.{source_odoo}/bin/activate && {command}"
        if new_window and self.cmd_source_default:
            command = self.cmd_source_default % command

        if not quiet:
            print("🏠 ⬇ Execute command :\n")
            print(redact_secrets(command))
        output_lines = []

        try:
            process = subprocess.Popen(
                command,
                shell=True,
                executable="/bin/bash",
                stdout=subprocess.PIPE,
                stderr=subprocess.STDOUT,
                text=True,
                bufsize=1,  # Disable buffering for live output
                universal_newlines=True,  # Handle line breaks correctly
                env=my_env,
            )

            while True:
                line = process.stdout.readline()
                if not line:
                    break
                if not quiet:
                    print(line, end="")
                if (
                    return_status_and_output
                    or return_status_and_output_and_command
                ):
                    # Remove last \n char
                    output_lines.append(
                        line.removesuffix("\r\n")
                        .removesuffix("\n")
                        .removesuffix("\r")
                    )

            process.wait()
            exit_code = process.returncode
            if process.returncode != 0 and not quiet:
                print("Command returned error code:" f" {process.returncode}")

        except FileNotFoundError:
            if not quiet:
                print(f"Error: Command '{redact_secrets(command)}' not found.")
        except Exception as e:
            if not quiet:
                print(f"An error occurred: {redact_secrets(str(e))}")
        process_end_time = time.time()
        duration_sec = process_end_time - process_start_time
        if humanize:
            duration_delta = datetime.timedelta(seconds=duration_sec)
            human_time = humanize.precisedelta(duration_delta)
            if not quiet:
                print(f"🏠 ⬆ Executed ({human_time}) :\n")
        else:
            if not quiet:
                print(f"🏠 ⬆ Executed ({duration_sec:.2f} sec.) :\n")
        if not quiet:
            print(redact_secrets(command))
            print()
        if return_status_and_output_and_command:
            return exit_code, command, output_lines
        if return_status_and_command:
            return exit_code, command
        if return_status_and_output:
            return exit_code, output_lines
        return exit_code