feat: rework prefix backups with libarchive progress UI #170

Merged
Boria138 merged 1 commits from libarchive-backup into main 2026-05-23 16:31:17 +00:00
12 changed files with 589 additions and 92 deletions

No files matched your search

+2 -1
View File
@@ -6,7 +6,7 @@ arch=('any')
url="https://git.linux-gaming.ru/Linux-Gaming/PortProtonQt"
license=('GPL-3.0')
depends=('python-requests' 'python-babel' 'python-evdev' 'python-pygame' 'python-orjson'
'python-psutil' 'python-tqdm' 'python-vdf' 'python-libarchive-c' 'pyside6' 'python-rapidfuzz' 'python-pefile' 'python-pillow' 'perl-image-exiftool' 'python-websocket-client' 'cabextract' 'unzip' 'curl' 'jq' 'file' 'findutils' 'gawk' 'grep' 'tar' 'xz' 'zstd' 'gzip' 'unrar' 'qt6-svg' 'qt6-imageformats' 'pciutils' 'mesa-utils' 'vulkan-icd-loader' 'procps-ng' 'psmisc' 'squashfs-tools' '7zip' 'python-dbus-fast')
'python-psutil' 'python-tqdm' 'python-vdf' 'python-libarchive-c' 'pyside6' 'python-rapidfuzz' 'python-pefile' 'python-pillow' 'perl-image-exiftool' 'python-websocket-client' 'cabextract' 'unzip' 'curl' 'jq' 'file' 'findutils' 'gawk' 'grep' 'tar' 'xz' 'zstd' 'gzip' 'unrar' 'qt6-svg' 'qt6-imageformats' 'pciutils' 'mesa-utils' 'vulkan-icd-loader' 'procps-ng' 'psmisc' '7zip' 'python-dbus-fast')
makedepends=('meson' 'ninja' 'vulkan-headers' 'gettext')
source=("git+https://git.linux-gaming.ru/Linux-Gaming/PortProtonQt#tag=v$pkgver")
sha256sums=('SKIP')
@@ -24,6 +24,7 @@ package_portprotonqt() {
'upower: System tab bluetooth battery level'
'libpulse: System tab audio volume/output management'
'python-qrcode: Wi-Fi QR code generation'
'squashfs-tools: for legacy PortProton prefix backup support'
)
DESTDIR="$pkgdir" meson install -C build --skip-subprojects
+2 -1
View File
@@ -6,7 +6,7 @@ arch=('any')
url="https://git.linux-gaming.ru/Linux-Gaming/PortProtonQt"
license=('GPL-3.0')
depends=('python-requests' 'python-babel' 'python-evdev' 'python-pygame' 'python-orjson'
'python-psutil' 'python-tqdm' 'python-vdf' 'python-libarchive-c' 'pyside6' 'python-rapidfuzz' 'python-pefile' 'python-pillow' 'perl-image-exiftool' 'python-websocket-client' 'cabextract' 'unzip' 'curl' 'jq' 'file' 'findutils' 'gawk' 'grep' 'tar' 'xz' 'zstd' 'gzip' 'unrar' 'qt6-svg' 'qt6-imageformats' 'pciutils' 'mesa-utils' 'vulkan-icd-loader' 'procps-ng' 'psmisc' 'squashfs-tools' '7zip' 'python-dbus-fast')
'python-psutil' 'python-tqdm' 'python-vdf' 'python-libarchive-c' 'pyside6' 'python-rapidfuzz' 'python-pefile' 'python-pillow' 'perl-image-exiftool' 'python-websocket-client' 'cabextract' 'unzip' 'curl' 'jq' 'file' 'findutils' 'gawk' 'grep' 'tar' 'xz' 'zstd' 'gzip' 'unrar' 'qt6-svg' 'qt6-imageformats' 'pciutils' 'mesa-utils' 'vulkan-icd-loader' 'procps-ng' 'psmisc' '7zip' 'python-dbus-fast')
makedepends=('meson' 'ninja' 'vulkan-headers' 'gettext')
source=("git+https://git.linux-gaming.ru/Linux-Gaming/PortProtonQt.git")
sha256sums=('SKIP')
@@ -29,6 +29,7 @@ package_portprotonqt-git() {
'upower: System tab bluetooth battery level'
'libpulse: System tab audio volume/output management'
'python-qrcode: Wi-Fi QR code generation'
'squashfs-tools: for legacy PortProton prefix backup support'
)
DESTDIR="$pkgdir" meson install -C build --skip-subprojects
+3 -1
View File
@@ -63,7 +63,6 @@ Requires: pciutils
Requires: vulkan-loader
Requires: procps-ng
Requires: psmisc
Requires: squashfs-tools
Requires: 7zip
Requires: python3-dbus-fast
@@ -74,6 +73,9 @@ Recommends: upower
Recommends: pulseaudio-utils
Recommends: python3-qrcode
# For legacy PortProton prefix backup support
Recommends: squashfs-tools
%description
A modern and intuitive interface for managing and launching games from PortProton and Steam. Combines libraries in one place and simplifies running Windows games on Linux.
+3 -1
View File
@@ -60,7 +60,6 @@ Requires: pciutils
Requires: vulkan-loader
Requires: procps-ng
Requires: psmisc
Requires: squashfs-tools
Requires: 7zip
Requires: python3-dbus-fast
@@ -71,6 +70,9 @@ Recommends: upower
Recommends: pulseaudio-utils
Recommends: python3-qrcode
# For legacy PortProton prefix backup support
Recommends: squashfs-tools
%description
A modern and intuitive interface for managing and launching games from PortProton and Steam. Combines libraries in one place and simplifies running Windows games on Linux.
@@ -4,6 +4,5 @@
<comment>PortProton prefix backup</comment>
<comment xml:lang="ru">Резервная копия префикса PortProton</comment>
<glob pattern="*.ppack"/>
<sub-class-of type="application/vnd.squashfs"/>
</mime-type>
</mime-info>
@@ -4668,65 +4668,25 @@ pw_create_prefix_backup_cli () {
PW_PREFIX_NAME="$1"
PW_PREFIX_TO_BACKUP="$2"
if [[ -n "$(grep "/${PW_PREFIX_NAME}/" "${PORT_DATA_PATH}"/*.desktop )" ]] ; then
try_remove_file "${PORT_DATA_PATH}/data/prefixes/${PW_PREFIX_NAME}/.create_shortcut"
grep "/${PW_PREFIX_NAME}/" "${PORT_DATA_PATH}"/*.desktop | awk -F"/${PW_PREFIX_NAME}/" '{print $2}' \
| awk -F\" '{print $1}' > "${PORT_DATA_PATH}/data/prefixes/${PW_PREFIX_NAME}/.create_shortcut"
fi
unset PW_SANDBOX_HOME_PATH
pw_init_runtime
if check_flatpak; then
TMP_ALL_PATH=""
else
TMP_ALL_PATH="LD_LIBRARY_PATH=\"${PW_LD_LIBRARY_PATH}\""
fi
chmod -R u+w "${PORT_DATA_PATH}/data/prefixes/${PW_PREFIX_NAME}"
echo "#!/usr/bin/env bash
${TMP_ALL_PATH} \"${PW_PLUGINS_PATH}/portable/bin/mksquashfs\" \"${PORT_DATA_PATH}/data/prefixes/${PW_PREFIX_NAME}\" \
\"${PW_PREFIX_TO_BACKUP}/${PW_PREFIX_NAME}.ppack.part\" -comp zstd \
|| rm -f \"${PW_PREFIX_TO_BACKUP}/${PW_PREFIX_NAME}.ppack.part\"
" > "${PORT_WINE_TMP_PATH}/pp_pfx_backup.sh"
chmod u+x "${PORT_WINE_TMP_PATH}/pp_pfx_backup.sh"
${pw_runtime} ${PW_TERM} "${PORT_WINE_TMP_PATH}/pp_pfx_backup.sh"
try_remove_file "${PORT_WINE_TMP_PATH}/pp_pfx_backup.sh"
if [[ -f "${PW_PREFIX_TO_BACKUP}/${PW_PREFIX_NAME}.ppack.part" ]]; then
mv -f "${PW_PREFIX_TO_BACKUP}/${PW_PREFIX_NAME}.ppack.part" \
"${PW_PREFIX_TO_BACKUP}/${PW_PREFIX_NAME}.ppack"
return 0
else
return 1
fi
python3 -m portprotonqt.scripts_utils.prefix_backup --port-data-path "${PORT_DATA_PATH}" \
backup "${PW_PREFIX_NAME}" "${PW_PREFIX_TO_BACKUP}"
return $?
}
pw_unpack_prefix () {
PW_BACKUP_FILE="$1"
pw_is_legacy_prefix_backup () {
[[ "$(head -c 4 "$1" 2>/dev/null)" == "hsqs" ]]
}
pw_unpack_legacy_prefix () {
PW_PREFIX_NAME=$(basename "${PW_BACKUP_FILE}")
PW_PREFIX_NAME="${PW_PREFIX_NAME%.[Pp][Pp][Aa][Cc][Kk]}"
unset PW_SANDBOX_HOME_PATH
pw_init_runtime
if check_flatpak; then
TMP_ALL_PATH=""
else
TMP_ALL_PATH="LD_LIBRARY_PATH=\"${PW_LD_LIBRARY_PATH}\""
fi
if check_selinux; then
NO_XATTRS_NEED="-no-xattrs"
else
NO_XATTRS_NEED=""
fi
PW_PREFIX_NAME=$(basename "${PW_BACKUP_FILE^^}" .PPACK)
try_remove_file "${PORT_WINE_TMP_PATH}/pp_pfx_unpack_error"
echo "#!/usr/bin/env bash
${TMP_ALL_PATH} \"${PW_PLUGINS_PATH}/portable/bin/unsquashfs\" $NO_XATTRS_NEED -f -d \
unsquashfs -no-xattrs -f -d \
\"${PORT_DATA_PATH}/data/prefixes/${PW_PREFIX_NAME}\" \"$PW_BACKUP_FILE\" \
|| echo \"ERROR\" > \"${PORT_WINE_TMP_PATH}/pp_pfx_unpack_error\"
" > "${PORT_WINE_TMP_PATH}/pp_pfx_unpack.sh"
@@ -4738,15 +4698,31 @@ pw_unpack_prefix () {
try_remove_file "${PORT_WINE_TMP_PATH}/pp_pfx_unpack_error"
try_remove_file "${PORT_WINE_TMP_PATH}/pp_pfx_unpack.sh"
return 1
else
try_remove_file "${PORT_WINE_TMP_PATH}/pp_pfx_unpack.sh"
if [[ -f "${PORT_DATA_PATH}/data/prefixes/${PW_PREFIX_NAME}/.create_shortcut" ]] ; then
while read -r line
do
export PW_EXE_FILE="$PORT_DATA_PATH/data/prefixes/$PW_PREFIX_NAME/$line"
portwine_output_yad_shortcut
done < "$PORT_DATA_PATH/data/prefixes/$PW_PREFIX_NAME/.create_shortcut"
fi
return 0
fi
try_remove_file "${PORT_WINE_TMP_PATH}/pp_pfx_unpack_error"
try_remove_file "${PORT_WINE_TMP_PATH}/pp_pfx_unpack.sh"
if [[ -f "${PORT_DATA_PATH}/data/prefixes/${PW_PREFIX_NAME}/.create_shortcut" ]] ; then
while read -r line
do
export PW_EXE_FILE="$PORT_DATA_PATH/data/prefixes/$PW_PREFIX_NAME/$line"
portwine_output_yad_shortcut
done < "$PORT_DATA_PATH/data/prefixes/$PW_PREFIX_NAME/.create_shortcut"
fi
return 0
}
pw_unpack_prefix () {
PW_BACKUP_FILE="$1"
PW_PREFIX_NAME=$(basename "${PW_BACKUP_FILE}")
PW_PREFIX_NAME="${PW_PREFIX_NAME%.[Pp][Pp][Aa][Cc][Kk]}"
if pw_is_legacy_prefix_backup "${PW_BACKUP_FILE}" ; then
pw_unpack_legacy_prefix
return $?
fi
python3 -m portprotonqt.scripts_utils.prefix_backup --port-data-path "${PORT_DATA_PATH}" \
restore "${PW_BACKUP_FILE}"
return $?
}
+44 -13
View File
@@ -3,6 +3,7 @@ import sys
import os
import importlib
import subprocess
from urllib.parse import quote, unquote
from logging import Logger
__app_id__ = "ru.linux_gaming.PortProtonQt"
@@ -203,19 +204,9 @@ def main():
if start_sh is None:
return
if parsed_args.create_backup:
setup_logger(parsed_args.debug_level)
prefix_name, backup_dir = parsed_args.create_backup
backup_exit_code = create_prefix_backup(start_sh, prefix_name, backup_dir)
sys.exit(backup_exit_code)
if parsed_args.restore_prefix and not parsed_args.file_or_url:
setup_logger(parsed_args.debug_level)
sys.exit(1)
if is_restore_prefix_request(parsed_args):
setup_logger(parsed_args.debug_level)
restore_exit_code = restore_prefix_backup(start_sh, parsed_args.file_or_url)
sys.exit(restore_exit_code)
# Handle Steam compatibility mode - launch game directly without GUI.
if is_steam_compat:
@@ -257,6 +248,8 @@ def main():
fullscreen = args.fullscreen or display_config.get_fullscreen()
ipc_message = "show:fullscreen" if fullscreen else "show"
backup_request = None
restore_prefix_path = None
resolution_from_args = None
if args.resolution:
resolution_from_args = parse_resolution(args.resolution)
@@ -266,9 +259,16 @@ def main():
window_config.set_geometry(resolution_from_args[0], resolution_from_args[1])
logger.info("Saved window resolution: %sx%s", resolution_from_args[0], resolution_from_args[1])
ipc_message = "noop"
if is_silent_launch and args.file_or_url and is_exe_file(args.file_or_url):
if args.create_backup:
prefix_name, backup_dir = args.create_backup
backup_request = (prefix_name, os.path.abspath(os.path.expanduser(backup_dir)))
ipc_message = "backup:{}:{}".format(quote(prefix_name, safe=""), quote(backup_request[1], safe=""))
elif is_silent_launch and args.file_or_url and is_exe_file(args.file_or_url):
run_silent_tray(app, start_sh, normalize_launch_path(args.file_or_url))
sys.exit(app.exec())
elif is_restore_prefix_request(args):
restore_prefix_path = normalize_launch_path(args.file_or_url)
ipc_message = f"restore:{quote(restore_prefix_path, safe='')}"
elif args.file_or_url and is_launch_file(args.file_or_url):
ipc_message = f"open:{normalize_launch_path(args.file_or_url)}"
@@ -319,7 +319,7 @@ def main():
logger.warning("Failed to persist PORT_DATA_PATH in PortProtonQt config")
# Check if we have a portproton:// URL or launch file to handle
if args.file_or_url:
if args.file_or_url and not restore_prefix_path:
if is_portproton_url(args.file_or_url):
# Parse the portproton:// URL to get the full download URL
download_url = parse_portproton_url(args.file_or_url)
@@ -366,6 +366,14 @@ def main():
def handle_launch_exe():
window.handle_launch_exe(exe_path)
QTimer.singleShot(0, handle_launch_exe)
elif backup_request:
def handle_create_backup():
window._perform_backup(backup_request[1], backup_request[0])
QTimer.singleShot(0, handle_create_backup)
elif restore_prefix_path:
def handle_restore_prefix():
window._perform_restore(restore_prefix_path)
QTimer.singleShot(0, handle_restore_prefix)
# --- Handle incoming connections ---
def handle_new_connection():
@@ -380,7 +388,12 @@ def main():
def restore_window():
try:
if msg.startswith("show") or msg.startswith("open:"):
if (
msg.startswith("show")
or msg.startswith("open:")
or msg.startswith("restore:")
or msg.startswith("backup:")
):
# Ensure the window is visible and not minimized
window.setWindowState(window.windowState() & ~Qt.WindowState.WindowMinimized)
window.show()
@@ -407,6 +420,22 @@ def main():
window.handle_launch_exe(launch_path)
else:
logger.warning("Invalid launch file via IPC: %s", launch_path)
elif msg.startswith("restore:"):
backup_path = unquote(msg[8:].strip())
if backup_path and is_prefix_backup_file(backup_path):
logger.info("Restoring prefix backup via IPC: %s", backup_path)
window._perform_restore(backup_path)
else:
logger.warning("Invalid prefix backup via IPC: %s", backup_path)
elif msg.startswith("backup:"):
parts = msg.split(":", 2)
if len(parts) == 3:
prefix_name = unquote(parts[1])
backup_dir = unquote(parts[2])
logger.info("Creating prefix backup via IPC: %s", prefix_name)
window._perform_backup(backup_dir, prefix_name)
else:
logger.warning("Invalid prefix backup request via IPC: %s", msg)
except Exception as e:
logger.warning(f"Failed to restore window: {e}")
@@ -424,6 +453,8 @@ def main():
and not args.fullscreen
and window_resolution is None
and exe_path is None
and backup_request is None
and restore_prefix_path is None
)
if launch_minimized:
logger.info("Launching in tray")
+4 -1
View File
@@ -588,7 +588,10 @@ class FileExplorer(DraggableDialog):
item = QListWidgetItem(f)
mime_type = self.mime_db.mimeTypeForFile(file_path)
if (
mime_type.name() in ("application/vnd.squashfs", "application/x-squashfs")
f.lower().endswith(".ppack")
or mime_type.name() == "application/x-portproton-prefix-backup"
or mime_type.inherits("application/x-portproton-prefix-backup")
or mime_type.name() in ("application/vnd.squashfs", "application/x-squashfs")
or mime_type.inherits("application/vnd.squashfs")
):
ppack_icon = theme_manager.get_icon("ppack")
+103
View File
@@ -0,0 +1,103 @@
"""Prefix backup progress dialog."""
from dataclasses import dataclass
from PySide6.QtCore import QThread, Signal, Qt
from PySide6.QtWidgets import QDialog, QLabel, QProgressBar, QVBoxLayout
from portprotonqt.config import ui_config
from portprotonqt.localization import _
from portprotonqt.scripts_utils.prefix_backup import BackupProgress, create_backup, restore_backup
from portprotonqt.theme_manager import ThemeManager
@dataclass(frozen=True)
class PrefixBackupJob:
operation: str
port_data_path: str
source: str
target: str = ""
class PrefixBackupThread(QThread):
"""Worker thread for prefix backup operations."""
progress = Signal(object)
completed = Signal(bool)
def __init__(self, job: PrefixBackupJob):
super().__init__()
self.job = job
self.success = False
def _emit_progress(self, progress: BackupProgress) -> None:
self.progress.emit(progress)
def run(self) -> None:
if self.job.operation == "backup":
exit_code = create_backup(
self.job.port_data_path,
self.job.source,
self.job.target,
self._emit_progress,
)
else:
exit_code = restore_backup(self.job.port_data_path, self.job.source, self._emit_progress)
self.success = exit_code == 0
self.completed.emit(self.success)
class PrefixBackupDialog(QDialog):
"""Dialog showing prefix backup or restore progress."""
def __init__(self, parent, worker: PrefixBackupThread, theme=None):
super().__init__(parent)
self.theme = theme if theme else ThemeManager().apply_theme(ui_config.get_theme())
self.worker = worker
title = _("Prefix backup") if worker.job.operation == "backup" else _("Prefix restore")
self.setWindowTitle(title)
self.setModal(True)
self.setWindowModality(Qt.WindowModality.ApplicationModal)
self.setStyleSheet(self.theme.MESSAGE_BOX_STYLE)
self.resize(740, 230)
layout = QVBoxLayout(self)
self.title_label = QLabel(title)
self.title_label.setStyleSheet(self.theme.PARAMS_TITLE_STYLE)
self.title_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
layout.addWidget(self.title_label)
self.progress_bar = QProgressBar()
self.progress_bar.setStyleSheet(self.theme.PROGRESS_BAR_STYLE)
self.progress_bar.setRange(0, 100)
layout.addWidget(self.progress_bar)
self.file_label = QLabel("")
self.file_label.setStyleSheet(self.theme.CONTENT_STYLE)
self.file_label.setWordWrap(True)
layout.addWidget(self.file_label)
self.stats_label = QLabel("")
self.stats_label.setStyleSheet(self.theme.CONTENT_STYLE)
layout.addWidget(self.stats_label)
self.worker.progress.connect(self.update_progress)
self.worker.completed.connect(self.finish)
def start(self) -> None:
self.worker.start()
self.exec()
def update_progress(self, progress: BackupProgress) -> None:
self.progress_bar.setValue(progress.percent)
self.file_label.setText(_("File: {0}").format(progress.path))
if progress.files_total:
files_text = _("{0}/{1} files").format(progress.files_done, progress.files_total)
else:
files_text = _("{0} files").format(progress.files_done)
self.stats_label.setText(_("{0}, Speed: {1:.1f} MB/s").format(files_text, progress.speed))
def finish(self, success: bool) -> None:
if success:
self.progress_bar.setValue(100)
self.accept()
+22 -11
View File
@@ -57,6 +57,8 @@ from portprotonqt.game_library_manager import GameLibraryManager
from portprotonqt.virtual_keyboard import VirtualKeyboard
from portprotonqt.disc_image_utils import DiscImageManager
from portprotonqt.dialogs.proton_manager import show_proton_manager
from portprotonqt.dialogs.prefix_backup import PrefixBackupDialog, PrefixBackupJob, PrefixBackupThread
from portprotonqt.scripts_utils.prefix_backup import is_legacy_squashfs_backup
from portprotonqt.tabs.control_hints import MainWindowControlHintsMixin
from portprotonqt.tabs.system_tab import MainWindowSystemTabMixin
from portprotonqt.tabs.workers import MainWindowWorkersMixin
@@ -1985,15 +1987,13 @@ class MainWindow(MainWindowControlHintsMixin, MainWindowSystemTabMixin, MainWind
def _perform_backup(self, backup_dir, prefix_name):
os.makedirs(backup_dir, exist_ok=True)
if not self.portproton_location or not self.start_sh:
if not self.portproton_location:
return
start_sh = self.start_sh
self.backup_process = QProcess(self)
self.backup_process.finished.connect(lambda exitCode: self._on_backup_finished(exitCode))
cmd = start_sh + ["--backup-prefix", prefix_name, backup_dir]
self.backup_process.start(cmd[0], cmd[1:])
if not self.backup_process.waitForStarted():
QMessageBox.warning(self, _("Error"), _("Failed to start backup process."))
job = PrefixBackupJob("backup", self.portproton_location, prefix_name, backup_dir)
worker = PrefixBackupThread(job)
dialog = PrefixBackupDialog(self, worker, self.theme)
dialog.start()
self._on_backup_finished(0 if worker.success else 1)
def load_prefix_backup(self):
file_explorer = FileExplorer(self, file_filter='.ppack')
@@ -2003,12 +2003,23 @@ class MainWindow(MainWindowControlHintsMixin, MainWindowSystemTabMixin, MainWind
def _perform_restore(self, file_path):
if not file_path or not os.path.exists(file_path):
return
if not self.portproton_location or not self.start_sh:
if not self.portproton_location:
return
if is_legacy_squashfs_backup(file_path):
self._perform_legacy_restore(file_path)
return
job = PrefixBackupJob("restore", self.portproton_location, file_path)
worker = PrefixBackupThread(job)
dialog = PrefixBackupDialog(self, worker, self.theme)
dialog.start()
self._on_restore_finished(0 if worker.success else 1)
def _perform_legacy_restore(self, file_path: str) -> None:
if not self.start_sh:
return
start_sh = self.start_sh
self.restore_process = QProcess(self)
self.restore_process.finished.connect(lambda exitCode: self._on_restore_finished(exitCode))
cmd = start_sh + ["--restore-prefix", file_path]
cmd = self.start_sh + ["--restore-prefix", file_path]
self.restore_process.start(cmd[0], cmd[1:])
if not self.restore_process.waitForStarted():
QMessageBox.warning(self, _("Error"), _("Failed to start restore process."))
+2
View File
@@ -65,6 +65,7 @@ install_data(
'dialogs/settings_mangohud.py',
'dialogs/settings_gamescope.py',
'dialogs/proton_manager.py',
'dialogs/prefix_backup.py',
'dialogs/wine_loader.py',
'dialogs/wine_downloader.py',
'dialogs/wine_extractor.py',
@@ -122,6 +123,7 @@ install_data(
['scripts_utils/__init__.py',
'scripts_utils/easyterm.py',
'scripts_utils/idle_inhibit.py',
'scripts_utils/prefix_backup.py',
'scripts_utils/power_profiles.py',
'scripts_utils/shortcut_tools.py',
],
+366
View File
@@ -0,0 +1,366 @@
"""Create and restore PortProton prefix backups."""
import argparse
import os
import shutil
import stat
import sys
import time
from collections.abc import Callable
from dataclasses import dataclass
import libarchive
from libarchive.entry import FileType
from libarchive.flags import READDISK_NO_XATTR
from portprotonqt.logger import get_logger
logger = get_logger(__name__)
BACKUP_EXTENSION = ".ppack"
ARCHIVE_FORMAT = "pax"
ARCHIVE_FILTER = "zstd"
SQUASHFS_MAGIC = b"hsqs"
@dataclass(frozen=True)
class BackupProgress:
percent: int
path: str
files_done: int
files_total: int
speed: float
ProgressCallback = Callable[[BackupProgress], None]
def _prefix_path(port_data_path: str, prefix_name: str) -> str:
safe_name = os.path.basename(prefix_name)
return os.path.join(port_data_path, "data", "prefixes", safe_name)
def _backup_path(backup_dir: str, prefix_name: str) -> tuple[str, str]:
safe_name = os.path.basename(prefix_name)
final_path = os.path.join(backup_dir, f"{safe_name}{BACKUP_EXTENSION}")
return final_path, f"{final_path}.part"
def is_legacy_squashfs_backup(backup_file: str) -> bool:
try:
with open(backup_file, "rb") as file:
return file.read(4) == SQUASHFS_MAGIC
except OSError as e:
logger.warning("Failed to read backup header %s: %s", backup_file, e)
return False
def _make_user_writable(path: str) -> None:
for root, dirs, files in os.walk(path):
for name in dirs + files:
item_path = os.path.join(root, name)
try:
item_stat = os.lstat(item_path)
if stat.S_ISLNK(item_stat.st_mode):
continue
os.chmod(item_path, item_stat.st_mode | stat.S_IWUSR)
except OSError as e:
logger.warning("Failed to update permissions for %s: %s", item_path, e)
def _collect_entries(prefix_dir: str) -> list[tuple[str, str, int]]:
entries = []
for root, dirs, files in os.walk(prefix_dir):
names = sorted(dirs) + sorted(files)
for name in names:
source_path = os.path.join(root, name)
archive_path = os.path.relpath(source_path, prefix_dir)
try:
item_stat = os.lstat(source_path)
size = item_stat.st_size if stat.S_ISREG(item_stat.st_mode) else 0
except OSError:
size = 0
entries.append((source_path, archive_path, size))
return entries
def _archive_symlink(archive, source_path: str, archive_path: str) -> bool:
try:
item_stat = os.lstat(source_path)
link_path = os.readlink(source_path)
except OSError as e:
logger.warning("Skipped symlink %s: %s", source_path, e)
return False
if not os.path.exists(source_path):
logger.warning("Skipped broken symlink %s -> %s", source_path, link_path)
return False
archive.add_file_from_memory(
archive_path,
0,
b"",
filetype=int(FileType.SYMBOLINK_LINK),
permission=stat.S_IMODE(item_stat.st_mode),
linkpath=link_path,
mtime=item_stat.st_mtime,
)
return True
def _archive_entry(archive, source_path: str, archive_path: str) -> bool:
try:
item_stat = os.lstat(source_path)
except OSError as e:
logger.warning("Skipped archive entry %s: %s", source_path, e)
return False
if stat.S_ISLNK(item_stat.st_mode):
return _archive_symlink(archive, source_path, archive_path)
archive.add_files(
source_path,
flags=READDISK_NO_XATTR,
pathname=archive_path,
recursive=False,
symlink_mode="physical",
)
return True
def _write_shortcut_manifest(port_data_path: str, prefix_name: str) -> None:
shortcuts = []
marker = f"/{prefix_name}/"
for name in os.listdir(port_data_path):
if not name.endswith(".desktop"):
continue
desktop_path = os.path.join(port_data_path, name)
try:
with open(desktop_path, encoding="utf-8", errors="ignore") as file:
for line in file:
if marker in line:
shortcuts.append(line.split(marker, 1)[1].split('"', 1)[0])
except OSError as e:
logger.warning("Failed to read desktop file %s: %s", desktop_path, e)
if not shortcuts:
return
manifest_path = os.path.join(_prefix_path(port_data_path, prefix_name), ".create_shortcut")
try:
os.remove(manifest_path)
except FileNotFoundError:
pass
with open(manifest_path, "w", encoding="utf-8") as file:
file.write("\n".join(shortcuts))
file.write("\n")
def _emit_progress(
callback: ProgressCallback | None,
progress: BackupProgress,
) -> None:
if callback:
callback(progress)
def create_backup(
port_data_path: str,
prefix_name: str,
backup_dir: str,
progress_callback: ProgressCallback | None = None,
) -> int:
prefix_dir = _prefix_path(port_data_path, prefix_name)
final_path, part_path = _backup_path(backup_dir, prefix_name)
if not os.path.isdir(prefix_dir):
logger.error("Prefix directory not found: %s", prefix_dir)
return 1
os.makedirs(backup_dir, exist_ok=True)
_write_shortcut_manifest(port_data_path, prefix_name)
_make_user_writable(prefix_dir)
try:
entries = _collect_entries(prefix_dir)
total_files = len(entries)
total_size = sum(entry[2] for entry in entries) or 1
processed_size = 0
start_time = time.monotonic()
with libarchive.file_writer(part_path, ARCHIVE_FORMAT, ARCHIVE_FILTER) as archive:
for index, (source_path, archive_path, size) in enumerate(entries, 1):
_archive_entry(archive, source_path, archive_path)
processed_size += size
elapsed = max(time.monotonic() - start_time, 0.001)
speed = processed_size / (1024 * 1024) / elapsed
percent = min(int((processed_size / total_size) * 100), 99)
progress = BackupProgress(percent, archive_path, index, total_files, speed)
_emit_progress(progress_callback, progress)
os.replace(part_path, final_path)
progress = BackupProgress(100, final_path, total_files, total_files, 0.0)
_emit_progress(progress_callback, progress)
except (libarchive.ArchiveError, OSError) as e:
logger.error("Failed to create prefix backup %s: %s", prefix_name, e)
try:
os.remove(part_path)
except FileNotFoundError:
pass
return 1
return 0
def _safe_entry_path(target_dir: str, entry_path: str) -> str:
if os.path.isabs(entry_path) or ".." in entry_path.split("/"):
raise ValueError(f"Unsafe archive path: {entry_path}")
destination = os.path.abspath(os.path.normpath(os.path.join(target_dir, entry_path)))
target_root = os.path.abspath(os.path.normpath(target_dir))
if destination != target_root and not destination.startswith(f"{target_root}/"):
raise ValueError(f"Unsafe archive path: {entry_path}")
return destination
def _apply_file_metadata(path: str, entry: libarchive.entry.ArchiveEntry) -> None:
if entry.mode and not os.path.islink(path):
try:
os.chmod(path, entry.mode)
except OSError as e:
logger.warning("Failed to set permissions for %s: %s", path, e)
if entry.mtime:
try:
os.utime(path, (entry.mtime, entry.mtime), follow_symlinks=False)
except OSError as e:
logger.warning("Failed to set mtime for %s: %s", path, e)
def _extract_entry(target_dir: str, entry: libarchive.entry.ArchiveEntry) -> None:
destination = _safe_entry_path(target_dir, entry.pathname)
parent_dir = os.path.dirname(destination)
if parent_dir:
os.makedirs(parent_dir, exist_ok=True)
if entry.isdir:
if os.path.islink(destination):
os.remove(destination)
os.makedirs(destination, exist_ok=True)
elif entry.islnk:
if os.path.lexists(destination):
os.remove(destination)
os.symlink(entry.linkpath, destination)
elif entry.isfile:
if os.path.islink(destination):
os.remove(destination)
with open(destination, "wb") as file:
for block in entry.get_blocks():
file.write(block)
else:
logger.warning("Skipped unsupported archive entry: %s", entry.pathname)
return
_apply_file_metadata(destination, entry)
def _read_xdg_desktop_dir() -> str:
config_path = os.path.join(os.path.expanduser("~"), ".config", "user-dirs.dirs")
try:
with open(config_path, encoding="utf-8") as file:
for line in file:
if line.startswith("XDG_DESKTOP_DIR="):
value = line.split("=", 1)[1].strip().strip('"')
return value.replace("$HOME", os.path.expanduser("~"))
except OSError:
pass
return ""
def _desktop_target_dir() -> str:
for path in (
os.path.join(os.path.expanduser("~"), "Desktop"),
os.path.join(os.path.expanduser("~"), "Рабочий стол"),
_read_xdg_desktop_dir(),
):
if path and os.path.isdir(path):
return path
return ""
def _copy_shortcut(desktop_path: str, target_dir: str) -> None:
if not target_dir:
return
os.makedirs(target_dir, exist_ok=True)
target_path = os.path.join(target_dir, os.path.basename(desktop_path))
shutil.copyfile(desktop_path, target_path)
os.chmod(target_path, 0o755)
def _create_restore_shortcuts(port_data_path: str, prefix_name: str) -> None:
from portprotonqt.scripts_utils.shortcut_tools import create_shortcut
manifest_path = os.path.join(_prefix_path(port_data_path, prefix_name), ".create_shortcut")
if not os.path.isfile(manifest_path):
return
menu_dir = os.path.join(os.path.expanduser("~"), ".local", "share", "applications")
desktop_dir = _desktop_target_dir()
with open(manifest_path, encoding="utf-8") as file:
for raw_line in file:
rel_path = raw_line.strip()
exe_path = os.path.join(_prefix_path(port_data_path, prefix_name), rel_path)
game_name = os.path.splitext(os.path.basename(exe_path))[0]
if not rel_path or not os.path.exists(exe_path):
continue
if not create_shortcut(exe_path, game_name):
continue
try:
desktop_path = os.path.join(port_data_path, f"{game_name}.desktop")
_copy_shortcut(desktop_path, menu_dir)
_copy_shortcut(desktop_path, desktop_dir)
except OSError as e:
logger.warning("Failed to create restored shortcut for %s: %s", exe_path, e)
def restore_backup(
port_data_path: str,
backup_file: str,
progress_callback: ProgressCallback | None = None,
) -> int:
prefix_name = os.path.basename(backup_file)[:-len(BACKUP_EXTENSION)]
target_dir = _prefix_path(port_data_path, prefix_name)
os.makedirs(target_dir, exist_ok=True)
try:
archive_size = os.path.getsize(backup_file) or 1
start_time = time.monotonic()
files_done = 0
with libarchive.file_reader(backup_file) as archive:
for entry in archive:
_extract_entry(target_dir, entry)
files_done += 1
bytes_read = archive.bytes_read
elapsed = max(time.monotonic() - start_time, 0.001)
speed = bytes_read / (1024 * 1024) / elapsed
percent = min(int((bytes_read / archive_size) * 100), 99)
progress = BackupProgress(percent, entry.pathname, files_done, 0, speed)
_emit_progress(progress_callback, progress)
_create_restore_shortcuts(port_data_path, prefix_name)
progress = BackupProgress(100, target_dir, files_done, files_done, 0.0)
_emit_progress(progress_callback, progress)
except (libarchive.ArchiveError, OSError, ValueError) as e:
logger.error("Failed to restore prefix backup %s: %s", backup_file, e)
return 1
return 0
def main() -> int:
parser = argparse.ArgumentParser(description="PortProton prefix backup helper")
parser.add_argument("--port-data-path", required=True)
subparsers = parser.add_subparsers(dest="command", required=True)
backup_parser = subparsers.add_parser("backup")
backup_parser.add_argument("prefix_name")
backup_parser.add_argument("backup_dir")
restore_parser = subparsers.add_parser("restore")
restore_parser.add_argument("backup_file")
args = parser.parse_args()
if args.command == "backup":
return create_backup(args.port_data_path, args.prefix_name, args.backup_dir)
return restore_backup(args.port_data_path, args.backup_file)
if __name__ == "__main__":
sys.exit(main())