backupmanager_ynh/sources/blueprints/overview.py

import json
import os
import shlex
import subprocess
import threading
from concurrent.futures import ThreadPoolExecutor

from flask import Blueprint, current_app, flash, redirect, render_template, request, url_for

from db import db, Job, Destination, RemoteInstance, _size_human
from helpers import get_ynh_apps

bp = Blueprint("overview", __name__)


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

def _retention_label(job):
    if job.retention_mode == "count":
        return f"{job.retention_value} dernières"
    if job.retention_mode == "daily":
        return f"1/j sur {job.retention_value} j"
    if job.retention_mode == "gfs":
        cfg = json.loads(job.retention_gfs_config or "{}") if job.retention_gfs_config else {}
        d = cfg.get("daily", 7)
        w = cfg.get("weekly", 4)
        m = cfg.get("monthly", 12)
        return f"GFS {d}j/{w}s/{m}m"
    return "—"


def _job_detail(job, ynh_app_labels=None):
    """Détail affiché à côté du nom du job pour distinguer des jobs homonymes
    (ex: instances multiples d'une même app YunoHost, toutes nommées pareil)."""
    if job.type not in ("ynh_app", "custom_dir", "mysql", "postgresql"):
        return ""
    cfg = json.loads(job.config_json or "{}")
    if job.type == "ynh_app":
        app_id = cfg.get("app_id", "")
        label = (ynh_app_labels or {}).get(app_id)
        return f"{label} ({app_id})" if label and label != app_id else app_id
    if job.type == "custom_dir":
        return cfg.get("source_path", "")
    return cfg.get("database", "")


def _group_archives(archive_list, jobs, instance, ynh_app_labels=None):
    """
    archive_list : [{"name": "jerry_nextcloud_20260716", "size_bytes": N}]
    Retourne (groups, orphaned) où chaque group est associé à un job.
    """
    from retention import _job_archive_prefix, _matches_job_prefix, _extract_date

    used = set()
    groups = []
    for job in jobs:
        prefix = _job_archive_prefix(job, instance)
        matching = [a for a in archive_list if _matches_job_prefix(a["name"], prefix)]
        matching.sort(key=lambda a: _extract_date(a["name"] + ".tar"))
        for a in matching:
            used.add(a["name"])
            d = _extract_date(a["name"] + ".tar")
            a["date_str"] = d.strftime("%d %b %Y") if d.year > 1 else "—"
            a["size_human"] = _size_human(a["size_bytes"])
        groups.append({
            "job": job,
            "retention_label": _retention_label(job),
            "detail": _job_detail(job, ynh_app_labels),
            "archives": matching,
        })

    orphaned = []
    for a in archive_list:
        if a["name"] not in used:
            from retention import _extract_date
            d = _extract_date(a["name"] + ".tar")
            a["date_str"] = d.strftime("%d %b %Y") if d.year > 1 else "—"
            a["size_human"] = _size_human(a["size_bytes"])
            orphaned.append(a)

    return groups, orphaned


def _collect_local(backup_dir):
    from jobs.utils import batch_list_archives
    stats = batch_list_archives(backup_dir)
    return [{"name": name, "size_bytes": info["size_bytes"]} for name, info in stats.items()]


def _ssh_base(destination, data_dir):
    key_path = os.path.join(data_dir, "keys", destination.key_name)
    return [
        "ssh", "-i", key_path,
        "-p", str(destination.port),
        "-o", "StrictHostKeyChecking=accept-new",
        "-o", "BatchMode=yes",
        "-o", "ConnectTimeout=15",
        f"{destination.user}@{destination.host}",
    ]


def _collect_ssh(destination, data_dir):
    remote_dir = shlex.quote(destination.remote_path)
    # find -printf n'est pas disponible sur tous les systèmes (BusyBox, BSD…)
    # On utilise cd + find + stat avec fallback BSD pour la portabilité
    remote_cmd = (
        f"cd {remote_dir} && find . -maxdepth 1 -name '*.tar' -type f 2>/dev/null"
        " | sed 's|^\\./||'"
        " | while IFS= read -r f; do"
        " s=$(stat -c%s \"$f\" 2>/dev/null || stat -f%z \"$f\" 2>/dev/null || echo 0);"
        " printf '%s\\t%s\\n' \"$f\" \"$s\";"
        " done; true"
    )
    result = subprocess.run(
        _ssh_base(destination, data_dir) + [remote_cmd],
        capture_output=True, text=True, timeout=20,
    )
    if result.returncode != 0:
        raise RuntimeError(result.stderr.strip() or "Connexion SSH échouée")
    archives = []
    for line in result.stdout.splitlines():
        parts = line.strip().split("\t")
        if len(parts) == 2 and parts[0].endswith(".tar"):
            try:
                archives.append({"name": parts[0][:-4], "size_bytes": int(parts[1])})
            except ValueError:
                pass
    return archives


def _collect_instance(instance):
    from federation.client import FederationClient
    raw = FederationClient(instance).get_archives()
    return [{"name": a["name"], "size_bytes": a.get("size_bytes", 0)} for a in raw]


def _fetch_ssh_file(destination, data_dir, remote_filename):
    remote_path = shlex.quote(f"{destination.remote_path}/{remote_filename}")
    result = subprocess.run(
        _ssh_base(destination, data_dir) + [f"cat {remote_path}"],
        capture_output=True, timeout=3600,
    )
    if result.returncode != 0:
        raise RuntimeError(result.stderr.decode(errors="replace").strip() or "Connexion SSH échouée")
    return result.stdout


def _do_pull(app, loc_type, archive_name, dest_id=None, instance_id=None):
    """Rapatrie une archive nommée (SSH ou instance) vers le backup_dir local."""
    with app.app_context():
        backup_dir = app.config["YUNOHOST_BACKUP_DIR"]
        data_dir = app.config["DATA_DIR"]
        try:
            if loc_type == "ssh":
                dest = db.session.get(Destination, dest_id)
                label = dest.host
                tar_bytes = _fetch_ssh_file(dest, data_dir, archive_name + ".tar")
                try:
                    info_bytes = _fetch_ssh_file(dest, data_dir, archive_name + ".info.json")
                except Exception:
                    info_bytes = None

            elif loc_type == "instance":
                inst = db.session.get(RemoteInstance, instance_id)
                label = inst.name
                from federation.client import FederationClient
                client = FederationClient(inst)
                tar_bytes = client.download_archive(archive_name)
                info_bytes = client.download_info_json(archive_name)

            else:
                return

            tmp_tar = f"/tmp/backupmanager_pull_{archive_name}.tar"
            with open(tmp_tar, "wb") as f:
                f.write(tar_bytes)
            subprocess.run(["sudo", "rsync", tmp_tar,
                            os.path.join(backup_dir, archive_name + ".tar")], check=True)
            os.unlink(tmp_tar)

            if info_bytes:
                tmp_info = f"/tmp/backupmanager_pull_{archive_name}.info.json"
                with open(tmp_info, "wb") as f:
                    f.write(info_bytes)
                subprocess.run(["sudo", "rsync", tmp_info,
                                os.path.join(backup_dir, archive_name + ".info.json")], check=True)
                os.unlink(tmp_info)

            app.logger.info(f"Pull {archive_name} ← {label} OK")
        except Exception as exc:
            app.logger.error(f"Pull {archive_name} échoué : {exc}")


# ---------------------------------------------------------------------------
# Routes
# ---------------------------------------------------------------------------

@bp.route("/overview")
def overview():
    backup_dir = current_app.config["YUNOHOST_BACKUP_DIR"]
    data_dir = current_app.config["DATA_DIR"]
    instance = current_app.config["INSTANCE_NAME"]
    jobs = Job.query.order_by(Job.name).all()
    ynh_app_labels = {a["id"]: a.get("label") or a.get("name", "") for a in get_ynh_apps()}

    # Construire la liste des tâches de collecte dans l'ordre d'affichage souhaité
    dests = Destination.query.filter_by(enabled=True).order_by(Destination.name).all()
    insts = RemoteInstance.query.order_by(RemoteInstance.name).all()

    tasks = [
        ({"type": "local", "label": "Local"},
         lambda: _collect_local(backup_dir)),
    ]
    for dest in dests:
        tasks.append((
            {"type": "ssh", "label": f"SSH → {dest.remote_str}", "dest_id": dest.id},
            lambda d=dest: _collect_ssh(d, data_dir),
        ))
    for inst in insts:
        tasks.append((
            {"type": "instance", "label": inst.name,
             "instance_id": inst.id, "instance_url": inst.url},
            lambda i=inst: _collect_instance(i),
        ))

    # Collecter toutes les destinations en parallèle pour éviter que les timeouts
    # d'une destination lente ne bloquent les autres
    def _run_task(args):
        meta, fn = args
        try:
            return meta, fn(), None
        except Exception as exc:
            return meta, None, str(exc)

    with ThreadPoolExecutor(max_workers=min(len(tasks), 10)) as pool:
        raw_results = list(pool.map(_run_task, tasks))

    locations = []
    for meta, archives, error in raw_results:
        if error:
            locations.append({**meta, "error": error, "groups": [], "orphaned": []})
        else:
            groups, orphaned = _group_archives(archives, jobs, instance, ynh_app_labels)
            locations.append({**meta, "error": None, "groups": groups, "orphaned": orphaned})

    # --- Vue par job : pivot de locations → jobs ---
    jobs_view = []
    for job in jobs:
        allowed = {(jd.dest_type, jd.dest_id) for jd in job.job_destinations}
        jlocs = []
        for loc in locations:
            if loc["type"] == "ssh" and ("ssh", loc.get("dest_id")) not in allowed:
                continue
            if loc["type"] == "instance" and ("instance", loc.get("instance_id")) not in allowed:
                continue
            group = next((g for g in loc.get("groups", []) if g["job"].id == job.id), None)
            jlocs.append({
                "type": loc["type"],
                "label": loc["label"],
                "dest_id": loc.get("dest_id"),
                "instance_id": loc.get("instance_id"),
                "error": loc["error"],
                "archives": group["archives"] if group else [],
            })
        jobs_view.append({
            "job": job,
            "retention_label": _retention_label(job),
            "detail": _job_detail(job, ynh_app_labels),
            "locations": jlocs,
        })

    return render_template("archives_overview.html", locations=locations, jobs_view=jobs_view)


@bp.route("/overview/delete", methods=["POST"])
def overview_delete():
    name = request.form["name"]
    loc_type = request.form["loc_type"]
    backup_dir = current_app.config["YUNOHOST_BACKUP_DIR"]
    data_dir = current_app.config["DATA_DIR"]

    try:
        if loc_type == "local":
            from jobs.utils import sudo_rm_archive
            sudo_rm_archive(name, backup_dir)
            flash(f"Archive « {name} » supprimée localement.", "success")

        elif loc_type == "ssh":
            dest = db.get_or_404(Destination, int(request.form["dest_id"]))
            tar_q = shlex.quote(f"{dest.remote_path}/{name}.tar")
            info_q = shlex.quote(f"{dest.remote_path}/{name}.info.json")
            r = subprocess.run(
                _ssh_base(dest, data_dir) + [f"rm -f {tar_q} {info_q}"],
                capture_output=True, timeout=20,
            )
            if r.returncode != 0:
                raise RuntimeError(r.stderr.decode().strip())
            flash(f"Archive « {name} » supprimée sur {dest.host}.", "success")

        elif loc_type == "instance":
            inst = db.get_or_404(RemoteInstance, int(request.form["instance_id"]))
            from federation.client import FederationClient
            FederationClient(inst).delete_archive(name)
            flash(f"Archive « {name} » supprimée sur {inst.name}.", "success")

    except Exception as exc:
        flash(f"Erreur lors de la suppression : {exc}", "error")

    return redirect(url_for("overview.overview"))


@bp.route("/overview/pull", methods=["POST"])
def overview_pull():
    name = request.form["name"]
    loc_type = request.form["loc_type"]
    dest_id = request.form.get("dest_id", type=int)
    instance_id = request.form.get("instance_id", type=int)

    app = current_app._get_current_object()
    threading.Thread(target=_do_pull, args=(app, loc_type, name, dest_id, instance_id), daemon=True).start()
    flash(f"Rapatriement de « {name} » démarré en arrière-plan.", "success")
    return redirect(url_for("overview.overview"))


@bp.route("/overview/retention", methods=["POST"])
def overview_retention():
    job_id = int(request.form["job_id"])
    loc_type = request.form["loc_type"]
    job = db.get_or_404(Job, job_id)
    backup_dir = current_app.config["YUNOHOST_BACKUP_DIR"]
    data_dir = current_app.config["DATA_DIR"]

    try:
        if loc_type == "local":
            from retention import apply_retention
            deleted, failed = apply_retention(job, "", backup_dir)
            msg = f"Rétention locale — {len(deleted)} archive(s) supprimée(s)"
            if failed:
                msg += f", {len(failed)} échec(s) : {', '.join(failed)}"
            flash(msg, "success" if not failed else "warning")

        elif loc_type == "ssh":
            dest = db.get_or_404(Destination, int(request.form["dest_id"]))
            from retention import apply_ssh_retention
            deleted = apply_ssh_retention(job, dest, data_dir)
            flash(
                f"Rétention SSH — {len(deleted)} archive(s) supprimée(s) sur {dest.host}.",
                "success",
            )

        elif loc_type == "instance":
            inst = db.get_or_404(RemoteInstance, int(request.form["instance_id"]))
            from federation.client import FederationClient
            from retention import apply_remote_retention
            deleted = apply_remote_retention(job, FederationClient(inst))
            flash(
                f"Rétention distante — {len(deleted)} archive(s) supprimée(s) sur {inst.name}.",
                "success",
            )

    except Exception as exc:
        flash(f"Erreur lors de la rétention : {exc}", "error")

    return redirect(url_for("overview.overview"))