[UPD] script execute to share exec_command_live

This commit is contained in:
Mathieu Benoit 2026-02-13 01:27:38 -05:00
parent 02582d3ad8
commit 54f8ba0feb
5 changed files with 218 additions and 174 deletions

View file

@ -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,

View file

184
script/execute/execute.py Normal file
View file

@ -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

View file

@ -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
)

View file

@ -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,