"""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 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 .launcher import ( LauncherError, configure_launcher_prefix, find_wine_tools, launch_official_updater, prepare_launcher_binary, ) from .locking import launcher_operation_lock from .login_diagnostics import inspect_login_environment from .login_trace import DEFAULT_TRACE_TIMEOUT_SECONDS, run_login_trace from .overlay import cache_compatibility_binaries, prepare_runtime_view from .paths import AppPaths from .runtime import ( prepare_game_runtime, run_game, select_runtime, ) from .settings import Settings, ensure_settings, load_settings, save_settings from .self_update import ( check_for_launcher_update, install_launcher_update, rollback_launcher_update, update_check_to_dict, ) 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)", ) 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="trace the post-authentication account check without retaining credentials", ) 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 or initialize settings") settings.add_argument("--init", action="store_true", help="write conservative defaults") launcher_update = subparsers.add_parser( "launcher-update", help="check, 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( "--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.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.publisher_dll, 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.json) if command == "launcher-update": return _launcher_update( paths, arguments.check, 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, as_json: bool) -> int: settings = _load_or_default_settings(paths, persist=False) report = inspect_login_environment( paths, settings, timeout_seconds=timeout_seconds, ) 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})") 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"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}") world_fix_status = "enabled" if report.world_fix_enabled else "retired" print(f"World fix: {world_fix_status}") if report.world_fix_disabled_reason: print(f"Reason: {report.world_fix_disabled_reason}") 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, ) _print_operation_status(as_json, 2, 3, "Synchronizing official game files…") result = sync_managed_game( paths, repair=repair, dry_run=dry_run, progress=None if as_json else _print_sync_progress, ) 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) 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), "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: 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, publisher_dll: bool, dry_run: bool, skip_runtime_setup: bool, as_json: bool, ) -> int: with _operation_guard(paths, "play", dry_run): return _play_locked( paths, game_dir, no_box, publisher_dll, dry_run, skip_runtime_setup, as_json, ) def _play_locked( paths: AppPaths, game_dir: Path | None, no_box: bool, publisher_dll: bool, dry_run: bool, skip_runtime_setup: bool, as_json: bool, ) -> int: # The legacy flag is retained for command-line compatibility. V21 performs # periodic module-code checks, so every production launch is publisher-pristine. _ = publisher_dll 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, "Preparing the managed game view…") if 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 not skip_runtime_setup: _print_operation_status(as_json, 3, 4, "Preparing Proton and DirectX components…") setup_results = prepare_game_runtime(paths, selection, runner) _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() try: outcome = run_game( paths, settings, selection, runner, use_gamebox=not no_box, use_world_fix=False, ) 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, ) world_fix_status = None _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, "world_fix": world_fix_status, }, 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, "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 _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: settings = _load_or_default_settings(paths, persist=True) source = _source_game(paths, settings, game_dir) runner = _runner(paths, "login-trace", False) with _operation_guard(paths, "login-trace", False): _print_operation_status(as_json, 1, 4, "Preparing the managed game view…") prepare_runtime_view(paths, source) _print_operation_status(as_json, 2, 4, "Selecting the configured Proton runtime…") selection = select_runtime(settings) if not skip_runtime_setup: _print_operation_status(as_json, 3, 4, "Checking Proton and DirectX components…") prepare_game_runtime(paths, selection, runner) _print_operation_status(as_json, 4, 4, "Starting the privacy-safe login trace…") if not as_json: print( "Enter the account name with its registered capitalization, press Login, " "leave the approval screen visible for five seconds, then close the game.", flush=True, ) report = run_login_trace( paths, settings, selection, runner, use_gamebox=not no_box, timeout_seconds=timeout_seconds, ) _emit({"ok": True, "login_trace": report.to_dict()}, as_json) return 0 def _settings(paths: AppPaths, initialize: bool, as_json: bool) -> int: if initialize: save_settings(paths, Settings()) settings = ensure_settings(paths) if initialize else 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, 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 _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") return CommandRunner(paths.logs / f"{timestamp}-{operation}.jsonl", dry_run=dry_run) 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))