1079 lines
31 KiB
Python
1079 lines
31 KiB
Python
"""Command-line interface for installation, updates, repair, and play."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
from contextlib import AbstractContextManager, nullcontext
|
|
from dataclasses import asdict
|
|
from datetime import datetime
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
import time
|
|
from typing import Callable, Sequence
|
|
|
|
from ptdiag.logs import (
|
|
LogFileState,
|
|
newest_current_native_log,
|
|
parse_current_run_log,
|
|
snapshot_native_logs,
|
|
)
|
|
|
|
from . import __version__
|
|
from .command import CommandError, CommandRunner
|
|
from .depot import SyncProgress, sync_managed_game
|
|
from .doctor import inspect_host
|
|
from .download import OFFICIAL_LAUNCHER_URL, fetch_official_launcher
|
|
from .importer import GameImportResult, discover_existing_game, import_existing_game
|
|
from .import_state import game_import_requires_verification, mark_game_import_verified
|
|
from .launcher import (
|
|
LauncherError,
|
|
configure_launcher_prefix,
|
|
find_wine_tools,
|
|
launch_official_updater,
|
|
prepare_launcher_binary,
|
|
)
|
|
from .locking import LauncherPlayLease, launcher_operation_lock, launcher_play_lock
|
|
from .login_diagnostics import inspect_login_environment
|
|
from .login_trace import DEFAULT_TRACE_TIMEOUT_SECONDS, LOGIN_TRACE_DISABLED_REASON
|
|
from .overlay import cache_compatibility_binaries, prepare_runtime_view
|
|
from .paths import AppPaths
|
|
from .runtime import (
|
|
game_runtime_ready,
|
|
prepare_game_runtime,
|
|
run_game,
|
|
select_runtime,
|
|
)
|
|
from .settings import Settings, ensure_settings, load_settings, reset_settings
|
|
from .self_update import (
|
|
check_for_launcher_update,
|
|
install_launcher_update,
|
|
rollback_launcher_update,
|
|
update_check_to_dict,
|
|
verify_launcher_update_channel,
|
|
)
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(
|
|
prog="pristontale-eu",
|
|
description="PristonTale EU all-in-one Linux installer and clean launcher",
|
|
)
|
|
parser.add_argument(
|
|
"--version",
|
|
action="version",
|
|
version=f"%(prog)s {__version__}",
|
|
)
|
|
parser.add_argument("--json", action="store_true", help="emit machine-readable output")
|
|
subparsers = parser.add_subparsers(dest="command")
|
|
|
|
subparsers.add_parser("gui", help="open the native Linux dashboard")
|
|
subparsers.add_parser("doctor", help="check cross-distribution dependencies")
|
|
login_diagnostics = subparsers.add_parser(
|
|
"login-diagnostics",
|
|
help="check the client, runtime, and publisher login route without credentials",
|
|
)
|
|
login_diagnostics.add_argument(
|
|
"--timeout",
|
|
type=float,
|
|
default=5.0,
|
|
help="TCP connection timeout in seconds (default: 5)",
|
|
)
|
|
login_diagnostics.add_argument(
|
|
"--runtime",
|
|
choices=("configured", "auto", "proton7", "proton7-container", "umu"),
|
|
default="configured",
|
|
help=(
|
|
"probe one runtime without changing settings or starting the game "
|
|
"(default: configured)"
|
|
),
|
|
)
|
|
|
|
fetch = subparsers.add_parser("fetch", help="fetch and safely prepare the latest official launcher")
|
|
fetch.add_argument("--dry-run", action="store_true")
|
|
|
|
setup = subparsers.add_parser("setup", help="configure Wine and .NET Desktop for the updater")
|
|
setup.add_argument("--dry-run", action="store_true")
|
|
setup.add_argument("--force-runtime", action="store_true")
|
|
|
|
update = subparsers.add_parser(
|
|
"update",
|
|
help="safely synchronize the official game depot without Wine",
|
|
)
|
|
update.add_argument("--dry-run", action="store_true")
|
|
update.add_argument("--repair", action="store_true", help="hash-check every game file")
|
|
import_options = update.add_mutually_exclusive_group()
|
|
import_options.add_argument(
|
|
"--seed-from",
|
|
type=Path,
|
|
help="import an existing game before updating",
|
|
)
|
|
import_options.add_argument(
|
|
"--no-import",
|
|
action="store_true",
|
|
help="disable automatic existing-game discovery",
|
|
)
|
|
|
|
sync = subparsers.add_parser(
|
|
"sync",
|
|
help="native official depot synchronization without bootstrap discovery",
|
|
)
|
|
sync.add_argument("--dry-run", action="store_true")
|
|
sync.add_argument("--repair", action="store_true", help="hash-check every game file")
|
|
|
|
legacy_update = subparsers.add_parser(
|
|
"wine-update",
|
|
help="legacy official WPF updater (normally unnecessary)",
|
|
)
|
|
legacy_update.add_argument("--dry-run", action="store_true")
|
|
legacy_update.add_argument("--force-runtime", action="store_true")
|
|
legacy_import_options = legacy_update.add_mutually_exclusive_group()
|
|
legacy_import_options.add_argument("--seed-from", type=Path)
|
|
legacy_import_options.add_argument("--no-import", action="store_true")
|
|
|
|
import_game = subparsers.add_parser(
|
|
"import-game",
|
|
help="safely seed the managed installation from an existing game",
|
|
)
|
|
import_game.add_argument("source", type=Path)
|
|
import_game.add_argument("--dry-run", action="store_true")
|
|
|
|
compatibility = subparsers.add_parser(
|
|
"cache-compatibility",
|
|
help="cache one exact verified Game.exe/game.dll compatibility pair",
|
|
)
|
|
compatibility.add_argument("source", type=Path)
|
|
prepare = subparsers.add_parser("prepare", help="build the read-only runtime view and scene fix")
|
|
prepare.add_argument("--game-dir", type=Path)
|
|
|
|
play = subparsers.add_parser("play", help="launch the game in the managed Proton window")
|
|
play.add_argument("--game-dir", type=Path)
|
|
play.add_argument("--no-box", action="store_true", help="launch without X11 window styling")
|
|
play.add_argument(
|
|
"--publisher-dll",
|
|
"--no-world-fix",
|
|
action="store_true",
|
|
dest="publisher_dll",
|
|
help=(
|
|
"compatibility option retained for older scripts; publisher code is "
|
|
"always used without the retired world fix"
|
|
),
|
|
)
|
|
play.add_argument("--dry-run", action="store_true")
|
|
play.add_argument("--skip-runtime-setup", action="store_true")
|
|
login_trace = subparsers.add_parser(
|
|
"login-trace",
|
|
help="retired unsafe authentication trace (always refuses to launch)",
|
|
)
|
|
login_trace.add_argument("--game-dir", type=Path)
|
|
login_trace.add_argument("--no-box", action="store_true", help="launch without X11 window styling")
|
|
login_trace.add_argument("--skip-runtime-setup", action="store_true")
|
|
login_trace.add_argument(
|
|
"--timeout",
|
|
type=float,
|
|
default=DEFAULT_TRACE_TIMEOUT_SECONDS,
|
|
help="maximum game runtime in seconds (default: 600)",
|
|
)
|
|
|
|
settings = subparsers.add_parser("settings", help="show, initialize, or reset settings")
|
|
settings_action = settings.add_mutually_exclusive_group()
|
|
settings_action.add_argument(
|
|
"--init",
|
|
action="store_true",
|
|
help="create missing settings and migrate legacy files without resetting choices",
|
|
)
|
|
settings_action.add_argument(
|
|
"--reset",
|
|
action="store_true",
|
|
help="explicitly replace all launcher settings with defaults",
|
|
)
|
|
|
|
launcher_update = subparsers.add_parser(
|
|
"launcher-update",
|
|
help="check, verify, install, or roll back the native Linux launcher",
|
|
)
|
|
launcher_action = launcher_update.add_mutually_exclusive_group()
|
|
launcher_action.add_argument(
|
|
"--check",
|
|
action="store_true",
|
|
help="authenticate the update manifest without installing it",
|
|
)
|
|
launcher_action.add_argument(
|
|
"--verify",
|
|
action="store_true",
|
|
help="download and fully validate the published release without installing it",
|
|
)
|
|
launcher_action.add_argument(
|
|
"--rollback",
|
|
action="store_true",
|
|
help="switch back to the previously installed launcher release",
|
|
)
|
|
launcher_update.add_argument(
|
|
"--manifest-url",
|
|
help="override the signed stable-manifest URL",
|
|
)
|
|
launcher_update.add_argument(
|
|
"--public-key",
|
|
type=Path,
|
|
help="override the pinned launcher-update public key",
|
|
)
|
|
return parser
|
|
|
|
|
|
def main(argv: Sequence[str] | None = None) -> int:
|
|
parser = build_parser()
|
|
arguments = parser.parse_args(argv)
|
|
command = arguments.command or "gui"
|
|
paths = AppPaths.from_environment()
|
|
try:
|
|
if command == "gui":
|
|
from .ui import run_dashboard
|
|
|
|
run_dashboard(paths)
|
|
return 0
|
|
if command == "doctor":
|
|
return _doctor(arguments.json)
|
|
if command == "login-diagnostics":
|
|
return _login_diagnostics(
|
|
paths,
|
|
arguments.timeout,
|
|
arguments.runtime,
|
|
arguments.json,
|
|
)
|
|
if command == "fetch":
|
|
return _fetch(paths, arguments.dry_run, arguments.json)
|
|
if command == "setup":
|
|
return _setup(
|
|
paths,
|
|
arguments.dry_run,
|
|
arguments.force_runtime,
|
|
arguments.json,
|
|
)
|
|
if command == "update":
|
|
return _native_update(
|
|
paths,
|
|
arguments.dry_run,
|
|
arguments.repair,
|
|
arguments.seed_from,
|
|
arguments.no_import,
|
|
arguments.json,
|
|
)
|
|
if command == "sync":
|
|
return _native_update(
|
|
paths,
|
|
arguments.dry_run,
|
|
arguments.repair,
|
|
None,
|
|
True,
|
|
arguments.json,
|
|
)
|
|
if command == "wine-update":
|
|
return _wine_update(
|
|
paths,
|
|
arguments.dry_run,
|
|
arguments.force_runtime,
|
|
arguments.seed_from,
|
|
arguments.no_import,
|
|
arguments.json,
|
|
)
|
|
if command == "import-game":
|
|
return _import_game(
|
|
paths,
|
|
arguments.source,
|
|
arguments.dry_run,
|
|
arguments.json,
|
|
)
|
|
if command == "cache-compatibility":
|
|
result = cache_compatibility_binaries(paths, arguments.source)
|
|
_emit({"ok": True, "compatibility": asdict(result)}, arguments.json)
|
|
return 0
|
|
if command == "prepare":
|
|
return _prepare(paths, arguments.game_dir, arguments.json)
|
|
if command == "play":
|
|
return _play(
|
|
paths,
|
|
arguments.game_dir,
|
|
arguments.no_box,
|
|
arguments.dry_run,
|
|
arguments.skip_runtime_setup,
|
|
arguments.json,
|
|
)
|
|
if command == "login-trace":
|
|
return _login_trace(
|
|
paths,
|
|
arguments.game_dir,
|
|
arguments.no_box,
|
|
arguments.skip_runtime_setup,
|
|
arguments.timeout,
|
|
arguments.json,
|
|
)
|
|
if command == "settings":
|
|
return _settings(paths, arguments.init, arguments.reset, arguments.json)
|
|
if command == "launcher-update":
|
|
return _launcher_update(
|
|
paths,
|
|
arguments.check,
|
|
arguments.verify,
|
|
arguments.rollback,
|
|
arguments.manifest_url,
|
|
arguments.public_key,
|
|
arguments.json,
|
|
)
|
|
parser.error(f"unknown command: {command}")
|
|
except KeyboardInterrupt:
|
|
message = "Interrupted; PristonTale and its compatibility runtime were stopped."
|
|
if arguments.json:
|
|
_print_json({"ok": False, "error": message, "interrupted": True})
|
|
else:
|
|
print(f"\n{message}", file=sys.stderr)
|
|
return 130
|
|
except (ValueError, OSError, RuntimeError) as exc:
|
|
if arguments.json:
|
|
_print_json({"ok": False, "error": str(exc)})
|
|
else:
|
|
print(f"Error: {exc}", file=sys.stderr)
|
|
if isinstance(exc, CommandError):
|
|
if exc.outcome.stdout:
|
|
print(exc.outcome.stdout.rstrip(), file=sys.stderr)
|
|
if exc.outcome.stderr:
|
|
print(exc.outcome.stderr.rstrip(), file=sys.stderr)
|
|
return 1
|
|
|
|
|
|
def _doctor(as_json: bool) -> int:
|
|
report = inspect_host()
|
|
if as_json:
|
|
_print_json(report.to_dict())
|
|
else:
|
|
print(f"Distribution: {report.distribution} {report.version} ({report.architecture})")
|
|
print(f"Display: {report.display_backend}")
|
|
for tool in report.tools:
|
|
status = tool.path or "MISSING"
|
|
required = "required" if tool.required else "optional"
|
|
print(f"{tool.name:12} {status} [{required}] — {tool.purpose}")
|
|
for warning in report.warnings:
|
|
print(f"Warning: {warning}")
|
|
if report.package_hint:
|
|
print(report.package_hint)
|
|
return 0 if report.ok else 1
|
|
|
|
|
|
def _login_diagnostics(
|
|
paths: AppPaths,
|
|
timeout_seconds: float,
|
|
runtime_choice: str,
|
|
as_json: bool,
|
|
) -> int:
|
|
settings = _load_or_default_settings(paths, persist=False)
|
|
runner = _runner(paths, "login-diagnostics", False)
|
|
with _operation_guard(paths, "login-diagnostics", False):
|
|
report = inspect_login_environment(
|
|
paths,
|
|
settings,
|
|
timeout_seconds=timeout_seconds,
|
|
runner=runner,
|
|
runtime_override=(
|
|
None if runtime_choice == "configured" else runtime_choice
|
|
),
|
|
)
|
|
if as_json:
|
|
_print_json(report.to_dict())
|
|
else:
|
|
probe = report.login_server
|
|
print(f"Launcher: {report.launcher_version} ({report.launcher_module})")
|
|
print(f"Python: {report.python_executable}")
|
|
print(f"Host: {report.distribution_id} ({report.kernel_release})")
|
|
print(
|
|
"Runtime test: "
|
|
f"{report.runtime_requested} "
|
|
f"(configured: {report.runtime_configured}; settings unchanged)"
|
|
)
|
|
status = "REACHABLE" if probe.reachable else "UNREACHABLE"
|
|
latency = f", {probe.latency_ms:.1f} ms" if probe.latency_ms is not None else ""
|
|
print(f"Login server: {status} ({probe.host}:{probe.port}{latency})")
|
|
if probe.error:
|
|
print(f"Network error: {probe.error}")
|
|
print("Authentication: not tested; no credentials or game-protocol data were sent")
|
|
print(f"Login responses: {report.login_response_ownership}")
|
|
identity_status = (
|
|
"ready"
|
|
if report.client_identity_ready is True
|
|
else "NOT READY"
|
|
if report.client_identity_ready is False
|
|
else "not probed"
|
|
)
|
|
print(f"Client identity: {identity_status}; raw identity values were not collected")
|
|
print(f"Privacy: {report.client_identity_privacy}")
|
|
if not report.login_trace_enabled:
|
|
print("Login trace: retired (will not launch or instrument the game)")
|
|
print(f"Publisher test: {report.publisher_comparison_command}")
|
|
print(f"Game profile: {report.game_profile}")
|
|
print(f"Game.exe: {report.game_exe}")
|
|
if report.game_exe_sha256:
|
|
print(f"Game SHA-256: {report.game_exe_sha256}")
|
|
print(f"game.dll: {report.game_dll}")
|
|
if report.game_dll_sha256:
|
|
print(f"DLL SHA-256: {report.game_dll_sha256}")
|
|
print(
|
|
"Managed source: "
|
|
+ ("matches runtime" if report.runtime_matches_managed_source else "DIFFERS")
|
|
)
|
|
if report.depot_build is not None:
|
|
print(
|
|
f"Depot: build {report.depot_build} "
|
|
f"({report.depot_version or 'unknown version'})"
|
|
)
|
|
if report.runtime_kind:
|
|
print(f"Runtime: {report.runtime_kind} ({report.runtime_path or 'unknown path'})")
|
|
else:
|
|
print(f"Runtime: unavailable ({report.runtime_error or 'selection failed'})")
|
|
if report.proton_sync_effective is not None:
|
|
print(
|
|
"Proton sync: "
|
|
f"{report.proton_sync_configured} -> {report.proton_sync_effective}"
|
|
)
|
|
prefix_status = "ready" if report.prefix_ready else "not ready"
|
|
directx_status = "ready" if report.directx_ready else "not ready"
|
|
print(f"Prefix: {prefix_status}; DirectX helpers {directx_status}")
|
|
print(f"Prefix path: {report.prefix or 'unavailable'}")
|
|
print(f"Latest log: {report.latest_game_log or 'not found'}")
|
|
if report.latest_game_log_modified_at:
|
|
print(f"Log modified: {report.latest_game_log_modified_at}")
|
|
for warning in report.warnings:
|
|
print(f"Warning: {warning}")
|
|
return 0 if report.preflight_ok else 1
|
|
|
|
|
|
def _fetch(paths: AppPaths, dry_run: bool, as_json: bool) -> int:
|
|
if not dry_run:
|
|
paths.create_base_directories()
|
|
if dry_run:
|
|
value = {
|
|
"ok": True,
|
|
"dry_run": True,
|
|
"url": OFFICIAL_LAUNCHER_URL,
|
|
"download": str(paths.launcher_pristine),
|
|
"prepared": str(paths.launcher_patched),
|
|
}
|
|
else:
|
|
download = fetch_official_launcher(
|
|
paths.launcher_pristine,
|
|
paths.launcher_metadata,
|
|
)
|
|
patch = prepare_launcher_binary(
|
|
paths.launcher_pristine,
|
|
paths.launcher_patched,
|
|
)
|
|
value = {
|
|
"ok": True,
|
|
"download": asdict(download),
|
|
"patch": asdict(patch),
|
|
"warning": None if patch.supported_profile else (
|
|
"The launcher is newer than the verified font-patch profile; "
|
|
"it was retained byte-for-byte."
|
|
),
|
|
}
|
|
_emit(value, as_json)
|
|
return 0
|
|
|
|
|
|
def _setup(
|
|
paths: AppPaths,
|
|
dry_run: bool,
|
|
force_runtime: bool,
|
|
as_json: bool,
|
|
) -> int:
|
|
tools = find_wine_tools()
|
|
if tools is None:
|
|
raise LauncherError("Wine, Wineboot, and Winetricks are required; run doctor")
|
|
runner = _runner(paths, "setup", dry_run)
|
|
with _operation_guard(paths, "setup", dry_run):
|
|
results = configure_launcher_prefix(
|
|
paths,
|
|
tools,
|
|
runner,
|
|
force_runtime=force_runtime,
|
|
)
|
|
_emit(
|
|
{
|
|
"ok": True,
|
|
"dry_run": dry_run,
|
|
"prefix": str(paths.launcher_prefix),
|
|
"commands": [asdict(result) for result in results],
|
|
},
|
|
as_json,
|
|
)
|
|
return 0
|
|
|
|
|
|
def _native_update(
|
|
paths: AppPaths,
|
|
dry_run: bool,
|
|
repair: bool,
|
|
seed_from: Path | None,
|
|
no_import: bool,
|
|
as_json: bool,
|
|
) -> int:
|
|
"""Use the native verified depot client for normal installs and updates."""
|
|
|
|
settings = _load_or_default_settings(paths, persist=not dry_run)
|
|
runner = _runner(paths, "native-update", dry_run)
|
|
with _operation_guard(paths, "native-update", dry_run):
|
|
_print_operation_status(as_json, 1, 3, "Checking local game files…")
|
|
game_import = None
|
|
if not no_import:
|
|
game_import = _maybe_import_game(
|
|
paths,
|
|
settings,
|
|
seed_from,
|
|
runner,
|
|
announce=not as_json,
|
|
)
|
|
verify_import = game_import is not None or game_import_requires_verification(paths)
|
|
_print_operation_status(as_json, 2, 3, "Synchronizing official game files…")
|
|
result = sync_managed_game(
|
|
paths,
|
|
repair=repair or verify_import,
|
|
quarantine_unmanaged=verify_import,
|
|
dry_run=dry_run,
|
|
progress=None if as_json else _print_sync_progress,
|
|
)
|
|
if verify_import and not dry_run:
|
|
mark_game_import_verified(
|
|
paths,
|
|
build=result.build,
|
|
files_quarantined=result.files_quarantined,
|
|
quarantine_directory=result.quarantine_directory,
|
|
)
|
|
overlay = None
|
|
if not dry_run and paths.source_game.is_dir():
|
|
_print_operation_status(as_json, 3, 3, "Preparing the clean game runtime view…")
|
|
overlay = asdict(prepare_runtime_view(paths, paths.source_game))
|
|
_emit(
|
|
{
|
|
"ok": True,
|
|
"dry_run": dry_run,
|
|
"game_import": asdict(game_import) if game_import else None,
|
|
"native_update": asdict(result),
|
|
"runtime_view": overlay,
|
|
},
|
|
as_json,
|
|
)
|
|
return 0
|
|
|
|
|
|
def _wine_update(
|
|
paths: AppPaths,
|
|
dry_run: bool,
|
|
force_runtime: bool,
|
|
seed_from: Path | None,
|
|
no_import: bool,
|
|
as_json: bool,
|
|
) -> int:
|
|
_fetch(paths, dry_run, True)
|
|
tools = find_wine_tools()
|
|
if tools is None:
|
|
raise LauncherError("Wine, Wineboot, and Winetricks are required; run doctor")
|
|
runner = _runner(paths, "update", dry_run)
|
|
settings = _load_or_default_settings(paths, persist=not dry_run)
|
|
with _operation_guard(paths, "update", dry_run):
|
|
configure_launcher_prefix(
|
|
paths,
|
|
tools,
|
|
runner,
|
|
force_runtime=force_runtime,
|
|
)
|
|
game_import = None
|
|
if not no_import:
|
|
game_import = _maybe_import_game(
|
|
paths,
|
|
settings,
|
|
seed_from,
|
|
runner,
|
|
announce=not as_json,
|
|
)
|
|
updater = launch_official_updater(paths, tools, runner)
|
|
native_verification = None
|
|
verify_import = not dry_run and (
|
|
game_import is not None or game_import_requires_verification(paths)
|
|
)
|
|
if verify_import:
|
|
native_verification = sync_managed_game(
|
|
paths,
|
|
repair=True,
|
|
quarantine_unmanaged=True,
|
|
progress=None if as_json else _print_sync_progress,
|
|
)
|
|
mark_game_import_verified(
|
|
paths,
|
|
build=native_verification.build,
|
|
files_quarantined=native_verification.files_quarantined,
|
|
quarantine_directory=native_verification.quarantine_directory,
|
|
)
|
|
overlay = None
|
|
if not dry_run and paths.source_game.is_dir():
|
|
overlay = asdict(prepare_runtime_view(paths, paths.source_game))
|
|
value = {
|
|
"ok": True,
|
|
"dry_run": dry_run,
|
|
"game_import": asdict(game_import) if game_import else None,
|
|
"updater": asdict(updater),
|
|
"native_verification": (
|
|
asdict(native_verification) if native_verification is not None else None
|
|
),
|
|
"runtime_view": overlay,
|
|
}
|
|
_emit(value, as_json)
|
|
return 0
|
|
|
|
|
|
def _print_sync_progress(progress: SyncProgress) -> None:
|
|
"""Emit line-delimited progress that the native Tk dashboard can consume."""
|
|
|
|
print(f"PROGRESS {json.dumps(asdict(progress), sort_keys=True)}", flush=True)
|
|
|
|
|
|
def _print_operation_status(
|
|
as_json: bool,
|
|
step: int,
|
|
total: int,
|
|
message: str,
|
|
) -> None:
|
|
"""Emit a structured operation stage for the native launcher dashboard."""
|
|
|
|
if not as_json:
|
|
print(
|
|
f"STATUS {json.dumps({'step': step, 'total': total, 'message': message})}",
|
|
flush=True,
|
|
)
|
|
|
|
|
|
def _import_game(
|
|
paths: AppPaths,
|
|
source: Path,
|
|
dry_run: bool,
|
|
as_json: bool,
|
|
) -> int:
|
|
runner = _runner(paths, "import-game", dry_run)
|
|
if not as_json:
|
|
print(f"Importing existing game from {source.expanduser()}…", flush=True)
|
|
with _operation_guard(paths, "import-game", dry_run):
|
|
result = import_existing_game(paths, source, runner)
|
|
_emit({"ok": True, "dry_run": dry_run, "game_import": asdict(result)}, as_json)
|
|
return 0
|
|
|
|
|
|
def _maybe_import_game(
|
|
paths: AppPaths,
|
|
settings: Settings,
|
|
seed_from: Path | None,
|
|
runner: CommandRunner,
|
|
*,
|
|
announce: bool,
|
|
) -> GameImportResult | None:
|
|
if paths.source_game.exists() or paths.source_game.is_symlink():
|
|
if seed_from is not None:
|
|
raise RuntimeError(
|
|
f"cannot seed an existing managed game directory: {paths.source_game}"
|
|
)
|
|
return None
|
|
candidate = seed_from
|
|
if candidate is None:
|
|
candidate = discover_existing_game(paths, settings)
|
|
if candidate is None:
|
|
return None
|
|
if announce:
|
|
print(f"Bootstrapping from existing game: {candidate}", flush=True)
|
|
return import_existing_game(paths, candidate, runner)
|
|
|
|
|
|
def _prepare(paths: AppPaths, game_dir: Path | None, as_json: bool) -> int:
|
|
with _operation_guard(paths, "prepare", False):
|
|
if game_dir is None and game_import_requires_verification(paths):
|
|
raise LauncherError(
|
|
"the imported game has not completed its full publisher verification; "
|
|
"run pristontale-eu update"
|
|
)
|
|
settings = ensure_settings(paths)
|
|
source = _source_game(paths, settings, game_dir)
|
|
result = prepare_runtime_view(paths, source)
|
|
_emit({"ok": True, "runtime_view": asdict(result)}, as_json)
|
|
return 0
|
|
|
|
|
|
def _play(
|
|
paths: AppPaths,
|
|
game_dir: Path | None,
|
|
no_box: bool,
|
|
dry_run: bool,
|
|
skip_runtime_setup: bool,
|
|
as_json: bool,
|
|
) -> int:
|
|
with _play_guard(paths, dry_run) as play_lease:
|
|
return _play_locked(
|
|
paths,
|
|
game_dir,
|
|
no_box,
|
|
dry_run,
|
|
skip_runtime_setup,
|
|
as_json,
|
|
reuse_prepared=not play_lease.preparation_allowed,
|
|
before_game=play_lease.allow_shared_clients,
|
|
)
|
|
|
|
|
|
def _play_locked(
|
|
paths: AppPaths,
|
|
game_dir: Path | None,
|
|
no_box: bool,
|
|
dry_run: bool,
|
|
skip_runtime_setup: bool,
|
|
as_json: bool,
|
|
*,
|
|
reuse_prepared: bool = False,
|
|
before_game: Callable[[], None] | None = None,
|
|
) -> int:
|
|
if game_dir is None and game_import_requires_verification(paths):
|
|
raise LauncherError(
|
|
"the imported game has not completed its full publisher verification; "
|
|
"run pristontale-eu update before playing"
|
|
)
|
|
settings = _load_or_default_settings(paths, persist=not dry_run)
|
|
source = _source_game(paths, settings, game_dir)
|
|
overlay = None
|
|
_print_operation_status(
|
|
as_json,
|
|
1,
|
|
4,
|
|
(
|
|
"Reusing the active managed game view…"
|
|
if reuse_prepared
|
|
else "Preparing the managed game view…"
|
|
),
|
|
)
|
|
if reuse_prepared:
|
|
if game_dir is not None:
|
|
raise LauncherError(
|
|
"an additional client cannot replace the active managed game view; "
|
|
"close every client before using --game-dir"
|
|
)
|
|
if not paths.runtime_view.is_dir():
|
|
raise LauncherError(
|
|
"another game client is active, but the prepared runtime view is missing"
|
|
)
|
|
elif not dry_run:
|
|
overlay = prepare_runtime_view(paths, source)
|
|
_print_operation_status(as_json, 2, 4, "Selecting the configured game runtime…")
|
|
selection = select_runtime(settings)
|
|
runner = _runner(paths, "play", dry_run)
|
|
setup_results = []
|
|
if reuse_prepared:
|
|
_print_operation_status(
|
|
as_json,
|
|
3,
|
|
4,
|
|
"Reusing the active Proton and DirectX runtime…",
|
|
)
|
|
if not game_runtime_ready(paths, selection):
|
|
raise LauncherError(
|
|
"another game client is active, but its prepared runtime no longer "
|
|
"matches the current settings; close every client and launch again"
|
|
)
|
|
elif not skip_runtime_setup:
|
|
_print_operation_status(as_json, 3, 4, "Preparing Proton and DirectX components…")
|
|
setup_results = prepare_game_runtime(paths, selection, runner)
|
|
else:
|
|
_print_operation_status(as_json, 3, 4, "Using the existing Proton runtime…")
|
|
_print_operation_status(as_json, 4, 4, "Starting the PristonTale game client…")
|
|
native_log_directory = paths.game_save / "logs"
|
|
native_log_baseline = snapshot_native_logs(
|
|
native_log_directory,
|
|
patterns=("log_*.txt",),
|
|
)
|
|
game_started_ns = time.time_ns()
|
|
if before_game is not None:
|
|
before_game()
|
|
try:
|
|
outcome = run_game(
|
|
paths,
|
|
settings,
|
|
selection,
|
|
runner,
|
|
use_gamebox=not no_box,
|
|
)
|
|
except KeyboardInterrupt:
|
|
current_game_log = _current_game_log(
|
|
native_log_directory,
|
|
native_log_baseline,
|
|
game_started_ns,
|
|
)
|
|
if current_game_log is not None and not as_json:
|
|
_print_interrupted_game_log(current_game_log)
|
|
raise
|
|
current_game_log = _current_game_log(
|
|
native_log_directory,
|
|
native_log_baseline,
|
|
game_started_ns,
|
|
)
|
|
runtime_issues = _runtime_issue_summary(outcome.stderr)
|
|
if (
|
|
selection.kind == "proton7-container"
|
|
and runtime_issues["host_futex_fatal_count"]
|
|
):
|
|
runtime_issues["recommended_one_variable_comparison"] = "proton7"
|
|
_emit(
|
|
{
|
|
"ok": outcome.returncode == 0,
|
|
"dry_run": dry_run,
|
|
"runtime": asdict(selection),
|
|
"runtime_view": asdict(overlay) if overlay else None,
|
|
"setup_commands": [asdict(result) for result in setup_results],
|
|
"game": asdict(outcome),
|
|
"current_game_log": current_game_log,
|
|
"runtime_issues": runtime_issues,
|
|
},
|
|
as_json,
|
|
)
|
|
return 0 if outcome.returncode == 0 else 1
|
|
|
|
|
|
def _current_game_log(
|
|
directory: Path,
|
|
baseline: dict[str, LogFileState],
|
|
started_ns: int,
|
|
) -> dict[str, object] | None:
|
|
state = newest_current_native_log(
|
|
directory,
|
|
baseline,
|
|
started_ns,
|
|
patterns=("log_*.txt",),
|
|
)
|
|
if state is None:
|
|
return None
|
|
analysis = parse_current_run_log(state, baseline)
|
|
return {
|
|
"path": str(state.path),
|
|
"modified_at": datetime.fromtimestamp(
|
|
state.mtime_ns / 1_000_000_000
|
|
).astimezone().isoformat(),
|
|
"size": state.size,
|
|
"load_texture_to_material_failed": analysis.load_texture_to_material_failed,
|
|
"import_scene_threw": analysis.import_scene_threw,
|
|
"failed_import_scenes": analysis.failed_import_scenes,
|
|
"stage_without_new_stage": analysis.stage_without_new_stage,
|
|
"module_code_sync_error": analysis.module_code_sync_error,
|
|
"module_code_sync_left": analysis.module_code_sync_left,
|
|
"module_code_sync_right": analysis.module_code_sync_right,
|
|
"module_code_sync_delta": analysis.module_code_sync_delta,
|
|
"banned_code_1": analysis.banned_code_1,
|
|
"banned_code_1_in_native_log": analysis.banned_code_1,
|
|
"banned_code_1_observation_scope": (
|
|
"native-game-log-only; server-provided login-screen text is not "
|
|
"written to this log"
|
|
),
|
|
"stage_without_new_stage_events": [
|
|
event.to_dict()
|
|
for event in analysis.events
|
|
if event.kind == "stage_without_new_stage"
|
|
],
|
|
"scene_imports": {
|
|
name: [scene.to_dict() for scene in imports]
|
|
for name, imports in sorted(analysis.scenes.items())
|
|
},
|
|
"failed_file_count": len(analysis.failed_files),
|
|
"failed_files": [event.to_dict() for event in analysis.failed_files[:64]],
|
|
}
|
|
|
|
|
|
def _runtime_issue_summary(stderr: str) -> dict[str, object]:
|
|
folded = stderr.casefold()
|
|
futex_message = "the futex facility returned an unexpected error code"
|
|
wrong_elf_message = "wrong elf class: elfclass64"
|
|
return {
|
|
"host_futex_fatal_count": folded.count(futex_message),
|
|
"gstreamer_wrong_elf_class_count": folded.count(wrong_elf_message),
|
|
"gstreamer_stream_error": "winegstreamer error" in folded,
|
|
"mesa_egl_initialization_failed": "egl: failed to create dri2 screen" in folded,
|
|
"steam_api_app_id_missing": "steamapi_init() failed; no appid found" in folded,
|
|
}
|
|
|
|
|
|
def _print_interrupted_game_log(report: dict[str, object]) -> None:
|
|
print(f"\nCurrent game log: {report['path']}", file=sys.stderr)
|
|
if report["module_code_sync_error"]:
|
|
print(
|
|
"Detected a publisher module-integrity mismatch; the game was launched "
|
|
"with publisher-pristine code in this launcher version.",
|
|
file=sys.stderr,
|
|
)
|
|
elif report["load_texture_to_material_failed"]:
|
|
print(
|
|
"Detected a texture/material failure while loading the game world.",
|
|
file=sys.stderr,
|
|
)
|
|
elif report["import_scene_threw"] or report["failed_import_scenes"]:
|
|
print("Detected a game-world scene import failure.", file=sys.stderr)
|
|
|
|
|
|
def _login_trace(
|
|
paths: AppPaths,
|
|
game_dir: Path | None,
|
|
no_box: bool,
|
|
skip_runtime_setup: bool,
|
|
timeout_seconds: float,
|
|
as_json: bool,
|
|
) -> int:
|
|
# Preserve the command name so old instructions fail safely before touching
|
|
# the prefix, preparing the runtime view, or starting Game.exe.
|
|
_ = paths, game_dir, no_box, skip_runtime_setup, timeout_seconds, as_json
|
|
raise RuntimeError(LOGIN_TRACE_DISABLED_REASON)
|
|
|
|
|
|
def _settings(
|
|
paths: AppPaths,
|
|
initialize: bool,
|
|
reset: bool,
|
|
as_json: bool,
|
|
) -> int:
|
|
if reset:
|
|
settings = reset_settings(paths)
|
|
elif initialize:
|
|
settings = ensure_settings(paths)
|
|
else:
|
|
settings = load_settings(paths)
|
|
_emit(
|
|
{
|
|
"ok": True,
|
|
"path": str(paths.settings_file),
|
|
"settings": asdict(settings),
|
|
},
|
|
as_json,
|
|
)
|
|
return 0
|
|
|
|
|
|
def _launcher_update(
|
|
paths: AppPaths,
|
|
check_only: bool,
|
|
verify: bool,
|
|
rollback: bool,
|
|
manifest_url: str | None,
|
|
public_key: Path | None,
|
|
as_json: bool,
|
|
) -> int:
|
|
if rollback:
|
|
result = rollback_launcher_update(paths)
|
|
_emit(
|
|
{
|
|
"ok": True,
|
|
"launcher_rollback": asdict(result),
|
|
},
|
|
as_json,
|
|
)
|
|
return 0
|
|
if check_only:
|
|
check = check_for_launcher_update(
|
|
manifest_url=manifest_url,
|
|
public_key_path=public_key,
|
|
)
|
|
_emit(
|
|
{
|
|
"ok": True,
|
|
"launcher_update": update_check_to_dict(check),
|
|
},
|
|
as_json,
|
|
)
|
|
return 0
|
|
if verify:
|
|
_print_operation_status(as_json, 1, 2, "Authenticating the launcher release…")
|
|
result = verify_launcher_update_channel(
|
|
manifest_url=manifest_url,
|
|
public_key_path=public_key,
|
|
)
|
|
_print_operation_status(as_json, 2, 2, "Published launcher release verified.")
|
|
_emit(
|
|
{
|
|
"ok": True,
|
|
"launcher_update_verification": asdict(result),
|
|
},
|
|
as_json,
|
|
)
|
|
return 0
|
|
_print_operation_status(as_json, 1, 2, "Verifying the signed launcher release…")
|
|
result = install_launcher_update(
|
|
paths,
|
|
manifest_url=manifest_url,
|
|
public_key_path=public_key,
|
|
)
|
|
_print_operation_status(as_json, 2, 2, "Launcher update installed safely.")
|
|
_emit(
|
|
{
|
|
"ok": True,
|
|
"launcher_update": asdict(result),
|
|
},
|
|
as_json,
|
|
)
|
|
return 0
|
|
|
|
|
|
def _source_game(
|
|
paths: AppPaths,
|
|
_settings: Settings,
|
|
argument: Path | None,
|
|
) -> Path:
|
|
"""Select the managed game unless an explicit diagnostic path was supplied.
|
|
|
|
The persisted ``game_source`` setting is an import seed only. It must not
|
|
redirect normal Play, Prepare, or Login Trace operations away from the
|
|
launcher-managed tree.
|
|
"""
|
|
|
|
if argument is not None:
|
|
return argument.expanduser().resolve()
|
|
return paths.source_game
|
|
|
|
|
|
def _load_or_default_settings(paths: AppPaths, *, persist: bool) -> Settings:
|
|
if paths.settings_file.is_file():
|
|
return load_settings(paths)
|
|
if persist:
|
|
return ensure_settings(paths)
|
|
return Settings()
|
|
|
|
|
|
def _runner(paths: AppPaths, operation: str, dry_run: bool) -> CommandRunner:
|
|
timestamp = datetime.now().astimezone().strftime("%Y%m%d-%H%M%S-%f")
|
|
return CommandRunner(paths.logs / f"{timestamp}-{operation}.jsonl", dry_run=dry_run)
|
|
|
|
|
|
def _play_guard(
|
|
paths: AppPaths,
|
|
dry_run: bool,
|
|
) -> AbstractContextManager[LauncherPlayLease]:
|
|
if dry_run:
|
|
return nullcontext(LauncherPlayLease.dry_run())
|
|
return launcher_play_lock(paths.launcher_operation_lock)
|
|
|
|
|
|
def _operation_guard(
|
|
paths: AppPaths,
|
|
operation: str,
|
|
dry_run: bool,
|
|
) -> AbstractContextManager[None]:
|
|
if dry_run:
|
|
return nullcontext()
|
|
return launcher_operation_lock(paths.launcher_operation_lock, operation)
|
|
|
|
|
|
def _emit(value: object, as_json: bool) -> None:
|
|
if as_json:
|
|
_print_json(value)
|
|
else:
|
|
if isinstance(value, dict) and value.get("ok") is True:
|
|
print("Done.")
|
|
print(json.dumps(value, indent=2, sort_keys=True))
|
|
|
|
|
|
def _print_json(value: object) -> None:
|
|
print(json.dumps(value, indent=2, sort_keys=True))
|