backupmanager_ynh/sources/jobs/utils.py

import os
import subprocess


def sudo_getsize(path):
    """Retourne la taille d'un fichier via sudo stat si non accessible directement."""
    try:
        return os.path.getsize(path)
    except OSError:
        result = subprocess.run(
            ["sudo", "stat", "--format=%s", path],
            capture_output=True, text=True,
        )
        if result.returncode == 0:
            try:
                return int(result.stdout.strip())
            except ValueError:
                pass
        return 0


def sudo_getmtime(path):
    """Retourne le mtime d'un fichier via sudo stat si non accessible directement."""
    try:
        return os.path.getmtime(path)
    except OSError:
        result = subprocess.run(
            ["sudo", "stat", "--format=%Y", path],
            capture_output=True, text=True,
        )
        if result.returncode == 0:
            try:
                return float(result.stdout.strip())
            except ValueError:
                pass
        return 0.0


def unique_archive_name(base_name, backup_dir):
    """Retourne un nom d'archive unique (ajoute _2, _3… si le nom du jour est déjà pris)."""
    name = base_name
    i = 2
    while sudo_exists(os.path.join(backup_dir, name + ".tar")):
        name = f"{base_name}_{i}"
        i += 1
    return name


def sudo_exists(path):
    """Vérifie l'existence d'un fichier via sudo si non accessible directement."""
    if os.path.exists(path):
        return True
    result = subprocess.run(
        ["sudo", "stat", path],
        capture_output=True,
    )
    return result.returncode == 0


def sudo_listdir(directory):
    """Liste les fichiers d'un répertoire via sudo find si non accessible directement."""
    try:
        return os.listdir(directory)
    except OSError:
        result = subprocess.run(
            ["sudo", "find", directory, "-maxdepth", "1", "-mindepth", "1",
             "-printf", "%f\n"],
            capture_output=True, text=True,
        )
        if result.returncode == 0:
            return [f for f in result.stdout.strip().split("\n") if f]
        return []