mirror of
https://github.com/Omni-guides/Jackify.git
synced 2026-06-07 22:57:45 +02:00
Sync from development - prepare for v0.2.2
This commit is contained in:
@@ -81,7 +81,7 @@ class LoggingHandler:
|
||||
if log_file or is_general:
|
||||
file_path = self.log_dir / (log_file if log_file else "jackify-cli.log")
|
||||
file_handler = logging.handlers.RotatingFileHandler(
|
||||
file_path, mode='a', encoding='utf-8', maxBytes=1024*1024, backupCount=5
|
||||
file_path, mode='a', encoding='utf-8', maxBytes=100*1024*1024, backupCount=5
|
||||
)
|
||||
file_handler.setLevel(logging.DEBUG)
|
||||
file_handler.setFormatter(file_formatter)
|
||||
@@ -90,7 +90,7 @@ class LoggingHandler:
|
||||
|
||||
return logger
|
||||
|
||||
def rotate_logs(self, max_bytes: int = 1024 * 1024, backup_count: int = 5) -> None:
|
||||
def rotate_logs(self, max_bytes: int = 100 * 1024 * 1024, backup_count: int = 5) -> None:
|
||||
"""Rotate log files based on size."""
|
||||
for log_file in self.get_log_files():
|
||||
try:
|
||||
|
||||
@@ -666,7 +666,36 @@ class ModlistMenuHandler:
|
||||
except Exception as e:
|
||||
self.logger.warning(f"ENB configuration skipped due to error: {e}")
|
||||
# Continue workflow - ENB config is optional
|
||||
|
||||
|
||||
# Run modlist-specific post-install automation (e.g., VNV) before showing completion
|
||||
# Only in CLI mode - GUI handles this in install_modlist.py
|
||||
if not gui_mode:
|
||||
from jackify.backend.services.vnv_integration_helper import run_vnv_automation_if_applicable
|
||||
from jackify.backend.services.automated_prefix_service import AutomatedPrefixService
|
||||
from pathlib import Path
|
||||
|
||||
modlist_name = context.get('name', '')
|
||||
modlist_path = Path(context.get('path', ''))
|
||||
|
||||
try:
|
||||
print("")
|
||||
print("Running VNV post-install automation...")
|
||||
automation_ran, error = run_vnv_automation_if_applicable(
|
||||
modlist_name=modlist_name,
|
||||
modlist_install_location=modlist_path,
|
||||
game_root=None, # Will be auto-detected
|
||||
ttw_installer_path=AutomatedPrefixService.get_ttw_installer_path(),
|
||||
progress_callback=lambda msg: print(msg),
|
||||
manual_file_callback=None, # CLI doesn't support manual file callback yet
|
||||
confirmation_callback=None # Will use default confirmation in CLI
|
||||
)
|
||||
if error:
|
||||
print(f"{COLOR_WARNING}VNV automation encountered an error: {error}{COLOR_RESET}")
|
||||
print("You can complete these steps manually by following: https://vivanewvegas.moddinglinked.com/wabbajack.html")
|
||||
except Exception as e:
|
||||
self.logger.debug(f"VNV automation check skipped: {e}")
|
||||
# Not an error - just means VNV automation wasn't applicable
|
||||
|
||||
print("")
|
||||
print("")
|
||||
print("") # Extra blank line before completion
|
||||
|
||||
@@ -353,7 +353,7 @@ class WinetricksHandler:
|
||||
for cmd in commands:
|
||||
bundled_tool = self._get_bundled_tool(cmd, fallback_to_system=False)
|
||||
if bundled_tool:
|
||||
dep_msg = f" ✓ {dep_name}: {bundled_tool} (bundled)"
|
||||
dep_msg = f" {dep_name}: {bundled_tool} (bundled)"
|
||||
self.logger.info(dep_msg)
|
||||
if status_callback:
|
||||
status_callback(dep_msg)
|
||||
@@ -367,7 +367,7 @@ class WinetricksHandler:
|
||||
result = subprocess.run(['which', cmd], capture_output=True, timeout=2)
|
||||
if result.returncode == 0:
|
||||
cmd_path = result.stdout.decode().strip()
|
||||
dep_msg = f" ✓ {dep_name}: {cmd_path} (system)"
|
||||
dep_msg = f" {dep_name}: {cmd_path} (system)"
|
||||
self.logger.info(dep_msg)
|
||||
if status_callback:
|
||||
status_callback(dep_msg)
|
||||
@@ -379,9 +379,9 @@ class WinetricksHandler:
|
||||
if not found:
|
||||
missing_deps.append(dep_name)
|
||||
if dep_name in bundled_tools_list:
|
||||
self.logger.warning(f" ✗ {dep_name}: NOT FOUND (neither bundled nor system)")
|
||||
self.logger.warning(f" {dep_name}: NOT FOUND (neither bundled nor system)")
|
||||
else:
|
||||
self.logger.warning(f" ✗ {dep_name}: NOT FOUND (system only - not bundled)")
|
||||
self.logger.warning(f" {dep_name}: NOT FOUND (system only - not bundled)")
|
||||
|
||||
if missing_deps:
|
||||
# Separate critical vs optional dependencies
|
||||
|
||||
@@ -1785,22 +1785,36 @@ echo Prefix creation complete.
|
||||
logger.error(f"Error setting Proton on shortcut: {e}")
|
||||
return False
|
||||
|
||||
def run_working_workflow(self, shortcut_name: str, modlist_install_dir: str,
|
||||
@staticmethod
|
||||
def get_ttw_installer_path() -> Optional[Path]:
|
||||
"""Get path to TTW_Linux_Installer if available"""
|
||||
try:
|
||||
from jackify.shared.paths import get_jackify_data_dir
|
||||
ttw_path = get_jackify_data_dir() / "TTW_Linux_Installer" / "ttw_linux_gui"
|
||||
if ttw_path.exists():
|
||||
return ttw_path
|
||||
except Exception:
|
||||
pass
|
||||
return None
|
||||
|
||||
def run_working_workflow(self, shortcut_name: str, modlist_install_dir: str,
|
||||
final_exe_path: str, progress_callback=None, steamdeck: Optional[bool] = None) -> Tuple[bool, Optional[Path], Optional[int], Optional[str]]:
|
||||
"""
|
||||
Run the proven working automated prefix creation workflow.
|
||||
|
||||
|
||||
This implements our tested and working approach:
|
||||
1. Create shortcut with native Steam service (pointing to ModOrganizer.exe initially)
|
||||
2. Restart Steam using Jackify's robust method
|
||||
3. Create Proton prefix invisibly using Proton wrapper with DISPLAY=
|
||||
4. Verify everything persists
|
||||
|
||||
|
||||
Args:
|
||||
shortcut_name: Name for the Steam shortcut
|
||||
modlist_install_dir: Directory where the modlist is installed
|
||||
final_exe_path: Path to ModOrganizer.exe
|
||||
|
||||
progress_callback: Optional callback for progress updates
|
||||
steamdeck: Optional Steam Deck detection override
|
||||
|
||||
Returns:
|
||||
Tuple of (success, prefix_path, appid, last_timestamp)
|
||||
"""
|
||||
@@ -1922,13 +1936,23 @@ echo Prefix creation complete.
|
||||
progress_callback(f"{self._get_progress_timestamp()} Injecting {special_game_type.upper()} game registry entries...")
|
||||
|
||||
if prefix_path:
|
||||
self._inject_game_registry_entries(str(prefix_path))
|
||||
self._inject_game_registry_entries(str(prefix_path), special_game_type)
|
||||
else:
|
||||
logger.warning("Could not find prefix path for registry injection")
|
||||
else:
|
||||
logger.info("Step 5: Skipping registry injection for standard modlist")
|
||||
if progress_callback:
|
||||
progress_callback(f"{self._get_progress_timestamp()} No special game registry injection needed")
|
||||
|
||||
# Step 5.5: Pre-create game-specific directories for all modlists
|
||||
logger.info(f"Step 5.5: Creating game-specific user directories")
|
||||
if progress_callback:
|
||||
progress_callback(f"{self._get_progress_timestamp()} Creating game user directories...")
|
||||
|
||||
if prefix_path:
|
||||
self._create_game_user_directories(str(prefix_path), special_game_type)
|
||||
else:
|
||||
logger.warning("Could not find prefix path for directory creation")
|
||||
|
||||
last_timestamp = self._get_progress_timestamp()
|
||||
logger.info(f" Working workflow completed successfully! AppID: {appid}, Prefix: {prefix_path}")
|
||||
@@ -1942,7 +1966,7 @@ echo Prefix creation complete.
|
||||
if progress_callback:
|
||||
progress_callback("") # Extra blank line to span across Configuration Summary
|
||||
progress_callback("") # And one more to create space before Prefix Configuration
|
||||
|
||||
|
||||
return True, prefix_path, appid, last_timestamp
|
||||
|
||||
except Exception as e:
|
||||
@@ -3250,7 +3274,7 @@ echo Prefix creation complete.
|
||||
logger.debug(f"Error during recursive wine search in {proton_path}: {e}")
|
||||
return None
|
||||
|
||||
def _inject_game_registry_entries(self, modlist_compatdata_path: str):
|
||||
def _inject_game_registry_entries(self, modlist_compatdata_path: str, special_game_type: str):
|
||||
"""Detect and inject FNV/Enderal game paths and apply universal dotnet4.x compatibility fixes"""
|
||||
system_reg_path = os.path.join(modlist_compatdata_path, "pfx", "system.reg")
|
||||
if not os.path.exists(system_reg_path):
|
||||
@@ -3290,15 +3314,6 @@ echo Prefix creation complete.
|
||||
)
|
||||
if success:
|
||||
logger.info(f"Updated registry entry for {config['name']}")
|
||||
|
||||
# Special handling for Enderal: Create required user directory
|
||||
if app_id == "976620": # Enderal Special Edition
|
||||
try:
|
||||
enderal_docs_path = os.path.join(modlist_compatdata_path, "pfx", "drive_c", "users", "steamuser", "Documents", "My Games", "Enderal Special Edition")
|
||||
os.makedirs(enderal_docs_path, exist_ok=True)
|
||||
logger.info(f"Created Enderal user directory: {enderal_docs_path}")
|
||||
except Exception as e:
|
||||
logger.warning(f"Failed to create Enderal user directory: {e}")
|
||||
else:
|
||||
logger.warning(f"Failed to update registry entry for {config['name']}")
|
||||
else:
|
||||
@@ -3306,6 +3321,49 @@ echo Prefix creation complete.
|
||||
|
||||
logger.info("Game registry injection completed")
|
||||
|
||||
def _create_game_user_directories(self, modlist_compatdata_path: str, special_game_type: str):
|
||||
"""
|
||||
Pre-create game-specific user directories to prevent first-launch issues.
|
||||
|
||||
Creates both My Documents/My Games and AppData/Local directories for the game.
|
||||
This prevents issues where games fail to create these on first launch under Proton.
|
||||
"""
|
||||
# Map game types to their directory names
|
||||
game_dir_names = {
|
||||
"skyrim": "Skyrim Special Edition",
|
||||
"fnv": "FalloutNV",
|
||||
"fo4": "Fallout4",
|
||||
"oblivion": "Oblivion",
|
||||
"oblivion_remastered": "Oblivion Remastered",
|
||||
"enderal": "Enderal Special Edition",
|
||||
"starfield": "Starfield"
|
||||
}
|
||||
|
||||
# Get the directory name for this game type
|
||||
game_dir_name = game_dir_names.get(special_game_type)
|
||||
if not game_dir_name:
|
||||
logger.debug(f"No user directory mapping for game type: {special_game_type}")
|
||||
return
|
||||
|
||||
base_path = os.path.join(modlist_compatdata_path, "pfx", "drive_c", "users", "steamuser")
|
||||
|
||||
directories_to_create = [
|
||||
os.path.join(base_path, "Documents", "My Games", game_dir_name),
|
||||
os.path.join(base_path, "AppData", "Local", game_dir_name)
|
||||
]
|
||||
|
||||
created_count = 0
|
||||
for directory in directories_to_create:
|
||||
try:
|
||||
os.makedirs(directory, exist_ok=True)
|
||||
logger.info(f"Created user directory: {directory}")
|
||||
created_count += 1
|
||||
except Exception as e:
|
||||
logger.warning(f"Failed to create directory {directory}: {e}")
|
||||
|
||||
if created_count > 0:
|
||||
logger.info(f"Created {created_count} user directories for {game_dir_name}")
|
||||
|
||||
def _get_lorerim_preferred_proton(self):
|
||||
"""Get Lorerim's preferred Proton 9 version with specific priority order"""
|
||||
try:
|
||||
|
||||
220
jackify/backend/services/nexus_download_service.py
Normal file
220
jackify/backend/services/nexus_download_service.py
Normal file
@@ -0,0 +1,220 @@
|
||||
"""
|
||||
Nexus Download Service
|
||||
|
||||
Handles downloading mod files from Nexus Mods using OAuth authentication.
|
||||
"""
|
||||
|
||||
import logging
|
||||
import requests
|
||||
import time
|
||||
from pathlib import Path
|
||||
from typing import Optional, Callable, Tuple
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class NexusDownloadService:
|
||||
"""Service for downloading files from Nexus Mods"""
|
||||
|
||||
NEXUS_API_BASE = "https://api.nexusmods.com/v1"
|
||||
|
||||
def __init__(self, auth_token: str):
|
||||
"""
|
||||
Initialize Nexus download service.
|
||||
|
||||
Args:
|
||||
auth_token: OAuth access token or API key
|
||||
"""
|
||||
self.auth_token = auth_token
|
||||
self.headers = {
|
||||
"Authorization": f"Bearer {auth_token}",
|
||||
"User-Agent": "jackify"
|
||||
}
|
||||
|
||||
def get_mod_files(self, game_domain: str, mod_id: int) -> Optional[list]:
|
||||
"""
|
||||
Get list of files for a mod.
|
||||
|
||||
Args:
|
||||
game_domain: Game domain (e.g., 'newvegas')
|
||||
mod_id: Mod ID number
|
||||
|
||||
Returns:
|
||||
List of file metadata dicts, or None if failed
|
||||
"""
|
||||
try:
|
||||
url = f"{self.NEXUS_API_BASE}/games/{game_domain}/mods/{mod_id}/files.json"
|
||||
|
||||
response = requests.get(url, headers=self.headers, timeout=30)
|
||||
response.raise_for_status()
|
||||
|
||||
data = response.json()
|
||||
files = data.get('files', [])
|
||||
|
||||
logger.info(f"Found {len(files)} files for mod {mod_id}")
|
||||
return files
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to get mod files: {e}")
|
||||
return None
|
||||
|
||||
def get_download_link(self, game_domain: str, mod_id: int, file_id: int) -> Optional[str]:
|
||||
"""
|
||||
Get download link for a specific file.
|
||||
|
||||
Args:
|
||||
game_domain: Game domain (e.g., 'newvegas')
|
||||
mod_id: Mod ID number
|
||||
file_id: File ID number
|
||||
|
||||
Returns:
|
||||
Download URL, or None if failed
|
||||
"""
|
||||
try:
|
||||
url = f"{self.NEXUS_API_BASE}/games/{game_domain}/mods/{mod_id}/files/{file_id}/download_link.json"
|
||||
|
||||
response = requests.get(url, headers=self.headers, timeout=30)
|
||||
|
||||
# Check for specific error codes
|
||||
if response.status_code == 403:
|
||||
logger.error(f"Download link request forbidden (403) - Nexus Premium required for file {file_id}")
|
||||
return None
|
||||
elif response.status_code == 404:
|
||||
logger.error(f"Download link request not found (404) - file {file_id} may not exist")
|
||||
return None
|
||||
|
||||
response.raise_for_status()
|
||||
|
||||
data = response.json()
|
||||
|
||||
# API returns list of download servers
|
||||
if isinstance(data, list) and len(data) > 0:
|
||||
download_url = data[0].get('URI')
|
||||
logger.info(f"Got download link for file {file_id}")
|
||||
return download_url
|
||||
else:
|
||||
logger.error(f"No download link returned for file {file_id}")
|
||||
return None
|
||||
|
||||
except requests.exceptions.HTTPError as e:
|
||||
logger.error(f"Failed to get download link: HTTP {e.response.status_code} - {e}")
|
||||
return None
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to get download link: {e}")
|
||||
return None
|
||||
|
||||
def download_file(
|
||||
self,
|
||||
download_url: str,
|
||||
output_path: Path,
|
||||
progress_callback: Optional[Callable[[int, int], None]] = None
|
||||
) -> bool:
|
||||
"""
|
||||
Download a file from Nexus.
|
||||
|
||||
Args:
|
||||
download_url: Download URL from get_download_link()
|
||||
output_path: Where to save the file
|
||||
progress_callback: Optional callback(downloaded_bytes, total_bytes)
|
||||
|
||||
Returns:
|
||||
True if successful
|
||||
"""
|
||||
try:
|
||||
output_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
response = requests.get(download_url, stream=True, timeout=60)
|
||||
response.raise_for_status()
|
||||
|
||||
total_size = int(response.headers.get('content-length', 0))
|
||||
downloaded = 0
|
||||
|
||||
with open(output_path, 'wb') as f:
|
||||
for chunk in response.iter_content(chunk_size=8192):
|
||||
if chunk:
|
||||
f.write(chunk)
|
||||
downloaded += len(chunk)
|
||||
|
||||
if progress_callback and total_size > 0:
|
||||
progress_callback(downloaded, total_size)
|
||||
|
||||
logger.info(f"Downloaded {output_path.name} ({downloaded} bytes)")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Download failed: {e}")
|
||||
if output_path.exists():
|
||||
output_path.unlink()
|
||||
return False
|
||||
|
||||
def download_latest_file(
|
||||
self,
|
||||
game_domain: str,
|
||||
mod_id: int,
|
||||
output_dir: Path,
|
||||
file_name_filter: Optional[str] = None,
|
||||
progress_callback: Optional[Callable[[str], None]] = None
|
||||
) -> Tuple[bool, Optional[Path], str]:
|
||||
"""
|
||||
Download the latest file from a mod.
|
||||
|
||||
Args:
|
||||
game_domain: Game domain (e.g., 'newvegas')
|
||||
mod_id: Mod ID number
|
||||
output_dir: Directory to save file
|
||||
file_name_filter: Optional substring to filter files (e.g., 'linux', 'mpi')
|
||||
progress_callback: Optional callback for status updates
|
||||
|
||||
Returns:
|
||||
Tuple of (success, file_path, message)
|
||||
"""
|
||||
def update_progress(msg: str):
|
||||
if progress_callback:
|
||||
progress_callback(msg)
|
||||
logger.info(msg)
|
||||
|
||||
try:
|
||||
update_progress(f"Fetching file list for mod {mod_id}...")
|
||||
|
||||
files = self.get_mod_files(game_domain, mod_id)
|
||||
if not files:
|
||||
return False, None, "Failed to get mod file list"
|
||||
|
||||
# Filter files if requested
|
||||
if file_name_filter:
|
||||
filtered = [f for f in files if file_name_filter.lower() in f.get('file_name', '').lower()]
|
||||
if not filtered:
|
||||
return False, None, f"No files found matching '{file_name_filter}'"
|
||||
files = filtered
|
||||
|
||||
# Get the most recent file
|
||||
files.sort(key=lambda f: f.get('uploaded_timestamp', 0), reverse=True)
|
||||
latest_file = files[0]
|
||||
|
||||
file_id = latest_file['file_id']
|
||||
file_name = latest_file['file_name']
|
||||
|
||||
update_progress(f"Downloading {file_name}...")
|
||||
|
||||
download_url = self.get_download_link(game_domain, mod_id, file_id)
|
||||
if not download_url:
|
||||
return False, None, "Failed to get download link"
|
||||
|
||||
output_path = output_dir / file_name
|
||||
|
||||
def download_progress(downloaded, total):
|
||||
if total > 0:
|
||||
percent = (downloaded / total) * 100
|
||||
update_progress(f"Downloading: {percent:.1f}%")
|
||||
|
||||
success = self.download_file(download_url, output_path, download_progress)
|
||||
|
||||
if success:
|
||||
return True, output_path, f"Downloaded {file_name}"
|
||||
else:
|
||||
return False, None, "Download failed"
|
||||
|
||||
except Exception as e:
|
||||
error_msg = f"Download failed: {str(e)}"
|
||||
logger.error(error_msg, exc_info=True)
|
||||
return False, None, error_msg
|
||||
198
jackify/backend/services/vnv_integration_helper.py
Normal file
198
jackify/backend/services/vnv_integration_helper.py
Normal file
@@ -0,0 +1,198 @@
|
||||
"""
|
||||
VNV Integration Helper
|
||||
|
||||
Helper functions to integrate VNV post-install automation into modlist workflows.
|
||||
Handles detection, confirmation, and execution for:
|
||||
- Install Modlist
|
||||
- Configure New Modlist
|
||||
- Configure Existing Modlist
|
||||
"""
|
||||
|
||||
import logging
|
||||
import configparser
|
||||
import re
|
||||
from pathlib import Path
|
||||
from typing import Optional, Callable, Tuple
|
||||
|
||||
from .vnv_post_install_service import VNVPostInstallService
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _parse_bytearray_value(value: str) -> str:
|
||||
"""
|
||||
Parse Qt @ByteArray format to extract the actual string value.
|
||||
|
||||
Format: @ByteArray(Viva New Vegas Extended)
|
||||
Returns: Viva New Vegas Extended
|
||||
"""
|
||||
match = re.match(r'@ByteArray\((.*)\)', value)
|
||||
if match:
|
||||
return match.group(1)
|
||||
return value
|
||||
|
||||
|
||||
def _check_modorganizer_ini_profile(modlist_install_location: Path) -> bool:
|
||||
"""
|
||||
Check ModOrganizer.ini for VNV profile names.
|
||||
|
||||
Args:
|
||||
modlist_install_location: Path to modlist installation directory
|
||||
|
||||
Returns:
|
||||
True if selected_profile is "Viva New Vegas" or "Viva New Vegas Extended"
|
||||
"""
|
||||
try:
|
||||
mo_ini_path = modlist_install_location / "ModOrganizer.ini"
|
||||
if not mo_ini_path.exists():
|
||||
logger.debug(f"ModOrganizer.ini not found at {mo_ini_path}")
|
||||
return False
|
||||
|
||||
config = configparser.ConfigParser()
|
||||
# Read with UTF-8-sig to handle BOM
|
||||
config.read(mo_ini_path, encoding='utf-8-sig')
|
||||
|
||||
if 'General' not in config:
|
||||
logger.debug("No [General] section in ModOrganizer.ini")
|
||||
return False
|
||||
|
||||
selected_profile_raw = config.get('General', 'selected_profile', fallback='')
|
||||
if not selected_profile_raw:
|
||||
logger.debug("No selected_profile in ModOrganizer.ini")
|
||||
return False
|
||||
|
||||
# Parse @ByteArray format
|
||||
selected_profile = _parse_bytearray_value(selected_profile_raw)
|
||||
logger.debug(f"Found selected_profile: {selected_profile}")
|
||||
|
||||
# Check if it's one of the VNV profiles
|
||||
vnv_profiles = ["Viva New Vegas", "Viva New Vegas Extended"]
|
||||
return selected_profile in vnv_profiles
|
||||
|
||||
except Exception as e:
|
||||
logger.debug(f"Error checking ModOrganizer.ini for VNV profile: {e}")
|
||||
return False
|
||||
|
||||
|
||||
def should_offer_vnv_automation(modlist_name: str, modlist_install_location: Optional[Path] = None) -> bool:
|
||||
"""
|
||||
Check if VNV automation should be offered for this modlist.
|
||||
|
||||
Detection methods (in order of reliability):
|
||||
1. Check ModOrganizer.ini selected_profile (most reliable)
|
||||
2. Check modlist name for VNV patterns
|
||||
|
||||
Args:
|
||||
modlist_name: Name of the modlist
|
||||
modlist_install_location: Optional path to modlist installation directory
|
||||
|
||||
Returns:
|
||||
True if VNV automation should be offered
|
||||
"""
|
||||
# Method 1: Check ModOrganizer.ini profile (most reliable)
|
||||
if modlist_install_location:
|
||||
if _check_modorganizer_ini_profile(modlist_install_location):
|
||||
logger.info(f"VNV detected via ModOrganizer.ini profile in {modlist_install_location}")
|
||||
return True
|
||||
|
||||
# Method 2: Check modlist name patterns
|
||||
modlist_name_lower = modlist_name.lower()
|
||||
vnv_patterns = [
|
||||
"viva new vegas",
|
||||
"vnv", # Common abbreviation
|
||||
"viva new vegas extended"
|
||||
]
|
||||
|
||||
for pattern in vnv_patterns:
|
||||
if pattern in modlist_name_lower:
|
||||
logger.info(f"VNV detected via name pattern '{pattern}' in '{modlist_name}'")
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
|
||||
def run_vnv_automation_if_applicable(
|
||||
modlist_name: str,
|
||||
modlist_install_location: Path,
|
||||
game_root: Path,
|
||||
ttw_installer_path: Optional[Path] = None,
|
||||
progress_callback: Optional[Callable[[str], None]] = None,
|
||||
manual_file_callback: Optional[Callable[[str, str], Optional[Path]]] = None,
|
||||
confirmation_callback: Optional[Callable[[str], bool]] = None
|
||||
) -> Tuple[bool, Optional[str]]:
|
||||
"""
|
||||
Check if VNV automation should run, get user confirmation, and execute if confirmed.
|
||||
|
||||
Args:
|
||||
modlist_name: Name of the installed modlist
|
||||
modlist_install_location: Path to modlist installation
|
||||
game_root: Path to game root directory
|
||||
ttw_installer_path: Optional path to TTW_Linux_Installer (for BSA decompression)
|
||||
progress_callback: Optional callback for progress updates
|
||||
manual_file_callback: Optional callback for manual file selection (non-Premium)
|
||||
confirmation_callback: Optional callback for user confirmation
|
||||
Takes description string, returns True if user confirms
|
||||
|
||||
Returns:
|
||||
Tuple of (automation_was_run: bool, error_message: Optional[str])
|
||||
"""
|
||||
try:
|
||||
# Check if this is VNV (pass install location for ModOrganizer.ini check)
|
||||
if not should_offer_vnv_automation(modlist_name, modlist_install_location):
|
||||
logger.debug(f"Modlist '{modlist_name}' does not require VNV automation")
|
||||
return False, None
|
||||
|
||||
logger.info(f"VNV detected: {modlist_name}")
|
||||
|
||||
# Initialize service
|
||||
vnv_service = VNVPostInstallService(
|
||||
modlist_install_location=modlist_install_location,
|
||||
game_root=game_root,
|
||||
ttw_installer_path=ttw_installer_path
|
||||
)
|
||||
|
||||
# Check what's already done
|
||||
completed = vnv_service.check_already_completed()
|
||||
# Only skip if ALL three steps are completed
|
||||
if completed['root_mods'] and completed['4gb_patch'] and completed['bsa_decompressed']:
|
||||
logger.info("VNV automation steps already completed")
|
||||
if progress_callback:
|
||||
progress_callback("VNV post-install steps already completed")
|
||||
return False, None
|
||||
|
||||
# Get confirmation from user (required)
|
||||
if not confirmation_callback:
|
||||
logger.error("VNV automation requires confirmation_callback")
|
||||
return False, "VNV automation requires user confirmation"
|
||||
|
||||
if confirmation_callback:
|
||||
description = vnv_service.get_automation_description()
|
||||
if not confirmation_callback(description):
|
||||
logger.info("User declined VNV automation")
|
||||
if progress_callback:
|
||||
progress_callback("VNV automation skipped by user")
|
||||
return False, None
|
||||
|
||||
# Run automation
|
||||
logger.info("Starting VNV post-install automation")
|
||||
if progress_callback:
|
||||
progress_callback("Running VNV post-install automation...")
|
||||
|
||||
success, message = vnv_service.run_all_steps(
|
||||
progress_callback=progress_callback,
|
||||
manual_file_callback=manual_file_callback
|
||||
)
|
||||
|
||||
if success:
|
||||
logger.info(f"VNV automation completed: {message}")
|
||||
if progress_callback:
|
||||
progress_callback(f"VNV automation: {message}")
|
||||
return True, None
|
||||
else:
|
||||
logger.error(f"VNV automation failed: {message}")
|
||||
return True, message
|
||||
|
||||
except Exception as e:
|
||||
error_msg = f"VNV automation error: {str(e)}"
|
||||
logger.error(error_msg, exc_info=True)
|
||||
return True, error_msg
|
||||
622
jackify/backend/services/vnv_post_install_service.py
Normal file
622
jackify/backend/services/vnv_post_install_service.py
Normal file
@@ -0,0 +1,622 @@
|
||||
"""
|
||||
Viva New Vegas Post-Install Service
|
||||
|
||||
Automates the post-installation steps required for Viva New Vegas modlist:
|
||||
1. Root Mods - Copy files from '__Files Requiring Manual Install' to game root
|
||||
2. 4GB Patcher - Download Linux version from Nexus, run natively
|
||||
3. BSA Decompression - Download FNV BSA Decompressor MPI, run via TTW_Linux_Installer
|
||||
|
||||
These steps are documented at: https://vivanewvegas.moddinglinked.com/wabbajack.html
|
||||
|
||||
Uses native Linux tools (no Wine required) by downloading from Nexus with OAuth.
|
||||
"""
|
||||
|
||||
import logging
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
import stat
|
||||
import tempfile
|
||||
import zipfile
|
||||
from pathlib import Path
|
||||
from typing import Optional, Callable
|
||||
|
||||
from ..handlers.subprocess_utils import get_clean_subprocess_env
|
||||
from .nexus_download_service import NexusDownloadService
|
||||
from .nexus_auth_service import NexusAuthService
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class VNVPostInstallService:
|
||||
"""Handles automated post-installation tasks for Viva New Vegas modlist."""
|
||||
|
||||
# Nexus mod IDs for required tools
|
||||
LINUX_4GB_PATCHER_MOD_ID = 62552
|
||||
FNV_BSA_DECOMPRESSOR_MOD_ID = 65854
|
||||
GAME_DOMAIN = "newvegas"
|
||||
|
||||
def __init__(self, modlist_install_location: Path, game_root: Path,
|
||||
ttw_installer_path: Optional[Path] = None):
|
||||
"""
|
||||
Initialize VNV post-install service.
|
||||
|
||||
Args:
|
||||
modlist_install_location: Path to the VNV installation (e.g., ~/VNV)
|
||||
game_root: Path to Fallout New Vegas game root
|
||||
ttw_installer_path: Path to TTW_Linux_Installer executable (for BSA decompression)
|
||||
"""
|
||||
self.modlist_install = modlist_install_location
|
||||
self.game_root = game_root
|
||||
self.ttw_installer_path = ttw_installer_path
|
||||
|
||||
# VNV-specific paths
|
||||
self.manual_install_dir = self.modlist_install / "__Files Requiring Manual Install"
|
||||
|
||||
# Download cache directory
|
||||
from jackify.shared.paths import get_jackify_data_dir
|
||||
self.cache_dir = get_jackify_data_dir() / "vnv_post_install_cache"
|
||||
self.cache_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Initialize authentication
|
||||
self.auth_service = NexusAuthService()
|
||||
self.download_service = None
|
||||
|
||||
def _ensure_auth(self, progress_callback: Optional[Callable[[str], None]] = None) -> bool:
|
||||
"""
|
||||
Ensure we have valid Nexus authentication for downloads.
|
||||
|
||||
Args:
|
||||
progress_callback: Optional callback for progress updates
|
||||
|
||||
Returns:
|
||||
True if authenticated
|
||||
"""
|
||||
auth_token = self.auth_service.ensure_valid_auth()
|
||||
|
||||
if not auth_token:
|
||||
if progress_callback:
|
||||
progress_callback("Nexus authentication required for post-install steps")
|
||||
logger.error("No Nexus authentication available")
|
||||
return False
|
||||
|
||||
self.download_service = NexusDownloadService(auth_token)
|
||||
return True
|
||||
|
||||
def should_run_automation(self, modlist_name: str) -> bool:
|
||||
"""
|
||||
Check if this modlist should trigger VNV automation.
|
||||
|
||||
Args:
|
||||
modlist_name: Name of the installed modlist
|
||||
|
||||
Returns:
|
||||
True if VNV automation should be offered
|
||||
"""
|
||||
return "viva new vegas" in modlist_name.lower()
|
||||
|
||||
def get_automation_description(self) -> str:
|
||||
"""
|
||||
Get user-friendly description of what VNV automation does.
|
||||
|
||||
Returns:
|
||||
Description string for confirmation dialog
|
||||
"""
|
||||
return (
|
||||
"Viva New Vegas Automation\n\n"
|
||||
"Jackify can automatically perform the following post-install steps:\n\n"
|
||||
"1. Copy root mods to game directory\n"
|
||||
"2. Download and run Linux 4GB patcher\n"
|
||||
"3. Download and run BSA decompressor (reduces loading times)\n\n"
|
||||
"Premium users: Downloads happen automatically\n"
|
||||
"Non-Premium users: You'll be prompted to download files manually\n\n"
|
||||
"Would you like Jackify to automate these steps?"
|
||||
)
|
||||
|
||||
def check_already_completed(self) -> dict:
|
||||
"""
|
||||
Check which VNV automation steps have already been completed.
|
||||
|
||||
Returns:
|
||||
Dict with keys: 'root_mods', '4gb_patch', 'bsa_decompressed'
|
||||
"""
|
||||
# Check if 4GB patch already applied
|
||||
backup_exe = self.game_root / "FalloutNV_backup.exe"
|
||||
already_patched = backup_exe.exists()
|
||||
|
||||
# Check if root mods copied (look for FNVpatch.exe in game root)
|
||||
root_mods_copied = (self.game_root / "FNVpatch.exe").exists()
|
||||
|
||||
# Check for BSA decompression marker file
|
||||
marker_file = self.game_root / ".jackify_bsa_decompressed"
|
||||
bsa_decompressed = marker_file.exists()
|
||||
|
||||
return {
|
||||
'root_mods': root_mods_copied,
|
||||
'4gb_patch': already_patched,
|
||||
'bsa_decompressed': bsa_decompressed
|
||||
}
|
||||
|
||||
def run_all_steps(self, progress_callback: Optional[Callable[[str], None]] = None,
|
||||
manual_file_callback: Optional[Callable[[str, str], Optional[Path]]] = None,
|
||||
skip_confirmation: bool = False) -> tuple[bool, str]:
|
||||
"""
|
||||
Run all VNV post-install steps in sequence.
|
||||
|
||||
Args:
|
||||
progress_callback: Optional callback for progress updates
|
||||
manual_file_callback: Optional callback for manual file selection (non-Premium users)
|
||||
Takes (title, instructions) returns Path or None
|
||||
skip_confirmation: Skip user confirmation (for programmatic use)
|
||||
|
||||
Returns:
|
||||
(success: bool, message: str)
|
||||
"""
|
||||
def update_progress(msg: str):
|
||||
if progress_callback:
|
||||
progress_callback(msg)
|
||||
logger.info(msg)
|
||||
|
||||
try:
|
||||
# Ensure authentication
|
||||
update_progress("Checking Nexus authentication...")
|
||||
if not self._ensure_auth(progress_callback):
|
||||
return False, "Nexus authentication required. Please authenticate in Settings."
|
||||
|
||||
# Step 1: Copy root mods
|
||||
update_progress("Step 1/3: Copying root mods to game directory...")
|
||||
success, msg = self.copy_root_mods()
|
||||
if not success:
|
||||
return False, f"Root mods failed: {msg}"
|
||||
update_progress(f"Root mods: {msg}")
|
||||
|
||||
# Step 2: Run 4GB patcher
|
||||
update_progress("Step 2/3: Downloading and running 4GB patcher...")
|
||||
success, msg = self.run_4gb_patcher(update_progress, manual_file_callback)
|
||||
if not success:
|
||||
return False, f"4GB patcher failed: {msg}"
|
||||
update_progress(f"4GB patcher: {msg}")
|
||||
|
||||
# Step 3: Run BSA decompressor
|
||||
update_progress("Step 3/3: Downloading and running BSA decompressor...")
|
||||
success, msg = self.run_bsa_decompressor(update_progress, manual_file_callback)
|
||||
if not success:
|
||||
return False, f"BSA decompression failed: {msg}"
|
||||
update_progress(f"BSA decompression: {msg}")
|
||||
|
||||
return True, "VNV post-install completed successfully"
|
||||
|
||||
except Exception as e:
|
||||
error_msg = f"VNV post-install failed: {str(e)}"
|
||||
logger.error(error_msg, exc_info=True)
|
||||
return False, error_msg
|
||||
|
||||
def copy_root_mods(self) -> tuple[bool, str]:
|
||||
"""
|
||||
Copy files from '__Files Requiring Manual Install' to game root.
|
||||
|
||||
Returns:
|
||||
(success: bool, message: str)
|
||||
"""
|
||||
try:
|
||||
if not self.manual_install_dir.exists():
|
||||
return False, f"Manual install directory not found: {self.manual_install_dir}"
|
||||
|
||||
if not self.game_root.exists():
|
||||
return False, f"Game root directory not found: {self.game_root}"
|
||||
|
||||
# Copy all files from manual install to game root
|
||||
copied_files = []
|
||||
for item in self.manual_install_dir.iterdir():
|
||||
dest = self.game_root / item.name
|
||||
|
||||
if item.is_file():
|
||||
shutil.copy2(item, dest)
|
||||
copied_files.append(item.name)
|
||||
logger.debug(f"Copied: {item.name}")
|
||||
elif item.is_dir():
|
||||
# Merge directories to preserve vanilla game files (e.g., BSA files in Data/)
|
||||
# dirs_exist_ok=True allows adding NVSE to Data/ without deleting vanilla BSAs
|
||||
shutil.copytree(item, dest, dirs_exist_ok=True)
|
||||
copied_files.append(f"{item.name}/")
|
||||
logger.debug(f"Copied directory: {item.name}/")
|
||||
|
||||
if not copied_files:
|
||||
return False, "No files found to copy"
|
||||
|
||||
logger.info(f"Copied {len(copied_files)} items to game root")
|
||||
return True, f"Copied {len(copied_files)} items to game root"
|
||||
|
||||
except Exception as e:
|
||||
error_msg = f"Failed to copy root mods: {str(e)}"
|
||||
logger.error(error_msg, exc_info=True)
|
||||
return False, error_msg
|
||||
|
||||
def run_4gb_patcher(self, progress_callback: Optional[Callable[[str], None]] = None,
|
||||
manual_file_callback: Optional[Callable[[str, str], Optional[Path]]] = None) -> tuple[bool, str]:
|
||||
"""
|
||||
Download and run native Linux 4GB patcher.
|
||||
|
||||
Args:
|
||||
progress_callback: Optional callback for progress updates
|
||||
manual_file_callback: Optional callback for manual file selection
|
||||
Takes (title, instructions) returns Path or None
|
||||
|
||||
Returns:
|
||||
(success: bool, message: str)
|
||||
"""
|
||||
try:
|
||||
# Check if already patched
|
||||
backup_exe = self.game_root / "FalloutNV_backup.exe"
|
||||
if backup_exe.exists():
|
||||
logger.info("Game already has 4GB patch (backup exists)")
|
||||
return True, "Game already patched (backup exists)"
|
||||
|
||||
# Check cache first - look for extracted executable or zip
|
||||
patcher_path = None
|
||||
cached_extracted = list(self.cache_dir.glob("*4gb*_extracted/*"))
|
||||
if cached_extracted:
|
||||
# Use already extracted executable
|
||||
for f in cached_extracted:
|
||||
if f.is_file():
|
||||
patcher_path = f
|
||||
logger.info(f"Using cached extracted 4GB patcher: {patcher_path}")
|
||||
break
|
||||
|
||||
if not patcher_path:
|
||||
cached_files = list(self.cache_dir.glob("*4gb*.zip"))
|
||||
if cached_files:
|
||||
patcher_path = cached_files[0]
|
||||
logger.info(f"Using cached 4GB patcher zip: {patcher_path}")
|
||||
|
||||
if not patcher_path:
|
||||
# Try to download from Nexus
|
||||
# Note: The Linux version is named "FNV4GB for Proton", not "linux"
|
||||
success, patcher_path, msg = self.download_service.download_latest_file(
|
||||
self.GAME_DOMAIN,
|
||||
self.LINUX_4GB_PATCHER_MOD_ID,
|
||||
self.cache_dir,
|
||||
file_name_filter="proton",
|
||||
progress_callback=progress_callback
|
||||
)
|
||||
|
||||
if not success:
|
||||
# Download failed - offer manual download
|
||||
logger.error(f"Automatic download failed: {msg}")
|
||||
logger.debug(f"Looking for file with 'proton' in name on mod {self.LINUX_4GB_PATCHER_MOD_ID}")
|
||||
|
||||
if not manual_file_callback:
|
||||
return False, f"Failed to download 4GB patcher: {msg}\n\nPlease download manually from: https://www.nexusmods.com/newvegas/mods/62552"
|
||||
|
||||
instructions = (
|
||||
"Automatic download failed (requires Nexus Premium).\n\n"
|
||||
"Please download the Linux 4GB Patcher manually:\n\n"
|
||||
"1. Visit: https://www.nexusmods.com/newvegas/mods/62552\n\n"
|
||||
"2. Download the file named 'FNV4GB for Linux'\n\n"
|
||||
"3. Select the downloaded file below"
|
||||
)
|
||||
|
||||
patcher_path = manual_file_callback("4GB Patcher Required", instructions)
|
||||
|
||||
if not patcher_path or not patcher_path.exists():
|
||||
return False, "4GB patcher file not provided"
|
||||
|
||||
# Copy to cache for future use
|
||||
cached_path = self.cache_dir / patcher_path.name
|
||||
shutil.copy2(patcher_path, cached_path)
|
||||
patcher_path = cached_path
|
||||
logger.info(f"Using manually selected 4GB patcher: {patcher_path}")
|
||||
|
||||
# Extract if it's a zip file and not already extracted
|
||||
if patcher_path.suffix.lower() == '.zip':
|
||||
extract_dir = self.cache_dir / f"{patcher_path.stem}_extracted"
|
||||
|
||||
# Extract if not already done
|
||||
if not extract_dir.exists():
|
||||
logger.info(f"Extracting {patcher_path.name}...")
|
||||
extract_dir.mkdir(parents=True, exist_ok=True)
|
||||
with zipfile.ZipFile(patcher_path, 'r') as zip_ref:
|
||||
zip_ref.extractall(extract_dir)
|
||||
logger.info(f"Extracted to {extract_dir}")
|
||||
|
||||
# Find the executable
|
||||
executables = list(extract_dir.glob("*"))
|
||||
if not executables:
|
||||
return False, "No files found in 4GB patcher zip"
|
||||
|
||||
# Look for executable file (FalloutNVPatcher or similar)
|
||||
patcher_exe = None
|
||||
for f in executables:
|
||||
if f.is_file() and ('fallout' in f.name.lower() or 'patcher' in f.name.lower() or 'fnv' in f.name.lower()):
|
||||
patcher_exe = f
|
||||
break
|
||||
|
||||
if not patcher_exe:
|
||||
# Use first file if no obvious match
|
||||
patcher_exe = next((f for f in executables if f.is_file()), None)
|
||||
|
||||
if not patcher_exe:
|
||||
return False, "No executable found in 4GB patcher zip"
|
||||
|
||||
patcher_path = patcher_exe
|
||||
logger.info(f"Using patcher executable: {patcher_path.name}")
|
||||
|
||||
# Make executable
|
||||
patcher_path.chmod(patcher_path.stat().st_mode | stat.S_IEXEC)
|
||||
|
||||
# Run patcher
|
||||
if progress_callback:
|
||||
progress_callback("Running 4GB patcher...")
|
||||
|
||||
result = subprocess.run(
|
||||
[str(patcher_path)],
|
||||
cwd=str(self.game_root),
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=60
|
||||
)
|
||||
|
||||
# Check if backup was created (indicates success)
|
||||
if backup_exe.exists():
|
||||
logger.info("4GB patch applied successfully")
|
||||
return True, "4GB patch applied successfully"
|
||||
else:
|
||||
logger.warning(f"Patcher output: {result.stdout}")
|
||||
if result.stderr:
|
||||
logger.warning(f"Patcher errors: {result.stderr}")
|
||||
return False, "Patcher ran but FalloutNV_backup.exe not created"
|
||||
|
||||
except subprocess.TimeoutExpired:
|
||||
return False, "4GB patcher timed out after 60 seconds"
|
||||
except Exception as e:
|
||||
error_msg = f"Failed to run 4GB patcher: {str(e)}"
|
||||
logger.error(error_msg, exc_info=True)
|
||||
return False, error_msg
|
||||
|
||||
def run_bsa_decompressor(self, progress_callback: Optional[Callable[[str], None]] = None,
|
||||
manual_file_callback: Optional[Callable[[str, str], Optional[Path]]] = None) -> tuple[bool, str]:
|
||||
"""
|
||||
Download FNV BSA Decompressor MPI and run via TTW_Linux_Installer.
|
||||
|
||||
Args:
|
||||
progress_callback: Optional callback for progress updates
|
||||
manual_file_callback: Optional callback for manual file selection
|
||||
Takes (title, instructions) returns Path or None
|
||||
|
||||
Returns:
|
||||
(success: bool, message: str)
|
||||
"""
|
||||
try:
|
||||
# Check if already completed
|
||||
marker_file = self.game_root / ".jackify_bsa_decompressed"
|
||||
if marker_file.exists():
|
||||
logger.info("BSA decompression already completed (marker file exists)")
|
||||
return True, "BSA decompression already completed"
|
||||
|
||||
if not self.ttw_installer_path or not self.ttw_installer_path.exists():
|
||||
logger.warning("TTW_Linux_Installer not found, skipping BSA decompression")
|
||||
return True, "BSA decompression skipped (TTW_Linux_Installer not available)"
|
||||
|
||||
# Check cache first
|
||||
cached_files = list(self.cache_dir.glob("*BSA*.mpi"))
|
||||
if cached_files:
|
||||
mpi_path = cached_files[0]
|
||||
logger.info(f"Using cached BSA Decompressor MPI: {mpi_path}")
|
||||
else:
|
||||
# Also check for exact filename match (handles spaces in filename)
|
||||
exact_path = self.cache_dir / "FNV BSA Decompressor.mpi"
|
||||
if exact_path.exists():
|
||||
mpi_path = exact_path
|
||||
logger.info(f"Using cached BSA Decompressor MPI: {mpi_path}")
|
||||
else:
|
||||
# Try to download from Nexus
|
||||
success, mpi_path, msg = self.download_service.download_latest_file(
|
||||
self.GAME_DOMAIN,
|
||||
self.FNV_BSA_DECOMPRESSOR_MOD_ID,
|
||||
self.cache_dir,
|
||||
file_name_filter="mpi",
|
||||
progress_callback=progress_callback
|
||||
)
|
||||
|
||||
if not success:
|
||||
# Download failed - offer manual download
|
||||
logger.warning(f"Automatic download failed: {msg}")
|
||||
|
||||
if not manual_file_callback:
|
||||
return False, f"Failed to download BSA Decompressor MPI: {msg}\n\nPlease download manually from: https://www.nexusmods.com/newvegas/mods/65854"
|
||||
|
||||
instructions = (
|
||||
"Automatic download failed (requires Nexus Premium).\n\n"
|
||||
"Please download the FNV BSA Decompressor manually:\n"
|
||||
"1. Visit: https://www.nexusmods.com/newvegas/mods/65854\n"
|
||||
"2. Download the .mpi file\n"
|
||||
"3. Select the downloaded file below"
|
||||
)
|
||||
|
||||
mpi_path = manual_file_callback("BSA Decompressor Required", instructions)
|
||||
|
||||
if not mpi_path or not mpi_path.exists():
|
||||
return False, "BSA Decompressor MPI file not provided"
|
||||
|
||||
# Validate it's an MPI file
|
||||
if not mpi_path.suffix.lower() == '.mpi':
|
||||
return False, f"Selected file is not an MPI file: {mpi_path}"
|
||||
|
||||
# Copy to cache for future use
|
||||
cached_path = self.cache_dir / mpi_path.name
|
||||
shutil.copy2(mpi_path, cached_path)
|
||||
mpi_path = cached_path
|
||||
logger.info(f"Using manually selected BSA Decompressor MPI: {mpi_path}")
|
||||
|
||||
# Create temp output directory
|
||||
with tempfile.TemporaryDirectory() as temp_output:
|
||||
temp_output_path = Path(temp_output)
|
||||
|
||||
# Create config file for TTW_Linux_Installer (handles spaces in paths better)
|
||||
config_file = self.ttw_installer_path.parent / "ttw-config.json"
|
||||
import json
|
||||
config_data = {
|
||||
"FalloutNVRoot": str(self.game_root),
|
||||
"MpiPackagePath": str(mpi_path),
|
||||
"DestinationPath": str(temp_output_path)
|
||||
}
|
||||
with open(config_file, 'w') as f:
|
||||
json.dump(config_data, f, indent=2)
|
||||
logger.debug(f"Created MPI config file: {config_file}")
|
||||
|
||||
# Run via TTW_Linux_Installer
|
||||
if progress_callback:
|
||||
progress_callback("Running BSA decompressor...")
|
||||
|
||||
cmd = [
|
||||
str(self.ttw_installer_path),
|
||||
"--start"
|
||||
]
|
||||
|
||||
logger.info(f"Running BSA decompressor: {' '.join(cmd)}")
|
||||
logger.debug(f"Using config file: {config_file}")
|
||||
logger.debug(f"Config: {json.dumps(config_data, indent=2)}")
|
||||
|
||||
env = get_clean_subprocess_env()
|
||||
|
||||
# Stream output and parse progress
|
||||
import re
|
||||
process = subprocess.Popen(
|
||||
cmd,
|
||||
cwd=str(self.ttw_installer_path.parent),
|
||||
env=env,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.STDOUT,
|
||||
text=True,
|
||||
bufsize=1
|
||||
)
|
||||
|
||||
# Pattern to match progress: "Assets processed: 12345/48649"
|
||||
progress_pattern = re.compile(r'Assets processed: (\d+)/(\d+)')
|
||||
last_progress = None
|
||||
|
||||
# Capture all output for diagnostics
|
||||
all_output = []
|
||||
already_modified_detected = False
|
||||
|
||||
# Stream output line by line
|
||||
for line in process.stdout:
|
||||
line = line.rstrip()
|
||||
all_output.append(line)
|
||||
|
||||
# Check for "already modified" messages
|
||||
if "already" in line.lower() and ("modified" in line.lower() or "decompressed" in line.lower()):
|
||||
already_modified_detected = True
|
||||
logger.info(f"BSA decompressor reports: {line}")
|
||||
|
||||
# Check for progress updates
|
||||
match = progress_pattern.search(line)
|
||||
if match:
|
||||
current = int(match.group(1))
|
||||
total = int(match.group(2))
|
||||
percent = (current / total * 100) if total > 0 else 0
|
||||
progress_msg = f"Decompressing BSA files: {current}/{total} ({percent:.1f}%)"
|
||||
|
||||
# Only send update if progress changed significantly
|
||||
if last_progress is None or current - last_progress >= total // 100:
|
||||
if progress_callback:
|
||||
progress_callback(progress_msg)
|
||||
# Log progress updates (not every single file)
|
||||
logger.debug(f"BSA decompression progress: {current}/{total} ({percent:.1f}%)")
|
||||
last_progress = current
|
||||
|
||||
# Wait for process to complete
|
||||
return_code = process.wait(timeout=600)
|
||||
|
||||
# Log full output for debugging failures
|
||||
if return_code != 0:
|
||||
logger.debug(f"BSA decompressor output:\n" + "\n".join(all_output[-50:])) # Last 50 lines
|
||||
|
||||
# Clean up config file after execution
|
||||
try:
|
||||
if config_file.exists():
|
||||
config_file.unlink()
|
||||
logger.debug(f"Cleaned up config file: {config_file}")
|
||||
except Exception as e:
|
||||
logger.warning(f"Failed to clean up config file: {e}")
|
||||
|
||||
if return_code == 0:
|
||||
# Check if files were actually extracted to temp directory
|
||||
extracted_files = list(temp_output_path.rglob("*"))
|
||||
if extracted_files:
|
||||
logger.info(f"BSA decompression extracted {len(extracted_files)} files")
|
||||
|
||||
# Copy extracted files back to game Data directory
|
||||
data_dir = self.game_root / "Data"
|
||||
copied_count = 0
|
||||
for extracted_file in extracted_files:
|
||||
if extracted_file.is_file():
|
||||
# Preserve relative path structure
|
||||
relative_path = extracted_file.relative_to(temp_output_path)
|
||||
dest_file = data_dir / relative_path
|
||||
dest_file.parent.mkdir(parents=True, exist_ok=True)
|
||||
shutil.copy2(extracted_file, dest_file)
|
||||
copied_count += 1
|
||||
|
||||
logger.info(f"Copied {copied_count} decompressed files to {data_dir}")
|
||||
|
||||
# Create marker file to indicate completion
|
||||
marker_file = self.game_root / ".jackify_bsa_decompressed"
|
||||
marker_file.touch()
|
||||
logger.info("BSA decompression completed successfully")
|
||||
return True, "BSA decompression completed successfully"
|
||||
else:
|
||||
# No files extracted - might be already decompressed or failed silently
|
||||
logger.warning("BSA decompressor returned 0 but no files were extracted")
|
||||
# Check if already decompressed by looking for marker
|
||||
marker_file = self.game_root / ".jackify_bsa_decompressed"
|
||||
if marker_file.exists():
|
||||
logger.info("BSA files already decompressed (marker file exists)")
|
||||
return True, "BSA files already decompressed"
|
||||
else:
|
||||
return False, "BSA decompressor completed but no files were extracted"
|
||||
else:
|
||||
# Exit code 1 often means "already decompressed" - check output and marker
|
||||
marker_file = self.game_root / ".jackify_bsa_decompressed"
|
||||
|
||||
# If output explicitly said "already modified/decompressed", treat as success
|
||||
if already_modified_detected:
|
||||
logger.info("BSA decompressor reports files already modified - marking as completed")
|
||||
marker_file.touch()
|
||||
return True, "BSA files already decompressed"
|
||||
|
||||
# Check marker file
|
||||
if marker_file.exists():
|
||||
logger.info("BSA decompressor returned error but marker file exists - assuming already completed")
|
||||
return True, "BSA decompression already completed"
|
||||
|
||||
# Try to provide helpful error message based on exit code and output
|
||||
logger.error(f"BSA decompressor failed with exit code {return_code}")
|
||||
|
||||
error_details = f"BSA decompressor failed with exit code {return_code}."
|
||||
|
||||
if return_code == 1:
|
||||
error_details += (
|
||||
"\n\nThis may indicate the BSA files are already decompressed or modified. "
|
||||
"If you've run this before, the step may have already completed. "
|
||||
"Otherwise, try running the decompressor manually from: "
|
||||
"https://www.nexusmods.com/newvegas/mods/65854"
|
||||
)
|
||||
else:
|
||||
error_details += (
|
||||
f"\n\nPlease check that:\n"
|
||||
f"1. Fallout New Vegas is properly installed at: {self.game_root}\n"
|
||||
f"2. The BSA files exist in the Data directory\n"
|
||||
f"3. You have write permissions to the game directory\n\n"
|
||||
f"You can complete this step manually using the guide at:\n"
|
||||
f"https://vivanewvegas.moddinglinked.com/wabbajack.html"
|
||||
)
|
||||
|
||||
return False, error_details
|
||||
|
||||
except subprocess.TimeoutExpired:
|
||||
return False, "BSA decompression timed out after 10 minutes"
|
||||
except Exception as e:
|
||||
error_msg = f"Failed to run BSA decompressor: {str(e)}"
|
||||
logger.error(error_msg, exc_info=True)
|
||||
return False, error_msg
|
||||
Reference in New Issue
Block a user