#!/usr/bin/env python3 # © 2021-2026 TechnoLibre (http://www.technolibre.ca) # License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl) import datetime import json import logging import os import shutil import sys import zipfile from uuid import uuid4 from script.todo.version_manager import get_odoo_version import click import todo_file_browser new_path = os.path.normpath( os.path.join(os.path.dirname(__file__), "..", "..") ) sys.path.append(new_path) from script.execute import execute from script.git.git_tool import GitTool _logger = logging.getLogger(__name__) PYTHON_BIN = ".venv.erplibre/bin/python3" UPGRADE_DATABASE_CONFIG_LOG = ".venv.erplibre/odoo_database_migration_log.json" # UPGRADE_MODULE_CONFIG_LOG = ".venv.erplibre/odoo_module_migration_log.json" VENV_NAME_MODULE_MIGRATOR = ".venv" LST_PATH_OCA_ODOO_MODULE_MIGRATOR = ["script", "OCA_odoo-module-migrator"] PATH_OCA_ODOO_MODULE_MIGRATOR = "./" + "/".join( LST_PATH_OCA_ODOO_MODULE_MIGRATOR ) PATH_VENV_MODULE_MIGRATOR = os.path.join( PATH_OCA_ODOO_MODULE_MIGRATOR, VENV_NAME_MODULE_MIGRATOR ) PATH_SOURCE_VENV_MODULE_MIGRATOR = os.path.join( PATH_VENV_MODULE_MIGRATOR, "bin", "activate" ) FILENAME_ODOO_VERSION = ".odoo-version" LOCAL_MANIFEST = os.path.join( ".repo", "local_manifests", "erplibre_manifest.xml" ) class TodoUpgrade: def __init__(self, todo): self.file_path = None self.dir_path = None self.todo = todo self.dct_progression = {} 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(): self.dct_progression["date_create"] = str(datetime.datetime.now()) self.dct_progression["date_update"] = str(datetime.datetime.now()) # Always put command_executed at the end if "command_executed" in self.dct_progression.keys(): value = self.dct_progression["command_executed"] del self.dct_progression["command_executed"] self.dct_progression["command_executed"] = value with open(UPGRADE_DATABASE_CONFIG_LOG, "w") as f: json.dump(self.dct_progression, f, indent=4) def on_file_selected(self, file_path): self.file_path = file_path todo_file_browser.exit_program() def on_dir_selected(self, dir_path): self.dir_path = dir_path todo_file_browser.exit_program() def execute_module_upgrade(self): print("Welcome to Odoo module upgrade processus with ERPLibre 🤖") 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." ) print("[m] Migrate one module") print("[r] Migrate one repo") print("[] Directory browser") lst_dir_path = [] from_version = 0 to_version = 0 status = input("💬 What do you choose : ").strip().lower() if status == "m": print("[p] Path (default)") print("[n] Name") status = input("💬 What do you choose : ").strip().lower() if status == "n": module_name = input("💬 Module name : ").strip().lower() from_version = int( input("💬 From odoo version (12 to 18) : ").strip() ) to_version = int( input("💬 To odoo version (12 to 18) : ").strip() ) self.switch_odoo(from_version) # TODO detect path ( lst_module_missing, lst_module_duplicate, lst_module_exist, lst_module_error, ) = self.check_addons_exist([module_name], get_all_info=True) if ( lst_module_missing or lst_module_duplicate or lst_module_error ): if lst_module_missing: print(f"Missing list : {lst_module_missing}") if lst_module_duplicate: print(f"Duplicate list : {lst_module_duplicate}") if lst_module_error: print(f"Error list : {lst_module_error}") return lst_dir_path.extend(lst_module_exist) else: self.dir_path = input("💬 Path : ").strip() elif status == "r": self.dir_path = input("💬 Path : ").strip() else: self.dir_path = None if not self.dir_path and not lst_dir_path: initial_dir = os.getcwd() file_browser = todo_file_browser.FileBrowser( initial_dir, self.on_dir_selected, open_dir=True ) file_browser.run_main_frame() if not lst_dir_path and not os.path.exists(self.dir_path): _logger.error(f"Path '{self.dir_path}' not exists.") return if not from_version: from_version = int( input("💬 From odoo version (12 to 18) : ").strip() ) if not to_version: to_version = int(input("💬 To odoo version (12 to 18) : ").strip()) # TODO ask option direct migration lst_version_to_update = [ (a, a + 1) for a in range(from_version, to_version) ] # lst_version_to_update = [(from_version, to_version)] # for actual_version in range(from_version, to_version): # next_version = actual_version + 1 for actual_version, next_version in lst_version_to_update: set_path_migrate_addons = set() if not lst_dir_path: if os.path.exists( os.path.join(self.dir_path, "__manifest__.py") ): lst_dir_path = [ [os.path.basename(self.dir_path), self.dir_path] ] else: lst_dir_path = [ [a, os.path.join(self.dir_path, a)] for a in os.listdir(self.dir_path) if os.path.exists( os.path.join(self.dir_path, a, "__manifest__.py") ) ] lst_path_git_clone_migrate = [] lst_module_to_migrate_all = [] for dir_path in lst_dir_path: source_manifest_path = os.path.join( dir_path[1], "__manifest__.py" ).replace(f"odoo{from_version}.0", f"odoo{actual_version}.0") is_dir_module = os.path.exists(source_manifest_path) if not is_dir_module: continue source_module_path = dir_path[1].replace( f"odoo{from_version}.0", f"odoo{actual_version}.0" ) source_addons_path = os.path.dirname(dir_path[1]).replace( f"odoo{from_version}.0", f"odoo{actual_version}.0" ) target_module_path = source_module_path.replace( f"odoo{actual_version}.0", f"odoo{next_version}.0" ) target_manifest_path = source_manifest_path.replace( f"odoo{actual_version}.0", f"odoo{next_version}.0" ) target_addons_path = source_addons_path.replace( f"odoo{actual_version}.0", f"odoo{next_version}.0" ) dct_module = { "source_module_path": source_module_path, "source_manifest_path": source_manifest_path, "source_addons_path": source_addons_path, "target_module_path": target_module_path, "target_manifest_path": target_manifest_path, "target_addons_path": target_addons_path, "module_name": dir_path[0], "source_version_odoo": actual_version, "target_version_odoo": next_version, } lst_module_to_migrate_all.append(dct_module) set_path_migrate_addons.add(target_addons_path) # lst_path_git_clone_migrate.append(target_addons_path) # TODO auto detect version from # TODO if detect from directory, check from repo list # TODO detect from manifest version self.switch_odoo(next_version) self.install_OCA_odoo_module_migrator() self.internal_module_upgrade( next_version, lst_module_to_migrate_all, lst_path_git_clone_migrate, ) for commit_path in set_path_migrate_addons: cmd = f"./script/code/git_commit_migration_addons_path.py --path {commit_path} --odoo_version {next_version}.0" self.todo_upgrade_execute(cmd) print(set_path_migrate_addons) status = input( f"💬 Please validate git commit on repos, press to continue : " ).strip() def internal_module_upgrade( self, next_version, lst_module_to_migrate_all, lst_path_git_clone_migrate, ): has_cmd = False cmd_parallel = "parallel :::" for dct_module in lst_module_to_migrate_all: target_addons_path = dct_module.get("target_addons_path") source_addons_path = dct_module.get("source_addons_path") module_name = dct_module.get("module_name") source_version_odoo = f'{dct_module.get("source_version_odoo")}.0' target_version_odoo = f'{dct_module.get("target_version_odoo")}.0' source_module_path_to_copy = dct_module.get("source_module_path") # Prepare git environment for target if target_addons_path not in lst_path_git_clone_migrate: lst_path_git_clone_migrate.append(target_addons_path) self.check_and_clone_source_to_target_migration_code( next_version, source_addons_path, target_addons_path, ) cmd_migration = ( f"echo 'odoo_module_migrate {module_name}' && " f"cp -r {source_module_path_to_copy} {target_addons_path} && " f"cd {PATH_OCA_ODOO_MODULE_MIGRATOR} && " f"source {VENV_NAME_MODULE_MIGRATOR}/bin/activate && " f"python -m odoo_module_migrate --directory {target_addons_path} --modules {module_name} " f"--init-version-name {source_version_odoo} --target-version-name {target_version_odoo} " f"--no-commit && " f"cd ~- " # f"cp -r {source_module_path_to_copy} {target_module_path_to_copy} && " # f"cd {target_module_path_to_copy} && git commit -am '[MIG] {module_name}: Migration to {target_version_odoo}' && cd ~-" ) cmd_parallel += f' "{cmd_migration}"' has_cmd = True if lst_module_to_migrate_all: if has_cmd: self.todo_upgrade_execute(cmd_parallel) print("List of path with migrate code :") print(lst_path_git_clone_migrate) print("ℹ To show repo status :\nmake repo_show_status") input("💬 Check migration code, press to continue : ") # source_module_path = dct_module_result.get( # "source_module_path" # ) # if not source_module_path: # _logger.error( # f"Missing source module path '{source_module_path}'" # ) # else: # if os.path.exists( # os.path.join(source_module_path, ".git") # ): # self.todo_upgrade_execute( # f"cd '{source_module_path}' && git stash && cd ~-", # ) # # target_module_path = dct_module_result.get( # "target_module_path" # ) # if not target_module_path: # _logger.error( # f"Missing target module path '{target_module_path}'" # ) # else: # # TODO check if has file to commit # self.todo_upgrade_execute( # f"cd '{target_module_path}' && git commit -am '[MIG] {len(lst_module_to_migrate_all)} modules: Migration to {next_version}' && cd ~-", # ) # TODO copie to next odoo version # do commit and continue # continue migration to loop if next_version in [18]: # TODO need odoo 18, validate python version without switch status = input( f"💬 Please validate repo is ready to run upgrade views_migration_18, press to continue : " ).strip() # Apply modification with views_migration_18 has_cmd = False # cmd_serial = "" cmd_parallel = "parallel :::" for path_git_clone_migrate in lst_path_git_clone_migrate: cmd_migration = ( f"echo 'views_migration_18 {path_git_clone_migrate}' && " f"./.venv.odoo18.0_python3.12.10/bin/python ./script/code/odoo_upgrade_code_with_dir_module.py --path {path_git_clone_migrate}" ) cmd_parallel += f' "{cmd_migration}"' # cmd_serial += f"{cmd_migration};" has_cmd = True if has_cmd: # self.todo_upgrade_execute( # cmd_serial # ) self.todo_upgrade_execute(cmd_parallel) print("List of module with migration 18 :") print(lst_module_to_migrate_all) print("ℹ To show repo status :\nmake repo_show_status") input("💬 Check migration 18 code, press to continue : ") if next_version == 17: status = input( f"💬 Please validate repo is ready to run upgrade views_migration_17, press to continue : " ).strip() # Apply modification with views_migration_17 has_cmd = False cmd_serial = "" cmd_parallel = "parallel :::" for dct_module in lst_module_to_migrate_all: database_migration_17_name = ( f"migration_odoo_{next_version}_{str(uuid4())[:6]}" ) module_name = dct_module.get("module_name") cmd_migration = ( f"echo 'views_migration_17 {module_name}' && " f"./run.sh -d {database_migration_17_name} -i {module_name} --load=base,web,views_migration_17 --dev upgrade --no-http --stop-after-init" ) cmd_parallel += f' "{cmd_migration}"' cmd_serial += f"{cmd_migration};" has_cmd = True if has_cmd: # self.todo_upgrade_execute( # cmd_serial # ) self.todo_upgrade_execute(cmd_parallel) print("List of module with migration 17 :") print(lst_module_to_migrate_all) print("ℹ To show repo status :\nmake repo_show_status") input("💬 Check migration 17 code, press to continue : ") def execute_odoo_upgrade(self): # TODO update dev environment for git project # TODO Redeploy new production after upgrade # 2 upgrades version = 5 environnement. 0-prod init, 1-dev init, 2-dev01, 3-dev02, 4-prod final print("Welcome to Odoo database upgrade processus with ERPLibre 🤖") self.lst_command_executed = [] self.dct_module_per_version = {} self.dct_module_per_dct_version_path = {} default_database_name = "test" 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 = { "migration_file": old_dct_progression.get( "migration_file" ), # More useful to ask this question each time "target_odoo_version": old_dct_progression.get( "target_odoo_version" ), "date_create": old_dct_progression.get("date_create"), } except json.decoder.JSONDecodeError: print( f'⚠️ The config file "{UPGRADE_DATABASE_CONFIG_LOG}" is invalid, ignore it.' ) print( f"✨ Detected migration \"{self.dct_progression.get('migration_file')}\" " f"to version \"{self.dct_progression.get('target_odoo_version')}\", " f"please select an option." ) print("[1] Erase progression for a new migration") print("[2] Reuse database with new process") print( "[3] Reuse database without state_4, before looping on next version" ) print("[4] Reuse database without configuration") erase_progression_input = ( input("💬 Select an option or press to continue : ") .strip() .lower() ) self.dct_progression = {} if erase_progression_input in ["2", "3", "4"]: with open(UPGRADE_DATABASE_CONFIG_LOG, "r") as f: try: old_dct_progression = json.load(f) self.dct_progression = { "migration_file": old_dct_progression.get( "migration_file" ), # More useful to ask this question each time # "target_odoo_version": old_dct_progression.get( # "target_odoo_version" # ), "date_create": old_dct_progression.get( "date_create" ), } for key, value in old_dct_progression.items(): if erase_progression_input == "3": if ( key.startswith("state_0") or key.startswith("state_1") or key.startswith("state_2") or key.startswith("state_3") ): self.dct_progression[key] = value if key.startswith(f"config_state") and not ( key.startswith(f"config_state_0") or key.startswith(f"config_state_1") or key.startswith(f"config_state_2") or key.startswith(f"config_state_3") ): continue if ( key.startswith("config_") and erase_progression_input != "4" ): self.dct_progression[key] = value # Force to search missing module to fill dict self.dct_progression[ "state_0_search_missing_module" ] = False except json.decoder.JSONDecodeError: print( f"⚠️ The config file '{UPGRADE_DATABASE_CONFIG_LOG}' is invalid, ignore it." ) self.write_config() elif erase_progression_input not in ["1"]: with open(UPGRADE_DATABASE_CONFIG_LOG, "r") as f: try: self.dct_progression = json.load(f) except json.decoder.JSONDecodeError: print( f"⚠️ The config file '{UPGRADE_DATABASE_CONFIG_LOG}' is invalid, ignore it." ) if "migration_file" in self.dct_progression: self.file_path = self.dct_progression["migration_file"] else: print("") print("Select the zip file of you database backup.") self.file_path = input( "💬 Give the path of file, or empty to use a File Browser, or type 'remote' to download from production : " ) if not self.file_path.strip(): self.file_path = None if not self.file_path: initial_dir = os.path.join(os.getcwd(), "image_db") file_browser = todo_file_browser.FileBrowser( initial_dir, self.on_file_selected ) file_browser.run_main_frame() elif self.file_path == "remote": status, self.file_path, default_database_name = ( self.todo.db_manager.download_database_backup_cli() ) if status: _logger.error( "Cannot retrieve database from remote, please retry migration." ) return self.dct_progression["migration_file"] = self.file_path self.write_config() print(f"✅ Open file {self.file_path}") with zipfile.ZipFile(self.file_path, "r") as zip_ref: manifest_file_1 = zip_ref.open("manifest.json") json_manifest_file_1 = json.load(manifest_file_1) odoo_actual_version = json_manifest_file_1.get("version") print(f"✅ Detect version Odoo CE '{odoo_actual_version}'.") # print("What is your actual Odoo version?") lst_version, lst_version_installed, odoo_installed_version = ( get_odoo_version() ) lst_odoo_version = [ {"prompt_description": a.get("odoo_version")} for a in lst_version if float(a.get("odoo_version")) > float(odoo_actual_version) ] help_info = self.todo.fill_help_info(lst_odoo_version) if "target_odoo_version" in self.dct_progression: odoo_target_version = self.dct_progression["target_odoo_version"] else: print("💬 Which version do you want to upgrade to?") odoo_target_version = None cmd_no_found = True while cmd_no_found: status = click.prompt(help_info) try: int_cmd = int(status) if 0 < int_cmd <= len(lst_odoo_version): cmd_no_found = False odoo_target_version = lst_odoo_version[ int_cmd - 1 ].get("prompt_description") except ValueError: pass if cmd_no_found: print("Commande non trouvée 🤖!") self.dct_progression["target_odoo_version"] = odoo_target_version self.write_config() # Search nb diff to use range start_version = int(float(odoo_actual_version)) end_version = int(float(odoo_target_version)) range_version = range(start_version, end_version) lst_module = sorted( list(set(json_manifest_file_1.get("modules").keys())) ) self.dct_module_per_version[start_version] = lst_module self.dct_progression["dct_module_per_version"] = ( self.dct_module_per_version ) self.dct_progression["lst_module_per_version_origin"] = lst_module # TODO need support minor version, example 18.2, the .2 (no need for OCE OCB) print("✨ Show documentation version :") # TODO Generate it locally and show it if asked for next_version in range_version: print( f"https://oca.github.io/OpenUpgrade/coverage_analysis/modules{next_version*10}-{(next_version+1)*10}.html" ) # ⚠️ ℹ 💬 ❗ 🔷 ✨ 🟦 🔹 🔵 ⟳ ⧖ ⚙ ✔ ✅ ❌ ⏵ ⏸ ⏹ ◆ ◇ … ➤ ⚑ ★ ☆ ☰ ⬍ ⍟ ⊗ ⌘ ⏻ ⍰ msg = "0 - Inspect zip" print(f"🔷 {msg}") self.add_comment_progression(msg) print("✅ -> Search odoo version") print("✅ -> Find good environment, read the .zip file") is_state_4_reach_open_upgrade = self.dct_progression.get( "state_4_reach_open_upgrade" ) if not is_state_4_reach_open_upgrade and not self.dct_progression.get( "state_0_install_odoo" ): lst_diff_version = sorted( list( set([f"odoo{a}.0" for a in range_version]).difference( set(lst_version_installed) ) ) ) for odoo_version_to_install in lst_diff_version: iter_range_version = odoo_version_to_install.replace( "odoo", "" ).replace(".0", "") want_continue = input( f"💬 Would you like to install '{odoo_version_to_install}' (y/Y) : " ) if want_continue.strip().lower() != "y": return self.todo_upgrade_execute( f"make install_odoo_{iter_range_version}" ) if not os.path.isfile(FILENAME_ODOO_VERSION): print( "⚠️ You need an installed system before continue, check your Odoo installation." ) return self.dct_progression["state_0_install_odoo"] = True self.write_config() # not self.dct_progression.get("state_0_switch_odoo") if not is_state_4_reach_open_upgrade: self.switch_odoo(odoo_actual_version) # self.dct_progression["state_0_switch_odoo"] = True # self.write_config() print("✅ -> Install environment if missing") if not self.dct_progression.get("state_0_search_missing_module"): self.switch_odoo(odoo_actual_version) dct_bd_modules = json_manifest_file_1.get("modules") lst_module_to_check = [a for a in dct_bd_modules.keys()] ( lst_module_missing, lst_module_duplicate, lst_module_exist, lst_module_error, ) = self.check_addons_exist(lst_module_to_check, get_all_info=True) if not lst_module_missing: lst_module_missing = [] dct_module_exist = {} if not lst_module_exist: lst_module_exist = [] else: for item_lst_module_exist in lst_module_exist: dct_module_exist[item_lst_module_exist[0]] = ( item_lst_module_exist[1].replace(os.getcwd(), ".") ) if not lst_module_duplicate: lst_module_duplicate = [] lst_module_missing = sorted(list(set(lst_module_missing))) self.dct_progression["len_lst_module_missing"] = len( lst_module_missing ) self.dct_progression["lst_module_missing"] = lst_module_missing self.dct_progression["len_dct_module_exist"] = len( lst_module_exist ) self.dct_progression["dct_module_exist"] = dct_module_exist self.dct_progression["len_lst_module_duplicate"] = len( lst_module_duplicate ) self.dct_progression["lst_module_duplicate"] = lst_module_duplicate self.write_config() if lst_module_missing or lst_module_duplicate: print("Cannot setup environment to begin.") if lst_module_missing: print("Missing module :") print(lst_module_missing) if lst_module_duplicate: print("Duplicate module :") print(lst_module_duplicate) want_continue = input( "💬 Detect error missing/duplicate module init, do you want to continue? (Y/N): " ) if want_continue.strip().lower() != "y": return self.dct_progression["state_0_search_missing_module"] = True self.write_config() else: # TODO fill from config lst_module_missing = [] print("✅ -> Search missing module") print( "❌ -> Install missing module, do a research or ask to uninstall it (can break data)" ) msg = "1 - Import database from zip" print(f"🔷 {msg}") self.add_comment_progression(msg) database_name = self.dct_progression.get("config_database_name") if not database_name: database_name = ( input( f"💬 Witch database name do you want to work with? Default ({default_database_name}) : " ).strip() or default_database_name ) self.dct_progression["config_database_name"] = database_name self.write_config() do_neutralize = False if not self.dct_progression.get("state_1_neutralize_database"): print("[1] Ignore neutralize database") wait_continue = ( input("💬 Neutralize database, press to continue : ") .strip() .lower() ) if wait_continue != "1": do_neutralize = True database_name += "_neutralize" self.dct_progression["config_database_name"] = database_name self.write_config() print(f"★ Work with database '{database_name}'") if not self.dct_progression.get("state_1_restore_database"): file_name = os.path.basename(self.file_path) image_db_file_path = os.path.join("image_db", file_name) str_will_copy = ( f"🤖 will copy '{self.file_path}' to '{image_db_file_path}'" ) if not shutil._samefile(self.file_path, image_db_file_path): do_copy = False if os.path.exists(image_db_file_path): status_overwrite_image_db = input( f"{str_will_copy}, " f"a file already exist, do you want to continue (y/Y) : " ).strip() if status_overwrite_image_db.lower() == "y": do_copy = True os.remove(image_db_file_path) else: print(str_will_copy) do_copy = True if do_copy: shutil.copy(self.file_path, image_db_file_path) neutralize_arg = "" if start_version >= 16 and do_neutralize: neutralize_arg = " --neutralize" status, cmd_executed = self.todo_upgrade_execute( f"./script/database/db_restore.py --database {database_name} --image {file_name} --ignore_cache{neutralize_arg}", single_source_odoo=True, ) if not status: self.dct_progression["state_1_restore_database"] = True self.write_config() print("✅ -> Restore database") print("✅ -> Neutralize database") if do_neutralize: status, cmd_executed = self.todo_upgrade_execute( f"./script/addons/update_prod_to_dev.sh {database_name}", single_source_odoo=True, ) # TODO exécuter next line si status != 0 et log contient # psycopg2.errors.UndefinedTable: relation "discuss_channel" does not exist # LIGNE 1 : SELECT "discuss_channel"."id" FROM "discuss_channel" WHERE (... # source ./.venv.odoo18.0_python3.12.10/bin/activate && cat script/postgresql/migration/fix_migration_postgresql_17_to_postgresql_18_module_mail_nov_2025.py | ./odoo18.0/odoo/odoo-bin shell -d ripbylop_stage_prod_17_nov_2025 # psycopg2.errors.ForeignKeyViolation: insert or update on table "discuss_channel_member" violates foreign key constraint "discuss_channel_member_channel_id_fkey" # DÉTAIL : Key (channel_id)=(20) is not present in table "discuss_channel". # ./script/database/migrate/process_backup_file.py --path_backup_zip image_db/db.zip --path_output_zip image_db/dbFIX.zip --word_to_delete discuss_channel_channel_type_not_null # Puis faire un retry de la commande, sinon rien self.dct_progression["state_1_neutralize_database"] = True self.write_config() if not status: self.dct_progression["state_1_neutralize_database"] = True self.write_config() else: self.dct_progression["state_1_neutralize_database"] = True self.write_config() config_state_1_uninstall_module = self.dct_progression.get( "config_state_1_uninstall_module" ) config_state_1_install_module = self.dct_progression.get( "config_state_1_install_module" ) if not is_state_4_reach_open_upgrade: lst_module_to_uninstall = [] uninstall_module_list_file = os.path.join( "script", "odoo", "migration", f"uninstall_module_list_odoo{start_version * 10}_to_odoo{(start_version + 1) * 10}.txt", ) if os.path.exists(uninstall_module_list_file): with open(uninstall_module_list_file, "r") as f: lst_module_to_uninstall = [ a.strip() for a in f.readline().split() ] if config_state_1_uninstall_module: lst_module_to_uninstall = ( lst_module_to_uninstall + config_state_1_uninstall_module ) if lst_module_to_uninstall: self.uninstall_from_database( lst_module_to_uninstall, database_name, start_version ) self.dct_progression["state_1_uninstall_module"] = True self.write_config() self.dct_progression["config_state_1_uninstall_module"] = ( config_state_1_uninstall_module ) self.write_config() print("✅ -> Uninstall module") print("✅ -> install module") if not is_state_4_reach_open_upgrade: lst_module_to_install = [] if config_state_1_install_module: lst_module_to_install = ( lst_module_to_install + config_state_1_install_module ) if lst_module_to_install: self.install_from_database( lst_module_to_install, database_name, start_version ) self.dct_progression["state_1_install_module"] = True self.write_config() self.dct_progression["config_state_1_install_module"] = ( config_state_1_install_module ) self.write_config() msg = "2 - Succeed update all addons" print(f"🔷 {msg}") self.add_comment_progression(msg) if not self.dct_progression.get("state_2_update_all"): status, cmd_executed = self.todo_upgrade_execute( f"./script/addons/update_addons_all.sh {database_name}", single_source_odoo=True, ) if not status: self.dct_progression["state_2_update_all"] = True self.write_config() msg = "3 - Clean up database before data migration" print(f"🔷 {msg}") self.add_comment_progression(msg) if not self.dct_progression.get("state_3_install_clean_database"): status, cmd_executed = self.todo_upgrade_execute( f"./script/addons/install_addons.sh {database_name} database_cleanup", single_source_odoo=True, ) if not status: self.dct_progression["state_3_install_clean_database"] = True self.write_config() if not self.dct_progression.get("state_3_clean_database"): print( "✨ Aller dans «configuration/Technique/Nettoyage.../Purger» les modules obsolètes" ) status = input( "💬 Did you finish to clean database? Press y/Y to open server with selenium, else ignore it : " ).strip() if status.lower().strip() == "y": self.todo.prompt_execute_selenium_and_run_db(database_name) status = input("💬 Press to continue state.3 : ").strip() self.dct_progression["state_3_clean_database"] = True self.write_config() self.install_OCA_odoo_module_migrator() msg = "4 - Upgrade version with OpenUpgrade" print(f"🔷 {msg}") self.add_comment_progression(msg) self.dct_progression["state_4_reach_open_upgrade"] = True self.write_config() lst_next_version = [ a for a in range(start_version + 1, end_version + 1) ] lst_database_name_upgrade = [ f"{database_name}_upgrade_{str(a)}" for a in lst_next_version ] # Setup lst_switch_odoo lst_clone_odoo = self.dct_progression.get( "state_4_clone_odoo_lst", [False] * len(lst_next_version) ) lst_switch_odoo = self.dct_progression.get( "state_4_switch_odoo_lst", [False] * len(lst_next_version) ) lst_module_migrate_odoo = self.dct_progression.get( "state_4_module_migrate_odoo_lst", [False] * len(lst_next_version) ) lst_module_uninstall_module = self.dct_progression.get( "state_4_uninstall_module", [False] * len(lst_next_version) ) lst_module_install_module = self.dct_progression.get( "state_4_install_module", [False] * len(lst_next_version) ) lst_module_search_missing_module = self.dct_progression.get( "state_4_search_missing_module", [False] * len(lst_next_version) ) nb_missing_value_switch_odoo = abs( len(lst_switch_odoo) - len(lst_next_version) ) if nb_missing_value_switch_odoo: lst_switch_odoo += [False] * nb_missing_value_switch_odoo # Setup lst_upgrade_odoo lst_upgrade_odoo = self.dct_progression.get( "state_4_upgrade_odoo_lst", [[]] * len(lst_next_version) ) lst_fix_migration_odoo = self.dct_progression.get( "state_4_fix_migration_odoo_lst", [[]] * len(lst_next_version) ) nb_missing_value_upgrade_odoo = abs( len(lst_upgrade_odoo) - len(lst_next_version) ) if nb_missing_value_upgrade_odoo: lst_upgrade_odoo += [[]] * nb_missing_value_upgrade_odoo database_name_upgrade = None lst_module_missing_next_version = [] lst_module_to_delete = [] lst_module_to_delete_last_version = [] for index, next_version in enumerate(lst_next_version): # Reinit the list lst_module_missing_last_version = lst_module_missing_next_version[ : ] lst_module_to_delete_last_version.extend(lst_module_to_delete) lst_module_to_delete = [] msg = f"4.{index} - Ready to work with version {next_version}" self.add_comment_progression(msg) option_comment = 0 msg = f"4.{index}.{chr(option_comment + 65)} - Search updated module list to next version" self.add_comment_progression(msg) if not database_name_upgrade: last_database_name = database_name else: last_database_name = database_name_upgrade database_name_upgrade = lst_database_name_upgrade[index] lst_module_to_uninstall = [] lst_module_to_install = [] lst_module_to_analyse = self.get_rename_module( self.dct_module_per_version[next_version - 1], next_version, ) self.dct_module_per_version[next_version] = sorted( list(set(lst_module_to_analyse)) ) self.dct_progression["dct_module_per_version"] = ( self.dct_module_per_version ) option_comment += 1 msg = f"4.{index}.{chr(option_comment + 65)} - Clone Odoo" self.add_comment_progression(msg) if not os.path.exists(f"odoo{next_version}.0/addons/addons"): os.makedirs(f"odoo{next_version}.0/addons/addons") if not lst_clone_odoo[index]: self.switch_odoo(next_version - 1) print( f"⧖ -> Clone to odoo.'{next_version}', from '{database_name}' to '{database_name_upgrade}'." ) # Delete if exist database self.todo_upgrade_execute( f"./script/database/db_restore.py -d {database_name_upgrade} --only_drop", ) # Duplicate database cmd_clone_database = f"./odoo_bin.sh db --clone --from_database {last_database_name} --database {database_name_upgrade}" self.todo_upgrade_execute(cmd_clone_database) lst_clone_odoo[index] = True self.dct_progression["state_4_clone_odoo_lst"] = lst_clone_odoo self.write_config() print(f"✅ -> Clone Odoo{next_version} done") else: print(f"✅ -> Clone Odoo{next_version} - nothing") option_comment += 1 msg = f"4.{index}.{chr(option_comment + 65)} - Uninstall module" self.add_comment_progression(msg) config_state_4_uninstall_module = self.dct_progression.get( "config_state_4_uninstall_module", [False] * len(lst_next_version), ) if not lst_module_uninstall_module[index]: lst_module_to_uninstall = config_state_4_uninstall_module[ index ] if lst_module_to_uninstall: self.uninstall_from_database( lst_module_to_uninstall, database_name_upgrade, next_version - 1, ) lst_module_uninstall_module[index] = True self.dct_progression["state_4_module_migrate_odoo_lst"] = ( lst_module_uninstall_module ) self.write_config() self.dct_progression["config_state_4_uninstall_module"] = ( config_state_4_uninstall_module ) self.dct_progression["state_4_uninstall_module"] = ( lst_module_uninstall_module ) self.write_config() option_comment += 1 msg = f"4.{index}.{chr(option_comment + 65)} - Install module" self.add_comment_progression(msg) config_state_4_install_module = self.dct_progression.get( "config_state_4_install_module", [False] * len(lst_next_version), ) # Special case to install module to fix migration if ( next_version == 13 and "dms" in lst_module_to_analyse and "muk_dms" not in self.dct_progression.get("dct_module_exist", {}).keys() ): # Force install dms into odoo 12 if not config_state_4_install_module[index]: config_state_4_install_module[index] = ["dms"] elif "dms" not in config_state_4_install_module[index]: config_state_4_install_module[index].append("dms") if not lst_module_install_module[index]: lst_module_to_install = config_state_4_install_module[index] if not lst_module_to_install: lst_module_to_install = [] if lst_module_to_install: self.install_from_database( lst_module_to_install, database_name_upgrade, next_version - 1, ) lst_module_install_module[index] = True self.dct_progression["state_4_module_migrate_odoo_lst"] = ( lst_module_install_module ) self.write_config() self.dct_progression["config_state_4_install_module"] = ( config_state_4_install_module ) self.dct_progression["state_4_install_module"] = ( lst_module_install_module ) self.write_config() option_comment += 1 msg = f"4.{index}.{chr(option_comment + 65)} - Switch Odoo" self.add_comment_progression(msg) if not lst_switch_odoo[index]: self.switch_odoo(next_version) lst_switch_odoo[index] = True self.dct_progression["state_4_switch_odoo_lst"] = ( lst_switch_odoo ) self.write_config() print(f"✅ -> Switch Odoo{next_version} done with update") else: print(f"✅ -> Switch Odoo{next_version} - nothing") lst_state_4_module_migrate_code = self.dct_progression.get( "config_state_4_module_to_migrate_code", [[]] * len(lst_next_version), ) if ( "config_state_4_module_to_migrate_code" not in self.dct_progression.keys() ): self.dct_progression[ "config_state_4_module_to_migrate_code" ] = lst_state_4_module_migrate_code lst_module_to_migrate = lst_state_4_module_migrate_code[index] option_comment += 1 msg = ( f"4.{index}.{chr(option_comment + 65)} - Search missing module" ) self.add_comment_progression(msg) if not lst_module_search_missing_module[index]: lst_module_to_analyse_updated = [] for bd_module in lst_module_to_analyse: if ( lst_module_to_uninstall and bd_module in lst_module_to_uninstall ): # Ignore check if uninstall before continue lst_module_to_analyse_updated.append(bd_module) # TODO remove from list past module deleted lst_module_to_check = [ a for a in lst_module_to_analyse_updated if a not in lst_module_to_delete_last_version ] ( lst_module_missing_next_version, lst_module_duplicate_next_version, ) = self.check_addons_exist(lst_module_to_check) lst_module_missing_next_version = sorted( list(set(lst_module_missing_next_version)) ) self.dct_progression["state_4_len_lst_module_missing"] = len( lst_module_missing_next_version ) self.dct_progression["state_4_lst_module_missing"] = ( lst_module_missing_next_version ) lst_module_duplicate = sorted( list(set(lst_module_duplicate_next_version)) ) self.dct_progression["state_4_len_lst_module_duplicate"] = len( lst_module_duplicate ) self.dct_progression["state_4_lst_module_duplicate"] = ( lst_module_duplicate ) if lst_module_duplicate: print(f"Duplicate module into odoo{next_version} : ") print(lst_module_duplicate) input( f"💬 Detect error duplicate module, manage this problem manually and press to continue." ) # if lst_module_missing_next_version and not lst_module_to_migrate: if lst_module_missing_next_version: # TODO support when lst_module_to_migrate is fill lst_module_to_migrate = [] print( f"👹 Detect error missing module, missing module into odoo{next_version} :" ) for index_missing_module, module_missing in enumerate( lst_module_missing_next_version ): old_path = self.dct_progression.get( "dct_module_exist", {} ).get(module_missing) print( f"[{index_missing_module}] {module_missing} - {old_path}" ) print("[a] All list above") print("[e] Add extra custom") want_continue = ( input( f"💬 Enumerate missing module separate by coma to delete it" f". The others will be migrate : " ) .strip() .lower() ) is_delete_all = False if want_continue: lst_want_continue = [ a.strip() for a in want_continue.split(",") ] if "a" in lst_want_continue: is_delete_all = True lst_module_to_delete = [ lst_module_missing_next_version[a] for a in range( len(lst_module_missing_next_version) ) ] else: # TODO show error if the index is wrong lst_want_continue_number = [ int(a) for a in lst_want_continue if a.isdigit() ] lst_module_to_delete = [ lst_module_missing_next_version[a] for a in lst_want_continue_number if 0 <= a < len(lst_module_missing_next_version) ] if len(lst_module_to_delete) == len( lst_module_missing_next_version ): is_delete_all = True if next_version == 15: lst_module_to_delete.append("users_default_groups") lst_module_to_delete.append( "web_editor_backend_context" ) lst_module_to_delete.append( "website_google_analytics_fixed" ) elif next_version == 17: lst_module_to_delete.append( "export_delete_login_log" ) elif next_version == 18: lst_module_to_delete.append( "base_attachment_object_storage" ) lst_module_to_delete.append( "user_password_strength" ) if "e" in lst_want_continue: want_continue = ( input( f"💬 Enumerate module name to delete, separate by coma : " ) .strip() .lower() ) lst_to_extend = [ a.strip() for a in want_continue.split(",") ] lst_module_to_delete.extend(lst_to_extend) if lst_module_to_delete: msg = f"4.{index}.{chr(option_comment + 65)}.option - Choose delete missing module" self.add_comment_progression(msg) self.switch_odoo(next_version - 1) if lst_module_to_delete: # Delete if exist database self.todo_upgrade_execute( f"./script/database/db_restore.py -d {database_name_upgrade} --only_drop", ) # Duplicate database cmd_clone_database = f"./odoo_bin.sh db --clone --from_database {last_database_name} --database {database_name_upgrade}" self.todo_upgrade_execute(cmd_clone_database) self.uninstall_from_database( lst_module_to_delete, database_name_upgrade, next_version, ) self.install_from_database( lst_module_to_install, database_name_upgrade, next_version - 1, ) if not is_delete_all: msg = f"4.{index}.{chr(option_comment + 65)}.option - Choose auto-fix (not implemented yet)" self.add_comment_progression(msg) lst_module_to_migrate_code = set( lst_module_missing_next_version ) - set(lst_module_to_delete) ( lst_module_missing_last, lst_module_duplicate_last, lst_module_exist_last, lst_module_error_last, ) = self.check_addons_exist( lst_module_to_migrate_code, get_all_info=True ) if lst_module_missing_last: print( f"Error missing module : {lst_module_missing_last}" ) if lst_module_duplicate_last: print( f"Error duplicate module : {lst_module_duplicate_last}" ) if lst_module_error_last: print( f"Error error module : {lst_module_error_last}" ) if lst_module_exist_last: odoo_name_last_version = ( f"odoo{next_version - 1}.0" ) odoo_name_actual_version = f"odoo{next_version}.0" for ( module_name, module_path, ) in lst_module_exist_last: module_dir_path = os.path.dirname(module_path) module_dir_path_new_version = ( module_dir_path.replace( odoo_name_last_version, odoo_name_actual_version, ) ) module_dir_path_manifest = os.path.join( module_path, "__manifest__.py" ) module_dir_new_version = os.path.join( module_dir_path_new_version, module_name ) module_dir_new_version_manifest = os.path.join( module_dir_new_version, "__manifest__.py" ) dct_module_to_migrate_module = { "source_module_path": module_path, "source_manifest_path": module_dir_path_manifest, "source_addons_path": module_dir_path, "target_module_path": module_dir_new_version, "target_manifest_path": module_dir_new_version_manifest, "target_addons_path": module_dir_path_new_version, "module_name": module_name, "source_version_odoo": next_version - 1, "target_version_odoo": next_version, } # TODO move this into config lst_module_to_migrate.append( dct_module_to_migrate_module ) self.dct_progression[ "config_state_4_module_to_migrate_code" ][index] = lst_module_to_migrate self.write_config() self.switch_odoo(next_version) # TODO auto-fix # TODO try to migrate module, find in previous version, application la migration vers une nouvelle version # TODO ajouté menu todo qui permet de faire une migration d'un module et migrer le générateur de code. # TODO when check module, reminder provenance # TODO implement asyncio instead of parallel # TODO detect when duplicate path module ou module manquant, prendre décision qui ont efface si dupliqué # TODO pourquoi web_ir_actions_act_multi est doublé dans odoo 13 lst_module_search_missing_module[index] = True self.dct_progression["state_4_search_missing_module"] = ( lst_module_search_missing_module ) self.write_config() option_comment += 1 msg = f"4.{index}.{chr(option_comment + 65)} - Migrate module" self.add_comment_progression(msg) lst_path_git_clone_migrate = [] if not lst_module_migrate_odoo[index]: # TODO Searching module # search addons/addons # search read manifest and detect branch difference, manifest into private? # Extract module name and run migration to another list # Maybe check if already exist and show list or continue with overwrite # Expliquer pourquoi on ne fait pas le oca-port, c'est config_migrate_repo = self.dct_progression.get( "config_migrate_repo", False ) self.dct_progression["config_migrate_repo"] = ( config_migrate_repo ) if config_migrate_repo: dct_module_result = self.search_module_to_move( next_version - 1, next_version ) # TODO code migration # git stash # call odoo-module-migrate, without commit source_module_path = dct_module_result.get( "source_module_path" ) if not source_module_path: _logger.error( f"Missing source module path '{source_module_path}'" ) else: if os.path.exists( os.path.join(source_module_path, ".git") ): self.todo_upgrade_execute( f"cd '{source_module_path}' && git stash && cd ~-" ) target_module_path = dct_module_result.get( "target_module_path" ) if not target_module_path: _logger.error( f"Missing target module path '{target_module_path}'" ) else: if os.path.exists( os.path.join(target_module_path, ".git") ): self.todo_upgrade_execute( f"cd '{target_module_path}' && git stash && cd ~-" ) lst_module_to_migrate_all = dct_module_result.get( "lst_module", [] ) else: lst_module_to_migrate_all = [] # TODO remove duplicate au lieu d'extend lst_module_to_migrate_all.extend(lst_module_to_migrate) self.internal_module_upgrade( next_version, lst_module_to_migrate_all, lst_path_git_clone_migrate, ) lst_module_migrate_odoo[index] = True self.dct_progression["state_4_module_migrate_odoo_lst"] = ( lst_module_migrate_odoo ) self.write_config() print(f"✅ -> Module upgrade Odoo{next_version} done") else: print(f"✅ -> Module upgrade Odoo{next_version} - nothing") option_comment += 1 msg = f"4.{index}.{chr(option_comment + 65)} - Fix migrate code" self.add_comment_progression(msg) if not lst_fix_migration_odoo[index]: print("") file_path_fix_migration = os.path.join( "script", "odoo", "migration", f"fix_migration_odoo{(next_version-1)*10}_to_odoo{next_version*10}.py", ) if os.path.exists(file_path_fix_migration): self.todo_upgrade_execute( f"cat ./{file_path_fix_migration} | ./odoo{next_version}.0/odoo/odoo-bin shell -d {database_name_upgrade}", single_source_odoo=True, ) lst_fix_migration_odoo[index] = file_path_fix_migration self.dct_progression["state_4_fix_migration_odoo_lst"] = ( lst_fix_migration_odoo ) self.write_config() print(f"✅ -> Fix migration Odoo{next_version} done") else: print( f"✅ -> Fix migration Odoo{next_version} - no fix to execute" ) else: print(f"✅ -> Fix migration Odoo{next_version} - nothing") for path_git_clone_migrate in lst_path_git_clone_migrate: cmd = f"./script/code/git_commit_migration_addons_path.py --path {path_git_clone_migrate} --odoo_version {next_version}.0" self.todo_upgrade_execute(cmd) option_comment += 1 msg = f"4.{index}.{chr(option_comment + 65)} - Migrate database" self.add_comment_progression(msg) if not lst_upgrade_odoo[index]: path_addons_openupgrade = os.path.join( os.getcwd(), f"odoo{next_version}.0", "OCA_OpenUpgrade" ) # Update config with OCA_OpenUpgrade ignore_path = ( "--ignore-odoo-path " if next_version <= 13 else "" ) extra_addons_path_extra = ( f",{path_addons_openupgrade}/addons,{path_addons_openupgrade}/odoo/addons" if next_version <= 13 else "" ) cmd_update_config = ( f"./script/git/git_repo_update_group.py {ignore_path}" f"--extra-addons-path {path_addons_openupgrade}{extra_addons_path_extra} " f"&& ./script/generate_config.sh" ) self.todo_upgrade_execute(cmd_update_config) print("🚸 Please, validate commits after code migration.") print("ℹ To show repo status :\nmake repo_show_status") print( f"🚸 Please, validate this path into config.conf : '{path_addons_openupgrade}'." ) status = input(f"💬 Press to continue {msg} : ").strip() # The technique change at version 14 if next_version <= 13: erplibre_version = self.install_OCA_openupgrade( next_version ) cmd_upgrade = f".venv.{erplibre_version}/bin/python ./odoo{next_version}.0/OCA_OpenUpgrade/odoo-bin -c ./config.conf --update all --no-http --stop-after-init -d {database_name_upgrade}" else: cmd_upgrade = f"./run.sh --upgrade-path=./odoo{next_version}.0/OCA_OpenUpgrade/openupgrade_scripts/scripts --update all -c config.conf --stop-after-init --no-http --load=base,web,openupgrade_framework -d {database_name_upgrade}" lst_upgrade_odoo[index] = cmd_upgrade self.todo_upgrade_execute( cmd_upgrade, new_env={ "OPENUPGRADE_TARGET_VERSION": f"{next_version}.0" }, ) # TODO detect error self.dct_progression["state_4_upgrade_odoo_lst"] = ( lst_upgrade_odoo ) self.write_config() str_wait_next_version = ( " (or wait next version 🤖)" if next_version != lst_next_version[-1] else "" ) status = ( input( f"💬 Do you want to upgrade all{str_wait_next_version}? Press y/Y to upgrade all addons database : " ) .strip() .lower() ) if status == "y": self.todo_upgrade_execute( f"./script/addons/update_addons_all.sh {database_name_upgrade}", ) print(f"✅ -> Database upgrade Odoo{next_version} done") # Update config without OCA_OpenUpgrade cmd_update_config = f"./script/git/git_repo_update_group.py && ./script/generate_config.sh" self.todo_upgrade_execute(cmd_update_config) print("[y] Open server with Selenium") status = ( input( "💬 Do you want to test this upgrade? Choose or press to ignore it : " ) .strip() .lower() ) "make repo_show_status" if status == "y": self.todo.prompt_execute_selenium_and_run_db( database_name_upgrade ) status = input( f"💬 Press to continue 4.{index} : " ).strip() else: print(f"✅ -> Database upgrade Odoo{next_version} - nothing") # # waiting_input = input("💬 Press any keyboard key to continue...") print("") msg = "5 - Cleaning up database after upgrade" print(f"🔷 {msg}") self.add_comment_progression(msg) print( "✨ Re-update i18n, purger data, tables (except mail_test and mail_test_full)" ) # waiting_input = input("💬print Press any keyboard key to continue...") msg = "6 - Migration finished" print(f"🔷 {msg}") self.add_comment_progression(msg) cmd_backup_template = f"./odoo_bin.sh db --backup --database {database_name_upgrade} --restore_image" cmd_backup = f"{cmd_backup_template} {database_name_upgrade}_finish_{datetime.datetime.now().strftime('%Y%m%d%H%M%S')}" print(f"✨ Can execute backup creation :\n{cmd_backup}") status = input( "💬 Press y/Y or write filename.zip to export or enter to continue : " ).strip() if status.lower(): if status.lower() != "y": cmd_backup = f"{cmd_backup_template} {status}" self.todo_upgrade_execute(cmd_backup) status = input("💬 Test the migration, press y/Y : ") if status.lower().strip() == "y": self.todo.prompt_execute_selenium_and_run_db(database_name_upgrade) def get_rename_module(self, lst_module, next_version): path_search = f"odoo{next_version}.0/OCA_OpenUpgrade/" status, cmd_executed, lst_output = self.todo_upgrade_execute( f"find {path_search} -name apriori.py", get_output=True, ) if not lst_output: _logger.error( f"Cannot find renamed module script apriori.py into path '{path_search}'" ) return lst_module apriory_py = lst_output[0].strip() with open(apriory_py, "r") as f: file_content = f.read() data_vars = {} exec(file_content, data_vars) renamed_modules = data_vars.get("renamed_modules", {}) merged_modules = data_vars.get("merged_modules", {}) deleted_modules = data_vars.get("deleted_modules", []) lst_index_to_delete = [] for index, module in enumerate(lst_module): renamed_module = renamed_modules.get(module) merged_module = merged_modules.get(module) if renamed_module: lst_module[index] = renamed_module if merged_module: lst_module[index] = merged_module if module in deleted_modules: lst_index_to_delete.append(index) for index_to_delete in lst_index_to_delete[::-1]: lst_module.pop(index_to_delete) return list(set(lst_module)) def search_module_to_move(self, source_version_odoo, target_version_odoo): lst_module = [] # lst_path_to_check = [ # os.path.join(f"odoo{actual_version_odoo}", "addons"), # os.path.join(f"odoo{target_version_odoo}", "addons"), # os.path.join(f"private", "addons"), # ] source_path_to_check = os.path.join( f"odoo{source_version_odoo}.0", "addons", "addons" ) target_path_to_check = os.path.join( f"odoo{target_version_odoo}.0", "addons", "addons" ) # Search is_moving_git = False if os.path.exists(source_path_to_check): if os.path.exists(os.path.join(source_path_to_check, ".git")): is_moving_git = True if not os.path.exists(target_path_to_check): shutil.copytree(source_path_to_check, target_path_to_check) # if not is_moving_git: # os.mkdir(path_to_check_target) # else: # # TODO clone # pass if not is_moving_git: # TODO do something # Time to compare os.listdir(source_path_to_check) if os.path.exists(target_path_to_check): for dir_name in os.listdir(target_path_to_check): source_module_path = os.path.join( source_path_to_check, dir_name ) source_manifest_path = os.path.join( source_module_path, "__manifest__.py" ) target_module_path = os.path.join( target_path_to_check, dir_name ) target_manifest_path = os.path.join( target_module_path, "__manifest__.py" ) if os.path.exists(target_manifest_path): # TODO remove from list when module already exist in version 15 dct_module = { "source_module_path": source_module_path, "source_manifest_path": source_manifest_path, "source_addons_path": source_path_to_check, "target_module_path": target_module_path, "target_manifest_path": target_manifest_path, "target_addons_path": target_path_to_check, "module_name": dir_name, "source_version_odoo": source_version_odoo, "target_version_odoo": target_version_odoo, } lst_module.append(dct_module) dct_module = { "lst_module": lst_module, "source_module_path": source_path_to_check, "target_module_path": target_path_to_check, } return dct_module def uninstall_from_database( self, lst_module_to_uninstall, database_name, actual_version ): if not lst_module_to_uninstall: return uninstall_module = ",".join(lst_module_to_uninstall) self.todo_upgrade_execute( f"./script/addons/uninstall_addons.sh {database_name} {uninstall_module}", single_source_odoo=True, ) # Update list installed module self.dct_module_per_version[actual_version] = sorted( list( set(self.dct_module_per_version[actual_version]) - set(lst_module_to_uninstall) ) ) self.dct_progression["dct_module_per_version"] = ( self.dct_module_per_version ) self.write_config() def install_from_database( self, lst_module_to_install, database_name, actual_version ): if not lst_module_to_install: return install_module = ",".join(lst_module_to_install) self.todo_upgrade_execute( f"./script/addons/install_addons.sh {database_name} {install_module}", single_source_odoo=True, ) # Update list installed module self.dct_module_per_version[actual_version] = sorted( list( set( self.dct_module_per_version[actual_version] + lst_module_to_install ) ) ) self.dct_progression["dct_module_per_version"] = ( self.dct_module_per_version ) self.write_config() def check_addons_exist( self, lst_module_to_check, ignore_error=True, get_all_info=False ): str_module_to_check = ",".join(sorted(lst_module_to_check)) status, cmd_executed, dct_output = self.todo_upgrade_execute( f"{PYTHON_BIN} ./script/addons/check_addons_exist.py --format_json --output_json -m {str_module_to_check}", get_output=True, output_is_json=True, wait_at_error=not ignore_error, ) lst_module_missing = dct_output.get("missing") lst_module_duplicate = dct_output.get("duplicate") if get_all_info: lst_module_error = dct_output.get("error") lst_module_exist = dct_output.get("exist") return ( lst_module_missing, lst_module_duplicate, lst_module_exist, lst_module_error, ) return lst_module_missing, lst_module_duplicate def switch_odoo(self, odoo_version): int_odoo_version = int(float(odoo_version)) # Expect odoo_version like 12.0 lst_version, lst_version_installed, odoo_installed_version = ( get_odoo_version() ) if odoo_installed_version != f"odoo{int_odoo_version}.0": print( f"⧖ -> Was '{odoo_installed_version}', Switch to odoo{int_odoo_version}.0" ) self.todo_upgrade_execute(f"make switch_odoo_{int_odoo_version}") self.todo_upgrade_execute("make config_gen_all") def install_OCA_odoo_module_migrator(self): if not os.path.exists(PATH_VENV_MODULE_MIGRATOR): self.todo_upgrade_execute( f"cd {PATH_OCA_ODOO_MODULE_MIGRATOR} && python -m venv {VENV_NAME_MODULE_MIGRATOR} && source {VENV_NAME_MODULE_MIGRATOR}/bin/activate && pip3 install -r requirements.txt" ) def install_OCA_openupgrade(self, next_version): # TODO install odoorpc==0.7.0 # openupgradelib # openupgrade_path = f"odoo{next_version}.0/OCA_OpenUpgrade" # venv_oca_path = f"{openupgrade_path}/.venv" # if os.path.exists(venv_oca_path): # return lst_version, lst_version_installed, odoo_installed_version = ( get_odoo_version() ) extract_version = f"{next_version}.0" dct_erplibre_info = [ a for a in lst_version if a.get("odoo_version") == extract_version ] if not dct_erplibre_info: raise Exception(f"Cannot extract {extract_version}") dct_erplibre_info = dct_erplibre_info[0] erplibre_version = dct_erplibre_info.get("erplibre_version") # self.todo_upgrade_execute( # f".venv.{erplibre_version}/bin/python -m venv {venv_oca_path} && {venv_oca_path}/bin/pip3 install -r {openupgrade_path}/requirements.txt" # ) self.todo_upgrade_execute( f".venv.{erplibre_version}/bin/pip install odoorpc==0.7.0" ) self.todo_upgrade_execute( f".venv.{erplibre_version}/bin/pip install openupgradelib" ) return erplibre_version def todo_upgrade_execute( self, cmd, single_source_odoo=False, new_env=None, quiet=False, get_output=False, output_is_json=False, wait_at_error=True, ): if output_is_json and not get_output: get_output = True output = None if get_output: status, cmd_executed, output = self.execute.exec_command_live( cmd, source_erplibre=False, single_source_odoo=single_source_odoo, new_env=new_env, return_status_and_output_and_command=True, quiet=quiet, ) else: status, cmd_executed = self.execute.exec_command_live( cmd, source_erplibre=False, single_source_odoo=single_source_odoo, new_env=new_env, return_status_and_command=True, quiet=quiet, ) self.lst_command_executed.append(cmd_executed) self.dct_progression["command_executed"] = self.lst_command_executed self.write_config() if status and wait_at_error: print("[1] to redo the command") wait_status = ( input( "💬 Error detected, press to continue or ctrl+c to stop : " ) .strip() .lower() ) # psycopg2.errors.UndefinedTable: relation "discuss_channel" does not exist # LIGNE 1 : SELECT "discuss_channel"."id" FROM "discuss_channel" WHERE (... if wait_status == "1": return self.todo_upgrade_execute( cmd, single_source_odoo=single_source_odoo, new_env=new_env, quiet=quiet, get_output=get_output, output_is_json=output_is_json, wait_at_error=wait_at_error, ) if get_output: if output_is_json: str_output = json.loads("".join(output)) return status, cmd_executed, str_output return status, cmd_executed, output return status, cmd_executed def check_and_clone_source_to_target_migration_code( self, next_version, source_addons_path, target_addons_path ): if not os.path.exists(os.path.join(source_addons_path, ".git")): return if os.path.exists(os.path.join(target_addons_path, ".git")): return # if not os.path.exists(target_addons_path): # cmd_mkdir = f"mkdir -p {target_addons_path}" # status, cmd_executed, lst_output = self.todo_upgrade_execute( # cmd_mkdir, # get_output=True, # ) # return source_dir_name = os.path.basename(source_addons_path) # Clone a project for next version # Get actual branch cmd_git_clone_migrate_source = ( f"cd {source_addons_path} && " f"git branch --show-current && " f"cd ~-" ) status, cmd_executed, lst_output = self.todo_upgrade_execute( cmd_git_clone_migrate_source, get_output=True, ) branch_source = lst_output[0].strip() if len(lst_output) else "" if not branch_source: # Get branch from repo branch_source = self.get_branch_name_from_local_manifest( source_addons_path, f"{next_version}.0" ) branch_target = branch_source.replace( str(next_version - 1), str(next_version) ) # Get remote branch for actual version cmd_git_clone_migrate_source_same_target = ( f"cd {source_addons_path} " f"&& git fetch --all " f'&& git branch -vv | grep "{branch_source}" ' f"&& cd ~-" ) cmd_git_clone_migrate_source_same_target_remote_only = ( f"cd {source_addons_path} && git remote && cd ~-" ) status, cmd_executed, lst_output = self.todo_upgrade_execute( cmd_git_clone_migrate_source_same_target, get_output=True, wait_at_error=False, ) if status == 1: status, cmd_executed, lst_output = self.todo_upgrade_execute( cmd_git_clone_migrate_source_same_target_remote_only, get_output=True, ) if lst_output: remote = lst_output[0].strip() # TODO write this modification into repo manifest else: remote = input( "👹 BUG, you need to push last branch. Please write the remote/ :" ) else: local_branch, remote, remote_branch = ( self.get_local_branch_remote_actual_branch_git( lst_output, ) ) # Get remote branch for next version remote_branch_target = f"{remote}/{branch_target}" cmd_git_clone_migrate_source_same_target = ( f"cd {source_addons_path} " f"&& git fetch --all " f'&& git branch --remotes -vv | grep "{remote_branch_target} " ' f"&& cd ~-" ) status, cmd_executed, lst_output = self.todo_upgrade_execute( cmd_git_clone_migrate_source_same_target, get_output=True, wait_at_error=False, ) has_existing_target_branch = any([a.strip() for a in lst_output]) # TODO check config if path is added # Get remote branch address cmd_remote_address = ( f"cd {source_addons_path} " f"&& git remote get-url {remote} " f"&& cd ~-" ) status, cmd_executed, lst_output = self.todo_upgrade_execute( cmd_remote_address, get_output=True, ) remote_address = lst_output[0].strip() # TODO some time, the clone has error, need to repeat branch_to_clone = ( branch_target if has_existing_target_branch else branch_source ) cmd_git_clone = ( f"cd {os.path.dirname(target_addons_path)} " f"&& git clone {remote_address} {source_dir_name} -b {branch_to_clone} && cd ~-" ) status, cmd_executed, lst_output = self.todo_upgrade_execute( cmd_git_clone, get_output=True, ) if not has_existing_target_branch: cmd_git_clone = ( f"cd {target_addons_path} " f"&& if git rev-parse --verify --quiet refs/heads/{branch_target}; then " f"git checkout {branch_target}; else git checkout -b {branch_target}; fi " f"&& cd ~-" ) status, cmd_executed, lst_output = self.todo_upgrade_execute( cmd_git_clone, get_output=True, ) def get_branch_name_from_local_manifest(self, addons_path, default_branch): relative_addons_path = addons_path.replace(os.getcwd() + "/", "") git_tool = GitTool() dct_remote, dct_project, default_remote = ( git_tool.get_manifest_xml_info(filename=LOCAL_MANIFEST) ) for dct_repo in dct_project.values(): path = dct_repo.get("@path") if path == relative_addons_path: revision = dct_repo.get("@revision") return revision return default_branch def get_local_branch_remote_actual_branch_git(self, lst_output): for line in lst_output: # The current branch is marked with an asterisk (*) at the start of the line if not line.strip().startswith("*"): continue # Split the line to isolate the remote and remote branch name parts = line.split() if len(parts) >= 4: # The remote and remote branch are in the 4th part, e.g., '[origin/main]' remote_info = parts[3].strip("[]") # Split 'origin/main' into 'origin' and 'main' # remote, remote_branch = remote_info.split("/", 1) try: remote, remote_branch = remote_info.split("/", 1) except ValueError as e: # # TODO this means no remote, take default one? or last one? # # TODO supporter la migration 17 vers 18 # print("👹 BUG, you need to push ") # TODO search remote and associate branch value = input( "👹 BUG, you need to push last branch. Please write the remote/branch_name :" ) remote, remote_branch = value.split("/", 1) return parts[1], remote, remote_branch return None, None, None def add_comment_progression(self, comment): comment_to_add = f"# {comment}" self.lst_command_executed.append(comment_to_add) self.dct_progression["command_executed"] = self.lst_command_executed self.write_config()