backupmanager_ynh/sources/jobs/db_dump.py

import json
import os
import subprocess
import tarfile
import tempfile
import time
from datetime import datetime


def run_db_dump(job, instance, backup_dir):
    """Point d'entrée commun mysql et postgresql."""
    if job.type == "mysql":
        return _run_mysql(job, instance, backup_dir)
    elif job.type == "postgresql":
        return _run_postgresql(job, instance, backup_dir)
    raise ValueError(f"Type inconnu pour db_dump : {job.type}")


# ---------------------------------------------------------------------------
# MySQL
# ---------------------------------------------------------------------------

def _run_mysql(job, instance, backup_dir):
    from flask import current_app
    cfg = json.loads(job.config_json or "{}")
    dbname = cfg.get("database", "")
    if not dbname:
        raise ValueError("Nom de base de données manquant dans la configuration du job.")

    archive_name = _archive_name(instance, "mysql", dbname)
    _abort_if_exists(archive_name, backup_dir)

    with tempfile.TemporaryDirectory() as tmpdir:
        dump_path = os.path.join(tmpdir, f"{dbname}.sql")

        result = subprocess.run(
            [
                "sudo", "mysqldump",
                "--single-transaction",
                "--routines",
                "--triggers",
                "--result-file", dump_path,
                dbname,
            ],
            capture_output=True,
            text=True,
            timeout=7200,
        )
        log = (result.stdout + result.stderr).strip()
        if result.returncode != 0:
            raise RuntimeError(f"mysqldump a échoué (code {result.returncode}) :\n{log}")

        _write_tar(tmpdir, dump_path, dbname, archive_name, backup_dir, job,
                   instance, current_app.config.get("INSTANCE_URL", ""))

    return archive_name, log or "mysqldump terminé sans sortie."


# ---------------------------------------------------------------------------
# PostgreSQL
# ---------------------------------------------------------------------------

def _run_postgresql(job, instance, backup_dir):
    from flask import current_app
    cfg = json.loads(job.config_json or "{}")
    dbname = cfg.get("database", "")
    if not dbname:
        raise ValueError("Nom de base de données manquant dans la configuration du job.")

    archive_name = _archive_name(instance, "postgresql", dbname)
    _abort_if_exists(archive_name, backup_dir)

    with tempfile.TemporaryDirectory() as tmpdir:
        dump_path = os.path.join(tmpdir, f"{dbname}.sql")

        # pg_dump doit tourner en tant qu'utilisateur postgres
        result = subprocess.run(
            ["sudo", "-u", "postgres", "pg_dump", "--format=plain", dbname],
            capture_output=True,
            timeout=7200,
        )
        if result.returncode != 0:
            log = result.stderr.decode("utf-8", errors="replace").strip()
            raise RuntimeError(f"pg_dump a échoué (code {result.returncode}) :\n{log}")

        with open(dump_path, "wb") as f:
            f.write(result.stdout)

        log = result.stderr.decode("utf-8", errors="replace").strip()

        _write_tar(tmpdir, dump_path, dbname, archive_name, backup_dir, job,
                   instance, current_app.config.get("INSTANCE_URL", ""))

    return archive_name, log or "pg_dump terminé sans sortie."


# ---------------------------------------------------------------------------
# Helpers partagés
# ---------------------------------------------------------------------------

def _archive_name(instance, db_type, dbname):
    date_str = datetime.utcnow().strftime("%Y%m%d")
    return f"{instance}_{db_type}_{dbname}_{date_str}"


def _abort_if_exists(archive_name, backup_dir):
    path = os.path.join(backup_dir, archive_name + ".tar")
    if os.path.exists(path):
        raise RuntimeError(
            f"L'archive {archive_name}.tar existe déjà. "
            "Supprimez-la ou attendez le prochain cycle."
        )


# ---------------------------------------------------------------------------
# Restore
# ---------------------------------------------------------------------------

def restore_db_dump(archive_name, backup_dir):
    """Restauration d'une base MySQL ou PostgreSQL depuis une archive BackupManager."""
    archive_path = os.path.join(backup_dir, archive_name + ".tar")
    if not os.path.exists(archive_path):
        raise FileNotFoundError(f"Archive introuvable : {archive_path}")

    info = _read_backup_info(archive_path)
    db_type = info.get("type")
    dbname = info.get("database", "")

    if not dbname:
        raise ValueError("Nom de base de données introuvable dans backup_info.json.")

    with tempfile.TemporaryDirectory() as tmpdir:
        dump_path = os.path.join(tmpdir, f"{dbname}.sql")

        # Extraction du dump depuis l'archive
        with tarfile.open(archive_path) as tar:
            member = tar.getmember(f"db/{dbname}.sql")
            with tar.extractfile(member) as src, open(dump_path, "wb") as dst:
                dst.write(src.read())

        if db_type == "mysql":
            return _restore_mysql(dbname, dump_path)
        elif db_type == "postgresql":
            return _restore_postgresql(dbname, dump_path)
        else:
            raise ValueError(f"Type de base inconnu dans l'archive : {db_type}")


def _restore_mysql(dbname, dump_path):
    log_lines = []

    # Suppression + recréation propre de la base
    result = subprocess.run(
        ["sudo", "mysql", "-e",
         f"DROP DATABASE IF EXISTS `{dbname}`; CREATE DATABASE `{dbname}` CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;"],
        capture_output=True, text=True, timeout=60,
    )
    if result.returncode != 0:
        raise RuntimeError(f"Impossible de recréer la base MySQL '{dbname}' :\n{result.stderr.strip()}")
    log_lines.append(f"Base MySQL '{dbname}' recréée.")

    # Restauration du dump
    with open(dump_path, "rb") as f:
        result = subprocess.run(
            ["sudo", "mysql", dbname],
            stdin=f,
            capture_output=True,
            timeout=7200,
        )
    log = result.stderr.decode("utf-8", errors="replace").strip()
    if result.returncode != 0:
        raise RuntimeError(f"mysql restore a échoué (code {result.returncode}) :\n{log}")

    log_lines.append(f"Dump restauré dans '{dbname}'.")
    if log:
        log_lines.append(log)
    return "\n".join(log_lines)


def _restore_postgresql(dbname, dump_path):
    log_lines = []

    # Terminer les connexions actives puis drop + recreate
    subprocess.run(
        ["sudo", "-u", "postgres", "psql", "-c",
         f"SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE datname = '{dbname}' AND pid <> pg_backend_pid();"],
        capture_output=True, timeout=30,
    )
    subprocess.run(
        ["sudo", "-u", "postgres", "dropdb", "--if-exists", dbname],
        capture_output=True, timeout=60,
    )
    result = subprocess.run(
        ["sudo", "-u", "postgres", "createdb", dbname],
        capture_output=True, text=True, timeout=60,
    )
    if result.returncode != 0:
        raise RuntimeError(f"Impossible de recréer la base PostgreSQL '{dbname}' :\n{result.stderr.strip()}")
    log_lines.append(f"Base PostgreSQL '{dbname}' recréée.")

    # Restauration du dump
    with open(dump_path, "rb") as f:
        result = subprocess.run(
            ["sudo", "-u", "postgres", "psql", "-d", dbname, "-v", "ON_ERROR_STOP=1"],
            stdin=f,
            capture_output=True,
            timeout=7200,
        )
    log = result.stderr.decode("utf-8", errors="replace").strip()
    if result.returncode != 0:
        raise RuntimeError(f"psql restore a échoué (code {result.returncode}) :\n{log}")

    log_lines.append(f"Dump restauré dans '{dbname}'.")
    if log:
        log_lines.append(log)
    return "\n".join(log_lines)


def _read_backup_info(archive_path):
    try:
        with tarfile.open(archive_path) as tar:
            member = tar.extractfile("backup_info.json")
            if member:
                return json.loads(member.read())
    except Exception:
        pass
    return {}


def _write_tar(tmpdir, dump_path, dbname, archive_name, backup_dir, job, instance, instance_url):
    """Crée le .tar dans backup_dir et le .info.json YunoHost à côté."""
    archive_path = os.path.join(backup_dir, archive_name + ".tar")

    # backup_info.json embarqué dans le tar (métadonnées BackupManager)
    import json as _json
    info = {
        "instance_name": instance,
        "instance_url": instance_url,
        "type": job.type,
        "database": dbname,
        "created_at": datetime.utcnow().isoformat(),
        "backupmanager_version": "1.0.0",
    }
    info_path = os.path.join(tmpdir, "backup_info.json")
    with open(info_path, "w") as f:
        _json.dump(info, f, indent=2)

    with tarfile.open(archive_path, "w") as tar:
        tar.add(dump_path, arcname=f"db/{dbname}.sql")
        tar.add(info_path, arcname="backup_info.json")

    # .info.json YunoHost (hors tar, requis pour listing webadmin)
    size = os.path.getsize(archive_path)
    ynh_info = {
        "created_at": int(time.time()),
        "description": f"BackupManager: {job.type} {dbname}",
        "size": size,
        "from_before_upgrade": False,
        "apps": {},
        "system": {},
    }
    ynh_info_path = os.path.join(backup_dir, archive_name + ".info.json")
    with open(ynh_info_path, "w") as f:
        _json.dump(ynh_info, f, indent=2)