From 54f8ba0feb8af3f08b177898a6a1274980f41d1d Mon Sep 17 00:00:00 2001 From: Mathieu Benoit Date: Fri, 13 Feb 2026 01:27:38 -0500 Subject: [PATCH] [UPD] script execute to share exec_command_live --- script/database/image_db.py | 1 + script/execute/__init__.py | 0 script/execute/execute.py | 184 +++++++++++++++++++++++++++++++++ script/todo/todo.py | 201 ++++++------------------------------ script/todo/todo_upgrade.py | 6 +- 5 files changed, 218 insertions(+), 174 deletions(-) create mode 100644 script/execute/__init__.py create mode 100644 script/execute/execute.py diff --git a/script/database/image_db.py b/script/database/image_db.py index a7ae69f..757debb 100755 --- a/script/database/image_db.py +++ b/script/database/image_db.py @@ -359,6 +359,7 @@ def run_cmd(cmd, quiet=False, sys_exit=True): if not quiet: _logger.info(f"Run cmd: {cmd}") debut = time.time() + # Maybe can use exec_command_live process = subprocess.Popen( cmd, shell=True, diff --git a/script/execute/__init__.py b/script/execute/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/script/execute/execute.py b/script/execute/execute.py new file mode 100644 index 0000000..3774252 --- /dev/null +++ b/script/execute/execute.py @@ -0,0 +1,184 @@ +#!/usr/bin/env python3 +# © 2021-2025 TechnoLibre (http://www.technolibre.ca) +# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl) + +import datetime +import logging +import os +import shutil +import subprocess +import sys +import time + +import humanize + +cst_venv_erplibre = ".venv.erplibre" + +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): + self.cmd_source_erplibre = "" + self.cmd_source_default = "" + exec_path_gnome_terminal = shutil.which("gnome-terminal") + if exec_path_gnome_terminal: + self.cmd_source_erplibre = ( + f"gnome-terminal -- bash -c 'source" + f" ./{cst_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 ./{cst_venv_erplibre}/bin/activate; %s\" in front window'" + self.cmd_source_erplibre += " -e 'end tell'" + else: + self.cmd_source_erplibre = ( + f"source ./{cst_venv_erplibre}/bin/activate;%s" + ) + + def exec_command_live( + self, + commande, + source_erplibre=True, + quiet=False, + single_source_erplibre=False, + new_window=False, + single_source_odoo=False, + source_odoo="", + new_env=None, + return_status_and_command=False, + return_status_and_output=False, + return_status_and_output_and_command=False, + ): + """ + Exécute une commande et affiche la sortie en direct. + + Args: + commande (str): La commande à exécuter (sous forme de chaîne de caractères). + """ + + my_env = os.environ.copy() + if new_env: + my_env.update(new_env) + + process_start_time = time.time() + return_status = None + if source_erplibre: + # commande = f"source ./{cst_venv_erplibre}/bin/activate && " + commande + # cmd = ( + # f"gnome-terminal --tab -- bash -c 'source" + # f" ./{cst_venv_erplibre}/bin/activate;{commande}'" + # ) + commande = self.cmd_source_erplibre % commande + # os.system(f"./script/terminal/open_terminal.sh {commande}") + elif single_source_erplibre: + commande = ( + f"source ./{cst_venv_erplibre}/bin/activate && %s" % commande + ) + 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( + f"You cannot execute Odoo command if no version is installed. Command : {commande}" + ) + return -1 + commande = ( + f"source ./.venv.{source_odoo}/bin/activate && {commande}" + ) + elif new_window and self.cmd_source_default: + commande = self.cmd_source_default % commande + + if not quiet: + print("🏠 ⬇ Execute command :") + print(commande) + lst_output = [] + + try: + process = subprocess.Popen( + commande, + shell=True, + executable="/bin/bash", + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + bufsize=1, # Désactive la mise en tampon pour la sortie en direct + universal_newlines=True, # Pour traiter les sauts de lignes correctement + env=my_env, + ) + + while True: + ligne = process.stdout.readline() + if not ligne: + break + if not quiet: + print(ligne, end="") + if ( + return_status_and_output + or return_status_and_output_and_command + ): + lst_output.append(ligne) + + process.wait() # Attendre la fin du process + return_status = process.returncode + if process.returncode != 0 and not quiet: + print( + "La commande a retourné un code d'erreur :" + f" {process.returncode}" + ) + + except FileNotFoundError: + if not quiet: + if "password" in commande: + print( + f"Erreur : La commande '{commande.split(' ')[0]}'[...] n'a" + " pas été trouvée." + ) + else: + print( + f"Erreur : La commande '{commande}' n'a pas été trouvée." + ) + except Exception as e: + if not quiet: + print(f"Une erreur s'est produite : {e}") + process_end_time = time.time() + duration_sec = process_end_time - process_start_time + if humanize: + duration_delta = datetime.timedelta(seconds=duration_sec) + humain_time = humanize.precisedelta(duration_delta) + if not quiet: + print(f"🏠 ⬆ Executed ({humain_time}) :") + else: + if not quiet: + print(f"🏠 ⬆ Executed ({duration_sec:.2f} sec.) :") + if not quiet: + print(commande) + print() + if return_status_and_output_and_command: + return return_status, commande, lst_output + if return_status_and_command: + return return_status, commande + if return_status_and_output: + return return_status, lst_output + return return_status diff --git a/script/todo/todo.py b/script/todo/todo.py index 8bceb5c..a9f4b26 100755 --- a/script/todo/todo.py +++ b/script/todo/todo.py @@ -54,6 +54,8 @@ try: # import rich import todo_upgrade from pykeepass import PyKeePass + + from script.execute import execute except ModuleNotFoundError as e: humanize = None ENABLE_CRASH = True @@ -81,34 +83,9 @@ class TODO: def __init__(self): self.dir_path = None self.kdbx = None - self.init() self.file_path = None self.config_file = config_file.ConfigFile() - - def init(self): - # Get command - self.cmd_source_erplibre = "" - self.cmd_source_default = "" - exec_path_gnome_terminal = shutil.which("gnome-terminal") - if exec_path_gnome_terminal: - self.cmd_source_erplibre = ( - f"gnome-terminal -- bash -c 'source" - f" ./{cst_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 ./{cst_venv_erplibre}/bin/activate; %s\" in front window'" - self.cmd_source_erplibre += " -e 'end tell'" - else: - self.cmd_source_erplibre = ( - f"source ./{cst_venv_erplibre}/bin/activate;%s" - ) + self.execute = execute.Execute() def run(self): with open(self.config_file.get_logo_ascii_file_path()) as my_file: @@ -150,7 +127,7 @@ class TODO: # f" ./{cst_venv_erplibre}/bin/activate;make todo'" # ) cmd = "make todo" - self.executer_commande_live(cmd, source_erplibre=True) + self.execute.exec_command_live(cmd, source_erplibre=True) # elif status == "3" or status == "install": # print("install") else: @@ -280,7 +257,7 @@ class TODO: ) if first_installation_input == "y": cmd = "./script/version/update_env_version.py --install" - self.executer_commande_live(cmd, source_erplibre=True) + self.execute.exec_command_live(cmd, source_erplibre=True) print("Wait after OS installation before continue.") # First detect pycharm, need to be open before installation and close to increase speed @@ -312,7 +289,7 @@ class TODO: pycharm_bin = "pycharm" if has_pycharm else "pycharm-community" cmd = f"cd {os.getcwd()} && {pycharm_bin} ./" - self.executer_commande_live( + self.execute.exec_command_live( cmd, source_erplibre=False, single_source_erplibre=False, @@ -432,7 +409,7 @@ class TODO: makefile_cmd = dct_instance.get("makefile_cmd") if makefile_cmd and not ignore_makefile: - status = self.executer_commande_live( + status = self.execute.exec_command_live( f"make {makefile_cmd}", source_erplibre=False, single_source_erplibre=True, @@ -538,7 +515,7 @@ class TODO: print("Commande non trouvée 🤖!") def prompt_execute_update(self): - # self.executer_commande_live(f"make {makefile_cmd}") + # self.execute.exec_command_live(f"make {makefile_cmd}") print("🤖 Mise à jour du développement") # TODO détecter les modules en modification pour faire la mise à jour en cours # TODO demander sur quel BD faire la mise à jour @@ -776,13 +753,13 @@ class TODO: # f'parallel ::: "./run.sh -d {bd}" "sleep' # f' 3;./script/selenium/web_login.py{extra_cmd_web_login}"' # ) - # self.executer_commande_live(cmd) + # self.execute.exec_command_live(cmd) cmd_server = f"./run.sh -d {bd};bash" - self.executer_commande_live(cmd_server) + self.execute.exec_command_live(cmd_server) cmd_client = ( f"sleep 3;./script/selenium/web_login.py{extra_cmd_web_login};bash" ) - self.executer_commande_live(cmd_client) + self.execute.exec_command_live(cmd_client) def prompt_execute_selenium(self, command=None, extra_cmd_web_login=""): lst_cmd = [] @@ -798,132 +775,12 @@ class TODO: lst_cmd.append(cmd + extra_cmd_web_login) if len(lst_cmd) == 1: - self.executer_commande_live(lst_cmd[0]) + self.execute.exec_command_live(lst_cmd[0]) elif len(lst_cmd) > 1: new_cmd = "parallel ::: " for i, cmd in enumerate(lst_cmd): new_cmd += f' "sleep {1 * i};{cmd}"' - self.executer_commande_live(new_cmd) - - def executer_commande_live( - self, - commande, - source_erplibre=True, - quiet=False, - single_source_erplibre=False, - new_window=False, - single_source_odoo=False, - source_odoo="", - new_env=None, - return_status_and_command=False, - return_status_and_output=False, - return_status_and_output_and_command=False, - ): - """ - Exécute une commande et affiche la sortie en direct. - - Args: - commande (str): La commande à exécuter (sous forme de chaîne de caractères). - """ - - my_env = os.environ.copy() - if new_env: - my_env.update(new_env) - - process_start_time = time.time() - return_status = None - if source_erplibre: - # commande = f"source ./{cst_venv_erplibre}/bin/activate && " + commande - # cmd = ( - # f"gnome-terminal --tab -- bash -c 'source" - # f" ./{cst_venv_erplibre}/bin/activate;{commande}'" - # ) - commande = self.cmd_source_erplibre % commande - # os.system(f"./script/terminal/open_terminal.sh {commande}") - elif single_source_erplibre: - commande = ( - f"source ./{cst_venv_erplibre}/bin/activate && %s" % commande - ) - 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( - f"You cannot execute Odoo command if no version is installed. Command : {commande}" - ) - return -1 - commande = ( - f"source ./.venv.{source_odoo}/bin/activate && {commande}" - ) - elif new_window and self.cmd_source_default: - commande = self.cmd_source_default % commande - - print("🏠 ⬇ Execute command :") - print(commande) - lst_output = [] - - try: - process = subprocess.Popen( - commande, - shell=True, - executable="/bin/bash", - stdout=subprocess.PIPE, - stderr=subprocess.STDOUT, - text=True, - bufsize=1, # Désactive la mise en tampon pour la sortie en direct - universal_newlines=True, # Pour traiter les sauts de lignes correctement - env=my_env, - ) - - while True: - ligne = process.stdout.readline() - if not ligne: - break - print(ligne, end="") - if ( - return_status_and_output - or return_status_and_output_and_command - ): - lst_output.append(ligne) - - process.wait() # Attendre la fin du process - return_status = process.returncode - if process.returncode != 0 and not quiet: - print( - "La commande a retourné un code d'erreur :" - f" {process.returncode}" - ) - - except FileNotFoundError: - if "password" in commande: - print( - f"Erreur : La commande '{commande.split(' ')[0]}'[...] n'a" - " pas été trouvée." - ) - else: - print( - f"Erreur : La commande '{commande}' n'a pas été trouvée." - ) - except Exception as e: - print(f"Une erreur s'est produite : {e}") - process_end_time = time.time() - duration_sec = process_end_time - process_start_time - if humanize: - duration_delta = datetime.timedelta(seconds=duration_sec) - humain_time = humanize.precisedelta(duration_delta) - print(f"🏠 ⬆ Executed ({humain_time}) :") - else: - print(f"🏠 ⬆ Executed ({duration_sec:.2f} sec.) :") - print(commande) - print() - if return_status_and_output_and_command: - return return_status, commande, lst_output - if return_status_and_command: - return return_status, commande - if return_status_and_output: - return return_status, lst_output - return return_status + self.execute.exec_command_live(new_cmd) def crash_diagnostic(self, e): # TODO show message at start if os.path.exists(file_error_path) @@ -948,7 +805,7 @@ class TODO: time.sleep(0.5) cmd = "./script/todo/source_todo.sh" # self.restart_script(e) - self.executer_commande_live(cmd, source_erplibre=True) + self.execute.exec_command_live(cmd, source_erplibre=True) sys.exit(1) if os.path.exists(cst_venv_erplibre): print("Import error : ") @@ -980,7 +837,7 @@ class TODO: from pykeepass import PyKeePass except ImportError: print("Rerun and exit") - self.executer_commande_live(cmd, source_erplibre=True) + self.execute.exec_command_live(cmd, source_erplibre=True) sys.exit(1) print("No error") else: @@ -992,7 +849,7 @@ class TODO: def upgrade_poetry(self): # Only show the version to the user - status = self.executer_commande_live( + status = self.execute.exec_command_live( f"make version", source_erplibre=False, ) @@ -1001,7 +858,7 @@ class TODO: "💬 Would you like to fetch all your git repositories, you need it (y/Y) : " ) if git_repo_update_input.strip().lower() == "y": - status = self.executer_commande_live( + status = self.execute.exec_command_live( f"./script/manifest/update_manifest_local_dev.sh", source_erplibre=False, ) @@ -1022,7 +879,7 @@ class TODO: except Exception as e: pass - status = self.executer_commande_live( + status = self.execute.exec_command_live( f"pip install -r requirement/erplibre_require-ments-poetry.txt && " f"./script/poetry/poetry_update.py -f", source_erplibre=False, @@ -1055,7 +912,7 @@ class TODO: if not database_name: _logger.error("Missing database name") return - status, lst_output = self.executer_commande_live( + status, lst_output = self.execute.exec_command_live( f"python3 ./script/database/db_restore.py -d {database_name} --ignore_cache --image {file_name}", return_status_and_output=True, single_source_erplibre=True, @@ -1067,7 +924,7 @@ class TODO: .lower() ) if status == "y": - status, lst_output = self.executer_commande_live( + status, lst_output = self.execute.exec_command_live( f"./script/addons/update_prod_to_dev.sh {database_name}", return_status_and_output=True, single_source_erplibre=True, @@ -1079,7 +936,7 @@ class TODO: .lower() ) if status == "y": - status, lst_output = self.executer_commande_live( + status, lst_output = self.execute.exec_command_live( f"./script/addons/update_addons_all.sh {database_name}", return_status_and_output=True, single_source_erplibre=True, @@ -1091,7 +948,7 @@ class TODO: cfg.read("./config.conf") http_port = cfg.getint("options", "http_port") - status = self.executer_commande_live( + status = self.execute.exec_command_live( f"./script/process/kill_process_by_port.py {http_port} --kill-tree --nb_parent 2", source_erplibre=False, ) @@ -1099,7 +956,7 @@ class TODO: def download_database_backup_cli(self, show_remote_list=True): database_domain = input("Domain Odoo (ex. https://mondomain.com) : ") if show_remote_list: - status, lst_output = self.executer_commande_live( + status, lst_output = self.execute.exec_command_live( f"python3 ./script/database/list_remote.py --raw --odoo-url {database_domain}", return_status_and_output=True, single_source_erplibre=True, @@ -1134,7 +991,7 @@ class TODO: my_env["DATABASE_NAME"] = database_name my_env["OUTPUT_FILE_PATH"] = output_path my_env["ODOO_URL"] = database_domain - status, cmd_executed = self.executer_commande_live( + status, cmd_executed = self.execute.exec_command_live( cmd, source_erplibre=False, return_status_and_command=True, @@ -1265,7 +1122,7 @@ class TODO: # Rename with script bash cmd_client = f'cd {MOBILE_HOME_PATH} && npx cap init "{project_name}" "{package_name}" && ./rename_android.sh "{project_name}" "{package_name}" && npx cap sync android' - self.executer_commande_live(cmd_client, source_erplibre=False) + self.execute.exec_command_live(cmd_client, source_erplibre=False) # dotenv_mobile = dotenv.dotenv_values(dotenv_file) # dotenv_mobile["VITE_TITLE"] = project_name @@ -1293,7 +1150,7 @@ class TODO: ) if do_change_picture_menu: - status = self.executer_commande_live( + status = self.execute.exec_command_live( f"cd {MOBILE_HOME_PATH} && npx cap open android;bash", source_erplibre=False, new_window=True, @@ -1305,11 +1162,11 @@ class TODO: "Did you finish to update image with Android-Studio ? Press to continue ..." ) cmd_client = "cp ./mobile/erplibre_home_mobile/android/app/src/main/ic_launcher-playstore.png ./mobile/erplibre_home_mobile/src/assets/company_logo.png" - self.executer_commande_live(cmd_client, source_erplibre=False) + self.execute.exec_command_live(cmd_client, source_erplibre=False) cmd_client = "cp ./mobile/erplibre_home_mobile/android/app/src/main/ic_launcher-playstore.png ./mobile/erplibre_home_mobile/src/assets/imgs/logo.png" - self.executer_commande_live(cmd_client, source_erplibre=False) + self.execute.exec_command_live(cmd_client, source_erplibre=False) - status = self.executer_commande_live( + status = self.execute.exec_command_live( "./mobile/compile_and_run.sh", source_erplibre=False ) diff --git a/script/todo/todo_upgrade.py b/script/todo/todo_upgrade.py index 4fb898c..9742be7 100755 --- a/script/todo/todo_upgrade.py +++ b/script/todo/todo_upgrade.py @@ -19,6 +19,7 @@ new_path = os.path.normpath( ) sys.path.append(new_path) +from script.execute import execute from script.git.git_tool import GitTool _logger = logging.getLogger(__name__) @@ -52,6 +53,7 @@ class TodoUpgrade: self.lst_command_executed = [] self.dct_module_per_version = {} self.dct_module_per_dct_version_path = {} + self.execute = execute.Execute() def write_config(self): if "date_create" not in self.dct_progression.keys(): @@ -1857,7 +1859,7 @@ class TodoUpgrade: get_output = True output = None if get_output: - status, cmd_executed, output = self.todo.executer_commande_live( + status, cmd_executed, output = self.execute.exec_command_live( cmd, source_erplibre=False, single_source_odoo=single_source_odoo, @@ -1866,7 +1868,7 @@ class TodoUpgrade: quiet=quiet, ) else: - status, cmd_executed = self.todo.executer_commande_live( + status, cmd_executed = self.execute.exec_command_live( cmd, source_erplibre=False, single_source_odoo=single_source_odoo,