backupmanager_ynh/sources/blueprints/destinations.py

import threading

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

from db import db, Destination, JobDestination

bp = Blueprint("dest", __name__)


@bp.route("/destinations")
def destinations_list():
    destinations = Destination.query.order_by(Destination.name).all()
    return render_template("destinations.html", destinations=destinations)


@bp.route("/destinations/new", methods=["GET", "POST"])
def destination_new():
    if request.method == "POST":
        return _save_destination(None)
    return render_template("destination_form.html", dest=None)


@bp.route("/destinations/<int:dest_id>/edit", methods=["GET", "POST"])
def destination_edit(dest_id):
    dest = db.get_or_404(Destination, dest_id)
    if request.method == "POST":
        return _save_destination(dest)
    pub_key = _get_pub_key(dest)
    return render_template("destination_form.html", dest=dest, pub_key=pub_key)


@bp.route("/destinations/<int:dest_id>/delete", methods=["POST"])
def destination_delete(dest_id):
    dest = db.get_or_404(Destination, dest_id)
    JobDestination.query.filter_by(dest_type="ssh", dest_id=dest_id).delete()
    db.session.delete(dest)
    db.session.commit()
    flash(f"Destination « {dest.name} » supprimée.", "success")
    return redirect(url_for("cfg.settings") + "?tab=destinations")


@bp.route("/destinations/<int:dest_id>/test", methods=["POST"])
def destination_test(dest_id):
    dest = db.get_or_404(Destination, dest_id)
    from jobs.transfer import test_connection
    ok, msg = test_connection(dest, current_app.config["DATA_DIR"])
    flash(msg, "success" if ok else "error")
    return redirect(url_for("cfg.settings") + "?tab=destinations")


@bp.route("/archives/<path:archive_name>/transfer", methods=["POST"])
def archive_transfer(archive_name):
    dest_id = request.form.get("destination_id", type=int)
    dest = db.get_or_404(Destination, dest_id)
    app = current_app._get_current_object()

    def _do_transfer():
        with app.app_context():
            try:
                from jobs.transfer import transfer_archive
                transfer_archive(archive_name, dest, app.config["YUNOHOST_BACKUP_DIR"],
                                 app.config["DATA_DIR"])
                app.logger.info(f"Transfert {archive_name} → {dest.remote_str} OK")
            except Exception as exc:
                app.logger.error(f"Transfert {archive_name} échoué : {exc}")

    threading.Thread(target=_do_transfer, daemon=True).start()
    flash(f"Transfert de « {archive_name} » vers {dest.remote_str} démarré.", "success")
    return redirect(request.referrer or url_for("jobs.index"))


def _save_destination(dest):
    f = request.form
    name = f.get("name", "").strip()
    host = f.get("host", "").strip()
    if not name or not host:
        flash("Nom et hôte sont requis.", "error")
        return render_template("destination_form.html", dest=dest)

    is_new = dest is None
    if is_new:
        dest = Destination()
        db.session.add(dest)

    dest.name = name
    dest.host = host
    dest.port = int(f.get("port", 22) or 22)
    dest.user = f.get("user", "root").strip() or "root"
    dest.remote_path = f.get("remote_path", "/home/yunohost.backup/archives").strip()
    dest.enabled = f.get("enabled") == "1"
    db.session.flush()

    if not dest.key_name:
        from jobs.transfer import generate_key
        dest.key_name = generate_key(dest.name, current_app.config["DATA_DIR"])

    db.session.commit()
    flash(f"Destination « {dest.name} » enregistrée.", "success")
    return redirect(url_for("dest.destination_edit", dest_id=dest.id))


def _get_pub_key(dest):
    if not dest.key_name:
        return None
    from jobs.transfer import get_public_key
    return get_public_key(dest.key_name, current_app.config["DATA_DIR"])