[FIX] script config: support merge dict configuration between 3 config
- fix upgrade when missing log
This commit is contained in:
parent
ff83d5e627
commit
41fbc39c21
4 changed files with 93 additions and 43 deletions
|
|
@ -5,6 +5,7 @@
|
|||
import json
|
||||
import logging
|
||||
import os
|
||||
from typing import Any, Dict, Literal, Mapping
|
||||
|
||||
CONFIG_FILE = "./script/todo/todo.json"
|
||||
CONFIG_OVERRIDE_FILE = "./private/todo/todo_override.json"
|
||||
|
|
@ -23,34 +24,74 @@ _logger = logging.getLogger(__name__)
|
|||
|
||||
|
||||
class ConfigFile:
|
||||
def get_config(self, lst_params):
|
||||
# Open file
|
||||
config_file = CONFIG_FILE
|
||||
if os.path.exists(CONFIG_OVERRIDE_FILE):
|
||||
config_file = CONFIG_OVERRIDE_FILE
|
||||
def get_config(self, key_param: str):
|
||||
# Open file and update dct_data
|
||||
dct_data_init = {}
|
||||
dct_data_second = {}
|
||||
dct_data_final = {}
|
||||
|
||||
if os.path.exists(CONFIG_FILE):
|
||||
with open(CONFIG_FILE) as cfg:
|
||||
dct_data_init = json.load(cfg)
|
||||
|
||||
if os.path.exists(CONFIG_OVERRIDE_FILE):
|
||||
with open(CONFIG_OVERRIDE_FILE) as cfg:
|
||||
dct_data_second = json.load(cfg)
|
||||
|
||||
find_in_private = False
|
||||
if os.path.exists(CONFIG_OVERRIDE_PRIVATE_FILE):
|
||||
with open(CONFIG_OVERRIDE_PRIVATE_FILE) as cfg:
|
||||
dct_data = json.load(cfg)
|
||||
for param in lst_params:
|
||||
if param in dct_data.keys():
|
||||
find_in_private = True
|
||||
dct_data = dct_data[param]
|
||||
dct_data_final = json.load(cfg)
|
||||
|
||||
if not find_in_private:
|
||||
with open(config_file) as cfg:
|
||||
dct_data = json.load(cfg)
|
||||
for param in lst_params:
|
||||
try:
|
||||
dct_data = dct_data[param]
|
||||
except KeyError:
|
||||
_logger.error(
|
||||
f"KeyError on file {config_file} with keys"
|
||||
f" {lst_params}"
|
||||
)
|
||||
return {}
|
||||
dct_data_first_merge = self.deep_merge_with_lists(
|
||||
dct_data_init, dct_data_final, list_strategy="extend"
|
||||
)
|
||||
dct_data = self.deep_merge_with_lists(
|
||||
dct_data_first_merge, dct_data_second, list_strategy="extend"
|
||||
)
|
||||
|
||||
return dct_data.get(key_param)
|
||||
|
||||
def get_config_value(self, lst_params: list):
|
||||
dct_data = self.get_config(lst_params[0])
|
||||
for param in lst_params[1:]:
|
||||
if param in dct_data.keys():
|
||||
find_in_private = True
|
||||
dct_data = dct_data.get(param)
|
||||
return dct_data
|
||||
|
||||
def get_logo_ascii_file_path(self):
|
||||
return LOGO_ASCII_FILE
|
||||
|
||||
def deep_merge_with_lists(
|
||||
self,
|
||||
dest: Mapping[str, Any],
|
||||
src: Mapping[str, Any],
|
||||
list_strategy: Literal["replace", "extend"] = "replace",
|
||||
) -> Dict[str, Any]:
|
||||
result: Dict[str, Any] = {}
|
||||
for k, v in dest.items():
|
||||
result[k] = v.copy() if isinstance(v, dict) else v
|
||||
|
||||
for k, v in src.items():
|
||||
if (
|
||||
k in result
|
||||
and isinstance(result[k], dict)
|
||||
and isinstance(v, dict)
|
||||
):
|
||||
result[k] = self.deep_merge_with_lists(
|
||||
result[k], v, list_strategy
|
||||
)
|
||||
elif (
|
||||
k in result
|
||||
and isinstance(result[k], list)
|
||||
and isinstance(v, list)
|
||||
and list_strategy == "extend"
|
||||
):
|
||||
# on étend : dest_list + src_list
|
||||
result[k] = result[k] + v
|
||||
elif k in result and isinstance(result[k], str):
|
||||
if v:
|
||||
result[k] = v
|
||||
else:
|
||||
result[k] = v
|
||||
return result
|
||||
|
|
|
|||
|
|
@ -397,7 +397,9 @@ class SeleniumLib(object):
|
|||
return self.kdbx
|
||||
print("Open KDBX")
|
||||
# Open file
|
||||
chemin_fichier_kdbx = self.config_file.get_config(["kdbx", "path"])
|
||||
chemin_fichier_kdbx = self.config_file.get_config_value(
|
||||
["kdbx", "path"]
|
||||
)
|
||||
if not chemin_fichier_kdbx:
|
||||
root = tk.Tk()
|
||||
root.withdraw() # Hide the main window
|
||||
|
|
@ -409,7 +411,9 @@ class SeleniumLib(object):
|
|||
# _logger.error(f"KDBX is not configured, please fill {CONFIG_FILE}")
|
||||
return
|
||||
|
||||
mot_de_passe_kdbx = self.config_file.get_config(["kdbx", "password"])
|
||||
mot_de_passe_kdbx = self.config_file.get_config_value(
|
||||
["kdbx", "password"]
|
||||
)
|
||||
if not mot_de_passe_kdbx:
|
||||
mot_de_passe_kdbx = getpass.getpass(
|
||||
prompt="Entrez votre mot de passe : "
|
||||
|
|
|
|||
|
|
@ -153,7 +153,9 @@ class TODO:
|
|||
if self.kdbx:
|
||||
return self.kdbx
|
||||
# Open file
|
||||
chemin_fichier_kdbx = self.config_file.get_config(["kdbx", "path"])
|
||||
chemin_fichier_kdbx = self.config_file.get_config_value(
|
||||
["kdbx", "path"]
|
||||
)
|
||||
if not chemin_fichier_kdbx:
|
||||
root = tk.Tk()
|
||||
root.withdraw() # Hide the main window
|
||||
|
|
@ -167,7 +169,9 @@ class TODO:
|
|||
)
|
||||
return
|
||||
|
||||
mot_de_passe_kdbx = self.config_file.get_config(["kdbx", "password"])
|
||||
mot_de_passe_kdbx = self.config_file.get_config_value(
|
||||
["kdbx", "password"]
|
||||
)
|
||||
if not mot_de_passe_kdbx:
|
||||
mot_de_passe_kdbx = getpass.getpass(
|
||||
prompt="Entrez votre mot de passe : "
|
||||
|
|
@ -191,7 +195,7 @@ class TODO:
|
|||
kp = self.get_kdbx()
|
||||
if not kp:
|
||||
return
|
||||
nom_configuration = self.config_file.get_config(
|
||||
nom_configuration = self.config_file.get_config_value(
|
||||
["kdbx_config", "openai", "kdbx_key"]
|
||||
)
|
||||
entry = kp.find_entries_by_title(nom_configuration, first=True)
|
||||
|
|
@ -445,7 +449,7 @@ class TODO:
|
|||
# TODO proposer le déploiement à distance
|
||||
# TODO proposer l'exécution de docker
|
||||
# TODO proposer la création de docker
|
||||
lst_instance = self.config_file.get_config(["instance"])
|
||||
lst_instance = self.config_file.get_config("instance")
|
||||
help_info = self.fill_help_info(lst_instance)
|
||||
|
||||
while True:
|
||||
|
|
@ -474,7 +478,7 @@ class TODO:
|
|||
print("Commande non trouvée 🤖!")
|
||||
|
||||
def prompt_execute_fonction(self):
|
||||
lst_instance = self.config_file.get_config(["function"])
|
||||
lst_instance = self.config_file.get_config("function")
|
||||
help_info = self.fill_help_info(lst_instance)
|
||||
|
||||
while True:
|
||||
|
|
@ -506,7 +510,7 @@ class TODO:
|
|||
# TODO faire la mise à jour de ERPLibre
|
||||
# TODO faire l'upgrade d'un odoo vers un autre
|
||||
|
||||
lst_instance = self.config_file.get_config(["update_from_makefile"])
|
||||
lst_instance = self.config_file.get_config("update_from_makefile")
|
||||
dct_upgrade_odoo_database = {
|
||||
"prompt_description": "Upgrade Odoo - Migration Database",
|
||||
}
|
||||
|
|
@ -554,7 +558,7 @@ class TODO:
|
|||
# [0] Retour
|
||||
# """
|
||||
|
||||
lst_instance = self.config_file.get_config(["code_from_makefile"])
|
||||
lst_instance = self.config_file.get_config("code_from_makefile")
|
||||
dct_upgrade_odoo_database = {
|
||||
"prompt_description": "Upgrade Module",
|
||||
}
|
||||
|
|
|
|||
|
|
@ -76,17 +76,18 @@ class TodoUpgrade:
|
|||
def execute_module_upgrade(self):
|
||||
print("Welcome to Odoo module upgrade processus with ERPLibre 🤖")
|
||||
|
||||
with open(UPGRADE_DATABASE_CONFIG_LOG, "r") as f:
|
||||
try:
|
||||
old_dct_progression = json.load(f)
|
||||
self.dct_progression = old_dct_progression
|
||||
self.lst_command_executed = old_dct_progression.get(
|
||||
"command_executed"
|
||||
)
|
||||
except json.decoder.JSONDecodeError:
|
||||
print(
|
||||
f'⚠️ The config file "{UPGRADE_DATABASE_CONFIG_LOG}" is invalid, ignore it.'
|
||||
)
|
||||
if os.path.exists(UPGRADE_DATABASE_CONFIG_LOG):
|
||||
with open(UPGRADE_DATABASE_CONFIG_LOG, "r") as f:
|
||||
try:
|
||||
old_dct_progression = json.load(f)
|
||||
self.dct_progression = old_dct_progression
|
||||
self.lst_command_executed = old_dct_progression.get(
|
||||
"command_executed"
|
||||
)
|
||||
except json.decoder.JSONDecodeError:
|
||||
print(
|
||||
f'⚠️ The config file "{UPGRADE_DATABASE_CONFIG_LOG}" is invalid, ignore it.'
|
||||
)
|
||||
|
||||
print(
|
||||
"Migrate a directory repo to migrate all module, or select a directory module."
|
||||
|
|
|
|||
Loading…
Reference in a new issue