From 817e8f5cd26739c69d930d21cd9dc4c0b6e4984e Mon Sep 17 00:00:00 2001 From: SulfurNitride Date: Sat, 14 Feb 2026 02:45:12 -0600 Subject: Add FUSE external mapping support, BG3/Oblivion Remastered fixes, fomod-plus and NaK integration FUSE VFS now deploys non-data-dir mod mappings (Paks, OBSE, UE4SS, etc.) via real symlinks and injects file-level data-dir mappings (plugins.txt, loadorder.txt) into the VFS tree. Fixes game launches for Oblivion Remastered (Root Builder path resolution, script extender support) and BG3 (Wine prefix documents directory, file mapper symlinks on Linux). Vendors mo2-fomod-plus plugin and NaK crate for FOMOD installer and game finder/runtime support. Co-Authored-By: Claude Opus 4.6 --- src/plugins/rootbuilder.py | 378 +++++++++++++++++++++++++++++++++------------ 1 file changed, 278 insertions(+), 100 deletions(-) (limited to 'src/plugins') diff --git a/src/plugins/rootbuilder.py b/src/plugins/rootbuilder.py index 7237400..fbe082c 100644 --- a/src/plugins/rootbuilder.py +++ b/src/plugins/rootbuilder.py @@ -15,6 +15,7 @@ import shutil import subprocess import mobase +from PyQt6.QtCore import qInfo, qWarning from PyQt6.QtGui import QIcon from PyQt6.QtWidgets import ( QCheckBox, @@ -26,8 +27,16 @@ from PyQt6.QtWidgets import ( QVBoxLayout, ) -MANIFEST_NAME = ".rootbuilder_manifest.json" -BACKUP_DIR_NAME = ".rootbuilder_backup" +# Storage lives under /rootbuilder/ — NOT the game directory, +# which Steam/Wine may make read-only during gameplay. +_STORAGE_SUBDIR = "rootbuilder" +_MANIFEST_NAME = "manifest.json" +_SETTINGS_NAME = "settings.json" +_BACKUP_SUBDIR = "backup" + +# Legacy names (stored in game dir by older versions) +_LEGACY_MANIFEST = ".rootbuilder_manifest.json" +_LEGACY_BACKUP = ".rootbuilder_backup" def _find_root_dir(mod_path: str) -> str | None: @@ -48,51 +57,116 @@ def _walk_files(root_dir: str): yield os.path.join(dirpath, name) +def _host_cp(src: str, dst: str) -> bool: + """Copy a file via the host OS (bypasses Flatpak sandbox restrictions).""" + try: + result = subprocess.run( + ["flatpak-spawn", "--host", "cp", "-f", "--reflink=auto", "--", src, dst], + capture_output=True, timeout=30, + ) + return result.returncode == 0 + except (subprocess.SubprocessError, OSError): + return False + + def _reflink_copy(src: str, dst: str): """Copy with reflink (CoW) if supported, fallback to regular copy.""" try: subprocess.run( - ["cp", "--reflink=auto", "--", src, dst], + ["cp", "--reflink=auto", "-f", "--", src, dst], check=True, capture_output=True, ) + return except (subprocess.CalledProcessError, FileNotFoundError): + pass + try: shutil.copy2(src, dst) + return + except OSError: + pass + if _IN_FLATPAK and _host_cp(src, dst): + return + raise OSError(f"Root Builder: failed to copy {src} -> {dst}") + +def _ensure_writable(path: str): + """Make a file or directory writable by the owner.""" + try: + st = os.stat(path) + os.chmod(path, st.st_mode | 0o200) + except OSError: + pass -def _manifest_path(game_dir: str) -> str: - return os.path.join(game_dir, MANIFEST_NAME) +_IN_FLATPAK = os.path.exists("/.flatpak-info") -def _load_manifest(game_dir: str) -> dict | None: - path = _manifest_path(game_dir) - if not os.path.isfile(path): - return None + +def _host_rm(path: str) -> bool: + """Remove a file via the host OS (bypasses Flatpak sandbox restrictions).""" try: - with open(path, "r") as f: - return json.load(f) - except (OSError, json.JSONDecodeError): - return None + result = subprocess.run( + ["flatpak-spawn", "--host", "rm", "-f", "--", path], + capture_output=True, timeout=10, + ) + return result.returncode == 0 + except (subprocess.SubprocessError, OSError): + return False -def _save_manifest(game_dir: str, manifest: dict): - path = _manifest_path(game_dir) - with open(path, "w") as f: - json.dump(manifest, f, indent=2) +def _host_mv(src: str, dst: str) -> bool: + """Move a file via the host OS (bypasses Flatpak sandbox restrictions).""" + try: + result = subprocess.run( + ["flatpak-spawn", "--host", "mv", "-f", "--", src, dst], + capture_output=True, timeout=10, + ) + return result.returncode == 0 + except (subprocess.SubprocessError, OSError): + return False -def _remove_manifest(game_dir: str): - path = _manifest_path(game_dir) - if os.path.isfile(path): +def _force_remove(path: str) -> bool: + """Remove a file, fixing permissions if needed. Returns True on success.""" + try: + os.remove(path) + return True + except PermissionError: + pass + try: + _ensure_writable(os.path.dirname(path)) + _ensure_writable(path) os.remove(path) + return True + except OSError: + pass + if _IN_FLATPAK and _host_rm(path): + return True + qWarning(f"Root Builder: could not remove {path}") + return False + + +def _rmtree_onerror(_func, path, _exc_info): + """onerror handler for shutil.rmtree that fixes permissions and retries.""" + try: + _ensure_writable(os.path.dirname(path)) + _ensure_writable(path) + os.remove(path) + except OSError: + pass -def _cleanup_empty_dirs(game_dir: str, deployed: list[str]): +def _cleanup_empty_dirs(base_dir: str, paths: list[str]): """Remove empty directories left behind after clearing deployed files.""" dirs_to_check = set() - for path in deployed: + for path in paths: parent = os.path.dirname(path) - while parent and parent != game_dir and not os.path.samefile(parent, game_dir): + while parent and parent != base_dir: + try: + if os.path.samefile(parent, base_dir): + break + except OSError: + break dirs_to_check.add(parent) parent = os.path.dirname(parent) @@ -104,15 +178,44 @@ def _cleanup_empty_dirs(game_dir: str, deployed: list[str]): pass +# --- Manifest helpers (operate on storage_dir, NOT game dir) --- + +def _load_manifest(storage_dir: str) -> dict | None: + path = os.path.join(storage_dir, _MANIFEST_NAME) + if not os.path.isfile(path): + return None + try: + with open(path, "r") as f: + return json.load(f) + except (OSError, json.JSONDecodeError): + return None + + +def _save_manifest(storage_dir: str, manifest: dict): + os.makedirs(storage_dir, exist_ok=True) + path = os.path.join(storage_dir, _MANIFEST_NAME) + with open(path, "w") as f: + json.dump(manifest, f, indent=2) + + +def _remove_manifest(storage_dir: str): + path = os.path.join(storage_dir, _MANIFEST_NAME) + if os.path.isfile(path): + try: + os.remove(path) + except OSError: + pass + + class RootBuilderDialog(QDialog): """Small settings/control dialog shown from the Tools menu.""" - def __init__(self, organizer: mobase.IOrganizer, build_fn, clear_fn, parent=None): + def __init__(self, settings: dict, save_fn, build_fn, clear_fn, parent=None): super().__init__(parent) - self._organizer = organizer + self._settings = settings + self._save_fn = save_fn self._build_fn = build_fn self._clear_fn = clear_fn - self._plugin_name = "Root Builder" self.setWindowTitle("Root Builder") self.resize(350, 220) @@ -125,9 +228,7 @@ class RootBuilderDialog(QDialog): # Enable checkbox self._enableCheck = QCheckBox("Auto-deploy on game launch") - self._enableCheck.setChecked( - bool(organizer.pluginSetting(self._plugin_name, "enabled")) - ) + self._enableCheck.setChecked(settings.get("enabled", False) is True) layout.addWidget(self._enableCheck) # Mode selector @@ -135,8 +236,7 @@ class RootBuilderDialog(QDialog): mode_layout.addWidget(QLabel("Deploy mode:")) self._modeCombo = QComboBox() self._modeCombo.addItems(["copy", "link"]) - current = organizer.pluginSetting(self._plugin_name, "mode") - self._modeCombo.setCurrentText(current if current else "copy") + self._modeCombo.setCurrentText(settings.get("mode", "copy")) mode_layout.addWidget(self._modeCombo) layout.addLayout(mode_layout) @@ -159,28 +259,23 @@ class RootBuilderDialog(QDialog): close_btn.clicked.connect(self.accept) layout.addWidget(close_btn) - def _save_settings(self): - self._organizer.setPluginSetting( - self._plugin_name, "enabled", self._enableCheck.isChecked() - ) - self._organizer.setPluginSetting( - self._plugin_name, "mode", self._modeCombo.currentText() - ) + # Save settings whenever the user changes them + self._enableCheck.stateChanged.connect(lambda _: self._do_save()) + self._modeCombo.currentTextChanged.connect(lambda _: self._do_save()) + + def _do_save(self): + self._settings["enabled"] = self._enableCheck.isChecked() + self._settings["mode"] = self._modeCombo.currentText() + self._save_fn(self._settings) def _on_build(self): - self._save_settings() count = self._build_fn() self._status.setText(f"Deployed {count} file(s).") def _on_clear(self): - self._save_settings() count = self._clear_fn() self._status.setText(f"Cleared {count} file(s).") - def accept(self): - self._save_settings() - super().accept() - class RootBuilder(mobase.IPluginTool): _organizer: mobase.IOrganizer @@ -193,24 +288,77 @@ class RootBuilder(mobase.IPluginTool): def init(self, organizer: mobase.IOrganizer) -> bool: self._organizer = organizer + self._migrate_legacy() self._check_third_party_rootbuilder() organizer.onAboutToRun(self._on_about_to_run) organizer.onFinishedRun(self._on_finished_run) return True + # --- Storage paths (instance dir, always writable) --- + + def _storage_dir(self) -> str: + d = os.path.join(self._organizer.basePath(), _STORAGE_SUBDIR) + os.makedirs(d, exist_ok=True) + return d + + # --- Settings (our own JSON, not pluginSetting) --- + + def _load_settings(self) -> dict: + path = os.path.join(self._storage_dir(), _SETTINGS_NAME) + try: + with open(path, "r") as f: + return json.load(f) + except (OSError, json.JSONDecodeError, ValueError): + return {"enabled": False, "mode": "copy"} + + def _save_settings(self, settings: dict): + path = os.path.join(self._storage_dir(), _SETTINGS_NAME) + with open(path, "w") as f: + json.dump(settings, f, indent=2) + + def _is_enabled(self) -> bool: + return self._load_settings().get("enabled", False) is True + + # --- Legacy migration --- + + def _migrate_legacy(self): + """Move legacy manifest/backup from game dir to our storage dir.""" + game = self._organizer.managedGame() + if game is None: + return + game_dir = game.gameDirectory().absolutePath() + storage = self._storage_dir() + + old_manifest = os.path.join(game_dir, _LEGACY_MANIFEST) + if os.path.isfile(old_manifest): + try: + new_path = os.path.join(storage, _MANIFEST_NAME) + if not os.path.isfile(new_path): + shutil.copy2(old_manifest, new_path) + _force_remove(old_manifest) + except OSError: + pass + + old_backup = os.path.join(game_dir, _LEGACY_BACKUP) + if os.path.isdir(old_backup): + try: + new_backup = os.path.join(storage, _BACKUP_SUBDIR) + if not os.path.isdir(new_backup): + shutil.copytree(old_backup, new_backup) + shutil.rmtree(old_backup, onerror=_rmtree_onerror) + except OSError: + pass + def _check_third_party_rootbuilder(self): """Move any third-party Root Builder plugins into DisabledPlugins/.""" plugins_dir = os.path.dirname(os.path.abspath(__file__)) disabled_dir = os.path.join(os.path.dirname(plugins_dir), "DisabledPlugins") my_file = os.path.basename(__file__) - # Collect conflicts first, then move (don't modify dir during iteration) conflicts = [] for entry in os.scandir(plugins_dir): - # Kezyma's standard install: plugins/rootbuilder/ if entry.is_dir() and entry.name.lower() == "rootbuilder": conflicts.append((entry.name, entry.path)) - # Other rootbuilder*.py files that aren't us elif ( entry.is_file() and entry.name.lower().startswith("rootbuilder") @@ -224,17 +372,15 @@ class RootBuilder(mobase.IPluginTool): try: os.makedirs(disabled_dir, exist_ok=True) shutil.move(path, dst) - mobase.log( - mobase.LogLevel.INFO, + qInfo( f"Root Builder: moved incompatible third-party plugin " f"'{name}' to DisabledPlugins/. " - f"It uses Windows-only USVFS and cannot work on Linux.", + f"It uses Windows-only USVFS and cannot work on Linux." ) except OSError as e: - mobase.log( - mobase.LogLevel.WARNING, + qWarning( f"Root Builder: failed to move third-party plugin " - f"'{name}' to DisabledPlugins/: {e}", + f"'{name}' to DisabledPlugins/: {e}" ) def name(self) -> str: @@ -256,15 +402,10 @@ class RootBuilder(mobase.IPluginTool): return mobase.VersionInfo(1, 0, 0) def enabledByDefault(self) -> bool: - return False + return True def settings(self) -> list[mobase.PluginSetting]: - return [ - mobase.PluginSetting("mode", "Deploy mode: copy or link", "copy"), - mobase.PluginSetting( - "enabled", "Auto-deploy root files on launch", True - ), - ] + return [] # --- IPluginTool --- @@ -281,20 +422,22 @@ class RootBuilder(mobase.IPluginTool): self.__parentWidget = widget def display(self): + settings = self._load_settings() dialog = RootBuilderDialog( - self._organizer, self._build, self._clear, self.__parentWidget + settings, self._save_settings, self._build, self._clear, + self.__parentWidget, ) dialog.exec() # --- Hooks --- def _on_about_to_run(self, executable: str) -> bool: - if self._organizer.pluginSetting(self.name(), "enabled"): + if self._is_enabled(): self._build() return True def _on_finished_run(self, executable: str, exit_code: int): - if self._organizer.pluginSetting(self.name(), "enabled"): + if self._is_enabled(): self._clear() # --- Build / Clear --- @@ -302,16 +445,17 @@ class RootBuilder(mobase.IPluginTool): def _build(self) -> int: """Deploy root files from all active mods. Returns number of files deployed.""" game_dir = self._organizer.managedGame().gameDirectory().absolutePath() + storage = self._storage_dir() mod_list = self._organizer.modList() mods = mod_list.allModsByProfilePriority() - mode = self._organizer.pluginSetting(self.name(), "mode") or "copy" + mode = self._load_settings().get("mode", "copy") # Clear any previous deployment first - if _load_manifest(game_dir) is not None: + if _load_manifest(storage) is not None: self._clear() manifest = {"deployed": [], "backups": {}} - backup_dir = os.path.join(game_dir, BACKUP_DIR_NAME) + backup_dir = os.path.join(storage, _BACKUP_SUBDIR) deployed_set = set() for mod_name in mods: @@ -333,62 +477,96 @@ class RootBuilder(mobase.IPluginTool): rel = os.path.relpath(src_file, root_dir) dst = os.path.join(game_dir, rel) - # Backup existing file if not already deployed by us - if os.path.exists(dst) and dst not in deployed_set: - bak = os.path.join(backup_dir, rel) - os.makedirs(os.path.dirname(bak), exist_ok=True) - shutil.move(dst, bak) - manifest["backups"][dst] = bak - - os.makedirs(os.path.dirname(dst), exist_ok=True) - - if os.path.lexists(dst): - os.remove(dst) - - # In link mode, .exe and .dll must be copied — Wine/Proton - # resolves a symlinked exe's path to the target, so the - # process can't find sibling files in the game directory. - ext = os.path.splitext(src_file)[1].lower() - if mode == "link" and ext not in (".exe", ".dll"): - os.symlink(src_file, dst) - else: - _reflink_copy(src_file, dst) - - if dst not in deployed_set: - manifest["deployed"].append(dst) - deployed_set.add(dst) - - _save_manifest(game_dir, manifest) + try: + # Backup existing file if not already deployed by us + if os.path.exists(dst) and dst not in deployed_set: + bak = os.path.join(backup_dir, rel) + os.makedirs(os.path.dirname(bak), exist_ok=True) + try: + shutil.copy2(dst, bak) + except PermissionError: + _ensure_writable(dst) + shutil.copy2(dst, bak) + manifest["backups"][dst] = bak + + os.makedirs(os.path.dirname(dst), exist_ok=True) + + if os.path.lexists(dst): + _force_remove(dst) + + # In link mode, .exe and .dll must be copied — Wine/Proton + # resolves a symlinked exe's path to the target, so the + # process can't find sibling files in the game directory. + ext = os.path.splitext(src_file)[1].lower() + if mode == "link" and ext not in (".exe", ".dll"): + os.symlink(src_file, dst) + else: + _reflink_copy(src_file, dst) + + if dst not in deployed_set: + manifest["deployed"].append(dst) + deployed_set.add(dst) + except OSError as e: + qWarning(f"Root Builder: could not deploy {rel}: {e}") + + _save_manifest(storage, manifest) return len(manifest["deployed"]) def _clear(self) -> int: """Remove deployed files and restore backups. Returns count of removed files.""" game_dir = self._organizer.managedGame().gameDirectory().absolutePath() - manifest = _load_manifest(game_dir) + storage = self._storage_dir() + manifest = _load_manifest(storage) if manifest is None: return 0 count = 0 + failed = [] # Remove deployed files for path in manifest["deployed"]: if os.path.lexists(path): - os.remove(path) - count += 1 + if _force_remove(path): + count += 1 + else: + failed.append(path) # Restore backups for dst, bak in manifest["backups"].items(): if os.path.exists(bak): - os.makedirs(os.path.dirname(dst), exist_ok=True) - shutil.move(bak, dst) + try: + parent = os.path.dirname(dst) + os.makedirs(parent, exist_ok=True) + _ensure_writable(parent) + if os.path.lexists(dst): + _force_remove(dst) + shutil.move(bak, dst) + except OSError: + if not (_IN_FLATPAK and _host_mv(bak, dst)): + qWarning( + f"Root Builder: could not restore backup " + f"{bak} -> {dst}" + ) # Clean up backup dir - backup_dir = os.path.join(game_dir, BACKUP_DIR_NAME) + backup_dir = os.path.join(storage, _BACKUP_SUBDIR) if os.path.isdir(backup_dir): shutil.rmtree(backup_dir, ignore_errors=True) - _remove_manifest(game_dir) - _cleanup_empty_dirs(game_dir, manifest["deployed"]) + if failed: + # Update manifest to only contain files we couldn't remove, + # so the next clear attempt can retry them. + manifest["deployed"] = failed + manifest["backups"] = {} + _save_manifest(storage, manifest) + qWarning( + f"Root Builder: {len(failed)} file(s) could not be removed. " + f"They will be retried on next clear." + ) + else: + _remove_manifest(storage) + + _cleanup_empty_dirs(game_dir, [p for p in manifest["deployed"] if p not in failed]) return count -- cgit v1.3.1