backupmanager_ynh/sources/jobs/utils.py

import json
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_read_backup_info(archive_path):
    """Lit backup_info.json depuis un .tar root-owned via sudo tar -xOf."""
    result = subprocess.run(
        ["sudo", "tar", "-xOf", archive_path, "backup_info.json"],
        capture_output=True,
    )
    if result.returncode == 0:
        try:
            return json.loads(result.stdout)
        except Exception:
            pass
    return {}


def batch_list_archives(backup_dir):
    """Retourne {name: {size_bytes, mtime}} pour tous les .tar — UN seul appel sudo find."""
    result = subprocess.run(
        ["sudo", "find", backup_dir, "-maxdepth", "1", "-name", "*.tar",
         "-printf", "%f\t%s\t%T@\n"],
        capture_output=True, text=True,
    )
    stats = {}
    for line in result.stdout.splitlines():
        parts = line.strip().split("\t")
        if len(parts) == 3:
            fname, size_str, mtime_str = parts
            if fname.endswith(".tar"):
                name = fname[:-4]
                try:
                    stats[name] = {"size_bytes": int(size_str), "mtime": float(mtime_str)}
                except ValueError:
                    pass
    return stats


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 []