1#@!#!123s

: / proc / 2 / cwd / opt / dedrads /
Filename : mysql_selector.py
back
#!/usr/bin/env python3


import sys, os, re, json, shlex, subprocess, datetime, time, socket, glob, shutil
from shutil import which

TARGETS = {
    "mariadb1011": ("mariadb", "10.11"),
    "mariadb106": ("mariadb", "10.6"),
    "mariadb114": ("mariadb", "11.4"),
    "mysql80": ("mysql", "8.0"),
    "mysql84": ("mysql", "8.4"),
}

CPANEL_CONFIG = "/var/cpanel/cpanel.config"
DATADIR = "/var/lib/mysql"
BACKUP_ROOT = "/root"
ROOT_MYCNF = "/root/.my.cnf"
LOG_FILE_DEFAULT = None  # initialized after ts() is defined
LOG_FH = None
DRY_RUN = False
FORCE_DELETE = False
MAX_ALLOWED_PACKET = "512M"
ORIGINAL_MAX_ALLOWED_PACKET = None
_RPM_CACHE = None


def print_help(prog):
    """Display detailed usage guidance."""
    help_text = f"""mysql_selector.py - Install or switch MySQL/MariaDB on cPanel (AlmaLinux 8+)

Usage:
  {prog} [--dry-run] [--force] [--log-file=/path/to/log] <target>

Targets (key -> engine/version):
  mariadb1011 -> MariaDB 10.11 (LTS)
  mariadb106  -> MariaDB 10.6
  mariadb114  -> MariaDB 11.4
  mysql80     -> MySQL 8.0
  mysql84     -> MySQL 8.4

What this script does:
  - Verifies the host is AlmaLinux 8 or newer.
  - Backs up all non-system databases to {BACKUP_ROOT}/mysql-backup-<timestamp>.
  - Stops existing MySQL/MariaDB services and removes conflicting packages.
  - Sets the requested version in cPanel config, then triggers WHM's installer.
  - Falls back to manual installs for supported versions if WHM install fails.
  - Ensures /root/.my.cnf credentials work and restores databases + grants.
  - Writes a log to {LOG_FILE_DEFAULT or "<default>"} (or a custom --log-file).

Expectations and prerequisites:
  - Run as root on a cPanel server (whmapi1 required for the primary path).
  - /root/.my.cnf must contain working root credentials; backups are not deleted.
  - Sufficient free space (>5%% headroom) is required on {BACKUP_ROOT} for backups.

Options:
  --dry-run          Skip destructive commands (package removals, globs) and prompt before deletion.
  --force            Proceed with deletions/removals without prompting (overrides prompts).
  --log-file=PATH    Write log output to PATH instead of the default in /root.
  -h, --help         Show this help.

Exit codes:
  0 on success, non-zero on error.

Examples:
  {prog} mariadb114
  {prog} mysql84
"""
    print(help_text)


def ts():
    return datetime.datetime.now().strftime("%Y-%m-%dT%H-%M-%S")


LOG_FILE_DEFAULT = f"/root/mysql_selector-{ts()}.log"
_REAL_PRINT = print


def init_logging(log_file=None):
    """Initialize logging to both stdout and a file."""
    global LOG_FH, LOG_FILE_DEFAULT, _REAL_PRINT
    target = log_file or LOG_FILE_DEFAULT
    try:
        LOG_FH = open(target, "a")
        LOG_FILE_DEFAULT = target
    except Exception as e:
        _REAL_PRINT(f"[!] Unable to open log file {target}: {e}")
        LOG_FH = None
        return

    def log_print(*args, **kwargs):
        _REAL_PRINT(*args, **kwargs)
        if LOG_FH:
            text = " ".join(str(a) for a in args)
            end = kwargs.get("end", "\n")
            try:
                LOG_FH.write(text + end)
                LOG_FH.flush()
            except Exception:
                pass

    globals()["print"] = log_print


def check_os_support():
    """Ensure we are running on AlmaLinux 8 or newer."""
    try:
        info = {}
        with open("/etc/os-release") as f:
            for line in f:
                if "=" in line:
                    k, v = line.strip().split("=", 1)
                    info[k] = v.strip('"')
        distro = info.get("NAME", "")
        version = info.get("VERSION_ID", "")
    except Exception:
        distro = ""
        version = ""
    if "almalinux" not in distro.lower():
        print("[!] This script supports only AlmaLinux systems.")
        sys.exit(1)
    try:
        major = int(version.split(".")[0])
    except (ValueError, IndexError):
        major = 0
    if major < 8:
        print(
            "[!] AlmaLinux version {} detected. Version 8 or newer required.".format(
                version or "<unknown>"
            )
        )
        sys.exit(1)


def run(cmd, check=True, destructive=False):
    if DRY_RUN and destructive:
        print(f"[dry-run] Skipping command: {cmd}")
        return 0
    if isinstance(cmd, str):
        cmd = shlex.split(cmd)
    print("+ " + shlex.join(cmd))
    p = subprocess.run(cmd)
    if check and p.returncode != 0:
        sys.exit(p.returncode)
    return p.returncode


def out(cmd):
    try:
        if isinstance(cmd, str):
            cmd = shlex.split(cmd)
        return subprocess.check_output(cmd, stderr=subprocess.STDOUT).decode(
            "utf-8", "replace"
        )
    except subprocess.CalledProcessError as e:
        return (e.output or b"").decode("utf-8", "replace")
    except FileNotFoundError:
        return ""


def run_background(cmd):
    """Run a command in the background, suppressing stdout/stderr, and return the Popen object."""
    if isinstance(cmd, str):
        cmd = shlex.split(cmd)
    if DRY_RUN:
        print(f"[dry-run] Skipping background command: {cmd}")
        return None
    print("+ " + shlex.join(cmd) + " (background)")
    try:
        return subprocess.Popen(
            cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL
        )
    except Exception as e:
        print(f"[!] Failed to start background command {cmd}: {e}")
        return None


def installed_family():
    pkgs = out(["rpm", "-qa"])
    if re.search(r"\bmysql-community-server\b", pkgs, re.I) or re.search(
        r"\bmysql-server\b", pkgs, re.I
    ):
        return "mysql"
    if re.search(r"\bMariaDB-server\b", pkgs, re.I):
        return "mariadb"
    return "none"


def ping_mysql():
    mysqladmin = which("mysqladmin")
    if not mysqladmin:
        return False
    rc = run([mysqladmin, "ping"], check=False)
    return rc == 0


def ensure_dir(d):
    if not os.path.isdir(d):
        os.makedirs(d, exist_ok=True)


def remove_matches(patterns, force=None):
    """Remove files/dirs matching glob patterns without invoking a shell."""
    force = FORCE_DELETE if force is None else force
    matches = []
    for pat in patterns:
        matches.extend(glob.glob(pat))
    matches = sorted(set(matches))
    if not matches:
        return
    print("[*] Paths matched for removal:")
    for path in matches:
        print(f"    {path}")
    if DRY_RUN:
        print("[dry-run] Skipping deletion of matched paths.")
        return
    if not force:
        ans = input("[?] Remove these paths? [y/N]: ").strip().lower()
        if ans not in ("y", "yes"):
            print("[!] Skipping removal by user request.")
            return
    for path in matches:
        try:
            if os.path.isdir(path) and not os.path.islink(path):
                shutil.rmtree(path, ignore_errors=True)
            else:
                os.remove(path)
        except FileNotFoundError:
            continue
        except Exception as e:
            print(f"[!] Unable to remove {path}: {e}")


def rpm_packages_matching(regex, refresh=False):
    """Return rpm -qa entries matching regex (case-insensitive)."""
    global _RPM_CACHE
    if refresh or _RPM_CACHE is None:
        _RPM_CACHE = out(["rpm", "-qa"]).splitlines()
    pattern = re.compile(regex, re.I)
    return [p for p in _RPM_CACHE if pattern.search(p)]


def invalidate_rpm_cache():
    """Clear cached rpm -qa results after package changes."""
    global _RPM_CACHE
    _RPM_CACHE = None


def estimate_datadir_size():
    """Return total size of DATADIR in bytes using du -sb."""
    try:
        output = subprocess.check_output(
            ["du", "-sb", DATADIR], stderr=subprocess.STDOUT
        ).decode("utf-8", "replace")
        size_str = output.strip().split()[0]
        return int(size_str)
    except subprocess.CalledProcessError as e:
        print(
            f"[!] Unable to measure {DATADIR} size: {e.output.decode('utf-8', 'replace')}"
        )
    except Exception as e:
        print(f"[!] Unable to measure {DATADIR} size: {e}")
    return None


def has_sufficient_space_for_backup():
    """Ensure creating a backup won't exceed 95% capacity on the backup filesystem."""
    try:
        stat = os.statvfs(BACKUP_ROOT)
    except Exception as e:
        print(
            f"[!] Could not determine filesystem stats for {BACKUP_ROOT}: {e}"
        )
        return False
    total = stat.f_blocks * stat.f_frsize
    avail = stat.f_bavail * stat.f_frsize
    datadir_size = estimate_datadir_size()
    if datadir_size is None:
        print("[!] Unable to determine MySQL data size; refusing to continue.")
        return False
    min_free = total * 0.05  # require 5% free space remaining
    projected_free = avail - datadir_size
    print(
        f"[i] Backup filesystem total: {total / (1024**3):.2f} GB, free: {avail / (1024**3):.2f} GB, datadir size: {datadir_size / (1024**3):.2f} GB"
    )
    if projected_free < min_free:
        print(
            f"[!] Not enough free space: projected free {projected_free / (1024**3):.2f} GB would drop below 5% of capacity."
        )
        return False
    return True


def backup_all(rootdir):
    """Dump all non-system databases and copy cPanel metadata if present.
    Returns (backup_directory, status) where status in {"ok","empty","error"}.
    """
    ensure_dir(rootdir)
    bdir = os.path.join(rootdir, "mysql-backup-" + ts())
    os.makedirs(bdir, exist_ok=True)
    print(f"[*] Backing up to: {bdir}")
    status = "ok"

    if not ping_mysql():
        print("[!] MySQL server is not running, cannot perform backup!")
        return bdir, "error"

    if not which("mysql") or not which("mysqldump"):
        print(
            "[!] mysql or mysqldump command not found, skipping database backup"
        )
        return bdir, "error"

    user_dbs = []
    try:
        # Get list of all databases
        print("[*] Getting list of databases...")
        dbs_cmd = [which("mysql"), "-NBe", "SHOW DATABASES"]
        dbs = out(dbs_cmd).splitlines()

        if not dbs:
            print("[*] No databases found to back up")
        else:
            # Filter out system databases
            system_dbs = {
                'mysql',
                'information_schema',
                'performance_schema',
                'sys',
            }
            user_dbs = [
                db.strip()
                for db in dbs
                if db.strip() and db.strip() not in system_dbs
            ]

        if not user_dbs:
            print("[*] No user databases found to back up")
            return bdir, "empty"
        else:
            print(
                "[*] Backing up user databases: {}".format(", ".join(user_dbs))
            )

            # Backup each database separately for better reliability
            for db in user_dbs:
                try:
                    backup_file = os.path.join(bdir, f"{db}.sql")
                    print(f"[*] Backing up database: {db}")

                    # Build the mysqldump command
                    dump_cmd = [
                        which("mysqldump"),
                        "--defaults-file=/root/.my.cnf",
                        "--single-transaction",
                        "--quick",
                        "--routines",
                        "--events",
                        "--triggers",
                        "--hex-blob",
                        db,
                    ]

                    # Execute the command and write output to file
                    with open(backup_file, 'w') as f:
                        subprocess.run(dump_cmd, stdout=f, check=True)

                    print(f"[+] Successfully backed up {db} to {backup_file}")

                except subprocess.CalledProcessError as e:
                    print(f"[!] Error backing up database {db}: {e}")
                    status = "error"
                    if os.path.exists(backup_file):
                        os.remove(backup_file)
                except Exception as e:
                    print(
                        f"[!] Unexpected error backing up database {db}: {str(e)}"
                    )
                    status = "error"
                    if os.path.exists(backup_file):
                        os.remove(backup_file)

    except Exception as e:
        print(f"[!] Error during backup process: {str(e)}")
        status = "error"

    # Backup cPanel databases configuration
    if os.path.isdir("/var/cpanel/databases"):
        run(
            [
                "/usr/bin/rsync",
                "-aH",
                "/var/cpanel/databases/",
                os.path.join(bdir, "cpanel_databases"),
            ],
            check=False,
        )

    return bdir, status


def restore_all_databases(backup_dir):
    """Restore all SQL dumps found in backup_dir (user DBs only). Returns True on success."""
    if not backup_dir or not os.path.isdir(backup_dir):
        print("[!] Backup directory missing or invalid, skipping restore")
        return False
    set_server_max_allowed_packet(MAX_ALLOWED_PACKET)
    if not wait_for_mysql_ready(timeout=600, interval=5, require_auth=True):
        print(
            "[!] MySQL/MariaDB service is not ready; cannot restore databases"
        )
        return False
    mysql_bin = which("mysql")
    if not mysql_bin:
        print("[!] mysql client not found, skipping restore")
        return False
    sql_files = sorted(
        [f for f in os.listdir(backup_dir) if f.endswith(".sql")]
    )
    if not sql_files:
        print(f"[*] No SQL dumps found to restore in {backup_dir}")
        return True
    restore_ok = True
    for fname in sql_files:
        db = os.path.splitext(fname)[0]
        if db in ("mysql", "sys"):
            continue
        fpath = os.path.join(backup_dir, fname)
        print(f"[*] Restoring database: {db} from {fpath}")
        try:
            # Ensure the database exists before import
            create_stmt = f"CREATE DATABASE IF NOT EXISTS `{db}` DEFAULT CHARACTER SET utf8mb4"
            run(
                [
                    mysql_bin,
                    "--defaults-file=/root/.my.cnf",
                    f"--max_allowed_packet={MAX_ALLOWED_PACKET}",
                    "-e",
                    create_stmt,
                ],
                check=False,
            )
            with open(fpath) as f:
                subprocess.run(
                    [
                        mysql_bin,
                        "--defaults-file=/root/.my.cnf",
                        f"--max_allowed_packet={MAX_ALLOWED_PACKET}",
                        db,
                    ],
                    stdin=f,
                    check=True,
                )
            print(f"[+] Restored {db}")
        except subprocess.CalledProcessError as e:
            print(f"[!] Failed to restore {db}: {e}")
            restore_ok = False
    # Restore cPanel database metadata if present
    cp_src = os.path.join(backup_dir, "cpanel_databases")
    if os.path.isdir(cp_src):
        run(
            ["/usr/bin/rsync", "-aH", cp_src + "/", "/var/cpanel/databases/"],
            check=False,
        )
        print("[*] Restored /var/cpanel/databases metadata")
    return restore_ok


def stop_db():
    run(["systemctl", "stop", "mysqld"], check=False)
    run(["systemctl", "stop", "mysql"], check=False)
    run(["systemctl", "stop", "mariadb"], check=False)


def park_datadir():
    if os.path.isdir(DATADIR) and not os.path.islink(DATADIR):
        parked = DATADIR + ".off-" + ts()
        print(f"[*] Parking datadir: {DATADIR} -> {parked}")
        run(["mv", DATADIR, parked], check=True, destructive=True)


def get_current_mysql_version():
    try:
        output = out(["mysql", "--version"])
        if not output:
            return None
        # Extract version number (e.g., 'mysql  Ver 8.0.33 for Linux on x86_64' -> '8.0')
        match = re.search(r'(?:Ver )?(\d+\.\d+)', output)
        if match:
            return match.group(1)
    except Exception as e:
        print(f"[!] Could not determine current MySQL version: {e}")
    return None


def remove_mysql_family():
    print("[*] Stopping MySQL service and parking datadir...")
    stop_db()
    park_datadir()

    print("[*] Disabling MySQL module and removing repositories...")
    run(["dnf", "-y", "module", "reset", "mysql"], check=False)
    run(["dnf", "-y", "module", "disable", "mysql"], check=False)
    remove_matches(
        [
            "/etc/yum.repos.d/mysql*.repo",
            "/etc/yum.repos.d/mysql*-community*.repo",
            "/etc/yum/repos.d/mysql*.repo",
        ]
    )

    print("[*] Removing MySQL packages...")
    # Find all MySQL packages
    mysql_pkgs = rpm_packages_matching(r"^(mysql|mysql-community|mysql[0-9])")
    if mysql_pkgs:
        print(f"[*] Found MySQL packages to remove: {' '.join(mysql_pkgs)}")
        # First try dnf remove
        run(["dnf", "-y", "remove"] + mysql_pkgs, check=False, destructive=True)
        invalidate_rpm_cache()
        # Then force remove any remaining packages
        remaining = rpm_packages_matching(
            r"^(mysql|mysql-community|mysql[0-9])", refresh=True
        )
        if remaining:
            print(
                f"[*] Force removing remaining MySQL packages: {' '.join(remaining)}"
            )
            run(
                ["rpm", "-e", "--nodeps"] + remaining,
                check=False,
                destructive=True,
            )
            invalidate_rpm_cache()

    # Additional cleanup for common MySQL-related packages
    extra_pkgs = rpm_packages_matching(
        r"^(mysql-community|mysql80-community|mysql84-community|numactl-libs)",
        refresh=True,
    )
    if extra_pkgs:
        run(["dnf", "-y", "remove"] + extra_pkgs, check=False, destructive=True)
        invalidate_rpm_cache()

    # Clean up any remaining MySQL files and directories
    print("[*] Cleaning up MySQL files and directories...")
    remove_matches(
        [
            "/var/lib/mysql*",
            "/var/run/mysql*",
            "/etc/my.cnf*",
            "/etc/mysql*",
            "/usr/lib64/mysql*",
            "/usr/share/mysql*",
        ]
    )

    # Clean up any remaining dependencies
    run(["dnf", "-y", "autoremove"], check=False, destructive=True)
    invalidate_rpm_cache()
    invalidate_rpm_cache()
    invalidate_rpm_cache()
    run(["dnf", "clean", "all"], check=False)

    print("[*] MySQL removal completed")


def remove_mariadb_family():
    print("[*] Stopping MariaDB service and parking datadir...")
    stop_db()
    park_datadir()

    print("[*] Disabling MariaDB module and removing repositories...")
    # Remove any MariaDB modules
    run(["dnf", "-y", "module", "reset", "mariadb"], check=False)
    run(["dnf", "-y", "module", "disable", "mariadb"], check=False)

    # Remove all MariaDB repositories
    remove_matches(
        [
            "/etc/yum.repos.d/MariaDB*.repo",
            "/etc/yum/repos.d/mariadb*.repo",
            "/etc/yum.repos.d/mysql*.repo",
        ]
    )

    print("[*] Removing all MySQL and MariaDB packages...")
    # First, try to remove everything using dnf
    pkg_list = rpm_packages_matching(
        r"(mariadb|mysql|galera|socat|boost-.*mariadb)", refresh=True
    )
    if pkg_list:
        run(["dnf", "-y", "remove"] + pkg_list, check=False, destructive=True)
        invalidate_rpm_cache()

    # Force remove any remaining packages
    remaining_pkgs = rpm_packages_matching(
        r"(mariadb|mysql|galera)", refresh=True
    )
    if remaining_pkgs:
        print(
            f"[*] Found remaining packages to remove: {' '.join(remaining_pkgs)}"
        )
        run(
            ["rpm", "-e", "--nodeps"] + remaining_pkgs,
            check=False,
            destructive=True,
        )
        invalidate_rpm_cache()

    print("[*] Cleaning up configuration files and directories...")
    # Clean up all MySQL/MariaDB related files and directories
    remove_matches(
        [
            "/var/lib/mysql*",
            "/var/run/mariadb*",
            "/var/run/mysqld*",
            "/etc/my.cnf*",
            "/etc/mysql*",
            "/etc/mariadb*",
            "/usr/lib64/mysql*",
            "/usr/share/mysql*",
            "/usr/bin/mysql*",
            "/usr/sbin/mysql*",
        ]
    )

    # Clean up any remaining dependencies and clear cache
    run(["dnf", "-y", "autoremove"], check=False, destructive=True)
    run(["dnf", "clean", "all"], check=False)

    print("[*] MariaDB/MySQL removal completed")


def ensure_perl_mysql_modules():
    """Install required Perl modules used by cPanel DB tooling."""
    # Use cPanel tooling to satisfy bundled Perl modules.
    check_pkgs = "/usr/local/cpanel/scripts/check_cpanel_pkgs"
    if os.path.exists(check_pkgs):
        print("[*] Ensuring cPanel Perl modules via check_cpanel_pkgs --fix")
        run([check_pkgs, "--fix"], check=False)
    else:
        print(
            "[!] /usr/local/cpanel/scripts/check_cpanel_pkgs not found; skipping Perl module fix"
        )


def set_cpanel_mysql_version(ver):
    print(
        "[*] Setting mysql-version={} in {} (Python write)".format(
            ver, CPANEL_CONFIG
        )
    )
    try:
        run(
            ["/bin/cp", "-a", CPANEL_CONFIG, CPANEL_CONFIG + ".bak-" + ts()],
            check=False,
        )
    except Exception:
        pass
    try:
        with open(CPANEL_CONFIG) as f:
            lines = f.read().splitlines(True)
    except Exception:
        lines = []

    found = False
    new_lines = []
    for line in lines:
        if line.strip().startswith("mysql-version="):
            new_lines.append(f"mysql-version={ver}\n")
            found = True
        else:
            new_lines.append(line)

    if not found:
        if new_lines and not new_lines[-1].endswith("\n"):
            new_lines[-1] = new_lines[-1] + "\n"
        new_lines.append(f"mysql-version={ver}\n")

    tmp = CPANEL_CONFIG + ".tmp"
    with open(tmp, "w") as f:
        f.writelines(new_lines)
        f.flush()
        os.fsync(f.fileno())
    os.replace(tmp, CPANEL_CONFIG)
    print(f"[+] mysql-version set to {ver}")
    # verify
    cur = ""
    try:
        with open(CPANEL_CONFIG) as f:
            for l in f:
                if l.strip().startswith("mysql-version="):
                    cur = l.strip()
                    break
    except Exception:
        cur = "<unreadable>"
    print("[=] [{}] -> {}".format(CPANEL_CONFIG, cur or "<missing>"))


def whm_allowed():
    raw = out(
        [
            "/usr/sbin/whmapi1",
            "start_background_mysql_upgrade",
            "version=__probe__",
            "--output=json",
        ]
    )
    allowed = []
    try:
        j = json.loads(raw)
        reason = (j.get("metadata") or {}).get("reason", "")
    except Exception:
        reason = raw
    vers = re.findall(r"[“\"]([0-9]+\.[0-9]+)[”\"]", reason)
    if vers:
        seen = set()
        allowed = [v for v in vers if not (v in seen or seen.add(v))]
    return allowed


def whm_start_upgrade(ver):
    print("[*] Triggering WHM MySQL/MariaDB install/upgrade …")
    # Remove --allow-downgrade as it's not a valid option
    rc = run(
        [
            "/usr/sbin/whmapi1",
            "start_background_mysql_upgrade",
            f"version={ver}",
        ],
        check=False,
    )
    if rc != 0:
        print("[!] WHM did not accept the request.")
        return False
    print("=== TRIGGERED ===")
    print("Track with: whmapi1 get_mysql_upgrade --output=json|yaml")
    print(
        "If datadir was parked, restore **after** engine is up, then run: /usr/local/cpanel/bin/restoregrants --force"
    )
    return True


def wait_for_mysql_ready(timeout=1800, interval=10, require_auth=False):
    """Wait until MySQL responds to mysqladmin ping.
    If require_auth=False, treat 'Access denied' as service-ready."""
    start = time.time()
    while time.time() - start < timeout:
        mysqladmin = which("mysqladmin")
        if not mysqladmin:
            time.sleep(interval)
            continue
        proc = subprocess.run(
            [mysqladmin, "ping"],
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
        )
        if proc.returncode == 0:
            return True
        output = proc.stdout.decode("utf-8", "replace")
        if not require_auth and "Access denied" in output:
            print(
                "[i] MySQL service responding but root authentication failed; continuing."
            )
            return True
        time.sleep(interval)
    return False


def install_with_cpanel_api(target_ver):
    print(f"[*] Requesting MySQL/MariaDB {target_ver} via cPanel API...")
    if not which("whmapi1"):
        print("[!] whmapi1 not found; cannot use cPanel API installer")
        return False
    if not whm_start_upgrade(target_ver):
        return False
    print("[*] Waiting for database service to come online...")
    if wait_for_mysql_ready():
        print("[+] Database service is online after API request")
        return True
    print(
        "[!] Database service did not come online in expected time; check whmapi1 get_mysql_upgrade"
    )
    return False


def set_server_max_allowed_packet(packet_size):
    """Raise server/client max_allowed_packet before imports."""
    global ORIGINAL_MAX_ALLOWED_PACKET

    def _current_packet():
        mysql_bin_inner = which("mysql")
        if not mysql_bin_inner or not os.path.exists(ROOT_MYCNF):
            return None
        try:
            raw = out(
                [
                    mysql_bin_inner,
                    "--defaults-file=" + ROOT_MYCNF,
                    "-NBe",
                    "SHOW VARIABLES LIKE 'max_allowed_packet'",
                ]
            )
            parts = raw.strip().split()
            if parts:
                return parts[-1]
        except Exception:
            return None
        return None

    mysql_bin = which("mysql")
    if not mysql_bin or not os.path.exists(ROOT_MYCNF):
        return
    if ORIGINAL_MAX_ALLOWED_PACKET is None:
        ORIGINAL_MAX_ALLOWED_PACKET = _current_packet()
    try:
        run(
            [
                mysql_bin,
                "--defaults-file=" + ROOT_MYCNF,
                "-e",
                f"SET GLOBAL max_allowed_packet={packet_size};",
            ],
            check=False,
        )
    except Exception as e:
        print(f"[!] Unable to set max_allowed_packet to {packet_size}: {e}")


def restore_max_allowed_packet():
    """Revert max_allowed_packet to original value if it was changed."""
    global ORIGINAL_MAX_ALLOWED_PACKET
    if not ORIGINAL_MAX_ALLOWED_PACKET:
        return
    mysql_bin = which("mysql")
    if not mysql_bin or not os.path.exists(ROOT_MYCNF):
        return
    try:
        run(
            [
                mysql_bin,
                "--defaults-file=" + ROOT_MYCNF,
                "-e",
                f"SET GLOBAL max_allowed_packet={ORIGINAL_MAX_ALLOWED_PACKET};",
            ],
            check=False,
        )
        print(
            f"[+] max_allowed_packet reverted to {ORIGINAL_MAX_ALLOWED_PACKET}"
        )
    except Exception as e:
        print(
            f"[!] Unable to restore max_allowed_packet to {ORIGINAL_MAX_ALLOWED_PACKET}: {e}"
        )


def run_upcp_force():
    if os.path.exists("/usr/local/cpanel/scripts/upcp"):
        print("[*] Running /usr/local/cpanel/scripts/upcp --force ...")
        run(["/usr/local/cpanel/scripts/upcp", "--force"], check=False)


def read_my_cnf_credentials():
    creds = {}
    if not os.path.exists(ROOT_MYCNF):
        return creds
    try:
        with open(ROOT_MYCNF) as f:
            for line in f:
                line = line.strip()
                if not line or line.startswith("#") or line.startswith("["):
                    continue
                if "=" in line:
                    k, v = line.split("=", 1)
                    k = k.strip().lower()
                    v = v.strip()
                    if k in ("user", "username"):
                        creds["user"] = v
                    elif k in ("password", "pass"):
                        creds["password"] = v
    except Exception as e:
        print(f"[!] Unable to read {ROOT_MYCNF}: {e}")
    return creds


def write_root_mycnf(user, password):
    """Write /root/.my.cnf with provided credentials."""
    try:
        with open(ROOT_MYCNF, "w") as f:
            f.write("[client]\n")
            f.write(f"user={user}\n")
            f.write(f"password={password}\n")
        os.chmod(ROOT_MYCNF, 0o600)
        print(f"[+] Wrote {ROOT_MYCNF} with provided credentials.")
        return True
    except Exception as e:
        print(f"[!] Failed to write {ROOT_MYCNF}: {e}")
        return False


def mysql_login_works():
    mysql_bin = which("mysql")
    if not mysql_bin or not os.path.exists(ROOT_MYCNF):
        return False
    proc = subprocess.run(
        [mysql_bin, "--defaults-file=" + ROOT_MYCNF, "-e", "SELECT 1;"],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
    )
    return proc.returncode == 0


def fetch_temp_root_password():
    log_path = os.path.join(DATADIR, f"{socket.gethostname()}.err")
    if not os.path.exists(log_path):
        return None
    temp_pw = None
    try:
        with open(log_path) as f:
            for line in f:
                if "temporary password" in line.lower():
                    temp_pw = line.strip().split()[-1]
    except Exception:
        return None
    return temp_pw


def ensure_root_password_matches_mycnf(max_wait=300, interval=5):
    """Ensure mysql root password matches credentials stored in /root/.my.cnf."""
    # Wait for service to be reachable before auth attempts
    print(
        f"[*] Waiting up to {max_wait}s for database service to accept connections before root auth sync..."
    )
    if not wait_for_mysql_ready(
        timeout=max_wait, interval=interval, require_auth=False
    ):
        print("[!] Database service did not become reachable in time.")
        return False

    # Try a few times to handle slow starts
    for attempt in range(1, 6):
        if mysql_login_works():
            return True

        creds = read_my_cnf_credentials()
        desired_user = creds.get("user", "root")
        desired_pass = creds.get("password")

        # If .my.cnf is missing or incomplete, attempt passwordless root login and write .my.cnf accordingly.
        if desired_pass is None:
            print(
                f"[i] Attempt {attempt}: {ROOT_MYCNF} missing or lacks password; probing for passwordless root access..."
            )
            proc = subprocess.run(
                ["mysql", "-uroot", "-e", "SELECT 1;"],
                stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL,
            )
            if proc.returncode == 0:
                print(
                    "[+] Passwordless root login succeeded; writing /root/.my.cnf with empty password."
                )
                if write_root_mycnf("root", ""):
                    return True
            else:
                print(f"[!] Attempt {attempt}: passwordless root login failed.")
        else:
            temp_password = fetch_temp_root_password()
            if temp_password:
                print(
                    f"[*] Attempt {attempt}: synchronizing MySQL root password using temporary password."
                )
                env = os.environ.copy()
                env["MYSQL_PWD"] = temp_password
                sql = "ALTER USER 'root'@'localhost' IDENTIFIED BY '{}'".format(
                    desired_pass.replace("'", "\\'")
                )
                subprocess.run(
                    [
                        "mysql",
                        "-uroot",
                        "--connect-expired-password",
                        "-e",
                        sql,
                    ],
                    env=env,
                    check=False,
                )
                if mysql_login_works():
                    print(
                        "[+] Root password synchronized with {}".format(
                            ROOT_MYCNF
                        )
                    )
                    return True

            setpass = "/usr/local/cpanel/scripts/setmysqlrootpass"
            if os.path.exists(setpass):
                print(
                    f"[*] Attempt {attempt}: setting MySQL root password via setmysqlrootpass."
                )
                run([setpass, desired_pass], check=False)
                if mysql_login_works():
                    print(
                        "[+] Root password synchronized with {}".format(
                            ROOT_MYCNF
                        )
                    )
                    return True

        time.sleep(interval)

    print(
        "[!] Unable to synchronize MySQL root password automatically. Please ensure {} credentials are valid.".format(
            ROOT_MYCNF
        )
    )
    return False


def enable_mysql_module_80():
    run(["dnf", "-y", "module", "reset", "mysql"], check=False)
    run(["dnf", "-y", "module", "enable", "mysql:8.0"], check=False)


def install_mysql80():
    # First, make sure we have a clean system
    print("[*] Ensuring clean system state for MySQL 8.0 installation...")

    # Stop any running database services
    stop_db()

    # Clean up any remaining packages and files
    print("[*] Removing existing MySQL/MariaDB packages...")

    # Remove all MySQL and MariaDB packages
    existing_pkgs = rpm_packages_matching(r"^(mariadb|mysql)")
    if existing_pkgs:
        run(
            ["rpm", "-e", "--nodeps"] + existing_pkgs,
            check=False,
            destructive=True,
        )
        invalidate_rpm_cache()

    # Clean up any remaining files
    print("[*] Cleaning up remaining database files...")
    remove_matches(
        [
            "/var/lib/mysql*",
            "/etc/my.cnf*",
            "/etc/mysql*",
        ]
    )

    # Remove any remaining packages that might conflict
    conflicting = rpm_packages_matching(r"^(mariadb|mysql)", refresh=True)
    if conflicting:
        run(
            ["dnf", "-y", "remove"] + conflicting, check=False, destructive=True
        )
        invalidate_rpm_cache()

    # Clean all dnf cache and metadata
    print("[*] Cleaning package manager cache...")
    run(["dnf", "clean", "all"], check=False)
    remove_matches(["/var/cache/dnf"])

    # Reset and enable MySQL 8.0 module
    print("[*] Configuring MySQL 8.0 repository...")
    run(["dnf", "-y", "module", "reset", "mysql"], check=False)
    run(["dnf", "-y", "module", "disable", "mysql"], check=False)

    # Remove any existing MySQL repositories to avoid conflicts
    remove_matches(
        [
            "/etc/yum.repos.d/mariadb*.repo",
            "/etc/yum.repos.d/mysql*.repo",
        ]
    )

    # Add MySQL 8.0 community repository
    mysql_repo = """[mysql80-community]
name=MySQL 8.0 Community Server
baseurl=http://repo.mysql.com/yum/mysql-8.0-community/el/8/$basearch/
enabled=1
gpgcheck=1
gpgkey=file:///etc/pki/rpm-gpg/RPM-GPG-KEY-mysql
"""
    with open("/etc/yum.repos.d/mysql-community.repo", "w") as f:
        f.write(mysql_repo)

    # Import MySQL GPG key
    run(
        ["rpm", "--import", "https://repo.mysql.com/RPM-GPG-KEY-mysql"],
        check=False,
    )

    # Clean and rebuild the dnf cache
    run(["dnf", "clean", "all"], check=False)
    run(["dnf", "makecache"], check=False)

    # Install MySQL 8.0 server with --allowerasing to handle any conflicts
    print("[*] Installing MySQL 8.0 server...")
    try:
        # Install MySQL server and client
        run(
            [
                "dnf",
                "-y",
                "install",
                "mysql-community-server",
                "--allowerasing",
            ],
            check=True,
        )

        # For cPanel systems, create necessary symlinks
        if os.path.exists("/usr/local/cpanel"):
            print("[*] Configuring for cPanel compatibility...")
            os.makedirs("/usr/local/mysql/bin", exist_ok=True)
            for cmd in ["mysql", "mysqladmin", "mysqldump"]:
                if os.path.exists(f"/usr/bin/{cmd}") and not os.path.exists(
                    f"/usr/local/mysql/bin/{cmd}"
                ):
                    os.symlink(f"/usr/bin/{cmd}", f"/usr/local/mysql/bin/{cmd}")

            # Update cPanel MySQL version
            with open("/var/cpanel/mysql/version", "w") as f:
                f.write("8.0")

    except Exception as e:
        print(f"[!] Failed to install MySQL 8.0: {str(e)}")
        print("[*] Attempting alternative installation method...")
        # Try with module if direct installation fails
        run(["dnf", "-y", "module", "reset", "mysql"], check=False)
        run(["dnf", "-y", "module", "enable", "mysql:8.0"], check=False)
        run(
            ["dnf", "-y", "install", "@mysql:8.0", "--allowerasing"], check=True
        )

    # Enable and start the service
    print("[*] Starting MySQL 8.0 service...")
    run(["systemctl", "daemon-reload"], check=True)
    run(["systemctl", "enable", "--now", "mysqld"], check=True)

    # Wait a moment for the service to start
    time.sleep(5)

    # Verify the service is running
    if not ping_mysql():
        print("[!] MySQL service failed to start. Checking status...")
        run(["systemctl", "status", "mysqld"], check=False)
        run(["journalctl", "-xe", "-u", "mysqld"], check=False)

        # Try to get the error log
        hostname_log = os.path.join(DATADIR, f"{socket.gethostname()}.err")
        if os.path.exists(hostname_log):
            print(f"\n[!] Error Log ({hostname_log} - last 20 lines):")
            run(["tail", "-n", "20", hostname_log], check=False)
        else:
            print(f"[!] Expected error log {hostname_log} not found.")

        print("\n[!] Trying to start MySQL manually...")
        run(["mysqld", "--initialize-insecure", "--user=mysql"], check=False)
        run(["systemctl", "start", "mysqld"], check=False)

        if not ping_mysql():
            raise Exception(
                "Failed to start MySQL service after multiple attempts"
            )

    # Get temporary root password if this is a fresh install
    temp_password = None
    hostname_log = os.path.join(DATADIR, f"{socket.gethostname()}.err")
    if os.path.exists(hostname_log):
        with open(hostname_log) as f:
            for line in f:
                if "temporary password" in line.lower():
                    temp_password = line.strip().split()[-1]
                    break

    # Secure the installation
    print("[*] Securing MySQL installation...")
    try:
        if temp_password:
            print(f"\n[!] IMPORTANT: Temporary root password: {temp_password}")
            # Change root password and secure installation
            cmds = [
                "ALTER USER 'root'@'localhost' IDENTIFIED BY '';",
                "DELETE FROM mysql.user WHERE User='';",
                "DELETE FROM mysql.user WHERE User='root' AND Host NOT IN ('localhost', '127.0.0.1', '::1');",
                "DROP DATABASE IF EXISTS test;",
                "DELETE FROM mysql.db WHERE Db='test' OR Db='test\\_%';",
                "FLUSH PRIVILEGES;",
            ]

            # Execute security commands
            for cmd in cmds:
                run(
                    [
                        "mysql",
                        "--user=root",
                        f"--password={temp_password}",
                        "--connect-expired-password",
                        "-e",
                        cmd,
                    ],
                    check=False,
                )
        else:
            print(
                "[*] No temporary password found, assuming existing installation"
            )

    except Exception as e:
        print(f"[!] Warning: Could not secure installation: {str(e)}")

    # For cPanel systems, update configuration
    if os.path.exists("/usr/local/cpanel"):
        print("[*] Updating cPanel configuration...")
        cpanel_scripts = [
            "/usr/local/cpanel/bin/update_local_rpm_versions",
            "/usr/local/cpanel/bin/check-db-stable-version",
            "/usr/local/cpanel/scripts/check_cpanel_mysql_routines",
        ]

        for script in cpanel_scripts:
            if os.path.exists(script):
                try:
                    run([script], check=False)
                except Exception as e:
                    print(f"[!] Warning: Failed to run {script}: {str(e)}")

        # Run upcp to restore cPanel users and databases
        print(
            "[*] Running cPanel update (upcp) to restore users and databases..."
        )
        print("    This may take several minutes to complete...")
        run(["/usr/local/cpanel/scripts/upcp", "--force"], check=False)

    print("\n[+] MySQL 8.0 installation completed successfully!")
    print("\n[!] IMPORTANT NEXT STEPS:")
    if temp_password:
        print(
            "    1. Set a root password using: mysqladmin -u root -p password 'new-password'"
        )
    else:
        print(
            "    1. If you need to reset the root password, use: mysql_secure_installation"
        )
    print("    2. Verify all databases and users are accessible")
    print("    3. Check application compatibility with MySQL 8.0")
    print("\n[!] If you encounter any issues, check these log files:")
    print("    - /var/lib/mysql/*.*.err")
    print("    - journalctl -xe | grep -i mysql")


def install_mariadb106():
    print("[*] Installing MariaDB 10.6...")

    # First, ensure we have a clean system
    print("[*] Ensuring clean system state for MariaDB 10.6 installation...")
    remove_mysql_family()  # Make sure MySQL is completely removed

    # Add MariaDB 10.6 repository
    print("[*] Setting up MariaDB 10.6 repository...")
    maria_repo = """[mariadb]
name = MariaDB
baseurl = http://yum.mariadb.org/10.6/rhel8-amd64
module_hotfixes=1
gpgkey=https://yum.mariadb.org/RPM-GPG-KEY-MariaDB
gpgcheck=1
"""
    with open("/etc/yum.repos.d/mariadb.repo", "w") as f:
        f.write(maria_repo)

    # Import GPG key
    run(
        ["rpm", "--import", "https://yum.mariadb.org/RPM-GPG-KEY-MariaDB"],
        check=False,
    )

    # Install required dependencies first
    print("[*] Installing required dependencies...")
    run(
        [
            "dnf",
            "-y",
            "install",
            "socat",
            "libnsl",
            "perl-DBD-MySQL",
            "policycoreutils-python-utils",
        ],
        check=True,
    )

    # Clean DNF cache
    run(["dnf", "clean", "all"], check=False)
    run(["dnf", "makecache"], check=False)

    # Install MariaDB 10.6 server
    print("[*] Installing MariaDB 10.6 server...")
    try:
        # Install base MariaDB packages
        run(
            [
                "dnf",
                "-y",
                "install",
                "MariaDB-server",
                "MariaDB-client",
                "MariaDB-shared",
                "MariaDB-common",
                "MariaDB-compat",
            ],
            check=True,
        )

        # Install cPanel specific packages if cPanel is detected
        if os.path.exists("/usr/local/cpanel"):
            print("[*] Configuring for cPanel compatibility...")
            # Create necessary symlinks for cPanel
            for cmd in ["mysql", "mysqladmin", "mysqldump"]:
                if os.path.exists(f"/usr/bin/{cmd}") and not os.path.exists(
                    f"/usr/local/mysql/bin/{cmd}"
                ):
                    os.makedirs("/usr/local/mysql/bin", exist_ok=True)
                    os.symlink(f"/usr/bin/{cmd}", f"/usr/local/mysql/bin/{cmd}")

            # Update cPanel MySQL version
            with open("/var/cpanel/mysql/version", "w") as f:
                f.write("10.6")

    except Exception as e:
        print(f"[!] Warning during MariaDB installation: {str(e)}")

    # Configure SELinux if it's enabled
    if os.path.exists("/usr/sbin/sestatus"):
        print("[*] Configuring SELinux for MariaDB...")
        try:
            # Set proper SELinux contexts
            run(
                [
                    "semanage",
                    "port",
                    "-a",
                    "-t",
                    "mysqld_port_t",
                    "-p",
                    "tcp",
                    "3306",
                ],
                check=False,
            )
            run(
                [
                    "semanage",
                    "port",
                    "-m",
                    "-t",
                    "mysqld_port_t",
                    "-p",
                    "tcp",
                    "3306",
                ],
                check=False,
            )

            # Set file contexts for MariaDB
            run(
                [
                    "semanage",
                    "fcontext",
                    "-a",
                    "-t",
                    "mysqld_db_t",
                    "/var/lib/mysql(/.*)?",
                ],
                check=False,
            )
            run(["restorecon", "-R", "/var/lib/mysql"], check=False)

        except Exception as e:
            print(f"[!] Warning: Could not configure SELinux: {str(e)}")

    # Enable and start the service
    print("[*] Starting MariaDB service...")
    run(["systemctl", "daemon-reload"], check=True)
    run(["systemctl", "enable", "--now", "mariadb"], check=True)

    # Verify the service is running
    if not ping_mysql():
        print("[!] MariaDB service failed to start. Checking status...")
        run(["systemctl", "status", "mariadb"], check=False)
        run(["journalctl", "-xe", "-u", "mariadb"], check=False)

        # Try to get the error log
        error_logs = [
            "/var/lig/mysql/*.*.err",
            f"/var/lib/mysql/{os.uname().nodename}.err",
            "/var/lib/mysql/*.*.err",
        ]

        for log in error_logs:
            if os.path.exists(log):
                print(f"\n[!] Error Log ({log} - last 20 lines):")
                run(["tail", "-n", "20", log], check=False)

        print("\n[!] Trying to start MariaDB with systemd...")
        run(["systemctl", "start", "mariadb"], check=False)

        if not ping_mysql():
            print("\n[!] Attempting to start MariaDB directly...")
            run_background(
                [
                    "/usr/libexec/mysqld",
                    "--skip-grant-tables",
                    "--skip-networking",
                ]
            )

    # Final verification
    if not ping_mysql():
        print(
            "\n[!] ERROR: Could not start MariaDB. Please check the error logs above."
        )
        print(
            "    You may need to manually configure MariaDB or check for port conflicts."
        )
        print("    Try running: journalctl -xe | grep -i mysql")
        return False

    # Secure the installation
    print("[*] Securing MariaDB installation...")
    try:
        # Set root password and secure installation
        cmds = [
            "SET PASSWORD FOR 'root'@'localhost' = PASSWORD('');",
            "DELETE FROM mysql.user WHERE User='';",
            "DELETE FROM mysql.user WHERE User='root' AND Host NOT IN ('localhost', '127.0.0.1', '::1');",
            "DROP DATABASE IF EXISTS test;",
            "DELETE FROM mysql.db WHERE Db='test' OR Db='test\\_%';",
            "FLUSH PRIVILEGES;",
        ]

        # Execute security commands
        for cmd in cmds:
            try:
                run(["mysql", "-e", cmd], check=False)
            except Exception as e:
                print(f"[!] Warning: Failed to execute: {cmd}")
                print(f"     Error: {str(e)}")

        # For cPanel systems, update the configuration
        if os.path.exists("/usr/local/cpanel"):
            print("[*] Updating cPanel MySQL configuration...")
            cpanel_scripts = [
                "/usr/local/cpanel/bin/update_local_rpm_versions",
                "/usr/local/cpanel/bin/check-db-stable-version",
                "/usr/local/cpanel/scripts/check_cpanel_mysql_routines",
            ]

            for script in cpanel_scripts:
                if os.path.exists(script):
                    try:
                        run([script], check=False)
                    except Exception as e:
                        print(f"[!] Warning: Failed to run {script}: {str(e)}")

        # Restart cPanel services and run upcp if cPanel is installed
        if os.path.exists("/usr/local/cpanel"):
            print("[*] Restarting cPanel services...")
            run(["/usr/local/cpanel/scripts/restartsrv_cpsrvd"], check=False)

            # Run upcp to restore cPanel users and databases
            print(
                "[*] Running cPanel update (upcp) to restore users and databases..."
            )
            print("    This may take several minutes to complete...")
            run(["/usr/local/cpanel/scripts/upcp", "--force"], check=False)

    except Exception as e:
        print(f"[!] Warning: Could not complete all security steps: {str(e)}")

    print("\n[+] MariaDB 10.6 installation completed successfully!")
    print("\n[!] IMPORTANT NEXT STEPS:")
    print(
        "    1. Set a root password using: mysqladmin -u root password 'your-new-password'"
    )
    print(
        "    2. If using cPanel, run: /usr/local/cpanel/bin/update_local_rpm_versions"
    )
    print("    3. Check MySQL users and permissions")
    print("\n[!] If you encounter any issues, check these log files:")
    print("    - /var/lib/mysql/*.*.err")
    print("    - journalctl -xe | grep -i mysql")


def discover_cpusers():
    """Return a sorted list of cPanel users inferred from DB metadata and user files."""
    users = set()
    meta_dir = "/var/cpanel/databases"
    users_dir = "/var/cpanel/users"
    if os.path.isdir(meta_dir):
        for entry in os.listdir(meta_dir):
            if entry.endswith(".yaml"):
                users.add(entry[:-5])
    if os.path.isdir(users_dir):
        for entry in os.listdir(users_dir):
            if re.match(r"^[A-Za-z0-9_]+$", entry):
                users.add(entry)
    return sorted(users)


def restoregrants_all_users():
    """Restore grants for each cPanel user, aligning with cPanel guidance."""
    script = "/usr/local/cpanel/bin/restoregrants"
    if not os.path.exists(script):
        print("[!] restoregrants script not found; skipping grant restoration.")
        return False
    cpusers = discover_cpusers()
    if not cpusers:
        print("[*] No cPanel users detected for grant restoration.")
        return True
    ok = True
    for user in cpusers:
        rc = run([script, f"--cpuser={user}"], check=False)
        if rc != 0:
            print(f"[!] restoregrants failed for cpuser {user}")
            ok = False
    return ok


def restart_mysql_service():
    """Restart MySQL/MariaDB service to pick up restored settings."""
    print("[*] Restarting database service...")
    # Try common service names; ignore failures to avoid aborting flow.
    run(["systemctl", "restart", "mysqld"], check=False)
    run(["systemctl", "restart", "mysql"], check=False)
    run(["systemctl", "restart", "mariadb"], check=False)


def post_restore_health_check():
    """Basic service + data sanity check after restore."""
    if DRY_RUN:
        print("[dry-run] Skipping post-restore health check.")
        return True
    mysql_bin = which("mysql")
    if not mysql_bin:
        print("[!] mysql client not found; skipping post-restore health check.")
        return False
    if not wait_for_mysql_ready(timeout=180, interval=5, require_auth=True):
        print("[!] MySQL/MariaDB did not become ready for health check.")
        return False

    dbs_raw = out(
        [mysql_bin, "--defaults-file=" + ROOT_MYCNF, "-NBe", "SHOW DATABASES"]
    )
    dbs = [d.strip() for d in dbs_raw.splitlines() if d.strip()]
    system_dbs = {"mysql", "information_schema", "performance_schema", "sys"}
    user_dbs = [d for d in dbs if d not in system_dbs]

    print(f"[*] Health check: found {len(user_dbs)} user databases.")
    if not user_dbs:
        return True

    sample_db = user_dbs[0]
    tables_raw = out(
        [
            mysql_bin,
            "--defaults-file=" + ROOT_MYCNF,
            "-NBe",
            f"SHOW TABLES IN `{sample_db}`",
        ]
    )
    tables = [t.strip() for t in tables_raw.splitlines() if t.strip()]
    if not tables:
        print(
            f"[i] Health check: sample DB {sample_db} has no tables to check."
        )
        return True

    sample_table = tables[0]
    rc = run(
        [
            mysql_bin,
            "--defaults-file=" + ROOT_MYCNF,
            "-e",
            f"CHECK TABLE `{sample_db}`.`{sample_table}`",
        ],
        check=False,
    )
    if rc != 0:
        print(f"[!] Health check failed on {sample_db}.{sample_table}")
        return False
    print(f"[+] Health check passed on {sample_db}.{sample_table}")
    return True


def get_current_mariadb_version():
    try:
        matches = rpm_packages_matching(r"^mariadb-server")
        if not matches:
            return None
        # Extract version number from package name (e.g., 'mariadb-server-10.11.5-1.el9_2.x86_64' -> '10.11')
        match = re.search(r'mariadb[^\d]*(\d+\.\d+)', matches[0])
        if match:
            return match.group(1)
    except Exception as e:
        print(f"[!] Could not determine current MariaDB version: {e}")
    return None


def needs_downgrade(current_ver, target_ver):
    if not current_ver or not target_ver:
        return False
    try:
        return float(current_ver) > float(target_ver)
    except (ValueError, TypeError):
        return False


def main():
    global DRY_RUN, FORCE_DELETE
    args = sys.argv[1:]
    positional = []
    log_path = None
    for arg in args:
        if arg in ("-h", "--help"):
            init_logging(log_path)
            print_help(sys.argv[0])
            sys.exit(0)
        elif arg == "--dry-run":
            DRY_RUN = True
        elif arg == "--force":
            FORCE_DELETE = True
        elif arg.startswith("--log-file="):
            log_path = arg.split("=", 1)[1]
        else:
            positional.append(arg)

    init_logging(log_path)
    print(f"[*] Logging to: {LOG_FILE_DEFAULT}")

    if not positional:
        print_help(sys.argv[0])
        sys.exit(1)

    check_os_support()

    target_key = positional[0]
    if target_key not in TARGETS:
        print(
            "[!] Invalid target: {}\n    Allowed: {}".format(
                target_key, ", ".join(TARGETS.keys())
            )
        )
        sys.exit(1)

    tfamily, tver = TARGETS[target_key]
    current_family = installed_family()
    current_ver = None
    if current_family == "mariadb":
        current_ver = get_current_mariadb_version()
    elif current_family == "mysql":
        current_ver = get_current_mysql_version()

    if current_family == tfamily and current_ver == tver:
        print(
            f"[*] {tfamily.title()} {current_ver} is already installed. No changes needed."
        )
        sys.exit(0)

    switching = current_family not in ("none", tfamily)
    downgrading = (current_family == tfamily) and needs_downgrade(
        current_ver, tver
    )

    print(f"Target: {target_key} ({tver})")
    print(f"When:   {ts()}")

    # 1. First backup all databases while the server is still running
    print("\n[1/5] Backing up all user databases...")
    if not has_sufficient_space_for_backup():
        print("[!] Aborting due to insufficient space for backups.")
        sys.exit(1)
    backup_dir, backup_status = backup_all(BACKUP_ROOT)
    if backup_status == "error":
        print(
            "[!] Backup failed; aborting to prevent data loss. Backup directory: {}".format(
                backup_dir
            )
        )
        sys.exit(1)
    if backup_status == "empty":
        ans = (
            input(
                "[?] No user databases were found to back up. Continue anyway? [y/N]: "
            )
            .strip()
            .lower()
        )
        if ans not in ("y", "yes"):
            print("[!] Aborting at user request due to missing backups.")
            sys.exit(1)

    # 2. Stop database services
    print("\n[2/5] Stopping database services...")
    stop_db()

    # 3. Remove existing installation
    print("\n[3/5] Removing existing installation (if any)...")
    if current_family == 'mariadb':
        print("[*] Removing existing MariaDB installation...")
        remove_mariadb_family()
    elif current_family == 'mysql':
        print("[*] Removing existing MySQL installation...")
        remove_mysql_family()
    else:
        print("[*] No existing MySQL/MariaDB packages detected.")

    # 4. Clean up any remaining packages
    print("\n[4/5] Cleaning up...")
    run(["dnf", "-y", "autoremove"], check=False, destructive=True)

    # Ensure Perl modules after cleanup, before installing new DB version
    print("[*] Ensuring Perl MySQL modules are present...")
    ensure_perl_mysql_modules()

    # Set the desired version in cPanel config
    set_cpanel_mysql_version(tver)

    # Check what WHM allows (informational only)
    allowed = whm_allowed()
    if allowed:
        print("[i] WHM (probe) allows: {}".format(", ".join(allowed)))

    # Use cPanel API to install requested version
    print("\n[5/5] Installing requested engine via cPanel API...")
    install_ok = install_with_cpanel_api(tver)

    # Fallbacks if API is not available or fails
    if not install_ok:
        if tfamily == "mariadb" and tver == "10.6":
            print(
                "\n[!] API install failed; falling back to manual MariaDB {} install...".format(
                    tver
                )
            )
            install_mariadb106()
            install_ok = True
        elif tfamily == "mysql" and tver == "8.0":
            print(
                "\n[!] API install failed; falling back to manual MySQL {} install...".format(
                    tver
                )
            )
            install_mysql80()
            install_ok = True
        else:
            print(f"[!] No manual fallback implemented for {tfamily} {tver}.")

    if not install_ok:
        print(
            "\n[!] Installation did not complete successfully for {}.".format(
                tver
            )
        )
        print(f"    - Backup was saved to: {backup_dir}")
        print(
            "    - Track WHM status with: whmapi1 get_mysql_upgrade --output=json|yaml"
        )
        sys.exit(1)

    # Ensure /root/.my.cnf credentials work with the new service
    if not ensure_root_password_matches_mycnf():
        print(
            "[!] Unable to authenticate with MySQL using /root/.my.cnf credentials. Aborting to prevent data issues."
        )
        sys.exit(1)

    # Restore data and finish up
    if not restore_all_databases(backup_dir):
        print(
            "[!] One or more database restores failed; check logs before proceeding."
        )
        sys.exit(1)
    restore_max_allowed_packet()
    restart_mysql_service()
    if not restoregrants_all_users():
        print(
            "[!] One or more grant restoration steps failed. Review output above."
        )
    run_upcp_force()
    health_ok = post_restore_health_check()
    if not health_ok:
        print(
            "[!] Post-restore health check reported issues. Review logs and database state."
        )
    else:
        print("[+] Post-restore health check passed.")
    print("[+] Requested database engine installed and data restore complete.")
    print(f"    - Backup directory retained at: {backup_dir}")


if __name__ == "__main__":
    main()