BorgBackup_gui/borgbackup_gui/BorgBackupGui.py

import os
import sys
from pathlib import Path
from borgbackup_gui.ConfigRepoGUI import ConfigRepoGUI
from borgbackup_gui.ListBackupGUI import ListBackupGUI
from PySide6.QtWidgets import QApplication, QWidget,QHBoxLayout, QVBoxLayout, QPlainTextEdit
from PySide6.QtCore import QObject, Slot, Signal, QThread
import configparser
import subprocess


class BorgBackupGui(QWidget):
    def __init__(self):
        super().__init__()
        self.config_file = configparser.ConfigParser()
        self.setWindowTitle("BorgBackup Gui")

        passphrase = ""
        depot = ""
        remote_path = "/usr/local/bin/borg"

        if Path("config.ini").exists():
            self.config_file.read("config.ini")
            try:
                passphrase = self.config_file["DEFAULT"]["passphrase"]
                depot = self.config_file["DEFAULT"]["depot"]
                remote_path = self.config_file["DEFAULT"]["remote_path"]
            except KeyError:
                pass

        self.backups:list[str] = []
        self.config = ConfigRepoGUI(
            passphrase=passphrase, 
            depot=depot,
            remote_path=remote_path)
        self.config.valider.connect(self.lister_depot_thread)
        self.log_console = QPlainTextEdit()
        self.log_console.setReadOnly(True)
        self.layout_h = QHBoxLayout()
        self.layout_v = QVBoxLayout()
        self.layout_v.addWidget(self.config)
        self.layout_v.addWidget(self.log_console)
        self.layout_h.addLayout(self.layout_v)
        self.setLayout(self.layout_h)
        self.show()

    @Slot()
    def lister_depot_thread(self):
        self.passphrase = self.config.get_passphrase()
        self.depot = self.config.get_repo()
        self.remote_path = self.config.get_remote_path()

        # On sauvegarde la configuration dans le fichier de config
        self.config_file['DEFAULT'] = {
                'passphrase': self.passphrase,
                'depot': self.depot,
                'remote_path': self.remote_path}
        with open('config.ini', 'w') as configfile:
            self.config_file.write(configfile)

        self.worker = ListerBackup(passphrase=self.passphrase,
                              depot = self.depot,
                              remote_path = self.remote_path)

        self.worker.finished.connect(self.afficher_liste_depot)
        self.worker.log_str.connect(self.log_console.appendPlainText)
        self.worker.start()
        
    def afficher_liste_depot(self):
        self.config.task_ended()
        self.list_backup_gui = ListBackupGUI(self.worker.get_backups())
        self.list_backup_gui.selected.connect(self.afficher_la_sauvegarde)
        self.layout_h.addWidget(self.list_backup_gui)
        self.adjustSize() 

    def afficher_la_sauvegarde(self, nom_backup):
        self.afficheur = MountAndDisplayBackup(self.passphrase, self.depot, self.remote_path, nom_backup)
        self.afficheur.log_str.connect(self.log_console.appendPlainText)
        self.afficheur.finished.connect(self.list_backup_gui.activate)
        self.afficheur.start()




class MountAndDisplayBackup(QThread):
    log_str = Signal(str)
    
    def __init__(self, passphrase, depot, remote_path, backup):
        super().__init__()
        self.passphrase = passphrase
        self.depot = depot
        self.remote_path = remote_path
        self.backup = backup

    def run(self):
        # mkdir /tmp/localborg/$(backup)
        base_dir = "/tmp/localborg/"
        backup_dir = base_dir + self.backup
        custom_env = os.environ.copy()
        custom_env["BORG_PASSPHRASE"] = self.passphrase
        self.log_str.emit("mkdir -p " + backup_dir)
        result = subprocess.run(["mkdir", "-p", backup_dir], capture_output=True, text=True)
        self.log_str.emit(result.stderr)
        self.log_str.emit(result.stdout)
        # borg mount nas.local:/volume1/NetBackup/chezmoi/client::2020-04-16 /tmp/localborg
        self.log_str.emit("borg mount " + self.depot + "::" + self.backup + " " + backup_dir)
        self.log_str.emit("Veuillez patienter, cette étape peut être longue")
        result = subprocess.run(["borg", "mount", self.depot + "::" + self.backup, "--remote-path", self.remote_path, backup_dir], env=custom_env, capture_output=True, text=True)
        self.log_str.emit(result.stderr)
        self.log_str.emit(result.stdout)
        self.log_str.emit("xdg-open" + backup_dir)
        result = subprocess.run(["xdg-open", backup_dir], capture_output=True, text=True)
        self.log_str.emit(result.stderr)
        self.log_str.emit(result.stdout)




class ListerBackup(QThread):
    log_str = Signal(str)

    def __init__(self, passphrase, depot, remote_path):
        super().__init__()
        self.passphrase = passphrase
        self.depot = depot
        self.remote_path = remote_path

    def run(self):
        self.lister_depot()

    def lister_depot(self):
        custom_env = os.environ.copy()
        custom_env["BORG_PASSPHRASE"] = self.passphrase

        self.log_str.emit("borg list " + self.depot + " --remote-path" + self.remote_path)
        result = subprocess.run(["borg", "list", self.depot, "--remote-path", self.remote_path], env=custom_env, capture_output=True, text=True)
        self.log_str.emit(result.stderr)
        self.log_str.emit(result.stdout)
        liste_backup = result.stdout.split("\n")
        self.backups:list[str] = []
        for backup in liste_backup:
            try:
                self.backups.append(backup.split()[0])
            except IndexError:
                pass

    def get_backups(self):
        return self.backups


if __name__ == "__main__":
    import sys
    app = QApplication([])
    widget = BorgBackupGui()

    sys.exit(app.exec())