backupmanager_ynh/sources/retention.py

import os
import re
import shlex
import subprocess
from datetime import datetime, timedelta


def apply_retention(job, new_archive_name, backup_dir):
    """Applique la politique de rétention après une sauvegarde réussie."""
    import json as _json
    archives = _list_archives_for_job(job, backup_dir)
    if job.retention_mode == "count":
        to_delete = _retention_count(archives, job.retention_value)
    elif job.retention_mode == "daily":
        to_delete = _retention_daily(archives, job.retention_value)
    elif job.retention_mode == "gfs":
        cfg = _json.loads(job.retention_gfs_config or "{}") if job.retention_gfs_config else {}
        to_delete = _retention_gfs(archives, cfg)
    else:
        return []

    from jobs.utils import sudo_rm
    deleted = []
    failed = []
    for archive_filename in to_delete:
        base = os.path.splitext(archive_filename)[0]
        tar_path = os.path.join(backup_dir, base + ".tar")
        if sudo_rm(tar_path):
            deleted.append(archive_filename)
            sudo_rm(os.path.join(backup_dir, base + ".info.json"))
        else:
            failed.append(archive_filename)
    return deleted, failed


def _job_archive_prefix(job, instance_name):
    """Retourne le préfixe des archives pour ce job (ex: jerry_nextcloud_)."""
    if job.type == "ynh_app":
        import json
        cfg = json.loads(job.config_json or "{}")
        return f"{instance_name}_{cfg.get('app_id', '')}_"
    elif job.type == "ynh_system":
        return f"{instance_name}_system_"
    elif job.type in ("mysql", "postgresql"):
        import json
        cfg = json.loads(job.config_json or "{}")
        return f"{instance_name}_{job.type}_{cfg.get('database', '')}_"
    elif job.type == "custom_dir":
        label = re.sub(r'[^a-z0-9]+', '-', job.name.lower().strip()).strip('-')
        return f"{instance_name}_{label}_"
    else:
        return f"{instance_name}_{job.name.lower().replace(' ', '-')}_"


def _matches_job_prefix(name, prefix):
    """True si `name` appartient bien à ce job (et pas à un job dont le préfixe
    n'est qu'un préfixe plus court, ex. "redirect_" vs "redirect__2_" pour les
    instances multiples d'une app YunoHost). Le préfixe est toujours suivi de
    la date (8 chiffres AAAAMMJJ)."""
    return name.startswith(prefix) and name[len(prefix):len(prefix) + 8].isdigit()


def _list_archives_for_job(job, backup_dir):
    """Liste les archives correspondant à ce job, triées par date (plus ancienne en premier)."""
    from flask import current_app
    instance = current_app.config["INSTANCE_NAME"]
    prefix = _job_archive_prefix(job, instance)

    from jobs.utils import sudo_listdir
    archives = [
        fname for fname in sudo_listdir(backup_dir)
        if _matches_job_prefix(fname, prefix) and fname.endswith(".tar")
    ]
    archives.sort(key=_extract_date)
    return archives


def apply_ssh_retention(job, destination, data_dir):
    """Applique la rétention sur une destination SSH après un transfert réussi."""
    import json as _json
    from flask import current_app

    instance = current_app.config["INSTANCE_NAME"]
    prefix = _job_archive_prefix(job, instance)
    key_path = os.path.join(data_dir, "keys", destination.key_name)

    ssh_base = [
        "ssh",
        "-i", key_path,
        "-p", str(destination.port),
        "-o", "StrictHostKeyChecking=accept-new",
        "-o", "BatchMode=yes",
        "-o", "ConnectTimeout=30",
        f"{destination.user}@{destination.host}",
    ]

    remote_dir = shlex.quote(destination.remote_path)
    result = subprocess.run(
        ssh_base + [
            f"cd {remote_dir} && find . -maxdepth 1 -name '*.tar' -type f 2>/dev/null"
            " | sed 's|^\\./||'; true"
        ],
        capture_output=True, text=True, timeout=30,
    )
    if result.returncode != 0:
        raise RuntimeError(f"Listage distant échoué : {result.stderr.strip()}")

    remote_archives = sorted(
        [f.strip() for f in result.stdout.splitlines()
         if _matches_job_prefix(f.strip(), prefix) and f.strip().endswith(".tar")],
        key=_extract_date,
    )

    if job.retention_mode == "count":
        to_delete = _retention_count(remote_archives, job.retention_value)
    elif job.retention_mode == "daily":
        to_delete = _retention_daily(remote_archives, job.retention_value)
    elif job.retention_mode == "gfs":
        cfg = _json.loads(job.retention_gfs_config or "{}") if job.retention_gfs_config else {}
        to_delete = _retention_gfs(remote_archives, cfg)
    else:
        return []

    deleted = []
    for archive_filename in to_delete:
        base = os.path.splitext(archive_filename)[0]
        tar_q = shlex.quote(f"{destination.remote_path}/{base}.tar")
        info_q = shlex.quote(f"{destination.remote_path}/{base}.info.json")
        r = subprocess.run(
            ssh_base + [f"rm -f {tar_q} {info_q}"],
            capture_output=True, timeout=30,
        )
        if r.returncode == 0:
            deleted.append(archive_filename)
    return deleted


def apply_remote_retention(job, client):
    """Applique la rétention sur l'instance distante après un push.

    Filtre les archives par le même préfixe que le job local et applique
    la même politique (count/daily). Ne touche pas aux archives des autres jobs.
    """
    from flask import current_app
    instance = current_app.config["INSTANCE_NAME"]
    prefix = _job_archive_prefix(job, instance)

    try:
        remote_archives = client.get_archives()
    except Exception:
        return []

    matching = sorted(
        [a["name"] + ".tar" for a in remote_archives if _matches_job_prefix(a["name"], prefix)],
        key=_extract_date,
    )

    if job.retention_mode == "count":
        to_delete = _retention_count(matching, job.retention_value)
    elif job.retention_mode == "daily":
        to_delete = _retention_daily(matching, job.retention_value)
    elif job.retention_mode == "gfs":
        import json as _json
        cfg = _json.loads(job.retention_gfs_config or "{}") if job.retention_gfs_config else {}
        to_delete = _retention_gfs(matching, cfg)
    else:
        return []

    deleted = []
    for archive_filename in to_delete:
        base = os.path.splitext(archive_filename)[0]
        try:
            client.delete_archive(base)
            deleted.append(base)
        except Exception:
            pass
    return deleted


def _extract_date(filename):
    match = re.search(r'(\d{8})', filename)
    if match:
        try:
            return datetime.strptime(match.group(1), "%Y%m%d")
        except ValueError:
            pass
    return datetime.min


def _retention_count(archives, keep_n):
    if len(archives) <= keep_n:
        return []
    return archives[: len(archives) - keep_n]


def _retention_daily(archives, days):
    cutoff = datetime.utcnow() - timedelta(days=days)
    to_delete = []
    seen_dates = set()
    for archive in reversed(archives):
        date = _extract_date(archive)
        if date < cutoff:
            to_delete.append(archive)
            continue
        date_key = date.date()
        if date_key in seen_dates:
            to_delete.append(archive)
        else:
            seen_dates.add(date_key)
    return to_delete


def _retention_gfs(archives, config):
    """Politique Grandfather-Father-Son.

    config: {"daily": N, "weekly": M, "monthly": P}
    - Fils (daily)      : conserve les N archives les plus récentes
    - Père (weekly)     : conserve 1 archive par semaine sur M semaines
    - Grand-Père (monthly): conserve 1 archive par mois sur P mois
    Une archive peut satisfaire plusieurs catégories simultanément.
    """
    daily_keep = int(config.get("daily", 7))
    weekly_keep = int(config.get("weekly", 4))
    monthly_keep = int(config.get("monthly", 12))

    dated = []
    for archive in archives:
        d = _extract_date(archive)
        if d != datetime.min:
            dated.append((d, archive))

    if not dated:
        return []

    # Trier du plus récent au plus ancien
    dated.sort(key=lambda x: x[0], reverse=True)

    keepers = set()

    # Fils : N archives les plus récentes
    for _, archive in dated[:daily_keep]:
        keepers.add(archive)

    # Père : 1 archive par semaine (la plus récente de chaque semaine), M semaines
    seen_weeks = {}
    for d, archive in dated:
        wk = (d.isocalendar()[0], d.isocalendar()[1])
        if wk not in seen_weeks:
            seen_weeks[wk] = archive  # premier = plus récent de la semaine
    for wk in sorted(seen_weeks, reverse=True)[:weekly_keep]:
        keepers.add(seen_weeks[wk])

    # Grand-Père : 1 archive par mois (la plus récente du mois), P mois
    seen_months = {}
    for d, archive in dated:
        mk = (d.year, d.month)
        if mk not in seen_months:
            seen_months[mk] = archive
    for mk in sorted(seen_months, reverse=True)[:monthly_keep]:
        keepers.add(seen_months[mk])

    return [archive for _, archive in dated if archive not in keepers]