backupmanager_ynh/sources/blueprints/settings.py

import subprocess

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

from db import db, Setting

bp = Blueprint("cfg", __name__)

_SETTING_KEYS = [
    "smtp_host", "smtp_port", "smtp_user", "smtp_password",
    "smtp_from", "smtp_to", "smtp_tls", "smtp_ssl",
    "notify_on_success", "notify_on_error",
]


def _get_setting(key, default=""):
    s = Setting.query.filter_by(key=key).first()
    return s.value if s else default


@bp.route("/settings", methods=["GET", "POST"])
def settings():
    if request.method == "POST":
        action = request.form.get("action")

        if action == "test_smtp":
            from notifications import send_test_email
            try:
                send_test_email(
                    host=request.form.get("smtp_host", "").strip(),
                    port=int(request.form.get("smtp_port", 587) or 587),
                    user=request.form.get("smtp_user", "").strip(),
                    password=request.form.get("smtp_password", ""),
                    from_addr=request.form.get("smtp_from", "").strip(),
                    to_addr=request.form.get("smtp_to", "").strip(),
                    use_ssl=request.form.get("smtp_ssl") == "1",
                    use_tls=request.form.get("smtp_tls") == "1",
                )
                flash("Email de test envoyé avec succès.", "success")
            except Exception as exc:
                flash(f"Échec du test SMTP : {exc}", "error")
        else:
            for key in _SETTING_KEYS:
                if key in ("smtp_tls", "smtp_ssl", "notify_on_success", "notify_on_error"):
                    value = "1" if request.form.get(key) == "1" else "0"
                else:
                    value = request.form.get(key, "").strip()
                s = Setting.query.filter_by(key=key).first()
                if s is None:
                    s = Setting(key=key, value=value)
                    db.session.add(s)
                else:
                    s.value = value
            db.session.commit()
            flash("Paramètres enregistrés.", "success")

        return redirect(url_for("cfg.settings"))

    cfg = {k: _get_setting(k) for k in _SETTING_KEYS}
    cfg.setdefault("smtp_port", "587")
    cfg["smtp_tls"] = cfg.get("smtp_tls") or "1"
    cfg["smtp_ssl"] = cfg.get("smtp_ssl") or "0"
    cfg["notify_on_error"] = cfg.get("notify_on_error") or "1"
    api_token = current_app.config.get("API_TOKEN", "")
    instance_url = current_app.config.get("INSTANCE_URL", "")
    return render_template("settings.html", cfg=cfg, api_token=api_token,
                           instance_url=instance_url)


@bp.route("/internal/databases/<db_type>")
def internal_databases(db_type):
    """Liste les bases de données disponibles pour le formulaire job."""
    databases = []
    try:
        if db_type == "mysql":
            result = subprocess.run(
                ["sudo", "mysql", "--skip-column-names", "-e", "SHOW DATABASES;"],
                capture_output=True, text=True, timeout=10,
            )
            if result.returncode == 0:
                exclude = {"information_schema", "performance_schema", "mysql", "sys"}
                databases = [d.strip() for d in result.stdout.splitlines()
                             if d.strip() and d.strip() not in exclude]
        elif db_type == "postgresql":
            result = subprocess.run(
                ["sudo", "-u", "postgres", "psql", "-Atc",
                 "SELECT datname FROM pg_database WHERE datistemplate = false;"],
                capture_output=True, text=True, timeout=10,
            )
            if result.returncode == 0:
                databases = [d.strip() for d in result.stdout.splitlines() if d.strip()]
    except Exception:
        pass
    return jsonify(databases)