Mathieu Benoit
935919e07b
Expose the QEMU deploy script from the interactive assistant so users can create, preview, download and manage Ubuntu VMs without memorising CLI flags. Adds a QEMU/KVM entry under Deploy, its fr/en translations, and an extensible qemu_from_makefile section in todo.json. Generated by Claude Code 2.1.210 claude-opus-4-8 Co-Authored-By: Mathieu Benoit <mathben@technolibre.ca>
2590 lines
91 KiB
Python
Executable file
2590 lines
91 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
# © 2021-2026 TechnoLibre (http://www.technolibre.ca)
|
|
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl)
|
|
|
|
import configparser
|
|
import datetime
|
|
import getpass
|
|
import json
|
|
import logging
|
|
import os
|
|
import shlex
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
import xml.etree.ElementTree as ET
|
|
import zipfile
|
|
|
|
new_path = os.path.normpath(
|
|
os.path.join(os.path.dirname(__file__), "..", "..")
|
|
)
|
|
sys.path.append(new_path)
|
|
|
|
from script.config import config_file
|
|
from script.execute import execute
|
|
from script.todo.database_manager import DatabaseManager
|
|
from script.todo.kdbx_manager import KdbxManager
|
|
from script.todo.todo_i18n import get_lang, lang_is_configured, set_lang, t
|
|
from script.todo.version_manager import get_odoo_version
|
|
|
|
ERROR_LOG_PATH = ".erplibre.error.txt"
|
|
VENV_ERPLIBRE = ".venv.erplibre"
|
|
ENABLE_CRASH = False
|
|
CRASH_E = None
|
|
# Support mobile ERPLibre
|
|
ANDROID_DIR = "android"
|
|
MOBILE_HOME_PATH = "./mobile/erplibre_home_mobile"
|
|
STRINGS_FILE = os.path.join(
|
|
MOBILE_HOME_PATH, ANDROID_DIR, "app/src/main/res/values/strings.xml"
|
|
)
|
|
GRADLE_FILE = os.path.join(MOBILE_HOME_PATH, ANDROID_DIR, "app/build.gradle")
|
|
|
|
|
|
try:
|
|
import click
|
|
import dotenv
|
|
import humanize
|
|
import openai
|
|
import todo_file_browser
|
|
|
|
# import urwid
|
|
# TODO implement rich for beautiful print and table
|
|
# import rich
|
|
import todo_upgrade
|
|
from pykeepass import PyKeePass
|
|
except ModuleNotFoundError as e:
|
|
humanize = None
|
|
ENABLE_CRASH = True
|
|
CRASH_E = e
|
|
|
|
if not ENABLE_CRASH:
|
|
print(t("Importation success!"))
|
|
|
|
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__)
|
|
|
|
CONFIG_FILE = "./script/todo/todo.json"
|
|
CONFIG_OVERRIDE_FILE = "./private/todo/todo.json"
|
|
LOGO_ASCII_FILE = "./script/todo/logo_ascii.txt"
|
|
|
|
|
|
class TODO:
|
|
def __init__(self):
|
|
self.dir_path = None
|
|
self.selected_file_path = None
|
|
self.config_file = config_file.ConfigFile()
|
|
self.execute = execute.Execute()
|
|
self.kdbx_manager = KdbxManager(self.config_file)
|
|
self.db_manager = DatabaseManager(self.execute, self.fill_help_info)
|
|
|
|
def _ask_language(self):
|
|
if not lang_is_configured():
|
|
print()
|
|
print("Choisir la langue / Choose language:")
|
|
print("[1] Francais")
|
|
print("[2] English")
|
|
choice = ""
|
|
while choice not in ("1", "2"):
|
|
choice = input("Select / Choisir : ").strip()
|
|
if choice == "1":
|
|
set_lang("fr")
|
|
else:
|
|
set_lang("en")
|
|
|
|
def _change_language(self):
|
|
print()
|
|
print(t("Choose language / Choisir la langue") + ":")
|
|
print(f"[1] {t('French')}")
|
|
print(f"[2] {t('English')}")
|
|
print(f"[0] {t('Back')}")
|
|
choice = ""
|
|
while choice not in ("0", "1", "2"):
|
|
choice = input(t("Select: ")).strip()
|
|
if choice == "0":
|
|
return False
|
|
elif choice == "1":
|
|
set_lang("fr")
|
|
else:
|
|
set_lang("en")
|
|
print(t("Language changed to: English"))
|
|
|
|
def run(self):
|
|
with open(self.config_file.get_logo_ascii_file_path()) as my_file:
|
|
print(my_file.read())
|
|
self._ask_language()
|
|
print(t("Opening TODO ..."))
|
|
print(f"🤖 {t('=> Enter your choice by number and press Enter!')}")
|
|
help_info = f"""{t("Command:")}
|
|
[1] {t("Execute")}
|
|
[2] {t("Install")}
|
|
[3] {t("Question")}
|
|
[4] {t("Fork - Open TODO in a new tab")}
|
|
[0] {t("Quit")}
|
|
"""
|
|
while True:
|
|
try:
|
|
status = click.prompt(help_info)
|
|
except NameError:
|
|
print("Do")
|
|
print(f"source ./{VENV_ERPLIBRE}/bin/activate && make")
|
|
sys.exit(1)
|
|
except ImportError:
|
|
print("Do")
|
|
print(f"source ./{VENV_ERPLIBRE}/bin/activate && make")
|
|
sys.exit(1)
|
|
except click.exceptions.Abort:
|
|
sys.exit(0)
|
|
print()
|
|
if status == "0":
|
|
break
|
|
elif status == "1":
|
|
self.prompt_execute()
|
|
elif status == "2":
|
|
self.prompt_install()
|
|
elif status == "3":
|
|
self.execute_prompt_ia()
|
|
elif status == "4":
|
|
# cmd = (
|
|
# f"gnome-terminal --tab -- bash -c 'source"
|
|
# f" ./{VENV_ERPLIBRE}/bin/activate;make todo'"
|
|
# )
|
|
cmd = "make todo"
|
|
self.execute.exec_command_live(cmd, source_erplibre=True)
|
|
# elif status == "3" or status == "install":
|
|
# print("install")
|
|
else:
|
|
print(t("Command not found !"))
|
|
|
|
print(status)
|
|
# manipuler()
|
|
|
|
def execute_prompt_ia(self):
|
|
while True:
|
|
help_info = f"""{t("Command:")}
|
|
[0] {t("Back")}
|
|
{t("Write your question ")}"""
|
|
status = click.prompt(help_info)
|
|
print()
|
|
if status == "0":
|
|
return
|
|
kp = self.kdbx_manager.get_kdbx()
|
|
if not kp:
|
|
return
|
|
config_name = self.config_file.get_config_value(
|
|
["kdbx_config", "openai", "kdbx_key"]
|
|
)
|
|
entry = kp.find_entries_by_title(config_name, first=True)
|
|
|
|
client = openai.OpenAI(api_key=entry.password)
|
|
prompt_update = status
|
|
completion = client.chat.completions.create(
|
|
model="gpt-4o",
|
|
messages=[{"role": "user", "content": prompt_update}],
|
|
)
|
|
|
|
print(completion.choices[0].message.content)
|
|
print()
|
|
|
|
def prompt_execute(self):
|
|
help_info = f"""{t("Command:")}
|
|
[1] {t("Automation - Demonstration of developed features")}
|
|
[2] {t("Code - Developer tools")}
|
|
[3] {t("Config - Configuration file management")}
|
|
[4] {t("Database - Database tools")}
|
|
[5] {t("Doc - Documentation search")}
|
|
[6] {t("Git - Git tools")}
|
|
[7] {t("GPT code - AI assistant tools")}
|
|
[8] {t("Language - Change language / Changer la langue")}
|
|
[9] {t("Network - Network tools")}
|
|
[10] {t("Process - Execution tools")}
|
|
[11] {t("Run - Execute and install an instance")}
|
|
[12] {t("Security - Dependency security audit")}
|
|
[13] {t("Test - Test an Odoo module")}
|
|
[14] {t("Update - Update all developed staging source code")}
|
|
[15] {t("Deploy - Deploy ERPLibre locally")}
|
|
[0] {t("Back")}
|
|
"""
|
|
while True:
|
|
status = click.prompt(help_info)
|
|
print()
|
|
if status == "0":
|
|
return
|
|
elif status == "1":
|
|
status = self.prompt_execute_function()
|
|
if status is not False:
|
|
return
|
|
elif status == "2":
|
|
status = self.prompt_execute_code()
|
|
if status is not False:
|
|
return
|
|
elif status == "3":
|
|
status = self.prompt_execute_config()
|
|
if status is not False:
|
|
return
|
|
elif status == "4":
|
|
status = self.prompt_execute_database()
|
|
if status is not False:
|
|
return
|
|
elif status == "5":
|
|
status = self.prompt_execute_doc()
|
|
if status is not False:
|
|
return
|
|
elif status == "6":
|
|
status = self.prompt_execute_git()
|
|
if status is not False:
|
|
return
|
|
elif status == "7":
|
|
status = self.prompt_execute_gpt_code()
|
|
if status is not False:
|
|
return
|
|
elif status == "8":
|
|
status = self._change_language()
|
|
if status is not False:
|
|
return
|
|
elif status == "9":
|
|
status = self.prompt_execute_network()
|
|
if status is not False:
|
|
return
|
|
elif status == "10":
|
|
status = self.prompt_execute_process()
|
|
if status is not False:
|
|
return
|
|
elif status == "11":
|
|
status = self.prompt_execute_instance()
|
|
if status is not False:
|
|
return
|
|
elif status == "12":
|
|
status = self.prompt_execute_security()
|
|
if status is not False:
|
|
return
|
|
elif status == "13":
|
|
status = self.prompt_execute_test()
|
|
if status is not False:
|
|
return
|
|
elif status == "14":
|
|
status = self.prompt_execute_update()
|
|
if status is not False:
|
|
return
|
|
elif status == "15":
|
|
status = self.prompt_execute_deploy()
|
|
if status is not False:
|
|
return
|
|
else:
|
|
print(t("Command not found !"))
|
|
|
|
def prompt_install(self):
|
|
print("Detect first installation from code source.")
|
|
|
|
first_installation_input = (
|
|
input(
|
|
"💬 First system installation? This will process system installation"
|
|
" before (Y/N): "
|
|
)
|
|
.strip()
|
|
.lower()
|
|
)
|
|
if first_installation_input == "y":
|
|
cmd = "./script/version/update_env_version.py --install"
|
|
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
|
|
has_pycharm = False
|
|
has_pycharm_community = False
|
|
result = subprocess.run(
|
|
["which", "pycharm"],
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
text=True,
|
|
)
|
|
if result.returncode == 0:
|
|
has_pycharm = True
|
|
else:
|
|
result = subprocess.run(
|
|
["which", "pycharm-community"],
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
text=True,
|
|
)
|
|
has_pycharm_community = result.returncode == 0
|
|
if (has_pycharm or has_pycharm_community) and not os.path.exists(
|
|
".idea"
|
|
):
|
|
pycharm_configuration_input = (
|
|
input("💬 Open Pycharm? (Y/N): ").strip().lower()
|
|
)
|
|
if pycharm_configuration_input == "y":
|
|
pycharm_bin = "pycharm" if has_pycharm else "pycharm-community"
|
|
|
|
cmd = f"cd {os.getcwd()} && {pycharm_bin} ./"
|
|
self.execute.exec_command_live(
|
|
cmd,
|
|
source_erplibre=False,
|
|
single_source_erplibre=False,
|
|
new_window=True,
|
|
)
|
|
print(
|
|
"👹 WAIT and Close Pycharm when processing is done before continue"
|
|
" this guide."
|
|
)
|
|
# TODO detect last version supported
|
|
# cmd_intern = "./script/install/install_erplibre.sh"
|
|
# TODO maybe update q to only install erplibre from install_locally
|
|
# TODO problem installing with q, the script depend on odoo
|
|
key_i = 0
|
|
commands_begin = {
|
|
"q": (
|
|
"q",
|
|
"q: ERPLibre only with system python without Odoo",
|
|
"./script/install/install_erplibre.sh",
|
|
),
|
|
"w": (
|
|
"w",
|
|
"w: Install all Odoo version with ERPLibre",
|
|
"make install_odoo_all_version",
|
|
),
|
|
"m": (
|
|
"m",
|
|
"m: ERPLibre with mobile home",
|
|
"./mobile/install_and_run.sh",
|
|
),
|
|
"0": (
|
|
"0",
|
|
f"0: {t('Quit')}",
|
|
),
|
|
}
|
|
commands_end = {}
|
|
versions, installed_versions, odoo_installed_version = (
|
|
get_odoo_version()
|
|
)
|
|
|
|
for version_info in versions[::-1]:
|
|
key_i += 1
|
|
key_s = str(key_i)
|
|
label = f"{key_s}: Odoo {version_info.get('odoo_version')}"
|
|
|
|
odoo_version = f"odoo{version_info.get('odoo_version')}"
|
|
if odoo_version in installed_versions:
|
|
label += " - Installed"
|
|
if odoo_version == odoo_installed_version:
|
|
label += " - Actual"
|
|
if version_info.get("Default"):
|
|
label += " - Default"
|
|
if version_info.get("is_deprecated"):
|
|
label += " - Deprecated"
|
|
erplibre_version = version_info.get("erplibre_version")
|
|
commands_begin[key_s] = (
|
|
key_s,
|
|
label,
|
|
f"./script/version/update_env_version.py --erplibre_version {erplibre_version} --install_dev",
|
|
)
|
|
|
|
# Add final command
|
|
install_commands = {**commands_begin, **commands_end}
|
|
|
|
# Show command
|
|
odoo_version_input = ""
|
|
while odoo_version_input not in install_commands:
|
|
if odoo_version_input:
|
|
print(
|
|
f"{t('Error, cannot understand value')} '{odoo_version_input}'"
|
|
)
|
|
str_input_dyn_odoo_version = (
|
|
f"💬 {t('Choose a version:')}\n\t"
|
|
+ "\n\t".join([a[1] for a in install_commands.values()])
|
|
+ f"\n{t('Select: ')}"
|
|
)
|
|
odoo_version_input = (
|
|
input(str_input_dyn_odoo_version).strip().lower()
|
|
)
|
|
|
|
if odoo_version_input == "0":
|
|
return
|
|
|
|
cmd_intern = install_commands.get(odoo_version_input)[2]
|
|
|
|
# For numbered version selections, offer extra modules sub-menu
|
|
if odoo_version_input.isdigit():
|
|
extra_choices = {
|
|
"1": (
|
|
"1",
|
|
f"1: {t('Standard install (without extra modules)')}",
|
|
),
|
|
"2": (
|
|
"2",
|
|
f"2: {t('Install with extra modules (CybroOdoo - large, slow)')}",
|
|
),
|
|
"0": ("0", f"0: {t('Back')}"),
|
|
}
|
|
extra_input = ""
|
|
while extra_input not in extra_choices:
|
|
if extra_input:
|
|
print(
|
|
f"{t('Error, cannot understand value')} '{extra_input}'"
|
|
)
|
|
str_extra = (
|
|
f"💬 {t('Install type:')}\n\t"
|
|
+ "\n\t".join([a[1] for a in extra_choices.values()])
|
|
+ f"\n{t('Select: ')}"
|
|
)
|
|
extra_input = input(str_extra).strip()
|
|
if extra_input == "0":
|
|
return
|
|
if extra_input == "2":
|
|
cmd_intern = cmd_intern + " --with_extra"
|
|
|
|
print(f"{t('Will execute:')}\n{cmd_intern}")
|
|
|
|
# TODO use external script to detect terminal to use on system
|
|
# TODO check script open_terminal_code_generator.sh
|
|
# cmd_extern = f"gnome-terminal -- bash -c '{cmd_intern};bash'"
|
|
try:
|
|
subprocess.run(
|
|
cmd_intern, shell=True, executable="/bin/bash", check=True
|
|
)
|
|
except subprocess.CalledProcessError as e:
|
|
print(
|
|
f"{t('The Bash script failed with return code')} {e.returncode}."
|
|
)
|
|
print("Wait after installation and open projects by terminal.")
|
|
print("make open_terminal")
|
|
self.restart_script(str(e))
|
|
|
|
def execute_from_configuration(
|
|
self, instance, exec_run_db=False, ignore_makefile=False
|
|
):
|
|
# exec_run_db need argument database
|
|
kdbx_key = instance.get("kdbx_key")
|
|
odoo_user = instance.get("user")
|
|
odoo_password = instance.get("password")
|
|
|
|
if kdbx_key:
|
|
extra_cmd_web_login = self.kdbx_manager.get_extra_command_user(
|
|
kdbx_key
|
|
)
|
|
elif odoo_user and odoo_password:
|
|
extra_cmd_web_login = (
|
|
f" --default_email_auth {odoo_user} --default_password_auth"
|
|
f" '{odoo_password}'"
|
|
)
|
|
else:
|
|
extra_cmd_web_login = ""
|
|
|
|
makefile_cmd = instance.get("makefile_cmd")
|
|
if makefile_cmd and not ignore_makefile:
|
|
status = self.execute.exec_command_live(
|
|
f"make {makefile_cmd}",
|
|
source_erplibre=False,
|
|
single_source_erplibre=True,
|
|
)
|
|
if status:
|
|
_logger.error(
|
|
f"Status {status} - exit execute_from_configuration"
|
|
)
|
|
return
|
|
|
|
if exec_run_db:
|
|
db_name = instance.get("database")
|
|
self.prompt_execute_selenium_and_run_db(
|
|
db_name, extra_cmd_web_login=extra_cmd_web_login
|
|
)
|
|
|
|
bash_command = instance.get("bash_command")
|
|
if bash_command:
|
|
print(f"{t('Will execute:')} {bash_command}")
|
|
self.execute.exec_command_live(bash_command, source_erplibre=False)
|
|
|
|
command = instance.get("Command:")
|
|
if command:
|
|
self.prompt_execute_selenium(
|
|
command=command, extra_cmd_web_login=extra_cmd_web_login
|
|
)
|
|
|
|
callback = instance.get("callback")
|
|
if callback:
|
|
callback(instance)
|
|
|
|
def fill_help_info(self, choices):
|
|
help_info = t("Command:") + "\n"
|
|
help_end = f"[0] {t('Back')}\n"
|
|
for i, instance in enumerate(choices):
|
|
desc_key = instance.get("prompt_description_key")
|
|
if desc_key:
|
|
desc = t(desc_key)
|
|
else:
|
|
desc = instance["prompt_description"]
|
|
help_info += f"[{i + 1}] " + desc + "\n"
|
|
help_info += help_end
|
|
return help_info
|
|
|
|
def prompt_execute_instance(self):
|
|
# TODO proposer le déploiement à distance
|
|
# TODO proposer l'exécution de docker
|
|
# TODO proposer la création de docker
|
|
choices = self.config_file.get_config("instance")
|
|
init_len = len(choices)
|
|
|
|
# Support mobile ERPLibre
|
|
if os.path.exists(MOBILE_HOME_PATH):
|
|
menu_entry = {
|
|
"prompt_description": t("Mobile - Compile and run software"),
|
|
"callback": self.callback_make_mobile_home,
|
|
}
|
|
choices.append(menu_entry)
|
|
|
|
# Support custom database to execute
|
|
menu_entry = {
|
|
"prompt_description": t("Choose your database"),
|
|
"callback": self.callback_execute_custom_database,
|
|
}
|
|
choices.insert(0, menu_entry)
|
|
help_info = self.fill_help_info(choices)
|
|
|
|
while True:
|
|
status = click.prompt(help_info)
|
|
print()
|
|
if status == "0":
|
|
return False
|
|
else:
|
|
cmd_no_found = True
|
|
try:
|
|
int_cmd = int(status)
|
|
if 1 < int_cmd <= init_len:
|
|
cmd_no_found = False
|
|
status = click.confirm(
|
|
t("Do you want a new instance?")
|
|
)
|
|
instance = choices[int_cmd - 1]
|
|
self.execute_from_configuration(
|
|
instance,
|
|
exec_run_db=True,
|
|
ignore_makefile=not bool(status),
|
|
)
|
|
elif int_cmd <= len(choices) or 1 == int_cmd:
|
|
cmd_no_found = False
|
|
# Execute dynamic instance
|
|
instance = choices[int_cmd - 1]
|
|
self.execute_from_configuration(
|
|
instance,
|
|
)
|
|
except ValueError:
|
|
pass
|
|
if cmd_no_found:
|
|
print(t("Command not found !"))
|
|
|
|
def prompt_execute_function(self):
|
|
choices = self.config_file.get_config("function")
|
|
help_info = self.fill_help_info(choices)
|
|
|
|
while True:
|
|
status = click.prompt(help_info)
|
|
print()
|
|
if status == "0":
|
|
return False
|
|
else:
|
|
cmd_no_found = True
|
|
try:
|
|
int_cmd = int(status)
|
|
if 0 < int_cmd <= len(choices):
|
|
cmd_no_found = False
|
|
instance = choices[int_cmd - 1]
|
|
self.execute_from_configuration(instance)
|
|
except ValueError:
|
|
pass
|
|
if cmd_no_found:
|
|
print(t("Command not found !"))
|
|
|
|
def prompt_execute_update(self):
|
|
# self.execute.exec_command_live(f"make {makefile_cmd}")
|
|
print(f"🤖 {t('Development update')}")
|
|
# TODO détecter les modules en modification pour faire la mise à jour en cours
|
|
# TODO demander sur quel BD faire la mise à jour
|
|
# TODO proposer les modules manuelles selon la configuration à mettre à jour
|
|
# TODO proposer la mise à jour de l'IDE
|
|
# TODO proposer la mise à jour des git-repo
|
|
# TODO faire la mise à jour de ERPLibre
|
|
# TODO faire l'upgrade d'un odoo vers un autre
|
|
|
|
choices = self.config_file.get_config("update_from_makefile")
|
|
menu_entry = {
|
|
"prompt_description": t("Upgrade Odoo - Migration Database"),
|
|
}
|
|
choices.append(menu_entry)
|
|
poetry_entry = {
|
|
"prompt_description": t("Upgrade Poetry - Dependency of Odoo"),
|
|
}
|
|
choices.append(poetry_entry)
|
|
help_info = self.fill_help_info(choices)
|
|
|
|
while True:
|
|
status = click.prompt(help_info)
|
|
print()
|
|
if status == "0":
|
|
return False
|
|
elif status == str(len(choices) - 1):
|
|
upgrade = todo_upgrade.TodoUpgrade(self)
|
|
upgrade.execute_odoo_upgrade()
|
|
elif status == str(len(choices)):
|
|
self.upgrade_poetry()
|
|
else:
|
|
cmd_no_found = True
|
|
try:
|
|
int_cmd = int(status) - 1
|
|
if 0 < int_cmd <= len(choices):
|
|
cmd_no_found = False
|
|
instance = choices[int_cmd - 1]
|
|
self.execute_from_configuration(instance)
|
|
except ValueError:
|
|
pass
|
|
if cmd_no_found:
|
|
print(t("Command not found !"))
|
|
|
|
def prompt_execute_deploy(self):
|
|
print(f"🤖 {t('Deploy ERPLibre to a local directory!')}")
|
|
choices = [
|
|
{"prompt_description": t("Clone ERPLibre locally (git clone)")},
|
|
{"prompt_description": t("Configure sshfs")},
|
|
{"prompt_description": t("SSH - Check connection")},
|
|
{"prompt_description": t("SSH - Sync files (rsync)")},
|
|
{"prompt_description": t("SSH - Install ERPLibre")},
|
|
{"prompt_description": t("SSH - Start Odoo")},
|
|
{"prompt_description": t("SSH - Stop Odoo")},
|
|
{"prompt_description": t("SSH - Restart Odoo")},
|
|
{"prompt_description": t("SSH - Service status")},
|
|
{"prompt_description": t("SSH - View logs")},
|
|
{"prompt_description": t("SSH - Run make target")},
|
|
{"prompt_description": t("SSH - Install systemd service")},
|
|
{"prompt_description": t("SSH - Configure nginx + SSL")},
|
|
{
|
|
"prompt_description": t(
|
|
"Deploy - Install NTFY notification server"
|
|
)
|
|
},
|
|
{
|
|
"prompt_description": t(
|
|
"QEMU/KVM - Deploy an Ubuntu VM (libvirt)"
|
|
)
|
|
},
|
|
]
|
|
help_info = self.fill_help_info(choices)
|
|
|
|
while True:
|
|
status = click.prompt(help_info)
|
|
print()
|
|
if status == "0":
|
|
return False
|
|
elif status == "1":
|
|
self._deploy_clone_erplibre()
|
|
elif status == "2":
|
|
self._configure_sshfs()
|
|
elif status == "3":
|
|
self._deploy_ssh_check()
|
|
elif status == "4":
|
|
self._deploy_ssh_push()
|
|
elif status == "5":
|
|
self._deploy_ssh_install()
|
|
elif status == "6":
|
|
self._deploy_ssh_run()
|
|
elif status == "7":
|
|
self._deploy_ssh_stop()
|
|
elif status == "8":
|
|
self._deploy_ssh_restart()
|
|
elif status == "9":
|
|
self._deploy_ssh_status()
|
|
elif status == "10":
|
|
self._deploy_ssh_logs()
|
|
elif status == "11":
|
|
self._deploy_ssh_make()
|
|
elif status == "12":
|
|
self._deploy_ssh_install_systemd()
|
|
elif status == "13":
|
|
self._deploy_ssh_install_nginx()
|
|
elif status == "14":
|
|
self._deploy_ntfy_server()
|
|
elif status == "15":
|
|
self.prompt_execute_qemu()
|
|
else:
|
|
print(t("Command not found !"))
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# QEMU / KVM (libvirt) VM deployment
|
|
# ------------------------------------------------------------------ #
|
|
def _qemu_script_path(self):
|
|
"""Chemin absolu vers script/qemu/deploy_qemu.py."""
|
|
path = os.path.join(
|
|
os.path.dirname(os.path.abspath(__file__)),
|
|
"..",
|
|
"qemu",
|
|
"deploy_qemu.py",
|
|
)
|
|
return os.path.realpath(path)
|
|
|
|
def _qemu_default_ssh_key(self):
|
|
"""Première clé publique SSH trouvée dans ~/.ssh, sinon ''."""
|
|
for name in ("id_ed25519.pub", "id_rsa.pub"):
|
|
path = os.path.expanduser(f"~/.ssh/{name}")
|
|
if os.path.exists(path):
|
|
return path
|
|
return ""
|
|
|
|
def _qemu_prompt_version(self):
|
|
"""Demande la version Ubuntu ; retourne une chaîne (défaut 24.04)."""
|
|
versions = ["20.04", "22.04", "24.04", "24.10", "25.04", "25.10"]
|
|
print(f"\n{t('Ubuntu version:')}")
|
|
for i, v in enumerate(versions, 1):
|
|
suffix = " (LTS)" if v in ("20.04", "22.04", "24.04") else ""
|
|
print(f" [{i}] {v}{suffix}")
|
|
sel = input(t("Choice (number or version, default: 24.04): ")).strip()
|
|
if not sel:
|
|
return "24.04"
|
|
try:
|
|
idx = int(sel) - 1
|
|
if 0 <= idx < len(versions):
|
|
return versions[idx]
|
|
except ValueError:
|
|
if sel in versions:
|
|
return sel
|
|
print(t("Invalid selection, using 24.04"))
|
|
return "24.04"
|
|
|
|
def prompt_execute_qemu(self):
|
|
print(f"🤖 {t('Deploy a QEMU/KVM virtual machine (libvirt)!')}")
|
|
script_path = self._qemu_script_path()
|
|
if not os.path.isfile(script_path):
|
|
print(f"{t('QEMU deploy script not found: ')}{script_path}")
|
|
return False
|
|
choices = [
|
|
{"prompt_description": t("Deploy a new Ubuntu VM")},
|
|
{
|
|
"prompt_description": t(
|
|
"Preview a deployment (dry-run, no sudo)"
|
|
)
|
|
},
|
|
{"prompt_description": t("Download an Ubuntu cloud image only")},
|
|
{"prompt_description": t("List VMs (virsh list --all)")},
|
|
{"prompt_description": t("Show a VM IP address")},
|
|
]
|
|
config_entries = self.config_file.get_config("qemu_from_makefile")
|
|
if config_entries:
|
|
choices.extend(config_entries)
|
|
help_info = self.fill_help_info(choices)
|
|
|
|
while True:
|
|
status = click.prompt(help_info)
|
|
print()
|
|
if status == "0":
|
|
return False
|
|
elif status == "1":
|
|
self._qemu_deploy_vm(dry_run=False)
|
|
elif status == "2":
|
|
self._qemu_deploy_vm(dry_run=True)
|
|
elif status == "3":
|
|
self._qemu_download_image()
|
|
elif status == "4":
|
|
self._qemu_list_vms()
|
|
elif status == "5":
|
|
self._qemu_show_ip()
|
|
else:
|
|
cmd_no_found = True
|
|
try:
|
|
int_cmd = int(status)
|
|
if 0 < int_cmd <= len(choices):
|
|
cmd_no_found = False
|
|
instance = choices[int_cmd - 1]
|
|
self.execute_from_configuration(instance)
|
|
except ValueError:
|
|
pass
|
|
if cmd_no_found:
|
|
print(t("Command not found !"))
|
|
|
|
def _qemu_deploy_vm(self, dry_run=False):
|
|
script_path = self._qemu_script_path()
|
|
name = input(t("VM name (required): ")).strip()
|
|
if not name:
|
|
print(t("VM name is required!"))
|
|
return
|
|
version = self._qemu_prompt_version()
|
|
memory = input(t("RAM in MB (default: 8192): ")).strip() or "8192"
|
|
vcpus = input(t("vCPUs (default: 4): ")).strip() or "4"
|
|
disk_size = input(t("Disk size (default: 20G): ")).strip() or "20G"
|
|
|
|
default_key = self._qemu_default_ssh_key()
|
|
key_hint = default_key or t("none")
|
|
ssh_key = input(f"{t('SSH public key path')} ({key_hint}): ").strip()
|
|
if not ssh_key:
|
|
ssh_key = default_key
|
|
if ssh_key:
|
|
# Résout ~ avec le HOME de l'utilisateur courant : le script tourne
|
|
# sous sudo (HOME=/root) et ne pourrait pas déduire ~ correctement.
|
|
ssh_key = os.path.expanduser(ssh_key)
|
|
|
|
use_password = False
|
|
if not ssh_key:
|
|
ans = (
|
|
input(t("No SSH key found. Set a password instead? (Y/n): "))
|
|
.strip()
|
|
.lower()
|
|
)
|
|
use_password = ans != "n"
|
|
|
|
force = False
|
|
if not dry_run:
|
|
ans = (
|
|
input(t("Overwrite existing VM disk if present? (y/N): "))
|
|
.strip()
|
|
.lower()
|
|
)
|
|
force = ans == "y"
|
|
|
|
parts = []
|
|
if not dry_run:
|
|
parts.append("sudo")
|
|
parts.append(script_path)
|
|
parts += ["--name", name, "--version", version]
|
|
parts += [
|
|
"--memory",
|
|
memory,
|
|
"--vcpus",
|
|
vcpus,
|
|
"--disk-size",
|
|
disk_size,
|
|
]
|
|
if ssh_key:
|
|
parts += ["--ssh-key", ssh_key]
|
|
if use_password:
|
|
parts.append("--ask-password")
|
|
if force:
|
|
parts.append("--force")
|
|
if dry_run:
|
|
parts.append("--dry-run")
|
|
else:
|
|
parts.append("-y") # accepte l'installation des dépendances
|
|
|
|
cmd = " ".join(shlex.quote(p) for p in parts)
|
|
print(f"{t('Will execute:')} {cmd}")
|
|
self.execute.exec_command_live(cmd, source_erplibre=False)
|
|
|
|
def _qemu_download_image(self):
|
|
script_path = self._qemu_script_path()
|
|
version = self._qemu_prompt_version()
|
|
ans = input(t("Verify SHA256 after download? (y/N): ")).strip().lower()
|
|
parts = [
|
|
"sudo",
|
|
script_path,
|
|
"--download-only",
|
|
"--version",
|
|
version,
|
|
]
|
|
if ans == "y":
|
|
parts.append("--verify")
|
|
cmd = " ".join(shlex.quote(p) for p in parts)
|
|
print(f"{t('Will execute:')} {cmd}")
|
|
self.execute.exec_command_live(cmd, source_erplibre=False)
|
|
|
|
def _qemu_list_vms(self):
|
|
cmd = "sudo virsh list --all"
|
|
print(f"{t('Will execute:')} {cmd}")
|
|
self.execute.exec_command_live(cmd, source_erplibre=False)
|
|
|
|
def _qemu_show_ip(self):
|
|
name = input(t("VM name: ")).strip()
|
|
if not name:
|
|
print(t("VM name is required!"))
|
|
return
|
|
cmd = f"sudo virsh domifaddr {shlex.quote(name)} --source lease"
|
|
print(f"{t('Will execute:')} {cmd}")
|
|
self.execute.exec_command_live(cmd, source_erplibre=False)
|
|
|
|
def _deploy_clone_erplibre(self):
|
|
default_path = os.path.expanduser("~/erplibre")
|
|
target_path = (
|
|
input(t("Target directory path (default: ~/erplibre): ")).strip()
|
|
or default_path
|
|
)
|
|
target_path = os.path.expanduser(target_path)
|
|
if os.path.exists(target_path):
|
|
print(f"{t('Directory already exists: ')}{target_path}")
|
|
return
|
|
print(t("Cloning ERPLibre..."))
|
|
cmd = (
|
|
"git clone"
|
|
" https://github.com/erplibre/erplibre"
|
|
f" {target_path}"
|
|
)
|
|
print(f"{t('Will execute:')} {cmd}")
|
|
try:
|
|
self.execute.exec_command_live(cmd, source_erplibre=False)
|
|
print(f"{t('ERPLibre cloned successfully to: ')}" f"{target_path}")
|
|
except Exception as e:
|
|
print(f"{t('Error cloning ERPLibre: ')}{e}")
|
|
|
|
def _deploy_ntfy_server(self):
|
|
print(
|
|
f"\n{t('Deploy a local NTFY push notification server (Ubuntu/Arch)')}"
|
|
)
|
|
port = input(t("NTFY server port (default: 8080): ")).strip() or "8080"
|
|
import socket
|
|
|
|
hostname = socket.gethostname()
|
|
try:
|
|
local_ip = socket.gethostbyname(hostname)
|
|
except Exception:
|
|
local_ip = "127.0.0.1"
|
|
|
|
default_url = f"http://{local_ip}:{port}"
|
|
base_url = (
|
|
input(f"{t('NTFY base URL')} (default: {default_url}): ").strip()
|
|
or default_url
|
|
)
|
|
|
|
script_path = os.path.join(
|
|
os.path.dirname(os.path.abspath(__file__)),
|
|
"..",
|
|
"install",
|
|
"install_ntfy.sh",
|
|
)
|
|
script_path = os.path.realpath(script_path)
|
|
|
|
if not os.path.isfile(script_path):
|
|
print(f"{t('NTFY install script not found: ')}{script_path}")
|
|
return
|
|
|
|
print(f"\n{t('Installing NTFY server (requires sudo)...')}")
|
|
cmd = (
|
|
f"sudo NTFY_PORT={port}"
|
|
f" NTFY_BASE_URL={base_url}"
|
|
f" bash {script_path}"
|
|
)
|
|
print(f"{t('Will execute:')} {cmd}\n")
|
|
try:
|
|
self.execute.exec_command_live(cmd, source_erplibre=False)
|
|
print(f"\n{t('NTFY server installed and started successfully!')}")
|
|
except Exception as e:
|
|
print(f"{t('Error installing NTFY server: ')}{e}")
|
|
|
|
def _configure_sshfs(self):
|
|
import getpass
|
|
import re
|
|
from datetime import datetime
|
|
|
|
print(f"\n{t('SSH address input method')}")
|
|
print(f"[1] {t('Manual entry')}")
|
|
print(f"[2] {t('From ~/.ssh/config')}")
|
|
choice = input(t("Your choice (1/2): ")).strip()
|
|
|
|
user = None
|
|
hostname = None
|
|
ssh_name = None
|
|
|
|
if choice == "2":
|
|
ssh_config_path = os.path.expanduser("~/.ssh/config")
|
|
hosts = []
|
|
if os.path.exists(ssh_config_path):
|
|
current_host = None
|
|
current_info = {}
|
|
with open(ssh_config_path) as f:
|
|
for line in f:
|
|
line = line.strip()
|
|
if line.lower().startswith("host "):
|
|
host_val = line.split(None, 1)[1].strip()
|
|
if host_val != "*":
|
|
if current_host:
|
|
hosts.append((current_host, current_info))
|
|
current_host = host_val
|
|
current_info = {}
|
|
elif current_host:
|
|
key = line.split(None, 1)
|
|
if len(key) == 2:
|
|
k = key[0].lower()
|
|
v = key[1].strip()
|
|
if k == "hostname":
|
|
current_info["hostname"] = v
|
|
elif k == "user":
|
|
current_info["user"] = v
|
|
if current_host:
|
|
hosts.append((current_host, current_info))
|
|
|
|
if not hosts:
|
|
print(t("No SSH hosts found in ~/.ssh/config"))
|
|
return
|
|
|
|
print()
|
|
for i, (host, info) in enumerate(hosts, 1):
|
|
hn = info.get("hostname", host)
|
|
u = info.get("user", "")
|
|
desc = host
|
|
if hn != host:
|
|
desc += f" ({hn})"
|
|
if u:
|
|
desc += f" [{u}]"
|
|
print(f"[{i}] {desc}")
|
|
|
|
sel = input(t("Select SSH host number: ")).strip()
|
|
try:
|
|
idx = int(sel) - 1
|
|
if idx < 0 or idx >= len(hosts):
|
|
print(t("Invalid selection!"))
|
|
return
|
|
except ValueError:
|
|
print(t("Invalid selection!"))
|
|
return
|
|
|
|
host_name, host_info = hosts[idx]
|
|
hostname = host_info.get("hostname", host_name)
|
|
user = host_info.get("user", getpass.getuser())
|
|
ssh_name = host_name
|
|
target = f"{host_name}:/"
|
|
else:
|
|
ssh_host = input(
|
|
t("SSH host (e.g.: user@192.168.1.100): ")
|
|
).strip()
|
|
if not ssh_host:
|
|
print(t("SSH host is required!"))
|
|
return
|
|
if "@" in ssh_host:
|
|
user, hostname = ssh_host.split("@", 1)
|
|
else:
|
|
hostname = ssh_host
|
|
user = getpass.getuser()
|
|
ssh_name = hostname
|
|
target = f"{user}@{hostname}:/"
|
|
|
|
safe_name = re.sub(r"[^a-zA-Z0-9_-]", "_", ssh_name)
|
|
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
mount_point = f"/tmp/sshfs_{safe_name}_{timestamp}"
|
|
os.makedirs(mount_point, exist_ok=True)
|
|
|
|
cmd = f"sshfs {target} {mount_point}"
|
|
print(f"{t('Mounting sshfs on: ')}{mount_point}")
|
|
print(f"{t('Will execute:')} {cmd}")
|
|
try:
|
|
self.execute.exec_command_live(cmd, source_erplibre=False)
|
|
print(f"{t('Mounted on: ')}{mount_point}")
|
|
print(f"mount | grep sshfs")
|
|
print(f"{t('To unmount: ')}" f"fusermount -u {mount_point}")
|
|
print(f"nautilus {mount_point}/home/{user}")
|
|
except Exception as e:
|
|
print(f"{t('Error mounting sshfs: ')}{e}")
|
|
|
|
def _get_ssh_params(self):
|
|
"""Prompt for SSH connection parameters. Returns dict or None on cancel."""
|
|
host = click.prompt(
|
|
t("Remote host (user@hostname or hostname): ")
|
|
).strip()
|
|
if not host:
|
|
print(t("SSH host is required!"))
|
|
return None
|
|
user = (
|
|
click.prompt(t("SSH user (default: erplibre): ")).strip()
|
|
or "erplibre"
|
|
)
|
|
port = click.prompt(t("SSH port (default: 22): ")).strip() or "22"
|
|
key = click.prompt(
|
|
t("SSH key path (default: ~/.ssh/id_rsa, empty for none): ")
|
|
).strip()
|
|
path = (
|
|
click.prompt(
|
|
t("Remote path (default: ~/erplibre_deploy_2): ")
|
|
).strip()
|
|
or "~/erplibre_deploy_2"
|
|
)
|
|
return {
|
|
"SSH_HOST": host,
|
|
"SSH_USER": user,
|
|
"SSH_PORT": port,
|
|
"SSH_KEY": key,
|
|
"SSH_PATH": path,
|
|
}
|
|
|
|
def _build_ssh_make_cmd(self, target, params, extra=None):
|
|
"""Build a make SSH command string from params dict."""
|
|
parts = [f"make {target}"]
|
|
for k, v in params.items():
|
|
if v:
|
|
parts.append(f'{k}="{v}"')
|
|
if extra:
|
|
for k, v in extra.items():
|
|
if v:
|
|
parts.append(f'{k}="{v}"')
|
|
return " ".join(parts)
|
|
|
|
def _deploy_ssh_check(self):
|
|
params = self._get_ssh_params()
|
|
if not params:
|
|
return
|
|
cmd = self._build_ssh_make_cmd("ssh_check", params)
|
|
print(f"{t('Will execute:')} {cmd}")
|
|
self.execute.exec_command_live(
|
|
cmd, source_erplibre=False, single_source_erplibre=True
|
|
)
|
|
|
|
def _deploy_ssh_push(self):
|
|
params = self._get_ssh_params()
|
|
if not params:
|
|
return
|
|
cmd = self._build_ssh_make_cmd("ssh_push", params)
|
|
print(f"{t('Will execute:')} {cmd}")
|
|
self.execute.exec_command_live(
|
|
cmd, source_erplibre=False, single_source_erplibre=True
|
|
)
|
|
|
|
def _deploy_ssh_install(self):
|
|
params = self._get_ssh_params()
|
|
if not params:
|
|
return
|
|
cmd = self._build_ssh_make_cmd("ssh_install", params)
|
|
print(f"{t('Will execute:')} {cmd}")
|
|
self.execute.exec_command_live(
|
|
cmd, source_erplibre=False, single_source_erplibre=True
|
|
)
|
|
|
|
def _deploy_ssh_run(self):
|
|
params = self._get_ssh_params()
|
|
if not params:
|
|
return
|
|
cmd = self._build_ssh_make_cmd("ssh_run", params)
|
|
print(f"{t('Will execute:')} {cmd}")
|
|
self.execute.exec_command_live(
|
|
cmd, source_erplibre=False, single_source_erplibre=True
|
|
)
|
|
|
|
def _deploy_ssh_stop(self):
|
|
params = self._get_ssh_params()
|
|
if not params:
|
|
return
|
|
cmd = self._build_ssh_make_cmd("ssh_stop", params)
|
|
print(f"{t('Will execute:')} {cmd}")
|
|
self.execute.exec_command_live(
|
|
cmd, source_erplibre=False, single_source_erplibre=True
|
|
)
|
|
|
|
def _deploy_ssh_restart(self):
|
|
params = self._get_ssh_params()
|
|
if not params:
|
|
return
|
|
cmd = self._build_ssh_make_cmd("ssh_restart", params)
|
|
print(f"{t('Will execute:')} {cmd}")
|
|
self.execute.exec_command_live(
|
|
cmd, source_erplibre=False, single_source_erplibre=True
|
|
)
|
|
|
|
def _deploy_ssh_status(self):
|
|
params = self._get_ssh_params()
|
|
if not params:
|
|
return
|
|
cmd = self._build_ssh_make_cmd("ssh_status", params)
|
|
print(f"{t('Will execute:')} {cmd}")
|
|
self.execute.exec_command_live(
|
|
cmd, source_erplibre=False, single_source_erplibre=True
|
|
)
|
|
|
|
def _deploy_ssh_logs(self):
|
|
params = self._get_ssh_params()
|
|
if not params:
|
|
return
|
|
cmd = self._build_ssh_make_cmd("ssh_logs", params)
|
|
print(f"{t('Will execute:')} {cmd}")
|
|
self.execute.exec_command_live(
|
|
cmd, source_erplibre=False, single_source_erplibre=True
|
|
)
|
|
|
|
def _deploy_ssh_make(self):
|
|
params = self._get_ssh_params()
|
|
if not params:
|
|
return
|
|
target = click.prompt(t("Make target to run remotely: ")).strip()
|
|
if not target:
|
|
print(t("SSH host is required!"))
|
|
return
|
|
cmd = self._build_ssh_make_cmd(
|
|
"ssh_make", params, extra={"SSH_TARGET": target}
|
|
)
|
|
print(f"{t('Will execute:')} {cmd}")
|
|
self.execute.exec_command_live(
|
|
cmd, source_erplibre=False, single_source_erplibre=True
|
|
)
|
|
|
|
def _deploy_ssh_install_systemd(self):
|
|
params = self._get_ssh_params()
|
|
if not params:
|
|
return
|
|
cmd = self._build_ssh_make_cmd("ssh_install_systemd", params)
|
|
print(f"{t('Will execute:')} {cmd}")
|
|
self.execute.exec_command_live(
|
|
cmd, source_erplibre=False, single_source_erplibre=True
|
|
)
|
|
|
|
def _deploy_ssh_install_nginx(self):
|
|
params = self._get_ssh_params()
|
|
if not params:
|
|
return
|
|
domain = click.prompt(t("Domain name (e.g.: example.com): ")).strip()
|
|
if not domain:
|
|
print(t("SSH host is required!"))
|
|
return
|
|
email = click.prompt(t("Admin email for SSL certificate: ")).strip()
|
|
cmd = self._build_ssh_make_cmd(
|
|
"ssh_install_nginx",
|
|
params,
|
|
extra={"SSH_DOMAIN": domain, "SSH_ADMIN_EMAIL": email},
|
|
)
|
|
print(f"{t('Will execute:')} {cmd}")
|
|
self.execute.exec_command_live(
|
|
cmd, source_erplibre=False, single_source_erplibre=True
|
|
)
|
|
|
|
def prompt_execute_code(self):
|
|
print(f"🤖 {t('What do you need for development?')}")
|
|
# help_info = """Commande :
|
|
# [1] Status Git local et distant
|
|
# [2] Démarrer le générateur de code
|
|
# [3] Format - Formatage automatique selon changement [ou manuelle]
|
|
# [4] Qualité - Qualité logiciel, détecter les fichiers qui manquent les licences AGPLv3
|
|
# [0] Retour
|
|
# """
|
|
# help_info = """Commande :
|
|
# [1] Status Git local et distant
|
|
# [0] Retour
|
|
# """
|
|
|
|
choices = self.config_file.get_config("code_from_makefile")
|
|
|
|
menu_entry = {
|
|
"prompt_description": t("Open SHELL"),
|
|
}
|
|
choices.append(menu_entry)
|
|
|
|
menu_entry = {
|
|
"prompt_description": t("Upgrade Module"),
|
|
}
|
|
choices.append(menu_entry)
|
|
|
|
choices.append(
|
|
{
|
|
"prompt_description": t("Debug"),
|
|
}
|
|
)
|
|
|
|
help_info = self.fill_help_info(choices)
|
|
|
|
while True:
|
|
status = click.prompt(help_info)
|
|
print()
|
|
if status == "0":
|
|
return False
|
|
elif status == str(len(choices)):
|
|
self.debug_ide()
|
|
elif status == str(len(choices) - 1):
|
|
self.upgrade_module()
|
|
elif status == str(len(choices) - 2):
|
|
self.open_shell_on_database()
|
|
else:
|
|
cmd_no_found = True
|
|
try:
|
|
int_cmd = int(status)
|
|
if 0 < int_cmd <= len(choices):
|
|
cmd_no_found = False
|
|
instance = choices[int_cmd - 1]
|
|
self.execute_from_configuration(instance)
|
|
except ValueError:
|
|
pass
|
|
if cmd_no_found:
|
|
print(t("Command not found !"))
|
|
|
|
def prompt_execute_git(self):
|
|
print(f"🤖 {t('Git management tools!')}")
|
|
choices = [
|
|
{"prompt_description": t("Local git server")},
|
|
{"prompt_description": t("Add a remote to a local repository")},
|
|
]
|
|
|
|
# Append config-driven entries
|
|
config_entries = self.config_file.get_config("git_from_makefile")
|
|
if config_entries:
|
|
choices.extend(config_entries)
|
|
|
|
help_info = self.fill_help_info(choices)
|
|
|
|
while True:
|
|
status = click.prompt(help_info)
|
|
print()
|
|
if status == "0":
|
|
return False
|
|
elif status == "1":
|
|
self.prompt_execute_git_local_server()
|
|
elif status == "2":
|
|
self._git_add_remote()
|
|
else:
|
|
cmd_no_found = True
|
|
try:
|
|
int_cmd = int(status)
|
|
if 0 < int_cmd <= len(choices):
|
|
cmd_no_found = False
|
|
instance = choices[int_cmd - 1]
|
|
self.execute_from_configuration(instance)
|
|
except ValueError:
|
|
pass
|
|
if cmd_no_found:
|
|
print(t("Command not found !"))
|
|
|
|
def _git_add_remote(self):
|
|
remote_name = (
|
|
input(t("Remote name (default: localhost): ")).strip()
|
|
or "localhost"
|
|
)
|
|
remote_url = input(
|
|
t("Repository address (e.g.: git://192.168.1.100/my-repo.git): ")
|
|
).strip()
|
|
if not remote_url:
|
|
print(t("Repository address is required!"))
|
|
return
|
|
cmd = f"git remote add {remote_name} {remote_url}"
|
|
print(f"{t('Will execute:')} {cmd}")
|
|
try:
|
|
self.execute.exec_command_live(cmd, source_erplibre=False)
|
|
print(t("Remote added successfully!"))
|
|
except Exception as e:
|
|
print(f"{t('Error adding remote: ')}{e}")
|
|
|
|
def prompt_execute_git_local_server(self):
|
|
print(f"🤖 {t('Manage local git repository server!')}")
|
|
choices = [
|
|
{
|
|
"prompt_description": t(
|
|
"Deploy a local git server (~/.git-server)"
|
|
)
|
|
},
|
|
{
|
|
"prompt_description": t(
|
|
"Deploy a production git server (/srv/git, root required)"
|
|
)
|
|
},
|
|
]
|
|
help_info = self.fill_help_info(choices)
|
|
|
|
while True:
|
|
status = click.prompt(help_info)
|
|
print()
|
|
if status == "0":
|
|
return False
|
|
elif status == "1":
|
|
self._prompt_git_server_actions(production_ready=False)
|
|
elif status == "2":
|
|
self._prompt_git_server_actions(production_ready=True)
|
|
else:
|
|
print(t("Command not found !"))
|
|
|
|
def _prompt_git_server_actions(self, production_ready=False):
|
|
mode = (
|
|
t("Production mode (/srv/git, root required)")
|
|
if production_ready
|
|
else t("Local mode (~/.git-server)")
|
|
)
|
|
print(f"🤖 {mode}")
|
|
choices = [
|
|
{
|
|
"prompt_description": t(
|
|
"Run all (init + remote + push + serve)"
|
|
)
|
|
},
|
|
{"prompt_description": t("Init - Create bare repos")},
|
|
{"prompt_description": t("Remote - Add local remotes")},
|
|
{"prompt_description": t("Push - Push to local server")},
|
|
{"prompt_description": t("Serve - Start git daemon")},
|
|
]
|
|
help_info = self.fill_help_info(choices)
|
|
|
|
while True:
|
|
status = click.prompt(help_info)
|
|
print()
|
|
if status == "0":
|
|
return False
|
|
elif status == "1":
|
|
self._deploy_git_server(
|
|
production_ready=production_ready,
|
|
action="all",
|
|
)
|
|
elif status == "2":
|
|
self._deploy_git_server(
|
|
production_ready=production_ready,
|
|
action="init",
|
|
)
|
|
elif status == "3":
|
|
self._deploy_git_server(
|
|
production_ready=production_ready,
|
|
action="remote",
|
|
)
|
|
elif status == "4":
|
|
self._deploy_git_server(
|
|
production_ready=production_ready,
|
|
action="push",
|
|
)
|
|
elif status == "5":
|
|
self._deploy_git_server(
|
|
production_ready=production_ready,
|
|
action="serve",
|
|
)
|
|
else:
|
|
print(t("Command not found !"))
|
|
|
|
def _deploy_git_server(self, production_ready=False, action="all"):
|
|
print(t("Starting git server deployment..."))
|
|
cmd = (
|
|
"python3 ./script/git/git_local_server.py -v" f" --action {action}"
|
|
)
|
|
if production_ready:
|
|
cmd += " --production-ready"
|
|
self.execute.exec_command_live(
|
|
cmd,
|
|
source_erplibre=False,
|
|
)
|
|
|
|
def prompt_execute_gpt_code(self):
|
|
print(f"🤖 {t('AI assistant tools for development!')}")
|
|
choices = [
|
|
{"prompt_description": t("Configure Claude Code configurations")},
|
|
{
|
|
"prompt_description": t(
|
|
"Add an automation with Claude in todo.py"
|
|
)
|
|
},
|
|
{
|
|
"prompt_description": t(
|
|
"RTK - CLI proxy to reduce LLM token consumption"
|
|
)
|
|
},
|
|
]
|
|
help_info = self.fill_help_info(choices)
|
|
|
|
while True:
|
|
status = click.prompt(help_info)
|
|
print()
|
|
if status == "0":
|
|
return False
|
|
elif status == "1":
|
|
self._prompt_claude_configs()
|
|
elif status == "2":
|
|
self._claude_add_automation()
|
|
elif status == "3":
|
|
self.prompt_execute_rtk()
|
|
else:
|
|
print(t("Command not found !"))
|
|
|
|
def _prompt_claude_configs(self):
|
|
print(f"🤖 {t('Deploy Claude Code commands!')}")
|
|
choices = [
|
|
{"prompt_description": t("Commit - OCA/Odoo commit command")},
|
|
{
|
|
"prompt_description": t(
|
|
"Todo Add Command - Add a command to todo.py menu"
|
|
)
|
|
},
|
|
{"prompt_description": t("Show installed custom commands")},
|
|
]
|
|
help_info = self.fill_help_info(choices)
|
|
|
|
while True:
|
|
status = click.prompt(help_info)
|
|
print()
|
|
if status == "0":
|
|
return False
|
|
elif status == "1":
|
|
self._setup_claude_command(
|
|
"commit",
|
|
"template_claude_commands_commit.md",
|
|
personalize=True,
|
|
)
|
|
elif status == "2":
|
|
self._setup_claude_command(
|
|
"todo_add_command",
|
|
"template_claude_commands_todo_add_command.md",
|
|
)
|
|
elif status == "3":
|
|
self._list_claude_commands()
|
|
else:
|
|
print(t("Command not found !"))
|
|
|
|
def _list_claude_commands(self):
|
|
commands_dir = os.path.expanduser("~/.claude/commands")
|
|
if not os.path.isdir(commands_dir):
|
|
print(t("No custom commands found in ~/.claude/commands/"))
|
|
return
|
|
files = sorted(
|
|
f for f in os.listdir(commands_dir) if f.endswith(".md")
|
|
)
|
|
if not files:
|
|
print(t("No custom commands found in ~/.claude/commands/"))
|
|
return
|
|
print(t("Claude Code custom commands:"))
|
|
print("-" * 50)
|
|
for f in files:
|
|
filepath = os.path.join(commands_dir, f)
|
|
mtime = os.path.getmtime(filepath)
|
|
date_str = datetime.datetime.fromtimestamp(mtime).strftime(
|
|
"%Y-%m-%d %H:%M"
|
|
)
|
|
name = f[:-3] # remove .md
|
|
print(f" /{name:<30} {date_str}")
|
|
print("-" * 50)
|
|
print(f"{t('Total:')}" f" {len(files)}")
|
|
|
|
def _setup_claude_command(
|
|
self, command_name, template_filename, personalize=False
|
|
):
|
|
dest_dir = os.path.expanduser("~/.claude/commands")
|
|
dest_file = os.path.join(dest_dir, f"{command_name}.md")
|
|
|
|
if os.path.exists(dest_file):
|
|
print(f"{t('File already exists: ')}{dest_file}")
|
|
overwrite = input(
|
|
t("Do you want to overwrite the file? (y/Y): ")
|
|
).strip()
|
|
if overwrite not in ("y", "Y"):
|
|
print(t("Nothing to do."))
|
|
return
|
|
|
|
template_path = os.path.join(
|
|
os.path.dirname(__file__),
|
|
"..",
|
|
"..",
|
|
"conf",
|
|
template_filename,
|
|
)
|
|
try:
|
|
with open(template_path) as f:
|
|
content = f.read()
|
|
|
|
if personalize:
|
|
name = input(t("Enter your full name: ")).strip()
|
|
email = input(t("Enter your email: ")).strip()
|
|
content = content.replace(
|
|
"Your Name <your@email.com>",
|
|
f"{name} <{email}>",
|
|
)
|
|
content = content.replace(
|
|
"Your Name ",
|
|
f"{name} ",
|
|
)
|
|
|
|
os.makedirs(dest_dir, exist_ok=True)
|
|
with open(dest_file, "w") as f:
|
|
f.write(content)
|
|
|
|
print(f"{t('File created successfully: ')}{dest_file}")
|
|
except Exception as e:
|
|
print(f"{t('Error creating file: ')}{e}")
|
|
|
|
def _claude_add_automation(self):
|
|
description = input(t("Description of the command to add: ")).strip()
|
|
if not description:
|
|
return
|
|
command = input(t("Bash command to execute: ")).strip()
|
|
if not command:
|
|
return
|
|
section = (
|
|
input(
|
|
t("Menu section (git/code/config/network/process): ")
|
|
).strip()
|
|
or "git"
|
|
)
|
|
section_key = f"{section}_from_makefile"
|
|
config_path = os.path.join(os.path.dirname(__file__), "todo.json")
|
|
try:
|
|
with open(config_path) as f:
|
|
config = json.load(f)
|
|
if section_key not in config:
|
|
config[section_key] = []
|
|
config[section_key].append(
|
|
{
|
|
"prompt_description": description,
|
|
"bash_command": command,
|
|
}
|
|
)
|
|
with open(config_path, "w") as f:
|
|
json.dump(config, f, indent=4, ensure_ascii=False)
|
|
f.write("\n")
|
|
print(t("Automation added successfully in todo.json!"))
|
|
except Exception as e:
|
|
print(f"{t('Error adding automation: ')}{e}")
|
|
|
|
def prompt_execute_doc(self):
|
|
print(f"🤖 {t('Looking for documentation?')}")
|
|
choices = [
|
|
{"prompt_description": t("Migration module coverage")},
|
|
{"prompt_description": t("What change between version")},
|
|
{"prompt_description": t("OCA guidelines")},
|
|
{"prompt_description": t("OCA migration Odoo 19 milestone")},
|
|
]
|
|
help_info = self.fill_help_info(choices)
|
|
|
|
while True:
|
|
status = click.prompt(help_info)
|
|
print()
|
|
if status == "0":
|
|
return False
|
|
elif status == "1":
|
|
str_version = input(
|
|
"Select version to upgrade Odoo CE (5-17) : "
|
|
)
|
|
try:
|
|
int_version = int(str_version)
|
|
print(
|
|
"https://oca.github.io/OpenUpgrade/coverage_analysis/modules"
|
|
f"{int_version * 10}-{(int_version + 1) * 10}.html"
|
|
)
|
|
except ValueError:
|
|
print(
|
|
"https://oca.github.io/OpenUpgrade/030_coverage_analysis.html"
|
|
)
|
|
elif status == "2":
|
|
str_version = input(
|
|
"Select version to show what change for Odoo CE version 8-18) : "
|
|
)
|
|
try:
|
|
int_version = int(str_version)
|
|
print(
|
|
f"https://github.com/OCA/maintainer-tools/wiki/Migration-to-version-{int_version}.0"
|
|
)
|
|
except ValueError:
|
|
print("https://github.com/OCA/maintainer-tools/wiki")
|
|
elif status == "3":
|
|
print(
|
|
"https://github.com/OCA/odoo-community.org/blob/master/website/Contribution/CONTRIBUTING.rst"
|
|
)
|
|
elif status == "4":
|
|
print("https://github.com/OCA/maintainer-tools/issues/658")
|
|
else:
|
|
print(t("Command not found !"))
|
|
|
|
def prompt_execute_database(self):
|
|
print(f"🤖 {t('Make changes to databases!')}")
|
|
choices = [
|
|
{
|
|
"prompt_description": t(
|
|
"Download database to create backup (.zip)"
|
|
)
|
|
},
|
|
{"prompt_description": t("Restore from backup (.zip)")},
|
|
{"prompt_description": t("Create backup (.zip)")},
|
|
{"prompt_description": t("Erase a database")},
|
|
]
|
|
help_info = self.fill_help_info(choices)
|
|
|
|
while True:
|
|
status = click.prompt(help_info)
|
|
print()
|
|
if status == "0":
|
|
return False
|
|
elif status == "1":
|
|
self.db_manager.download_database_backup_cli()
|
|
elif status == "2":
|
|
self.db_manager.restore_from_database()
|
|
elif status == "3":
|
|
self.db_manager.create_backup_from_database()
|
|
elif status == "4":
|
|
self.db_manager.drop_database()
|
|
else:
|
|
print(t("Command not found !"))
|
|
|
|
def prompt_execute_process(self):
|
|
print(f"🤖 {t('Manage execution processes!')}")
|
|
choices = [
|
|
{"prompt_description": t("Kill Odoo process from actual port")},
|
|
{"prompt_description": t("Kill git daemon server process")},
|
|
]
|
|
help_info = self.fill_help_info(choices)
|
|
|
|
while True:
|
|
status = click.prompt(help_info)
|
|
print()
|
|
if status == "0":
|
|
return False
|
|
elif status == "1":
|
|
self.process_kill_from_port()
|
|
elif status == "2":
|
|
self.process_kill_git_daemon()
|
|
else:
|
|
print(t("Command not found !"))
|
|
|
|
def process_kill_git_daemon(self):
|
|
self.execute.exec_command_live(
|
|
"pkill -f 'git daemon'",
|
|
source_erplibre=False,
|
|
)
|
|
print(t("Git daemon process killed."))
|
|
|
|
def prompt_execute_rtk(self):
|
|
print(
|
|
f"🤖 {t('Manage RTK (Rust Token Killer) for token optimization!')}"
|
|
)
|
|
choices = [
|
|
{"prompt_description": t("Install RTK")},
|
|
{"prompt_description": t("Check RTK version")},
|
|
{"prompt_description": t("Show cumulative token savings")},
|
|
{"prompt_description": t("Discover optimization opportunities")},
|
|
{"prompt_description": t("Initialize global auto-rewrite hook")},
|
|
{"prompt_description": t("Check RTK status")},
|
|
]
|
|
help_info = self.fill_help_info(choices)
|
|
|
|
while True:
|
|
status = click.prompt(help_info)
|
|
print()
|
|
if status == "0":
|
|
return False
|
|
elif status == "1":
|
|
self.rtk_install()
|
|
elif status == "2":
|
|
self.rtk_check_version()
|
|
elif status == "3":
|
|
self.rtk_show_gain()
|
|
elif status == "4":
|
|
self.rtk_discover()
|
|
elif status == "5":
|
|
self.rtk_init_global()
|
|
elif status == "6":
|
|
self.rtk_check_status()
|
|
else:
|
|
print(t("Command not found !"))
|
|
|
|
def rtk_install(self):
|
|
print(f"🤖 {t('Installation method:')}")
|
|
choices = [
|
|
{"prompt_description": t("curl - Automatic install script")},
|
|
{"prompt_description": t("brew - Homebrew (macOS/Linux)")},
|
|
{
|
|
"prompt_description": t(
|
|
"cargo - Build from source (Rust required)"
|
|
)
|
|
},
|
|
]
|
|
help_info = self.fill_help_info(choices)
|
|
status = click.prompt(help_info)
|
|
print()
|
|
if status == "0":
|
|
return
|
|
elif status == "1":
|
|
self.execute.exec_command_live(
|
|
"curl -fsSL https://raw.githubusercontent.com/rtk-ai/rtk/refs/heads/master/install.sh | sh",
|
|
source_erplibre=False,
|
|
)
|
|
elif status == "2":
|
|
self.execute.exec_command_live(
|
|
"brew install rtk",
|
|
source_erplibre=False,
|
|
)
|
|
elif status == "3":
|
|
self.execute.exec_command_live(
|
|
"cargo install --git https://github.com/rtk-ai/rtk",
|
|
source_erplibre=False,
|
|
)
|
|
else:
|
|
print(t("Command not found !"))
|
|
|
|
def rtk_check_version(self):
|
|
self.execute.exec_command_live(
|
|
"rtk --version",
|
|
source_erplibre=False,
|
|
)
|
|
|
|
def rtk_show_gain(self):
|
|
self.execute.exec_command_live(
|
|
"rtk gain",
|
|
source_erplibre=False,
|
|
)
|
|
|
|
def rtk_discover(self):
|
|
self.execute.exec_command_live(
|
|
"rtk discover",
|
|
source_erplibre=False,
|
|
)
|
|
|
|
def rtk_init_global(self):
|
|
self.execute.exec_command_live(
|
|
"rtk init --global",
|
|
source_erplibre=False,
|
|
)
|
|
|
|
def rtk_check_status(self):
|
|
rtk_path = shutil.which("rtk")
|
|
if rtk_path is None:
|
|
print(t("RTK is not installed. Use option 1 to install it."))
|
|
return
|
|
|
|
result = self.execute.exec_command_live(
|
|
"rtk --version",
|
|
source_erplibre=False,
|
|
quiet=True,
|
|
return_status_and_output=True,
|
|
)
|
|
if isinstance(result, tuple) and result[0] == 0:
|
|
version_output = " ".join(result[1]).strip()
|
|
print(f"{t('RTK is installed, version: ')}{version_output}")
|
|
else:
|
|
print(f"{t('RTK is installed, version: ')}?")
|
|
|
|
config_path = os.path.expanduser("~/.config/rtk/config.toml")
|
|
if os.path.exists(config_path):
|
|
print(t("Global auto-rewrite hook: active"))
|
|
else:
|
|
print(t("Global auto-rewrite hook: inactive"))
|
|
|
|
def prompt_execute_config(self):
|
|
print(f"🤖 {t('Manage ERPLibre and Odoo configuration!')}")
|
|
choices = [
|
|
{"prompt_description": t("Generate all configuration")},
|
|
{"prompt_description": t("Generate from pre-configuration")},
|
|
{"prompt_description": t("Generate from backup file")},
|
|
{"prompt_description": t("Generate from database")},
|
|
{"prompt_description": t("Setup queue job for parallelism")},
|
|
]
|
|
help_info = self.fill_help_info(choices)
|
|
|
|
while True:
|
|
status = click.prompt(help_info)
|
|
print()
|
|
if status == "0":
|
|
return False
|
|
elif status == "1":
|
|
self.generate_config()
|
|
elif status == "2":
|
|
self.generate_config_from_preconfiguration()
|
|
elif status == "3":
|
|
self.generate_config_from_backup()
|
|
elif status == "4":
|
|
self.generate_config_from_database()
|
|
elif status == "5":
|
|
self.generate_config_queue_job()
|
|
else:
|
|
print(t("Command not found !"))
|
|
|
|
def prompt_execute_network(self):
|
|
print(f"🤖 {t('Network tools!')}")
|
|
choices = [
|
|
{"prompt_description": t("SSH port-forwarding")},
|
|
{
|
|
"prompt_description": t(
|
|
"Network performance request per second"
|
|
)
|
|
},
|
|
]
|
|
help_info = self.fill_help_info(choices)
|
|
|
|
while True:
|
|
status = click.prompt(help_info)
|
|
print()
|
|
if status == "0":
|
|
return False
|
|
elif status == "1":
|
|
self.generate_network_port_forwarding()
|
|
elif status == "2":
|
|
self.generate_network_performance_test()
|
|
else:
|
|
print(t("Command not found !"))
|
|
|
|
def generate_network_port_forwarding(self, add_arg=None):
|
|
# ssh -L local_port:localhost:remote_port SSH_connection
|
|
ssh_connection = click.prompt(
|
|
"SSH connection, check ~/.ssh/config or user@address"
|
|
)
|
|
local_port = click.prompt("local port (8069)")
|
|
remote_port = click.prompt("remote port (8069)")
|
|
cmd = f"ssh -L {local_port}:localhost:{remote_port} {ssh_connection}"
|
|
self.execute.exec_command_live(
|
|
cmd,
|
|
source_erplibre=False,
|
|
single_source_erplibre=False,
|
|
)
|
|
|
|
def generate_network_performance_test(self, add_arg=None):
|
|
# ./script/performance/test_performance.sh
|
|
address = click.prompt("https address, like https://erplibre.com")
|
|
cmd = f"./script/performance/test_performance.sh {address}"
|
|
self.execute.exec_command_live(
|
|
cmd,
|
|
source_erplibre=False,
|
|
single_source_erplibre=True,
|
|
)
|
|
|
|
def prompt_execute_security(self):
|
|
print(f"🤖 {t('Dependency security audit!')}")
|
|
choices = [
|
|
{
|
|
"prompt_description": t(
|
|
"pip-audit - Check vulnerabilities on Python environments"
|
|
)
|
|
},
|
|
]
|
|
help_info = self.fill_help_info(choices)
|
|
|
|
while True:
|
|
status = click.prompt(help_info)
|
|
print()
|
|
if status == "0":
|
|
return False
|
|
elif status == "1":
|
|
self.execute_pip_audit()
|
|
else:
|
|
print(t("Command not found !"))
|
|
|
|
def prompt_execute_test(self):
|
|
print(f"🤖 {t('Test an Odoo module on a temporary database!')}")
|
|
choices = [
|
|
{"prompt_description": t("Test a module")},
|
|
{"prompt_description": t("Test a module with code coverage")},
|
|
{"prompt_description": t("ERPLibre unit tests")},
|
|
]
|
|
help_info = self.fill_help_info(choices)
|
|
|
|
while True:
|
|
status = click.prompt(help_info)
|
|
print()
|
|
if status == "0":
|
|
return False
|
|
elif status == "1":
|
|
self.execute_test_module(coverage=False)
|
|
elif status == "2":
|
|
self.execute_test_module(coverage=True)
|
|
elif status == "3":
|
|
self.execute_unit_tests()
|
|
else:
|
|
print(t("Command not found !"))
|
|
|
|
def execute_test_module(self, coverage=False):
|
|
# Module name
|
|
module_name = input(t("Module name to test: ")).strip()
|
|
if not module_name:
|
|
print(t("Module name is required!"))
|
|
return
|
|
|
|
# Database name
|
|
db_name = input(
|
|
t("Temporary database name (default: test_todo_tmp): ")
|
|
).strip()
|
|
if not db_name:
|
|
db_name = "test_todo_tmp"
|
|
|
|
# Extra modules
|
|
extra_modules = input(
|
|
t("Extra modules to install (comma-separated, empty for none): ")
|
|
).strip()
|
|
|
|
# Log level
|
|
log_level = input(t("Log level (default: test): ")).strip()
|
|
if not log_level:
|
|
log_level = "test"
|
|
|
|
# Build module list
|
|
modules_to_install = module_name
|
|
if extra_modules:
|
|
modules_to_install += f",{extra_modules}"
|
|
|
|
# Step 1: Create temp DB
|
|
print(f"\n--- {t('Creating temporary database')} '{db_name}' ---")
|
|
cmd_restore = f"./script/database/db_restore.py --database {db_name}"
|
|
self.execute.exec_command_live(
|
|
cmd_restore,
|
|
source_erplibre=False,
|
|
single_source_erplibre=True,
|
|
)
|
|
|
|
# Step 2: Install modules
|
|
print(f"\n--- {t('Installing modules')}: {modules_to_install} ---")
|
|
cmd_install = (
|
|
f"./script/addons/install_addons.sh"
|
|
f" {db_name} {modules_to_install}"
|
|
)
|
|
self.execute.exec_command_live(
|
|
cmd_install,
|
|
source_erplibre=False,
|
|
single_source_erplibre=True,
|
|
)
|
|
|
|
# Step 3: Run tests
|
|
print(f"\n--- {t('Running tests')}: {module_name} ---")
|
|
cmd_test = (
|
|
f"ODOO_MODE_TEST=true"
|
|
f" ./run.sh"
|
|
f" -d {db_name}"
|
|
f" -u {module_name}"
|
|
f" --log-level={log_level}"
|
|
)
|
|
if coverage:
|
|
cmd_test = f"ODOO_MODE_COVERAGE=true {cmd_test}"
|
|
status_code, output = self.execute.exec_command_live(
|
|
cmd_test,
|
|
return_status_and_output=True,
|
|
source_erplibre=False,
|
|
single_source_erplibre=True,
|
|
)
|
|
|
|
if status_code == 0:
|
|
print(f"\n✅ {t('Tests completed successfully!')}")
|
|
else:
|
|
print(f"\n❌ {t('Tests failed with return code')} {status_code}")
|
|
|
|
# Step 4: Cleanup
|
|
lang = get_lang()
|
|
keep_input = (
|
|
input(t("Keep the temporary database? (y/N): ")).strip().lower()
|
|
)
|
|
keep = keep_input in (("o", "oui") if lang == "fr" else ("y", "yes"))
|
|
if keep:
|
|
print(f"{t('Database kept')}: {db_name}")
|
|
else:
|
|
print(
|
|
f"\n--- {t('Cleaning up temporary database')} '{db_name}' ---"
|
|
)
|
|
cmd_drop = f"./odoo_bin.sh db --drop --database {db_name}"
|
|
self.execute.exec_command_live(
|
|
cmd_drop,
|
|
source_erplibre=False,
|
|
single_source_erplibre=True,
|
|
)
|
|
|
|
def execute_unit_tests(self):
|
|
print(f"\n--- {t('Running unit tests')} ---")
|
|
cmd = (
|
|
".venv.erplibre/bin/python -m unittest discover"
|
|
" -s test -p 'test_*.py' -v"
|
|
)
|
|
status_code, output = self.execute.exec_command_live(
|
|
cmd,
|
|
source_erplibre=False,
|
|
return_status_and_output=True,
|
|
)
|
|
if status_code == 0:
|
|
print(f"\n✅ {t('All unit tests passed')}")
|
|
else:
|
|
print(
|
|
f"\n❌ {t('Some unit tests failed, exit code')}: {status_code}"
|
|
)
|
|
|
|
def execute_pip_audit(self):
|
|
versions, installed_versions, odoo_installed_version = (
|
|
get_odoo_version()
|
|
)
|
|
|
|
# Build list of installed environments
|
|
environments = {}
|
|
key_i = 0
|
|
for version_info in versions[::-1]:
|
|
erplibre_version = version_info.get("erplibre_version")
|
|
venv_path = f".venv.{erplibre_version}"
|
|
req_path = f"requirement/requirements.{erplibre_version}.txt"
|
|
odoo_version = f"odoo{version_info.get('odoo_version')}"
|
|
|
|
if not os.path.isdir(venv_path):
|
|
continue
|
|
|
|
key_i += 1
|
|
key_s = str(key_i)
|
|
label = f"{key_s}: {erplibre_version}"
|
|
if odoo_version == odoo_installed_version:
|
|
label += f" - {t('Current')}"
|
|
if version_info.get("Default"):
|
|
label += f" - {t('Default')}"
|
|
|
|
environments[key_s] = {
|
|
"label": label,
|
|
"venv_path": venv_path,
|
|
"req_path": req_path,
|
|
"erplibre_version": erplibre_version,
|
|
}
|
|
|
|
if not environments:
|
|
print(
|
|
t(
|
|
"No installed environment found. Install an Odoo version first."
|
|
)
|
|
)
|
|
return
|
|
|
|
# Show selection menu
|
|
str_input = (
|
|
f"💬 {t('Choose an environment for the audit:')}\n\t"
|
|
+ "\n\t".join([v["label"] for v in environments.values()])
|
|
+ f"\n\t0: {t('Back')}"
|
|
+ f"\n{t('Select: ')}"
|
|
)
|
|
env_input = ""
|
|
while env_input not in environments and env_input != "0":
|
|
if env_input:
|
|
print(
|
|
f"{t('Error, cannot understand value')}" f" '{env_input}'"
|
|
)
|
|
env_input = input(str_input).strip()
|
|
|
|
if env_input == "0":
|
|
return
|
|
|
|
selected = environments[env_input]
|
|
venv_path = selected["venv_path"]
|
|
req_path = selected["req_path"]
|
|
|
|
if not os.path.isfile(req_path):
|
|
print(f"{t('Dependencies file not found: ')}{req_path}")
|
|
return
|
|
|
|
# TODO support bash from parameter if open gnome-terminal
|
|
cmd = f"pip-audit -r {req_path} -l;bash"
|
|
print(f"{t('Execution: ')}{cmd}")
|
|
self.execute.exec_command_live(
|
|
cmd,
|
|
source_erplibre=True,
|
|
single_source_erplibre=False,
|
|
)
|
|
|
|
def generate_config(self, add_arg=None):
|
|
# Repeating to get all item before get group
|
|
cmd = (
|
|
f"./script/git/git_merge_repo_manifest.py --output .repo/local_manifests/erplibre_manifest.xml --with_OCA;"
|
|
f"./script/git/git_repo_update_group.py;"
|
|
f"./script/generate_config.sh"
|
|
)
|
|
if add_arg:
|
|
cmd += (
|
|
f";./script/git/git_repo_update_group.py {add_arg};"
|
|
f"./script/generate_config.sh"
|
|
)
|
|
self.execute.exec_command_live(
|
|
cmd,
|
|
source_erplibre=False,
|
|
single_source_erplibre=True,
|
|
)
|
|
|
|
def generate_config_from_preconfiguration(self):
|
|
choices = [
|
|
{"prompt_description": t("base")},
|
|
{"prompt_description": t("base + code_generator")},
|
|
{"prompt_description": t("base + image_db")},
|
|
{"prompt_description": t("all")},
|
|
# {"prompt_description": "base + migration"},
|
|
]
|
|
help_info = self.fill_help_info(choices)
|
|
|
|
while True:
|
|
status = click.prompt(help_info)
|
|
print()
|
|
if status == "0":
|
|
return False
|
|
elif status == "1":
|
|
group = "base"
|
|
str_group = f"--group {group}"
|
|
self.generate_config(add_arg=str_group)
|
|
elif status == "2":
|
|
group = "base,code_generator"
|
|
str_group = f"--group {group}"
|
|
self.generate_config(add_arg=str_group)
|
|
elif status == "3":
|
|
group = "base,image_db"
|
|
str_group = f"--group {group}"
|
|
self.generate_config(add_arg=str_group)
|
|
elif status == "4":
|
|
self.generate_config()
|
|
# elif status == "5":
|
|
# group = "base,migration"
|
|
# str_group = f"--group {group}"
|
|
# self.generate_config(add_arg=str_group)
|
|
else:
|
|
print(t("Command not found !"))
|
|
|
|
def debug_ide(self):
|
|
choices = [
|
|
{"prompt_description": t("Debug todo.py")},
|
|
]
|
|
help_info = self.fill_help_info(choices)
|
|
|
|
while True:
|
|
status = click.prompt(help_info)
|
|
print()
|
|
if status == "0":
|
|
return False
|
|
elif status == "1":
|
|
self.open_pycharm_file(
|
|
os.getcwd(),
|
|
os.path.join(os.getcwd(), "script/todo/todo.py"),
|
|
)
|
|
else:
|
|
print(t("Command not found !"))
|
|
|
|
def generate_config_from_backup(self):
|
|
file_name = self.db_manager.open_file_image_db()
|
|
add_arg = f"--from_backup_name {file_name} --add_repo odoo18.0/addons/MathBenTech_development"
|
|
self.generate_config(add_arg=add_arg)
|
|
|
|
def generate_config_from_database(self):
|
|
database_name = self.db_manager.select_database()
|
|
str_arg = f"--database {database_name}"
|
|
self.generate_config(add_arg=str_arg)
|
|
return False
|
|
|
|
def generate_config_queue_job(self):
|
|
cmd = "./script/config/setup_odoo_config_conf_devops.py"
|
|
self.execute.exec_command_live(
|
|
cmd,
|
|
source_erplibre=False,
|
|
single_source_erplibre=True,
|
|
)
|
|
|
|
def prompt_execute_selenium_and_run_db(
|
|
self, db_name, extra_cmd_web_login=""
|
|
):
|
|
cmd_server = f"./run.sh -d {db_name};bash"
|
|
self.execute.exec_command_live(cmd_server)
|
|
cmd_client = (
|
|
f"sleep 3;./script/selenium/web_login.py{extra_cmd_web_login};bash"
|
|
)
|
|
self.execute.exec_command_live(cmd_client)
|
|
|
|
def prompt_execute_selenium(self, command=None, extra_cmd_web_login=""):
|
|
commands = []
|
|
if not command:
|
|
cmd = "./script/selenium/web_login.py"
|
|
else:
|
|
cmd = command
|
|
|
|
if type(extra_cmd_web_login) is list:
|
|
for item in extra_cmd_web_login:
|
|
commands.append(cmd + item)
|
|
else:
|
|
commands.append(cmd + extra_cmd_web_login)
|
|
|
|
if len(commands) == 1:
|
|
self.execute.exec_command_live(commands[0])
|
|
elif len(commands) > 1:
|
|
new_cmd = "parallel ::: "
|
|
for i, cmd in enumerate(commands):
|
|
new_cmd += f' "sleep {1 * i};{cmd}"'
|
|
self.execute.exec_command_live(new_cmd)
|
|
|
|
def crash_diagnostic(self, e):
|
|
# TODO show message at start if os.path.exists(ERROR_LOG_PATH)
|
|
if os.path.exists(ERROR_LOG_PATH) and not os.path.exists(
|
|
VENV_ERPLIBRE
|
|
):
|
|
print("Got error : ")
|
|
print(e)
|
|
print("Got error at first execution.", ERROR_LOG_PATH)
|
|
try:
|
|
file = open(ERROR_LOG_PATH, "r")
|
|
content = file.read()
|
|
# TODO si vide, ajouter notre erreur
|
|
print(content)
|
|
except FileNotFoundError:
|
|
print("Error: File not found.")
|
|
finally:
|
|
if "file" in locals() and file:
|
|
file.close()
|
|
# Force auto installation
|
|
print("Auto installation")
|
|
time.sleep(0.5)
|
|
cmd = "./script/todo/source_todo.sh"
|
|
# self.restart_script(e)
|
|
self.execute.exec_command_live(cmd, source_erplibre=True)
|
|
sys.exit(1)
|
|
if os.path.exists(VENV_ERPLIBRE):
|
|
print("Import error : ")
|
|
print(e)
|
|
# TODO auto-detect gnome-terminal, or choose another. Is it done already?
|
|
self.restart_script(e)
|
|
# self.prompt_install()
|
|
|
|
# print(
|
|
# f"You forgot to activate source \nsource ./{VENV_ERPLIBRE}/bin/activate"
|
|
# )
|
|
# time.sleep(0.5)
|
|
# cmd = "./script/todo/source_todo.sh"
|
|
print("Re-execute TODO 🤖 or execute :")
|
|
print()
|
|
print(f"source {VENV_ERPLIBRE}/bin/activate;make")
|
|
print()
|
|
cmd = "./script/todo/todo.py"
|
|
# # self.restart_script(e)
|
|
try:
|
|
# TODO duplicate
|
|
import click
|
|
import humanize
|
|
import openai
|
|
import urwid
|
|
from pykeepass import PyKeePass
|
|
except ImportError:
|
|
print("Rerun and exit")
|
|
self.execute.exec_command_live(cmd, source_erplibre=True)
|
|
sys.exit(1)
|
|
print("No error")
|
|
else:
|
|
self.prompt_install()
|
|
|
|
def open_shell_on_database(self):
|
|
database = self.db_manager.select_database()
|
|
if database:
|
|
cmd_server = f"./odoo_bin.sh shell -d {database}"
|
|
status, databases = self.execute.exec_command_live(
|
|
cmd_server,
|
|
return_status_and_output=True,
|
|
source_erplibre=False,
|
|
single_source_erplibre=True,
|
|
new_window=True,
|
|
)
|
|
|
|
def open_pycharm_file(self, folder, filename):
|
|
cmd = "~/.local/share/JetBrains/Toolbox/scripts/pycharm"
|
|
# cmd = "/snap/bin/pycharm-community"
|
|
# if pycharm_arg:
|
|
# cmd += f" {pycharm_arg}"
|
|
if folder:
|
|
cmd += f" {folder}"
|
|
if filename:
|
|
cmd += f" --line 1 {filename}"
|
|
self.execute.exec_command_live(cmd, source_erplibre=False)
|
|
|
|
def upgrade_module(self):
|
|
upgrade = todo_upgrade.TodoUpgrade(self)
|
|
upgrade.execute_module_upgrade()
|
|
|
|
def upgrade_poetry(self):
|
|
# Only show the version to the user
|
|
status = self.execute.exec_command_live(
|
|
f"make version",
|
|
source_erplibre=False,
|
|
)
|
|
# TODO maybe autodetect to update it
|
|
git_repo_update_input = input(
|
|
"💬 Would you like to fetch all your git repositories, you need it (y/Y) : "
|
|
)
|
|
if git_repo_update_input.strip().lower() == "y":
|
|
status = self.execute.exec_command_live(
|
|
f"./script/manifest/update_manifest_local_dev.sh",
|
|
source_erplibre=False,
|
|
)
|
|
|
|
poetry_lock = "./poetry.lock"
|
|
try:
|
|
os.remove(poetry_lock)
|
|
except Exception as e:
|
|
pass
|
|
odoo_long_version = ""
|
|
if os.path.exists("./.erplibre-version"):
|
|
with open("./.erplibre-version") as f:
|
|
odoo_long_version = f.read()
|
|
path_file_odoo_lock = f"./requirement/poetry.{odoo_long_version}.lock"
|
|
if odoo_long_version:
|
|
try:
|
|
os.remove(path_file_odoo_lock)
|
|
except Exception as e:
|
|
pass
|
|
|
|
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,
|
|
single_source_erplibre=False,
|
|
single_source_odoo=True,
|
|
source_odoo=odoo_long_version,
|
|
)
|
|
|
|
if os.path.exists(poetry_lock):
|
|
shutil.copy2(poetry_lock, path_file_odoo_lock)
|
|
|
|
def callback_execute_custom_database(self, config):
|
|
database_name = self.db_manager.select_database()
|
|
self.prompt_execute_selenium_and_run_db(database_name)
|
|
|
|
def process_kill_from_port(self):
|
|
cfg = configparser.ConfigParser()
|
|
cfg.read("./config.conf")
|
|
http_port = cfg.getint("options", "http_port")
|
|
|
|
status = self.execute.exec_command_live(
|
|
f"./script/process/kill_process_by_port.py {http_port} --kill-tree --nb_parent 2",
|
|
source_erplibre=False,
|
|
)
|
|
|
|
def restart_script(self, last_error):
|
|
print(f"🤖 {t('Reboot TODO ...')}")
|
|
# os.execv(sys.executable, ['python'] + sys.argv)
|
|
# TODO mettre check que le répertoire est créé, s'il existe, auto-loop à corriger
|
|
if os.path.exists(VENV_ERPLIBRE) and not os.path.exists(
|
|
ERROR_LOG_PATH
|
|
):
|
|
# TODO mettre check import suivant ne vont pas planter
|
|
try:
|
|
with open(ERROR_LOG_PATH, "w") as f_file:
|
|
f_file.write(str(last_error))
|
|
pass # The file is created and closed here, no content is written
|
|
print(
|
|
f"Try to reopen process with before :\nsource ./{VENV_ERPLIBRE}/bin/activate && exec python "
|
|
+ " ".join(sys.argv)
|
|
)
|
|
os.execv(
|
|
"/bin/bash",
|
|
[
|
|
"/bin/bash",
|
|
"-c",
|
|
f"source ./{VENV_ERPLIBRE}/bin/activate && exec python "
|
|
+ " ".join(sys.argv),
|
|
],
|
|
)
|
|
except Exception as e:
|
|
print("Error detect at first execution.")
|
|
print(e)
|
|
|
|
def on_dir_selected(self, dir_path):
|
|
self.dir_path = dir_path
|
|
todo_file_browser.exit_program()
|
|
|
|
def callback_make_mobile_home(self, config):
|
|
# Read file
|
|
default_project_name = "ERPLibre"
|
|
default_package_name = "ca.erplibre.home"
|
|
# Read default information
|
|
if os.path.exists(STRINGS_FILE):
|
|
tree = ET.parse(STRINGS_FILE)
|
|
root = tree.getroot()
|
|
for elem in root.findall("string"):
|
|
if elem.get("name") == "app_name":
|
|
default_project_name = elem.text
|
|
if elem.get("name") == "package_name":
|
|
default_package_name = elem.text
|
|
|
|
default_project_url_name = "https://erplibre.ca"
|
|
# Read default information
|
|
dotenv_file = dotenv.find_dotenv(
|
|
filename=os.path.join(MOBILE_HOME_PATH, "src", ".env.production")
|
|
)
|
|
default_project_url_name = dotenv.get_key(
|
|
dotenv_file, "VITE_WEBSITE_URL"
|
|
)
|
|
default_project_note_subject = dotenv.get_key(
|
|
dotenv_file, "VITE_LABEL_NOTE"
|
|
)
|
|
|
|
default_debug = False
|
|
project_name = default_project_name
|
|
project_url_name = default_project_url_name
|
|
project_principal_subject = default_project_note_subject
|
|
package_name = default_package_name
|
|
do_debug = default_debug
|
|
do_change_picture_menu = False
|
|
|
|
do_personalize = input(
|
|
"Do you want to personalize the mobile application (Y) : "
|
|
)
|
|
if do_personalize.strip().lower() == "y":
|
|
project_name = (
|
|
input(
|
|
f'Your project name (Separate by space in title), default "{default_project_name}" : '
|
|
).strip()
|
|
or default_project_name
|
|
)
|
|
package_name = (
|
|
input(
|
|
f'Your package name (separate by . lower case, 3 works like DOMAIN.NAME.OBJECT), default "{default_package_name}" : '
|
|
).strip()
|
|
or default_package_name
|
|
)
|
|
project_url_name = (
|
|
input(
|
|
f'Your project url website, default "{default_project_url_name}" : '
|
|
).strip()
|
|
or default_project_url_name
|
|
)
|
|
project_principal_subject = (
|
|
input(
|
|
f'Your project subject, default "{default_project_note_subject}" : '
|
|
).strip()
|
|
or default_project_note_subject
|
|
)
|
|
do_debug = (
|
|
input("Compilation with debug information, default No (Y) : ")
|
|
.strip()
|
|
.lower()
|
|
== "y"
|
|
)
|
|
do_change_picture_menu = (
|
|
input(
|
|
"Want to change picture from menu, you need android-studio (Y) : "
|
|
)
|
|
.strip()
|
|
.lower()
|
|
== "y"
|
|
)
|
|
|
|
# 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.execute.exec_command_live(cmd_client, source_erplibre=False)
|
|
|
|
# dotenv_mobile = dotenv.dotenv_values(dotenv_file)
|
|
# dotenv_mobile["VITE_TITLE"] = project_name
|
|
# dotenv_mobile["VITE_WEBSITE_URL"] = project_url_name
|
|
dotenv.set_key(
|
|
dotenv_file, "VITE_TITLE", project_name, quote_mode="always"
|
|
)
|
|
dotenv.set_key(
|
|
dotenv_file,
|
|
"VITE_WEBSITE_URL",
|
|
project_url_name,
|
|
quote_mode="always",
|
|
)
|
|
dotenv.set_key(
|
|
dotenv_file,
|
|
"VITE_LABEL_NOTE",
|
|
project_principal_subject,
|
|
quote_mode="always",
|
|
)
|
|
dotenv.set_key(
|
|
dotenv_file,
|
|
"VITE_DEBUG_DEV",
|
|
"true" if do_debug else "false",
|
|
quote_mode="never",
|
|
)
|
|
|
|
if do_change_picture_menu:
|
|
status = self.execute.exec_command_live(
|
|
f"cd {MOBILE_HOME_PATH} && npx cap open android;bash",
|
|
source_erplibre=False,
|
|
new_window=True,
|
|
)
|
|
print(
|
|
"Guide for Android-Studio, wait loading is finish. Right-click to app/New/Image Asset and load your image."
|
|
)
|
|
input(
|
|
"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.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.execute.exec_command_live(cmd_client, source_erplibre=False)
|
|
|
|
status = self.execute.exec_command_live(
|
|
"./mobile/compile_and_run.sh", source_erplibre=False
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
start_time = time.time()
|
|
try:
|
|
todo = TODO()
|
|
if ENABLE_CRASH:
|
|
todo.crash_diagnostic(CRASH_E)
|
|
todo.run()
|
|
except KeyboardInterrupt:
|
|
print(t("Keyboard interrupt"))
|
|
finally:
|
|
end_time = time.time()
|
|
duration_sec = end_time - start_time
|
|
if humanize:
|
|
duration_delta = datetime.timedelta(seconds=duration_sec)
|
|
humain_time = humanize.precisedelta(duration_delta)
|
|
print(f"\n{t('TODO execution time')} {humain_time}\n")
|
|
else:
|
|
print(f"\n{t('TODO execution time')} {duration_sec:.2f} sec.\n")
|