421 lines
17 KiB
Python
421 lines
17 KiB
Python
import os
|
|
import configparser
|
|
import json
|
|
from distutils import util
|
|
from datetime import datetime
|
|
import subprocess
|
|
|
|
from PyQt5.QtCore import Qt
|
|
from PyQt5.QtWidgets import QDialog, QFileDialog
|
|
from PyQt5 import uic
|
|
|
|
from borg_qt.helper import BorgException
|
|
from borg_qt.systemd import SystemdFile
|
|
|
|
|
|
class Config(QDialog):
|
|
"""A class to read, display and write the Borg-Qt configuration."""
|
|
|
|
def __init__(self):
|
|
super(QDialog, self).__init__()
|
|
# Load the UI file to get the dialogs layout.
|
|
dir_path = os.path.dirname(os.path.realpath(__file__))
|
|
ui_path = os.path.join(dir_path + '/static/UI/Settings.ui')
|
|
uic.loadUi(ui_path, self)
|
|
|
|
# Connect all the button and actions.
|
|
self.button_box.accepted.connect(self.accept)
|
|
self.button_include_file.clicked.connect(self.include_file)
|
|
self.button_include_directory.clicked.connect(self.include_directory)
|
|
self.button_exclude_file.clicked.connect(self.exclude_file)
|
|
self.button_exclude_directory.clicked.connect(self.exclude_directory)
|
|
self.button_remove_include.clicked.connect(self.remove_include)
|
|
self.button_remove_exclude.clicked.connect(self.remove_exclude)
|
|
self.button_restore_exclude_defaults.clicked.connect(
|
|
self.restore_exclude_defaults)
|
|
|
|
weekdays = ['', 'Monday', 'Thuesday', 'Weekdays', 'Thursday', 'Friday',
|
|
'Saturday', 'Sunday']
|
|
self.combo_schedule_weekday.addItems(weekdays)
|
|
months = ['', 'January', 'February', 'March', 'April', 'May', 'June',
|
|
'July', 'August', 'September', 'October', 'November',
|
|
'December']
|
|
self.combo_schedule_month.addItems(months)
|
|
schedules = ['hourly', 'daily', 'weekly', 'monthly']
|
|
self.combo_schedule_predefined.addItems(schedules)
|
|
|
|
@property
|
|
def full_path(self):
|
|
"""returns the repository path or the repository server path if a
|
|
server was provided in the configuration."""
|
|
if self._return_single_option('repository_path'):
|
|
if self._return_single_option('server'):
|
|
return self._create_server_path()
|
|
else:
|
|
return self._return_single_option('repository_path')
|
|
else:
|
|
return ""
|
|
|
|
@property
|
|
def repository_path(self):
|
|
return self._return_single_option('repository_path')
|
|
|
|
@property
|
|
def password(self):
|
|
return self._return_single_option('password')
|
|
|
|
@property
|
|
def includes(self):
|
|
return self._return_list_option('includes')
|
|
|
|
@property
|
|
def excludes(self):
|
|
return self._return_list_option('excludes')
|
|
|
|
@property
|
|
def server(self):
|
|
return self._return_single_option('server')
|
|
|
|
@property
|
|
def port(self):
|
|
return self._return_single_option('port')
|
|
|
|
@property
|
|
def user(self):
|
|
return self._return_single_option('user')
|
|
|
|
@property
|
|
def prefix(self):
|
|
return self._return_single_option('prefix')
|
|
|
|
@property
|
|
def schedule_enabled(self):
|
|
return util.strtobool(self._return_single_option('schedule_enabled'))
|
|
|
|
@property
|
|
def schedule_predefined_enabled(self):
|
|
return util.strtobool(
|
|
self._return_single_option('schedule_predefined_enabled'))
|
|
|
|
@property
|
|
def schedule_custom_enabled(self):
|
|
return util.strtobool(
|
|
self._return_single_option('schedule_custom_enabled'))
|
|
|
|
@property
|
|
def schedule_time(self):
|
|
return self._return_single_option('schedule_time')
|
|
|
|
@property
|
|
def schedule_weekday(self):
|
|
return int(self._return_single_option('schedule_weekday'))
|
|
|
|
@property
|
|
def schedule_month(self):
|
|
return int(self._return_single_option('schedule_month'))
|
|
|
|
@property
|
|
def schedule_date(self):
|
|
return int(self._return_single_option('schedule_date'))
|
|
|
|
@property
|
|
def schedule_predefined_name(self):
|
|
return self._return_single_option('schedule_predefined_name')
|
|
|
|
@property
|
|
def hide_help(self):
|
|
return util.strtobool(self._return_single_option('hide_help'))
|
|
|
|
@property
|
|
def retention_policy_enabled(self):
|
|
return util.strtobool(self._return_single_option('retention_policy_enabled'))
|
|
|
|
@property
|
|
def retention_policy(self):
|
|
return self._return_dict_option('retention_policy')
|
|
|
|
def _return_single_option(self, option):
|
|
"""Gets the provided option from the configparser object."""
|
|
if option in self.config['borgqt']:
|
|
return self.config['borgqt'][option]
|
|
else:
|
|
return ""
|
|
|
|
def _return_list_option(self, option):
|
|
"""Reads the provided option from the configparser object and returns
|
|
it as a list."""
|
|
if option in self.config['borgqt']:
|
|
return json.loads(self.config['borgqt'][option])
|
|
else:
|
|
return []
|
|
|
|
def _return_dict_option(self, option):
|
|
"""Reads the provided option from the configparser object and returns
|
|
it as a dict."""
|
|
if option in self.config['borgqt']:
|
|
return json.loads(self.config['borgqt'][option])
|
|
else:
|
|
return {}
|
|
|
|
def _get_path(self):
|
|
"""searches for the configuration file and returns its full path."""
|
|
home = os.environ['HOME']
|
|
dir_path = os.path.dirname(os.path.realpath(__file__))
|
|
|
|
if os.path.exists(os.path.join(home, '.config/borg_qt/borg_qt.conf')):
|
|
return os.path.join(home, '.config/borg_qt/borg_qt.conf')
|
|
elif os.path.exists(os.path.join(dir_path, 'borg_qt.conf')):
|
|
return os.path.join(dir_path, 'borg_qt.conf')
|
|
else:
|
|
raise BorgException("Configuration file not found!")
|
|
|
|
def _set_environment_variables(self):
|
|
os.environ['BORG_REPO'] = self.full_path
|
|
os.environ['BORG_PASSPHRASE'] = self.password
|
|
|
|
def _create_server_path(self):
|
|
"""creates the full server path from the server, user and port
|
|
options."""
|
|
if not self._return_single_option('user'):
|
|
raise BorgException("User is missing in config.")
|
|
if not self._return_single_option('port'):
|
|
raise BorgException("Port is missing in config.")
|
|
server_path = ('ssh://'
|
|
+ self.config['borgqt']['user']
|
|
+ "@"
|
|
+ self.config['borgqt']['server']
|
|
+ ":"
|
|
+ self.config['borgqt']['port']
|
|
+ self.config['borgqt']['repository_path'])
|
|
return server_path
|
|
|
|
def _select_file(self):
|
|
"""Qt dialog to select an exisiting file."""
|
|
dialog = QFileDialog
|
|
dialog.ExistingFile
|
|
file_path, _ = dialog.getOpenFileName(
|
|
self, "Select Directory", os.getenv('HOME'), "All Files (*)")
|
|
return file_path
|
|
|
|
def _select_directory(self):
|
|
"""Qt dialog to select directories."""
|
|
dialog = QFileDialog
|
|
dialog.DirectoryOnly
|
|
return dialog.getExistingDirectory(
|
|
self, "Select Directory", os.getenv('HOME'))
|
|
|
|
def _create_service(self):
|
|
self.service = SystemdFile('borg_qt.service')
|
|
self.service.content['Service'] = {}
|
|
self.service.content['Unit']['Description'] = ("Runs Borg-Qt once in "
|
|
"the backround to take "
|
|
"a backup according to "
|
|
"the configuration.")
|
|
self.service.content['Service']['Type'] = 'oneshot'
|
|
process = subprocess.run(["which", "borg_qt"], stdout=subprocess.PIPE,
|
|
encoding='utf8')
|
|
output = process.stdout.strip()
|
|
self.service.content['Service']['ExecStart'] = output + ' -B'
|
|
self.service.write()
|
|
|
|
def _create_timer(self, schedule_interval):
|
|
self.timer = SystemdFile('borg_qt.timer')
|
|
self.timer.content['Timer'] = {}
|
|
self.timer.content['Install'] = {}
|
|
self.timer.content['Unit']['Description'] = ("Starts the "
|
|
"borg_qt.service "
|
|
"according to the "
|
|
"configured "
|
|
"schedule.")
|
|
self.timer.content['Timer']['OnCalendar'] = schedule_interval
|
|
self.timer.content['Timer']['Persistent'] = 'true'
|
|
self.timer.content['Install']['WantedBy'] = 'timers.target'
|
|
self.timer.write()
|
|
|
|
def _parse_schedule_interval(self):
|
|
if self.schedule_predefined_enabled:
|
|
return self.schedule_predefined_name
|
|
if self.schedule_custom_enabled:
|
|
if self.schedule_date > 0:
|
|
date = str(self.schedule_date)
|
|
else:
|
|
date = "*"
|
|
if self.schedule_month > 0:
|
|
month = str(self.schedule_month)
|
|
else:
|
|
month = "*"
|
|
if self.schedule_weekday > 0:
|
|
weekday = self.combo_schedule_weekday.currentText()
|
|
else:
|
|
weekday = ""
|
|
|
|
date_string = (weekday
|
|
+ " "
|
|
+ "*"
|
|
+ "-"
|
|
+ month
|
|
+ "-"
|
|
+ date
|
|
+ " "
|
|
+ self.schedule_time)
|
|
return date_string
|
|
|
|
def include_file(self):
|
|
"""add a file to the include list if the selected path is not empty."""
|
|
file_path = self._select_file()
|
|
if file_path:
|
|
self.list_include.addItem(file_path)
|
|
|
|
def include_directory(self):
|
|
"""add a directory to the include list if the selected path is not
|
|
empty."""
|
|
directory_path = self._select_directory()
|
|
if directory_path:
|
|
self.list_include.addItem(directory_path)
|
|
|
|
def exclude_file(self):
|
|
"""add a file to the exclude list if the selected path is not empty."""
|
|
file_path = self._select_file()
|
|
if file_path:
|
|
self.list_exclude.addItem(file_path)
|
|
|
|
def exclude_directory(self):
|
|
"""add a file to the exclude list if the selected path is not empty."""
|
|
directory_path = self._select_directory()
|
|
if directory_path:
|
|
self.list_exclude.addItem(directory_path)
|
|
|
|
def remove_include(self):
|
|
self.list_include.takeItem(self.list_include.currentRow())
|
|
|
|
def remove_exclude(self):
|
|
self.list_exclude.takeItem(self.list_exclude.currentRow())
|
|
|
|
def restore_exclude_defaults(self):
|
|
self.list_exclude.clear()
|
|
default_excludes = json.loads(self.config['DEFAULT']['excludes'])
|
|
self.list_exclude.addItems(default_excludes)
|
|
|
|
def read(self):
|
|
"""Reads the config file"""
|
|
self.path = self._get_path()
|
|
self.config = configparser.ConfigParser()
|
|
self.config.read(self.path)
|
|
|
|
def set_form_values(self):
|
|
# set the general tab values
|
|
self.line_edit_repository_path.setText(self.repository_path)
|
|
self.line_edit_password.setText(self.password)
|
|
self.line_edit_prefix.setText(self.prefix)
|
|
self.line_edit_server.setText(self.server)
|
|
self.line_edit_port.setText(self.port)
|
|
self.line_edit_user.setText(self.user)
|
|
# set the include tab values
|
|
self.list_include.clear()
|
|
self.list_include.addItems(self.includes)
|
|
# set the include tab values
|
|
self.list_exclude.clear()
|
|
self.list_exclude.addItems(self.excludes)
|
|
# set schedule tab values
|
|
if self.schedule_enabled:
|
|
self.check_schedule_enabled.setChecked(True)
|
|
if self.schedule_custom_enabled:
|
|
self.radio_schedule_custom_enabled.setChecked(True)
|
|
_time = datetime.strptime(self.schedule_time, '%H:%M:%S')
|
|
self.time_schedule_time.setTime(_time.time())
|
|
self.combo_schedule_weekday.setCurrentIndex(self.schedule_weekday)
|
|
self.combo_schedule_month.setCurrentIndex(self.schedule_month)
|
|
index = self.combo_schedule_predefined.findText(
|
|
self.schedule_predefined_name, Qt.MatchFixedString)
|
|
self.combo_schedule_predefined.setCurrentIndex(index)
|
|
self.spin_schedule_date.setValue(self.schedule_date)
|
|
if self.retention_policy_enabled:
|
|
self.check_policy_enabled.setChecked(True)
|
|
policy = self.retention_policy
|
|
self.spin_policy_hourly.setValue(int(policy['hourly']))
|
|
self.spin_policy_daily.setValue(int(policy['daily']))
|
|
self.spin_policy_weekly.setValue(int(policy['weekly']))
|
|
self.spin_policy_monthly.setValue(int(policy['monthly']))
|
|
self.spin_policy_yearly.setValue(int(policy['yearly']))
|
|
|
|
def apply_options(self):
|
|
"""Writes the changed options back into the configparser object."""
|
|
self.config['borgqt']['repository_path'] = (
|
|
self.line_edit_repository_path.text())
|
|
self.config['borgqt']['password'] = self.line_edit_password.text()
|
|
self.config['borgqt']['prefix'] = self.line_edit_prefix.text()
|
|
self.config['borgqt']['server'] = self.line_edit_server.text()
|
|
self.config['borgqt']['port'] = self.line_edit_port.text()
|
|
self.config['borgqt']['user'] = self.line_edit_user.text()
|
|
self.config['borgqt']['schedule_enabled'] = (
|
|
str(self.check_schedule_enabled.isChecked()))
|
|
self.config['borgqt']['schedule_predefined_enabled'] = (
|
|
str(self.radio_schedule_predefined_enabled.isChecked()))
|
|
self.config['borgqt']['schedule_custom_enabled'] = (
|
|
str(self.radio_schedule_custom_enabled.isChecked()))
|
|
self.config['borgqt']['schedule_time'] = (
|
|
self.time_schedule_time.time().toString())
|
|
self.config['borgqt']['schedule_weekday'] = (
|
|
str(self.combo_schedule_weekday.currentIndex()))
|
|
self.config['borgqt']['schedule_month'] = (
|
|
str(self.combo_schedule_month.currentIndex()))
|
|
self.config['borgqt']['schedule_date'] = self.spin_schedule_date.text()
|
|
self.config['borgqt']['schedule_predefined_name'] = (
|
|
self.combo_schedule_predefined.currentText())
|
|
|
|
# Workaraound to get all items of a QListWidget as a list
|
|
excludes = []
|
|
for index in range(self.list_exclude.count()):
|
|
excludes.append(self.list_exclude.item(index).text())
|
|
|
|
# Workaraound to get all items of a QListWidget as a list
|
|
includes = []
|
|
for index in range(self.list_include.count()):
|
|
includes.append(self.list_include.item(index).text())
|
|
|
|
# Configparser doesn't know about list therefore we store them as json
|
|
# strings
|
|
self.config['borgqt']['includes'] = json.dumps(includes,
|
|
indent=4,
|
|
sort_keys=True)
|
|
self.config['borgqt']['excludes'] = json.dumps(excludes,
|
|
indent=4,
|
|
sort_keys=True)
|
|
self.config['borgqt']['retention_policy_enabled'] = (
|
|
str(self.check_policy_enabled.isChecked()))
|
|
retention_policy = {}
|
|
retention_policy['hourly'] = self.spin_policy_hourly.text()
|
|
retention_policy['daily'] = self.spin_policy_daily.text()
|
|
retention_policy['weekly'] = self.spin_policy_weekly.text()
|
|
retention_policy['monthly'] = self.spin_policy_monthly.text()
|
|
retention_policy['yearly'] = self.spin_policy_yearly.text()
|
|
self.config['borgqt']['retention_policy'] = json.dumps(retention_policy,
|
|
indent=4,
|
|
sort_keys=True)
|
|
self._set_environment_variables()
|
|
|
|
# create and enable the required systemd files
|
|
# if it is not enable make sure that the timer is disabled.
|
|
if self.schedule_enabled:
|
|
self._create_service()
|
|
self._create_timer(self._parse_schedule_interval())
|
|
self.timer.enable()
|
|
else:
|
|
active_timer_path = os.path.join(
|
|
os.environ['HOME'],
|
|
'.config/systemd/user/timer.targets.wants/borg_qt.timer')
|
|
if os.path.exists(active_timer_path):
|
|
self.timer.disable()
|
|
|
|
def write(self):
|
|
"""Write the configparser object back to the config file."""
|
|
with open(self.path, 'w+') as configfile:
|
|
self.config.write(configfile)
|
|
|
|
def accept(self):
|
|
"""Extend the built in accept method to apply and write the new
|
|
options."""
|
|
super().accept()
|
|
self.apply_options()
|
|
self.write()
|