1040 lines
44 KiB
Python
1040 lines
44 KiB
Python
"""Commands for installer scripts"""
|
|
|
|
import glob
|
|
import json
|
|
import multiprocessing
|
|
import os
|
|
import shlex
|
|
import shutil
|
|
from gettext import gettext as _
|
|
from pathlib import Path
|
|
|
|
from lutris import runtime
|
|
from lutris.cache import is_file_in_custom_cache
|
|
from lutris.exceptions import MissingExecutableError, UnspecifiedVersionError
|
|
from lutris.installer.errors import ScriptingError
|
|
from lutris.installer.installer import LutrisInstaller
|
|
from lutris.monitored_command import MonitoredCommand
|
|
from lutris.runners import InvalidRunnerError, import_runner, import_task
|
|
from lutris.runners.wine import wine
|
|
from lutris.util import extract, linux, selective_merge, system
|
|
from lutris.util.fileio import EvilConfigParser, MultiOrderedDict
|
|
from lutris.util.gog import apply_gog_config, find_gog_config_dir
|
|
from lutris.util.gogdl import clear_stale_manifest, run_gogdl
|
|
from lutris.util.jobs import schedule_repeating_at_idle
|
|
from lutris.util.log import logger
|
|
from lutris.util.wine.wine import WINE_DEFAULT_ARCH, get_default_wine_version, get_wine_path_for_version
|
|
|
|
|
|
class CommandsMixin:
|
|
"""The directives for the `installer:` part of the install script."""
|
|
|
|
# pylint: disable=no-member
|
|
installer: LutrisInstaller = NotImplemented
|
|
|
|
def get_wine_path(self) -> str:
|
|
"""Return absolute path of wine version used during the installation, but
|
|
None if the wine exe can't be located."""
|
|
runner = self.get_runner_class(self.installer.runner)()
|
|
version = runner.get_installer_runner_version(self.installer, use_runner_config=False)
|
|
if version:
|
|
wine_path = get_wine_path_for_version(version)
|
|
return wine_path
|
|
|
|
# Special case that lets the Wine configuration explicit specify the path
|
|
# to the Wine executable, not just a version number.
|
|
if self.installer.runner == "wine":
|
|
try:
|
|
config_version, runner_config = wine.get_runner_version_and_config()
|
|
wine_path = get_wine_path_for_version(config_version, config=runner_config.runner_level["wine"])
|
|
return wine_path
|
|
except UnspecifiedVersionError:
|
|
pass
|
|
|
|
version = get_default_wine_version()
|
|
wine_path = get_wine_path_for_version(version)
|
|
return wine_path
|
|
|
|
def get_runner_class(self, runner_name):
|
|
"""Runner the runner class from its name"""
|
|
try:
|
|
runner = import_runner(runner_name)
|
|
except InvalidRunnerError as err:
|
|
raise ScriptingError(_("Invalid runner provided %s") % runner_name) from err
|
|
return runner
|
|
|
|
@staticmethod
|
|
def _check_required_params(params, command_data, command_name):
|
|
"""Verify presence of a list of parameters required by a command."""
|
|
if isinstance(params, str):
|
|
params = [params]
|
|
for param in params:
|
|
if isinstance(param, tuple):
|
|
param_present = False
|
|
for key in param:
|
|
if key in command_data:
|
|
param_present = True
|
|
if not param_present:
|
|
raise ScriptingError(
|
|
_("One of {params} parameter is mandatory for the {cmd} command").format(
|
|
params=_(" or ").join(param), cmd=command_name
|
|
),
|
|
faulty_data=command_data,
|
|
)
|
|
else:
|
|
if param not in command_data:
|
|
raise ScriptingError(
|
|
_("The {param} parameter is mandatory for the {cmd} command").format(
|
|
param=param, cmd=command_name
|
|
),
|
|
faulty_data=command_data,
|
|
)
|
|
|
|
def chmodx(self, filename):
|
|
"""Make filename executable"""
|
|
filename = self._substitute(filename)
|
|
if not system.path_exists(filename):
|
|
raise ScriptingError(_("Invalid file '%s'. Can't make it executable") % filename)
|
|
system.make_executable(filename)
|
|
|
|
def execute(self, data):
|
|
"""Run an executable file."""
|
|
args = []
|
|
terminal = None
|
|
working_dir = None
|
|
env = {}
|
|
if isinstance(data, dict):
|
|
self._check_required_params([("file", "command")], data, "execute")
|
|
if "command" in data and "file" in data:
|
|
raise ScriptingError(
|
|
_("Parameters file and command can't be used at the same time for the execute command"),
|
|
faulty_data=data,
|
|
)
|
|
|
|
# Accept return codes other than 0
|
|
if "return_code" in data:
|
|
return_code = data.pop("return_code")
|
|
else:
|
|
return_code = "0"
|
|
|
|
exec_path = data.get("file", "")
|
|
command = data.get("command", "")
|
|
if not isinstance(command, str):
|
|
raise ScriptingError(
|
|
_("The command parameter of the execute command must be a string"),
|
|
faulty_data=data,
|
|
)
|
|
args_string = data.get("args", "")
|
|
for arg in shlex.split(args_string):
|
|
args.append(self._substitute(arg))
|
|
terminal = data.get("terminal")
|
|
working_dir = data.get("working_dir")
|
|
if working_dir:
|
|
working_dir = self._substitute(working_dir)
|
|
if not data.get("disable_runtime"):
|
|
# Possibly need to handle prefer_system_libs here
|
|
env.update(runtime.get_env())
|
|
|
|
# Loading environment variables set in the script
|
|
env.update(self.script_env)
|
|
|
|
# Environment variables can also be passed to the execute command
|
|
local_env = data.get("env") or {}
|
|
env.update({key: self._substitute(value) for key, value in local_env.items()})
|
|
include_processes = shlex.split(data.get("include_processes", ""))
|
|
exclude_processes = shlex.split(data.get("exclude_processes", ""))
|
|
elif isinstance(data, str):
|
|
command = data
|
|
include_processes = []
|
|
exclude_processes = []
|
|
else:
|
|
raise ScriptingError(_("No parameters supplied to execute command."), faulty_data=data)
|
|
|
|
if command:
|
|
exec_path = "bash"
|
|
args = ["-c", self._get_file_path(command.strip())]
|
|
include_processes.append("bash")
|
|
else:
|
|
# Determine whether 'file' value is a file id or a path
|
|
exec_path = self._get_file_path(exec_path)
|
|
if system.path_exists(exec_path) and not system.is_executable(exec_path):
|
|
logger.warning("Making %s executable", exec_path)
|
|
system.make_executable(exec_path)
|
|
|
|
try:
|
|
exec_abs_path = system.find_required_executable(exec_path)
|
|
except MissingExecutableError as ex:
|
|
raise ScriptingError(_("Unable to find executable %s") % exec_path) from ex
|
|
|
|
if terminal:
|
|
terminal = linux.get_default_terminal()
|
|
if working_dir and not os.path.isdir(working_dir):
|
|
logger.warning(
|
|
"Working directory %s doesn't exist or is not a directory, using target path instead", working_dir
|
|
)
|
|
working_dir = None
|
|
if not working_dir:
|
|
working_dir = self.target_path
|
|
|
|
command = MonitoredCommand(
|
|
[exec_abs_path] + args,
|
|
env=env,
|
|
term=terminal,
|
|
cwd=working_dir,
|
|
include_processes=include_processes,
|
|
exclude_processes=exclude_processes,
|
|
)
|
|
command.accepted_return_code = return_code
|
|
command.start()
|
|
self.interpreter_ui_delegate.attach_log(command)
|
|
schedule_repeating_at_idle(self._monitor_task, command, interval_seconds=1.0)
|
|
return "STOP"
|
|
|
|
def extract(self, data):
|
|
"""Extract a file, guessing the compression method."""
|
|
self._check_required_params([("file", "src")], data, "extract")
|
|
src_param = data.get("file") or data.get("src")
|
|
filespec = self._get_file_path(src_param)
|
|
|
|
if os.path.exists(filespec):
|
|
filenames = [filespec]
|
|
else:
|
|
filenames = glob.glob(filespec)
|
|
|
|
if not filenames:
|
|
raise ScriptingError(_("%s does not exist") % filespec)
|
|
if "dst" in data:
|
|
dest_path = self._substitute(data["dst"])
|
|
else:
|
|
dest_path = self.target_path
|
|
for filename in filenames:
|
|
msg = _("Extracting %s") % os.path.basename(filename)
|
|
logger.debug(msg)
|
|
self.interpreter_ui_delegate.report_status(msg)
|
|
merge_single = "nomerge" not in data
|
|
extractor = data.get("format")
|
|
logger.debug("extracting file %s to %s", filename, dest_path)
|
|
self._killable_process(extract.extract_archive, filename, dest_path, merge_single, extractor)
|
|
logger.debug("Extract done")
|
|
|
|
def input_menu(self, data):
|
|
"""Display an input request as a dropdown menu with options."""
|
|
self._check_required_params("options", data, "input_menu")
|
|
identifier = data.get("id")
|
|
alias = "INPUT_%s" % identifier if identifier else None
|
|
options = data["options"]
|
|
preselect = self._substitute(data.get("preselect", ""))
|
|
self.interpreter_ui_delegate.begin_input_menu(alias, options, preselect, self._on_input_menu_validated)
|
|
return "STOP"
|
|
|
|
def _on_input_menu_validated(self, alias, menu):
|
|
choosen_option = menu.get_active_id()
|
|
if choosen_option:
|
|
self.user_inputs.append({"alias": alias, "value": choosen_option})
|
|
self._iter_commands()
|
|
else:
|
|
raise RuntimeError("A required input option was not selected, so the installation can't continue.")
|
|
|
|
def insert_disc(self, data):
|
|
"""Request user to insert an optical disc"""
|
|
self._check_required_params("requires", data, "insert_disc")
|
|
requires = data.get("requires")
|
|
message = data.get(
|
|
"message",
|
|
_(
|
|
"Insert or mount game disc and click Autodetect or\n"
|
|
"use Browse if the disc is mounted on a non standard location."
|
|
),
|
|
)
|
|
message += (
|
|
_(
|
|
"\n\nLutris is looking for a mounted disk drive or image \n"
|
|
"containing the following file or folder:\n"
|
|
"<i>%s</i>"
|
|
)
|
|
% requires
|
|
)
|
|
self.interpreter_ui_delegate.begin_disc_prompt(message, requires, self.installer, self._find_matching_disc)
|
|
return "STOP"
|
|
|
|
def _find_matching_disc(self, _widget, requires, extra_path=None):
|
|
if extra_path:
|
|
drives = [extra_path]
|
|
else:
|
|
drives = system.get_mounted_discs()
|
|
for drive in drives:
|
|
required_abspath = os.path.join(drive, requires)
|
|
required_abspath = system.fix_path_case(required_abspath)
|
|
if system.path_exists(required_abspath):
|
|
logger.debug("Found %s on cdrom %s", requires, drive)
|
|
self.game_disc = drive
|
|
self._iter_commands()
|
|
return
|
|
|
|
raise RuntimeError(_("The required file '%s' could not be located.") % requires)
|
|
|
|
def mkdir(self, directory):
|
|
"""Create directory"""
|
|
directory = self._substitute(directory)
|
|
try:
|
|
os.makedirs(directory)
|
|
except OSError:
|
|
logger.debug("Directory %s already exists", directory)
|
|
else:
|
|
logger.debug("Created directory %s", directory)
|
|
|
|
def merge(self, params):
|
|
"""Merge the contents given by src to destination folder dst"""
|
|
self._check_required_params(["src", "dst"], params, "merge")
|
|
src, dst = self._get_move_paths(params)
|
|
logger.debug("Merging %s into %s", src, dst)
|
|
if not os.path.exists(src):
|
|
if params.get("optional"):
|
|
logger.info("Optional path %s not present", src)
|
|
return
|
|
raise ScriptingError(_("Source does not exist: %s") % src, params)
|
|
os.makedirs(dst, exist_ok=True)
|
|
if os.path.isfile(src):
|
|
# If single file, copy it and change reference in game file so it
|
|
# can be used as executable. Skip copying if the source is the same
|
|
# as destination.
|
|
if os.path.dirname(src) != dst:
|
|
self._killable_process(shutil.copy, src, dst)
|
|
if params["src"] in self.game_files.keys():
|
|
self.game_files[params["src"]] = os.path.join(dst, os.path.basename(src))
|
|
return
|
|
self._killable_process(system.merge_folders, src, dst)
|
|
# Optionally drop the source tree once its contents have been merged.
|
|
# Useful when the source is throwaway scaffolding (e.g. the
|
|
# installDirectory subfolder a gogdl depot download creates) that
|
|
# would otherwise duplicate the merged data.
|
|
if params.get("remove_source") and os.path.isdir(src):
|
|
self._killable_process(shutil.rmtree, src)
|
|
|
|
def copy(self, params):
|
|
"""Alias for merge"""
|
|
self.merge(params)
|
|
|
|
def move(self, params):
|
|
"""Move a file or directory into a destination folder."""
|
|
self._check_required_params(["src", "dst"], params, "move")
|
|
src, dst = self._get_move_paths(params)
|
|
logger.debug("Moving %s to %s", src, dst)
|
|
if not os.path.exists(src):
|
|
if params.get("optional"):
|
|
logger.info("Optional path %s not present", src)
|
|
return
|
|
raise ScriptingError(_("Invalid source for 'move' operation: %s") % src)
|
|
|
|
if os.path.isfile(src):
|
|
if os.path.dirname(src) == dst:
|
|
logger.info("Source file is the same as destination, skipping")
|
|
return
|
|
|
|
if os.path.exists(os.path.join(dst, os.path.basename(src))):
|
|
# May not be the best choice, but it's the safest.
|
|
# Maybe should display confirmation dialog (Overwrite / Skip) ?
|
|
logger.info("Destination file exists, skipping")
|
|
return
|
|
try:
|
|
if is_file_in_custom_cache(src):
|
|
action = shutil.copy
|
|
else:
|
|
action = shutil.move
|
|
self._killable_process(action, src, dst)
|
|
except shutil.Error as err:
|
|
raise ScriptingError(_("Can't move {src} \nto destination {dst}").format(src=src, dst=dst)) from err
|
|
|
|
def rename(self, params):
|
|
"""Rename file or folder."""
|
|
self._check_required_params(["src", "dst"], params, "rename")
|
|
src, dst = self._get_move_paths(params)
|
|
if not os.path.exists(src):
|
|
raise ScriptingError(_("Rename error, source path does not exist: %s") % src)
|
|
if os.path.isdir(dst):
|
|
try:
|
|
os.rmdir(dst) # Remove if empty
|
|
except OSError:
|
|
pass
|
|
if os.path.exists(dst):
|
|
raise ScriptingError(_("Rename error, destination already exists: %s") % src)
|
|
dst_dir = os.path.dirname(dst)
|
|
|
|
# Pre-move on dest filesystem to avoid error with
|
|
# os.rename through different filesystems
|
|
temp_dir = os.path.join(dst_dir, "lutris_rename_temp")
|
|
os.makedirs(temp_dir)
|
|
self._killable_process(shutil.move, src, temp_dir)
|
|
src = os.path.join(temp_dir, os.path.basename(src))
|
|
os.renames(src, dst)
|
|
|
|
def _get_move_paths(self, params):
|
|
"""Process raw 'src' and 'dst' data."""
|
|
try:
|
|
src_ref = params["src"]
|
|
except KeyError as err:
|
|
raise ScriptingError(_("Missing parameter src")) from err
|
|
src = self.game_files.get(src_ref) or self._substitute(src_ref)
|
|
if not src:
|
|
raise ScriptingError(_("Wrong value for 'src' param"), src_ref)
|
|
dst_ref = params["dst"]
|
|
dst = self._substitute(dst_ref)
|
|
if not dst:
|
|
raise ScriptingError(_("Wrong value for 'dst' param"), dst_ref)
|
|
return src.rstrip("/"), dst.rstrip("/")
|
|
|
|
def substitute_vars(self, data):
|
|
"""Substitute variable names found in given file."""
|
|
self._check_required_params("file", data, "substitute_vars")
|
|
filename = self._substitute(data["file"])
|
|
logger.debug("Substituting variables for file %s", filename)
|
|
tmp_filename = filename + ".tmp"
|
|
with open(filename, "r", encoding="utf-8") as source_file:
|
|
with open(tmp_filename, "w", encoding="utf-8") as dest_file:
|
|
line = "."
|
|
while line:
|
|
line = source_file.readline()
|
|
line = self._substitute(line)
|
|
dest_file.write(line)
|
|
os.rename(tmp_filename, filename)
|
|
|
|
def _get_task_runner_and_name(self, task_name):
|
|
if "." in task_name:
|
|
# Run a task from a different runner
|
|
# than the one for this installer
|
|
runner_name, task_name = task_name.split(".")
|
|
else:
|
|
runner_name = self.installer.runner
|
|
|
|
task_name = task_name.replace("-", "_")
|
|
# Prevent private functions from being accessed as commands
|
|
task_name = task_name.strip("_")
|
|
return runner_name, task_name
|
|
|
|
def task(self, data):
|
|
"""Directive triggering another function specific to a runner.
|
|
|
|
The 'name' parameter is mandatory. If 'args' is provided it will be
|
|
passed to the runner task.
|
|
"""
|
|
self._check_required_params("name", data, "task")
|
|
runner_name, task_name = self._get_task_runner_and_name(data.pop("name"))
|
|
|
|
# Accept return codes other than 0
|
|
if "return_code" in data:
|
|
return_code = data.pop("return_code")
|
|
else:
|
|
return_code = "0"
|
|
|
|
if runner_name.startswith("wine"):
|
|
data["wine_path"] = self.get_wine_path()
|
|
data["prefix"] = data.get("prefix") or self.installer.script.get("game", {}).get("prefix") or "$GAMEDIR"
|
|
data["arch"] = data.get("arch") or self.installer.script.get("game", {}).get("arch") or WINE_DEFAULT_ARCH
|
|
if task_name == "wineexec":
|
|
data["env"] = self.script_env
|
|
|
|
for key in data:
|
|
value = data[key]
|
|
if value is None and task_name == "set_regedit":
|
|
continue
|
|
if isinstance(value, dict):
|
|
for inner_key in value:
|
|
value[inner_key] = self._substitute(value[inner_key])
|
|
elif isinstance(value, list):
|
|
for index, elem in enumerate(value):
|
|
value[index] = self._substitute(elem)
|
|
else:
|
|
value = self._substitute(data[key])
|
|
data[key] = value
|
|
|
|
task = import_task(runner_name, task_name)
|
|
command = task(**data)
|
|
if isinstance(command, MonitoredCommand):
|
|
# Monitor thread and continue when task has executed
|
|
self.interpreter_ui_delegate.attach_log(command)
|
|
command.accepted_return_code = return_code
|
|
schedule_repeating_at_idle(self._monitor_task, command, interval_seconds=1.0)
|
|
return "STOP"
|
|
return None
|
|
|
|
def _monitor_task(self, command):
|
|
if not command.is_running:
|
|
logger.debug("Return code: %s", command.return_code)
|
|
if command.return_code not in (str(command.accepted_return_code), "0"):
|
|
raise ScriptingError(_("Command exited with code %s") % command.return_code)
|
|
|
|
self._iter_commands()
|
|
return False
|
|
return True # keep checking
|
|
|
|
def write_file(self, params):
|
|
"""Write text to a file."""
|
|
self._check_required_params(["file", "content"], params, "write_file")
|
|
|
|
# Get file
|
|
dest_file_path = self._get_file_path(params["file"])
|
|
|
|
# Create dir if necessary
|
|
basedir = os.path.dirname(dest_file_path)
|
|
os.makedirs(basedir, exist_ok=True)
|
|
|
|
mode = params.get("mode", "w")
|
|
if not mode.startswith(("a", "w")):
|
|
raise ScriptingError(_("Wrong value for write_file mode: '%s'") % mode)
|
|
|
|
with open(dest_file_path, mode, encoding="utf-8") as dest_file:
|
|
dest_file.write(self._substitute(params["content"]))
|
|
|
|
def write_json(self, params):
|
|
"""Write data into a json file."""
|
|
self._check_required_params(["file", "data"], params, "write_json")
|
|
|
|
# Get file
|
|
filename = self._get_file_path(params["file"])
|
|
|
|
# Create dir if necessary
|
|
basedir = os.path.dirname(filename)
|
|
os.makedirs(basedir, exist_ok=True)
|
|
|
|
merge = params.get("merge", True)
|
|
|
|
# create an empty file if it doesn't exist
|
|
Path(filename).touch(exist_ok=True)
|
|
|
|
with open(filename, "r+" if merge else "w", encoding="utf-8") as json_file:
|
|
json_data = {}
|
|
if merge:
|
|
try:
|
|
json_data = json.load(json_file)
|
|
except ValueError:
|
|
logger.error("Failed to parse JSON from file %s", filename)
|
|
|
|
json_data = selective_merge(json_data, params.get("data", {}))
|
|
json_file.seek(0)
|
|
json_file.truncate()
|
|
json_file.write(json.dumps(json_data, indent=2))
|
|
|
|
def write_config(self, params):
|
|
"""Write a key-value pair into an INI type config file."""
|
|
if params.get("data"):
|
|
self._check_required_params(["file", "data"], params, "write_config")
|
|
else:
|
|
self._check_required_params(["file", "section", "key", "value"], params, "write_config")
|
|
# Get file
|
|
config_file_path = self._get_file_path(params["file"])
|
|
|
|
# Create dir if necessary
|
|
basedir = os.path.dirname(config_file_path)
|
|
os.makedirs(basedir, exist_ok=True)
|
|
|
|
merge = params.get("merge", True)
|
|
|
|
parser = EvilConfigParser(allow_no_value=True, dict_type=MultiOrderedDict, strict=False)
|
|
parser.optionxform = str # Preserve text case
|
|
if merge:
|
|
parser.read(config_file_path)
|
|
|
|
data = {}
|
|
if params.get("data"):
|
|
data = params["data"]
|
|
else:
|
|
data[params["section"]] = {}
|
|
data[params["section"]][params["key"]] = params["value"]
|
|
|
|
for section, keys in data.items():
|
|
if not parser.has_section(section):
|
|
parser.add_section(section)
|
|
for key, value in keys.items():
|
|
value = self._substitute(value)
|
|
parser.set(section, key, value)
|
|
|
|
with open(config_file_path, "wb") as config_file:
|
|
parser.write(config_file)
|
|
|
|
def _get_file_path(self, fileid):
|
|
file_path = self.game_files.get(fileid)
|
|
if not file_path:
|
|
file_path = self._substitute(fileid)
|
|
return file_path
|
|
|
|
def _killable_process(self, func, *args, **kwargs):
|
|
"""Run function `func` in a separate, killable process."""
|
|
with multiprocessing.Pool(1) as process:
|
|
result_obj = process.apply_async(func, args, kwargs)
|
|
self.abort_current_task = process.terminate
|
|
result = result_obj.get() # Wait process end & re-raise exceptions
|
|
self.abort_current_task = None
|
|
logger.debug("Process %s returned: %s", func, result)
|
|
return result
|
|
|
|
def _extract_innosetup(self, file_id):
|
|
self.extract({"src": file_id, "dst": "$GAMEDIR", "extractor": "innoextract"})
|
|
app_path = os.path.join(self.target_path, "app")
|
|
if system.path_exists(app_path):
|
|
for app_content in os.listdir(app_path):
|
|
source_path = os.path.join(app_path, app_content)
|
|
if os.path.exists(os.path.join(self.target_path, app_content)):
|
|
self.merge({"src": source_path, "dst": self.target_path})
|
|
else:
|
|
self.move({"src": source_path, "dst": self.target_path})
|
|
support_path = os.path.join(self.target_path, "__support/app")
|
|
if system.path_exists(support_path):
|
|
self.merge({"src": support_path, "dst": self.target_path})
|
|
|
|
def _get_scummvm_arguments(self, gog_config_path):
|
|
"""Return a ScummVM configuration from the GOG config files"""
|
|
with open(gog_config_path, encoding="utf-8") as gog_config_file:
|
|
gog_config = json.loads(gog_config_file.read())
|
|
game_tasks = [task for task in gog_config["playTasks"] if task["category"] == "game"]
|
|
arguments = game_tasks[0]["arguments"]
|
|
game_id = arguments.split()[-1]
|
|
arguments = " ".join(arguments.split()[:-1])
|
|
base_dir = os.path.dirname(gog_config_path)
|
|
return {"game_id": game_id, "path": base_dir, "args": arguments}
|
|
|
|
def autosetup_gog_game(self, file_id, silent=False):
|
|
"""Automatically guess the best way to install a GOG game by inspecting its contents.
|
|
This chooses the right runner (DOSBox, Wine) for Windows game files.
|
|
Linux setup files don't use innosetup, they can be unzipped instead.
|
|
"""
|
|
file_path = self.game_files[file_id]
|
|
file_list = extract.get_innoextract_list(file_path)
|
|
dosbox_found = False
|
|
scummvm_found = False
|
|
windows_override_found = False # DOS games that also have a Windows executable
|
|
for filename in file_list:
|
|
if "dosbox.exe" in filename.casefold():
|
|
dosbox_found = True
|
|
if "scummvm.exe" in filename.casefold():
|
|
scummvm_found = True
|
|
if "_some_windows.exe" in filename.casefold():
|
|
# There's not a good way to handle exceptions without extracting the .info file
|
|
# before extracting the game. Added for Quake but GlQuake.exe doesn't run on modern wine
|
|
windows_override_found = True
|
|
if dosbox_found and not windows_override_found:
|
|
self._extract_innosetup(file_id)
|
|
if "DOSBOX" in os.listdir(self.target_path):
|
|
dosbox_config = {
|
|
"working_dir": "$GAMEDIR/DOSBOX",
|
|
}
|
|
else:
|
|
dosbox_config = {}
|
|
single_conf = None
|
|
config_file = None
|
|
for filename in os.listdir(self.target_path):
|
|
if filename == "dosbox.conf":
|
|
dosbox_config["main_file"] = filename
|
|
elif filename.endswith("_single.conf"):
|
|
single_conf = filename
|
|
elif filename.endswith(".conf"):
|
|
config_file = filename
|
|
if single_conf:
|
|
dosbox_config["main_file"] = single_conf
|
|
if config_file:
|
|
if dosbox_config.get("main_file"):
|
|
dosbox_config["config_file"] = config_file
|
|
else:
|
|
dosbox_config["main_file"] = config_file
|
|
self.installer.script["game"] = dosbox_config
|
|
self.installer.runner = "dosbox"
|
|
elif scummvm_found:
|
|
self._extract_innosetup(file_id)
|
|
for filename in os.listdir(self.target_path):
|
|
if filename.startswith("goggame") and filename.endswith(".info"):
|
|
arguments = self._get_scummvm_arguments(os.path.join(self.target_path, filename))
|
|
if not arguments:
|
|
raise RuntimeError("Unable to get ScummVM arguments")
|
|
logger.info("ScummVM config: %s", arguments)
|
|
self.installer.script["game"] = arguments
|
|
self.installer.runner = "scummvm"
|
|
else:
|
|
args = "/SP- /NOCANCEL"
|
|
if silent:
|
|
args += " /SUPPRESSMSGBOXES /VERYSILENT /NOGUI"
|
|
self.installer.post_install_hooks.append(apply_gog_config)
|
|
return self.task({"name": "wineexec", "prefix": "$GAMEDIR", "executable": file_id, "args": args})
|
|
|
|
def _get_dosbox_arguments(self, zoom_bat_path):
|
|
"""Return the arguments for the DOSBox executable from a ZOOM Plaform installer"""
|
|
with open(zoom_bat_path, encoding="utf-8") as zoom_bat_file:
|
|
zoom_bat = zoom_bat_file.read()
|
|
|
|
# Find the line that starts with ".\DOSBOX\dosbox.exe" and extract the arguments
|
|
lines = zoom_bat.splitlines()
|
|
arguments = []
|
|
for line in lines:
|
|
if line.startswith(".\\DOSBOX\\dosbox.exe"):
|
|
# Extract the arguments from the line
|
|
arguments = line.split(" ", 1)[1].strip()
|
|
break
|
|
|
|
return arguments
|
|
|
|
def autosetup_zoom_platform(self, file_id, silent=False):
|
|
"""Automatically guess the best way to install a ZOOM Platofrm game by inspecting its contents.
|
|
This chooses the right runner (DOSBox, Wine) for Windows game files.
|
|
Linux setup files don't use innosetup, they can be unzipped instead.
|
|
"""
|
|
file_path = self.game_files[file_id]
|
|
file_list = extract.get_innoextract_list(file_path)
|
|
dosbox_found = False
|
|
scummvm_found = False
|
|
for filename in file_list:
|
|
if "dosbox.exe" in filename.casefold():
|
|
dosbox_found = True
|
|
if "scummvm.exe" in filename.casefold():
|
|
scummvm_found = True
|
|
if dosbox_found:
|
|
self._extract_innosetup(file_id)
|
|
dosbox_config = {}
|
|
single_conf = None
|
|
config_file = None
|
|
for filename in os.listdir(self.target_path):
|
|
if filename == "dosbox.conf":
|
|
dosbox_config["main_file"] = filename
|
|
elif filename.endswith("_single.conf"):
|
|
single_conf = filename
|
|
elif filename.endswith(".conf"):
|
|
config_file = filename
|
|
if single_conf:
|
|
dosbox_config["main_file"] = single_conf
|
|
if config_file:
|
|
if dosbox_config.get("main_file"):
|
|
dosbox_config["config_file"] = config_file
|
|
else:
|
|
dosbox_config["main_file"] = config_file
|
|
|
|
for filename in os.listdir(self.target_path):
|
|
if filename.startswith("Launch") and filename.endswith(".bat"):
|
|
arguments = self._get_dosbox_arguments(os.path.join(self.target_path, filename))
|
|
|
|
if len(arguments) > 0:
|
|
# Add the arguments to the dosbox config
|
|
dosbox_config["args"] = arguments
|
|
# Remove "config_file" from the dosbox config
|
|
dosbox_config.pop("config_file", None)
|
|
|
|
self.installer.script["game"] = dosbox_config
|
|
self.installer.runner = "dosbox"
|
|
elif scummvm_found:
|
|
self._extract_innoextract_setup(file_id)
|
|
arguments = {"path": os.path.join(self.target_path, "Data"), "args": "--auto-detect"}
|
|
logger.info("ScummVM config: %s", arguments)
|
|
self.installer.script["game"] = arguments
|
|
self.installer.runner = "scummvm"
|
|
else:
|
|
args = "/SP- /NOCANCEL"
|
|
if silent:
|
|
args += " /SUPPRESSMSGBOXES /VERYSILENT /NOGUI"
|
|
return self.task({"name": "wineexec", "prefix": "$GAMEDIR", "executable": file_id, "args": args})
|
|
|
|
def autosetup_amazon(self, file_and_dir_dict):
|
|
files = file_and_dir_dict["files"]
|
|
directories = file_and_dir_dict["directories"]
|
|
|
|
# create directories
|
|
for directory in directories:
|
|
self.mkdir(f"$GAMEDIR/drive_c/game/{directory}")
|
|
|
|
# move installed files from CACHE to game folder
|
|
for file_hash, file_data in files.items():
|
|
if file_hash not in self.game_files:
|
|
logger.warning("Amazon: Missing file hash %s (expected at %s)", file_hash, file_data["paths"][0])
|
|
continue
|
|
|
|
source_file = self.game_files[file_hash]
|
|
paths = file_data["paths"]
|
|
|
|
for path in paths:
|
|
dest_path = f"$GAMEDIR/drive_c/game/{path}"
|
|
|
|
abs_dest_path = self._substitute(dest_path)
|
|
if os.path.isdir(abs_dest_path):
|
|
logger.warning("Amazon: Removing conflicting directory %s", abs_dest_path)
|
|
shutil.rmtree(abs_dest_path)
|
|
|
|
self._killable_process(shutil.copyfile, source_file, abs_dest_path)
|
|
|
|
if os.path.exists(source_file):
|
|
os.remove(source_file)
|
|
|
|
def gogdl_setup(self, data):
|
|
"""Download and set up a GOG game via depot/CDN using heroic-gogdl.
|
|
|
|
Params:
|
|
game_id: GOG product ID (defaults to the installer's resolved
|
|
gogid, so scripts only need to set it once; an explicit value
|
|
still wins, e.g. for downloading a DLC)
|
|
platform: "windows" or "linux" (default: based on runner)
|
|
lang: language code (default: system locale)
|
|
dlcs: "all", "none", or comma-separated DLC IDs (optional)
|
|
preserve_manifest: True to keep an existing gogdl manifest (use when
|
|
adding DLCs to an installed base game, so gogdl diffs old→new
|
|
and only downloads the DLC files instead of treating the base
|
|
as missing)
|
|
command: "download" | "update" | "repair" (default: "download")
|
|
"""
|
|
game_id = data.get("game_id") or self.installer.service_appid
|
|
if not game_id:
|
|
raise ScriptingError(
|
|
_("gogdl_setup requires a game_id, or a gogid set on the installer."),
|
|
data,
|
|
)
|
|
|
|
game_id = str(game_id)
|
|
command = data.get("command", "download")
|
|
platform = data.get("platform")
|
|
lang = data.get("lang")
|
|
dlcs_param = data.get("dlcs")
|
|
preserve_manifest = bool(data.get("preserve_manifest"))
|
|
|
|
if not platform:
|
|
platform = "linux" if self.installer.runner == "linux" else "windows"
|
|
|
|
# Determine install path
|
|
# gogdl appends the game's installDirectory (from the depot manifest)
|
|
# to the path on fresh downloads, so we point it at the parent only.
|
|
install_path = self.target_path
|
|
|
|
os.makedirs(install_path, exist_ok=True)
|
|
|
|
# Handle DLC parameter
|
|
if dlcs_param == "all":
|
|
dlcs = True
|
|
elif dlcs_param == "none":
|
|
dlcs = False
|
|
elif dlcs_param:
|
|
dlcs = str(dlcs_param)
|
|
else:
|
|
dlcs = None
|
|
|
|
# Clear stale manifest for fresh downloads so gogdl doesn't
|
|
# think the game is already installed from a previous attempt.
|
|
if command == "download" and not preserve_manifest:
|
|
clear_stale_manifest(game_id, install_path, runner="linux" if platform == "linux" else "wine")
|
|
|
|
msg = _("Downloading via GOG depot...")
|
|
logger.info(msg)
|
|
self.interpreter_ui_delegate.report_status(msg)
|
|
|
|
def progress_callback(progress):
|
|
fraction = progress.percent / 100.0
|
|
parts = []
|
|
if progress.speed_mib > 0:
|
|
parts.append(_("{speed:.1f} MiB/s").format(speed=progress.speed_mib))
|
|
if progress.eta and progress.eta != "00:00:00":
|
|
parts.append(_("ETA: {eta}").format(eta=progress.eta))
|
|
text = " — ".join(parts) if parts else ""
|
|
self.interpreter_ui_delegate.report_progress(fraction, text)
|
|
|
|
try:
|
|
run_gogdl(
|
|
command=command,
|
|
game_id=game_id,
|
|
path=install_path,
|
|
platform=platform,
|
|
lang=lang,
|
|
dlcs=dlcs,
|
|
progress_callback=progress_callback,
|
|
)
|
|
except FileNotFoundError as err:
|
|
raise ScriptingError(_("gogdl runtime component not found. Please update your Lutris runtime.")) from err
|
|
except Exception as err:
|
|
raise ScriptingError(_("GOG depot download failed: %s") % err) from err
|
|
|
|
self.interpreter_ui_delegate.report_progress(None)
|
|
|
|
# Post-install: find where gogdl actually put the files and apply config.
|
|
# For a fresh "download", gogdl appends installDirectory to the path, so we
|
|
# need to discover the actual game directory. For updates and DLC adds
|
|
# against an existing install, we wrote into the install dir directly.
|
|
if command == "download" and not preserve_manifest:
|
|
game_path = self._find_gogdl_game_path(install_path)
|
|
else:
|
|
game_path = install_path
|
|
|
|
self._apply_gogdl_post_install(game_path, platform)
|
|
logger.info("GOG depot download completed for %s", game_id)
|
|
|
|
def _find_gogdl_game_path(self, parent_path):
|
|
"""Find the actual game directory under parent_path after a gogdl download.
|
|
gogdl creates a subdirectory named after the game's installDirectory."""
|
|
if not os.path.isdir(parent_path):
|
|
return parent_path
|
|
entries = [e for e in os.listdir(parent_path) if os.path.isdir(os.path.join(parent_path, e))]
|
|
if len(entries) == 1:
|
|
return os.path.join(parent_path, entries[0])
|
|
# Multiple dirs or none — look for goggame-*.info to find the right one
|
|
for entry in entries:
|
|
candidate = os.path.join(parent_path, entry)
|
|
if any(f.startswith("goggame") and f.endswith(".info") for f in os.listdir(candidate)):
|
|
return candidate
|
|
return parent_path
|
|
|
|
def _apply_gogdl_post_install(self, install_path, platform):
|
|
"""Detect game configuration from goggame-*.info after a depot download."""
|
|
from lutris.util.gog import convert_gog_config_to_lutris, get_gog_config
|
|
|
|
config_path = find_gog_config_dir(install_path)
|
|
if not config_path:
|
|
return
|
|
|
|
gog_config = get_gog_config(config_path)
|
|
if not gog_config:
|
|
return
|
|
|
|
play_tasks = gog_config.get("playTasks", [])
|
|
if not play_tasks:
|
|
return
|
|
|
|
# Detect DOSBox/ScummVM games from playTasks
|
|
for task in play_tasks:
|
|
task_path = task.get("path", "").casefold()
|
|
if "dosbox" in task_path:
|
|
self._configure_dosbox_from_depot(install_path, gog_config)
|
|
return
|
|
if "scummvm" in task_path:
|
|
self._configure_scummvm_from_depot(install_path, gog_config)
|
|
return
|
|
|
|
# Normal game — apply config from goggame-*.info
|
|
lutris_config = convert_gog_config_to_lutris(gog_config, config_path)
|
|
if lutris_config and "game" in self.installer.script:
|
|
# For Linux GOG games, prefer start.sh over the binary from goggame info.
|
|
# GOG's start.sh handles working directory setup and permissions.
|
|
if platform == "linux":
|
|
start_sh = os.path.join(install_path, "start.sh")
|
|
if os.path.isfile(start_sh):
|
|
lutris_config["exe"] = start_sh
|
|
system.make_executable(start_sh)
|
|
# Don't clobber config the script set explicitly: a community
|
|
# installer may ship its own engine (e.g. a native source port
|
|
# run over GOG game data), in which case its exe must win over the
|
|
# Windows exe detected from goggame-*.info. Auto-generated GOG
|
|
# installers leave these unset, so detection still fills them in.
|
|
game_config = self.installer.script["game"]
|
|
for key, value in lutris_config.items():
|
|
game_config.setdefault(key, value)
|
|
|
|
def _configure_dosbox_from_depot(self, install_path, gog_config):
|
|
"""Configure a DOSBox game from depot download using playTasks args.
|
|
|
|
In depot layouts, the -conf paths in goggame-*.info are relative to
|
|
the GOG Galaxy working dir and may not resolve directly. The actual
|
|
conf files are typically in gog-support/{id}/app/. We parse the
|
|
conf filenames from the arguments and search the install tree for them.
|
|
"""
|
|
game_task = next(
|
|
(t for t in gog_config.get("playTasks", []) if t.get("category") == "game"),
|
|
None,
|
|
)
|
|
if not game_task:
|
|
logger.warning("No game playTask found in goggame info for DOSBox game")
|
|
return
|
|
|
|
arguments = game_task.get("arguments", "")
|
|
|
|
# Parse -conf arguments to get the config filenames
|
|
try:
|
|
args = shlex.split(arguments)
|
|
except ValueError:
|
|
args = arguments.split()
|
|
|
|
conf_names = []
|
|
i = 0
|
|
while i < len(args):
|
|
if args[i] == "-conf" and i + 1 < len(args):
|
|
conf_name = os.path.basename(args[i + 1].replace("\\", "/"))
|
|
conf_names.append(conf_name)
|
|
i += 2
|
|
else:
|
|
i += 1
|
|
|
|
# Search install tree for the actual conf files
|
|
conf_files = []
|
|
for conf_name in conf_names:
|
|
for dirpath, _dirnames, filenames in os.walk(install_path):
|
|
if conf_name in filenames:
|
|
conf_files.append(os.path.join(dirpath, conf_name))
|
|
break
|
|
|
|
dosbox_config = {}
|
|
|
|
if conf_files:
|
|
single = next((c for c in conf_files if "_single" in c), None)
|
|
dosbox_config["main_file"] = single or conf_files[0]
|
|
remaining = [c for c in conf_files if c != dosbox_config["main_file"]]
|
|
if remaining:
|
|
dosbox_config["config_file"] = remaining[0]
|
|
|
|
self.installer.script["game"] = dosbox_config
|
|
self.installer.runner = "dosbox"
|
|
|
|
def _configure_scummvm_from_depot(self, install_path, gog_config):
|
|
"""Configure a ScummVM game from depot download using playTasks args."""
|
|
game_task = next(
|
|
(t for t in gog_config.get("playTasks", []) if t.get("category") == "game"),
|
|
None,
|
|
)
|
|
if not game_task:
|
|
return
|
|
arguments = game_task.get("arguments", "")
|
|
parts = arguments.split()
|
|
if parts:
|
|
game_id = parts[-1]
|
|
args = " ".join(parts[:-1])
|
|
self.installer.script["game"] = {
|
|
"game_id": game_id,
|
|
"path": install_path,
|
|
"args": args,
|
|
}
|
|
self.installer.runner = "scummvm"
|
|
|
|
def install_or_extract(self, file_id):
|
|
"""Runs if file is executable or extracts if file is archive"""
|
|
file_path = self._get_file_path(file_id)
|
|
runner = self.installer.runner
|
|
if runner != "wine":
|
|
raise ScriptingError(_("install_or_extract only works with wine!"))
|
|
if file_path.endswith(".exe"):
|
|
params = {"name": "wineexec", "executable": file_id}
|
|
return self.task(params)
|
|
|
|
slug = self.installer.game_slug
|
|
params = {"file": file_id, "dst": f"$GAMEDIR/drive_c/{slug}"}
|
|
return self.extract(params)
|
|
|
|
@staticmethod
|
|
def _is_installer_exe(path):
|
|
"""Check if an .exe file is a known installer (NSIS or Inno Setup)."""
|
|
try:
|
|
with open(path, "rb") as f:
|
|
header = f.read(512 * 1024)
|
|
return b"NullsoftInst" in header or b"Inno Setup" in header
|
|
except OSError:
|
|
return False
|
|
|
|
def extract_or_run(self, data):
|
|
"""Extract an archive. If extraction yields an installer exe (NSIS/Inno),
|
|
run it with Wine instead of placing the files."""
|
|
self._check_required_params("file", data, "extract_or_run")
|
|
file_id = data["file"]
|
|
dst = self._substitute(data.get("dst", "$CACHE"))
|
|
file_path = self._get_file_path(file_id)
|
|
|
|
# Extract the archive (e.g. tar.gz wrapper)
|
|
self._killable_process(extract.extract_archive, file_path, dst, True)
|
|
|
|
# Check if any extracted .exe is an installer
|
|
for entry in os.listdir(dst):
|
|
entry_path = os.path.join(dst, entry)
|
|
if entry.lower().endswith(".exe") and os.path.isfile(entry_path) and self._is_installer_exe(entry_path):
|
|
logger.info("Detected installer exe: %s, running with Wine", entry)
|
|
params = {"name": "wineexec", "executable": entry_path}
|
|
return self.task(params)
|
|
|
|
# No installer found, merge the extracted files to the destination
|
|
merge_dst = self._substitute(data.get("merge_dst", data.get("dst", "$GAMEDIR")))
|
|
if merge_dst != dst:
|
|
self.merge({"src": dst, "dst": merge_dst})
|