erplibre/script/database/db_restore.py

#!/usr/bin/env python3
# © 2021-2026 TechnoLibre (http://www.technolibre.ca)
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl)

import argparse
import configparser
import getpass
import logging
import os
import shutil
import sys
import subprocess
from subprocess import check_output

sys.path.append(
    os.path.normpath(os.path.join(os.path.dirname(__file__), "..", ".."))
)

from script.execute.execute import redact_secrets

logging.basicConfig(level=os.environ.get("LOGLEVEL", "INFO"))

_logger = logging.getLogger(__name__)


def get_config():
    """Parse command line arguments, extracting the config file name,
    returning the union of config file and command line arguments

    :return: dict of config file settings and command line arguments
    """
    # TODO update description
    parser = argparse.ArgumentParser(
        formatter_class=argparse.RawDescriptionHelpFormatter,
        description="""\
DESCRIPTION
    Restore database, use cache to clone to improve speed.

SUGGESTION
    ./script/database/db_restore.py -d test
""",
        epilog="""\
""",
    )
    # parser.add_argument('-d', '--dir', dest="dir", default="./",
    #                     help="Path of repo to change remote, including submodule.")
    parser.add_argument("-d", "--database", help="Database to manipulate.")
    parser.add_argument(
        "--image",
        help=(
            "Image name to restore, from directory image_db, filename without"
            " '.zip'. Example, use odoo12.0_base to use image"
            " odoo12.0_base.zip. Default value is odoo12.0_base"
        ),
    )
    parser.add_argument(
        "--clean_cache",
        action="store_true",
        help="Delete all database cache to clone, begin by _cache_.",
    )
    parser.add_argument(
        "--ignore_cache",
        action="store_true",
        help="Ignore creating _cache_ when restoring.",
    )
    parser.add_argument(
        "--only_drop",
        action="store_true",
        help="Will only drop database if exist.",
    )
    parser.add_argument(
        "--neutralize",
        action="store_true",
        help="Will disable all cron.",
    )
    args = parser.parse_args()
    return args


def get_master_password():
    try:
        # _logger.info("You have 5 seconds to add master password...")
        pa = getpass.getpass(prompt="\nEnter master password... ")
        return pa
    except getpass.GetPassWarning:
        _logger.error("Password echoed, danger!")


# Assez pour une faute de frappe répétée, pas assez pour qu'une boucle
# oubliée tourne toute la nuit devant une invite que personne ne lit.
MAX_ESSAIS_MOT_DE_PASSE = 10


def password_refused(sortie):
    """Odoo a-t-il refusé le mot de passe maître, ou autre chose ?

    La distinction porte tout. Reposer la question sur n'importe quel
    échec cacherait la vraie panne derrière dix invites, et l'on
    chercherait un mot de passe alors que la base est cassée.

    Odoo lève `AccessDenied` — la classe apparaît dans la trace, et son
    message traduit peut varier. On reconnaît donc la CLASSE.
    """
    return "AccessDenied" in (sortie or "")


def probe_master_password(arg_base):
    """(accepté, sortie) — éprouver le mot de passe sur `--list`.

    La commande la plus inoffensive : elle ne touche à rien et rend le
    même refus qu'une restauration. Valider ici évite d'échouer à
    mi-parcours, une fois la base déjà supprimée.
    """
    done = subprocess.run(
        f"{arg_base} --list".split(" "),
        capture_output=True,
        text=True,
    )
    return done.returncode == 0, (done.stdout or "") + (done.stderr or "")


def ask_master_password(arg_base, essais=MAX_ESSAIS_MOT_DE_PASSE):
    """Le mot de passe maître, redemandé tant qu'Odoo le refuse.

    None si l'on renonce — invite vide, essais épuisés, ou panne qui
    n'a rien à voir avec le mot de passe.

    Une faute de frappe arrêtait la migration net, sur une trace
    `CalledProcessError` que rien n'attrapait. Après une heure de
    paliers, c'est cher payé pour une lettre.
    """
    for tour in range(1, essais + 1):
        mot = get_master_password()
        if not mot:
            return None
        candidat = f"{arg_base} --master_password={mot}"
        accepte, sortie = probe_master_password(candidat)
        if accepte:
            return mot
        if not password_refused(sortie):
            # Autre chose est cassé : le dire, et ne pas noyer la panne
            # sous dix invites de mot de passe.
            _logger.error(redact_secrets(sortie.strip()[-1500:]))
            return None
        restants = essais - tour
        if restants:
            _logger.warning(
                f"Master password refused, {restants} attempt(s) left."
            )
    _logger.error("Master password refused too many times.")
    return None


def get_list_db_cache(arg_base):
    arg = f"{arg_base} --list"
    out = check_output(arg.split(" ")).decode()
    lst_db = out.strip().split("\n")
    lst_db_cache = [a for a in lst_db if a.startswith("_cache_")]
    return lst_db, lst_db_cache


def verify_filestore(database, image):
    """Contrôler qu'une restauration a bien posé ses fichiers.

    Une seule fois, à la restauration d'origine. Après un clone il n'y a
    rien à vérifier : `copytree` recopie la source telle quelle, défauts
    compris — le contrôle appartient à ce qui a créé le défaut, pas à ce
    qui l'a dupliqué.

    Le contrôle n'interrompt pas : la base est restaurée et utilisable,
    c'est la DISPOSITION des fichiers qui est suspecte. Refuser ici
    casserait des chaînes qui marchent, pour un défaut qui se répare
    d'une commande.
    """
    chemin = os.path.join("image_db", f"{image}.zip")
    if not os.path.isfile(chemin):
        return
    try:
        from script.analyse import check_filestore
    except Exception:  # pragma: no cover - l'outil d'analyse est optionnel
        return
    rapport = check_filestore.verify_restore(database, chemin)
    for ligne in check_filestore.render_verify(rapport):
        print(ligne)
    if rapport.get("nested"):
        offer_tidy(check_filestore, rapport)


def offer_tidy(check_filestore, rapport):
    """Proposer de ranger TOUT DE SUITE, là où le défaut naît.

    C'est le seul endroit qui vaille. Le nichage se produit une fois, à
    la restauration, puis le clone le recopie tel quel : mesuré, les six
    bases de la chaîne portaient les mêmes 1168 fichiers. Ranger ici,
    c'est ranger une fois ; ranger plus tard, c'est six fois.

    Rien ne se fait sans réponse humaine, et rien du tout hors d'un
    terminal : ce script tourne aussi sans personne devant, et une
    question posée à un `stdin` fermé arrêterait la migration.
    """
    if not sys.stdin.isatty():
        return
    remonter, doublons = check_filestore.tidy_nested_plan(rapport)
    if not remonter and not doublons:
        return
    print(f"   {len(remonter)} à remonter, {len(doublons)} doublons purs")
    try:
        reponse = input("💬 Ranger maintenant ? (y/N) : ").strip().lower()
    except EOFError:
        return
    if reponse not in ("y", "yes", "o"):
        return
    for source, cible in remonter:
        os.makedirs(os.path.dirname(cible), exist_ok=True)
        shutil.move(source, cible)
    for source, _cible in doublons:
        os.remove(source)
    shutil.rmtree(check_filestore.nested_dir(rapport), ignore_errors=True)
    print(f"✅ {len(remonter)} remontés, {len(doublons)} doublons supprimés.")


def restore_or_clone(config, arg_base, cache_database, lst_db_cache):
    """Restaurer depuis l'image, ou cloner le cache déjà restauré.

    Le contrôle du filestore ne suit QUE les vraies restaurations. Le
    clone recopie sa source telle quelle : contrôler le miroir dirait
    deux fois la même chose, et la seconde au mauvais endroit.
    """
    if cache_database not in lst_db_cache and not config.ignore_cache:
        _logger.info(
            f"## Create cache {cache_database} from image {config.image} ##"
        )
        arg = (
            f"{arg_base} --restore"
            f" --restore_image {config.image} --database {cache_database}"
        )
        print(redact_secrets(check_output(arg.split(" ")).decode()))
        verify_filestore(cache_database, config.image)

    if config.ignore_cache:
        _logger.info(
            f"## Restoring {config.image} to database {config.database} ##"
        )
        arg = (
            f"{arg_base} --restore --restore_image"
            f" {config.image} --database {config.database}"
        )
    else:
        _logger.info(
            f"## Clone cache {cache_database} to database"
            f" {config.database} ##"
        )
        arg = (
            f"{arg_base} --clone --from_database"
            f" {cache_database} --database {config.database}"
        )
    if config.neutralize:
        arg += " --neutralize"
    # « arg_base » porte --master_password des que la base en exige un.
    print(redact_secrets(arg))
    print(redact_secrets(check_output(arg.split(" ")).decode()))
    if config.ignore_cache:
        verify_filestore(config.database, config.image)


def main():
    config = get_config()

    arg_base = "./odoo_bin.sh db"

    if not config.image:
        with open(".odoo-version", "r") as f:
            odoo_version = f.readline()
            config.image = f"odoo{odoo_version}_base"

    # check if it needs master password from config file
    has_config_file = True
    config_path = "./config.conf"
    if not os.path.isfile(config_path):
        config_path = "/etc/odoo/odoo.conf"
        if not os.path.isfile(config_path):
            has_config_file = False
    if has_config_file:
        config_parser = configparser.ConfigParser()
        config_parser.read(config_path)

        has_admin_password = config_parser.get("options", "admin_passwd")
        if has_admin_password and has_admin_password != "admin":
            master_password = ask_master_password(arg_base)
            if not master_password:
                _logger.error("Missing master password, cancel transaction.")
                sys.exit(1)
            arg_base += f" --master_password={master_password}"
        else:
            _logger.info("No master password needed... Continue")

    # Get list of database
    lst_db, lst_db_cache = get_list_db_cache(arg_base)

    if config.clean_cache:
        for db in lst_db_cache:
            _logger.info(f"## Delete {db} ##")
            arg = f"{arg_base} --drop --database {db}"
            out = redact_secrets(check_output(arg.split(" ")).decode())
            print(out)
        lst_db, lst_db_cache = get_list_db_cache(arg_base)

    if config.database:
        cache_database = f"_cache_{config.image}"
        # Drop db
        if config.database in lst_db:
            _logger.info(f"## Drop {config.database} ##")
            arg = f"{arg_base} --drop --database {config.database}"
            out = redact_secrets(check_output(arg.split(" ")).decode())
            print(out)
        if config.only_drop:
            return
        restore_or_clone(config, arg_base, cache_database, lst_db_cache)

    if not config.clean_cache and not config.database:
        print("Nothing to do.")


if __name__ == "__main__":
    main()