Files
pristontale-eu-linux/ptlinux/cli.py

918 lines
26 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 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,
)
from .world_fix import (
create_world_fix_status_target,
read_world_fix_status,
)
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=(
"diagnostic fallback: disable the in-memory post-login world fix "
"(the publisher DLL is always used)"
),
)
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 = "ready" if report.world_fix_helper_ready else "not ready"
print(f"World fix: {world_fix_status} ({report.world_fix_helper})")
if report.world_fix_helper_sha256:
print(f"Helper SHA: {report.world_fix_helper_sha256}")
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:
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()
world_fix_status_target = None
if not dry_run and not publisher_dll:
world_fix_status_target = create_world_fix_status_target(
paths.game_save,
game_started_ns,
)
try:
outcome = run_game(
paths,
settings,
selection,
runner,
use_gamebox=not no_box,
use_world_fix=not publisher_dll,
world_fix_status_target=world_fix_status_target,
)
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)
world_fix_status = read_world_fix_status(
world_fix_status_target,
game_started_ns,
)
if world_fix_status is not None and not as_json:
print(
f"World-fix status: {str(world_fix_status['status'])} "
f"({str(world_fix_status['path'])})",
file=sys.stderr,
)
raise
current_game_log = _current_game_log(
native_log_directory,
native_log_baseline,
game_started_ns,
)
world_fix_status = read_world_fix_status(
world_fix_status_target,
game_started_ns,
)
_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,
"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["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))