Compare commits
20 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 6900c20734 | |||
| f96aba9d99 | |||
| 78dade0d86 | |||
| d819809197 | |||
| dadae8f60a | |||
| 99982b12d9 | |||
| 75b21acf87 | |||
| b16540099d | |||
| b0855ccd9b | |||
| 87dee93b86 | |||
| fe883047e0 | |||
| 20a0bbcd75 | |||
| ed0c278a8b | |||
| 094c881798 | |||
| bc87f10661 | |||
| 562c924809 | |||
| 94a5e81555 | |||
| 94f2260f43 | |||
| 994fda5cc7 | |||
| 2a7da33160 |
23
.gitignore
vendored
23
.gitignore
vendored
@@ -3,8 +3,25 @@ __pycache__/
|
||||
*.py[cod]
|
||||
.pytest_cache/
|
||||
.idea/
|
||||
dist/
|
||||
.run/
|
||||
release-keys/
|
||||
config/local.toml
|
||||
results/*
|
||||
!results/.gitkeep
|
||||
|
||||
# Local development, investigation, and release tooling. The production
|
||||
# repository intentionally contains only the installer/runtime sources.
|
||||
AGENTS.md
|
||||
AIO_INSTALLER.md
|
||||
BOOTSTRAP_WORKPLAN.md
|
||||
DEBUGGING.md
|
||||
GITEA_UPDATES.md
|
||||
README.md
|
||||
config/
|
||||
configs/
|
||||
findings/
|
||||
replay_pid*.log
|
||||
results/
|
||||
testcases/
|
||||
tests/
|
||||
tools/
|
||||
dist/
|
||||
/native/
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
# PristonTale EU for Linux
|
||||
|
||||
Version 0.6.5 installs, updates, repairs, and launches PristonTale EU through a
|
||||
Version 0.8.1 installs, updates, repairs, and launches PristonTale EU through a
|
||||
native Linux dashboard. Game files come directly from the publisher and are not
|
||||
included in this archive.
|
||||
|
||||
@@ -15,11 +15,11 @@ Then run:
|
||||
```text
|
||||
openssl pkey -pubin -in launcher-update-public.pem -outform DER | sha256sum
|
||||
openssl dgst -sha256 -verify launcher-update-public.pem \
|
||||
-signature pristontale-eu-linux-installer-v0-6-5.tar.gz.sig \
|
||||
pristontale-eu-linux-installer-v0-6-5.tar.gz
|
||||
sha256sum -c pristontale-eu-linux-installer-v0-6-5.tar.gz.sha256
|
||||
tar -xzf pristontale-eu-linux-installer-v0-6-5.tar.gz
|
||||
cd pristontale-eu-linux-installer-v0-6-5
|
||||
-signature pristontale-eu-linux-installer-v0-8-1.tar.gz.sig \
|
||||
pristontale-eu-linux-installer-v0-8-1.tar.gz
|
||||
sha256sum -c pristontale-eu-linux-installer-v0-8-1.tar.gz.sha256
|
||||
tar -xzf pristontale-eu-linux-installer-v0-8-1.tar.gz
|
||||
cd pristontale-eu-linux-installer-v0-8-1
|
||||
./install.sh
|
||||
```
|
||||
|
||||
@@ -34,6 +34,11 @@ Existing installations at the previous default location are preserved during
|
||||
upgrades. The installer adds **PristonTale EU** to the application menu and
|
||||
installs the `pristontale-eu` command in `~/.local/bin`.
|
||||
|
||||
Steam Proton 7.0 is discovered in both the default Steam library and secondary
|
||||
libraries registered with the Steam client. If Proton uses an unregistered
|
||||
manual layout, open **Settings → Optional Overrides** and select the exact
|
||||
`Proton 7.0/proton` script.
|
||||
|
||||
## Use
|
||||
|
||||
Open **PristonTale EU** from the application menu or run:
|
||||
|
||||
@@ -8,3 +8,4 @@ Terminal=false
|
||||
Categories=Game;RolePlaying;
|
||||
Keywords=PristonTale;MMORPG;Wine;Proton;
|
||||
StartupNotify=true
|
||||
StartupWMClass=PristonTaleEU
|
||||
|
||||
@@ -33,8 +33,9 @@ fi
|
||||
|
||||
# Never let an unpacked older installer in the caller's current directory
|
||||
# shadow the selected installed release. Children started by the dashboard
|
||||
# inherit the same safe-path policy and exact package root.
|
||||
# inherit the same safe-path policy, package root, and no-bytecode policy.
|
||||
export PYTHONSAFEPATH=1
|
||||
export PYTHONDONTWRITEBYTECODE=1
|
||||
export PYTHONPATH="${release_root}"
|
||||
|
||||
exec "$python_bin" -B -P -m ptlinux "$@"
|
||||
|
||||
@@ -4,4 +4,4 @@ from .paths import AppPaths
|
||||
|
||||
__all__ = ["AppPaths", "__version__"]
|
||||
|
||||
__version__ = "0.6.5"
|
||||
__version__ = "0.8.1"
|
||||
|
||||
332
ptlinux/changelog.py
Normal file
332
ptlinux/changelog.py
Normal file
@@ -0,0 +1,332 @@
|
||||
"""Official patch changelog support for the native Linux dashboard."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import asdict, dataclass
|
||||
from datetime import UTC, datetime
|
||||
import json
|
||||
import math
|
||||
import os
|
||||
from pathlib import Path
|
||||
import re
|
||||
from typing import Callable, Protocol
|
||||
from urllib.request import Request, urlopen
|
||||
|
||||
|
||||
PATCHES_ENDPOINT = "https://pristontale.eu/api/changelog/patches"
|
||||
PATCH_NEWS_URL = "https://pristontale.eu/news#"
|
||||
MAX_PATCHES = 10
|
||||
MAX_PATCH_BYTES = 2 * 1024 * 1024
|
||||
MAX_PATCH_SECTIONS = 32
|
||||
MAX_PATCH_CHANGES = 200
|
||||
_MAX_PATCH_CANDIDATES = 100
|
||||
_SLUG_PATTERN = re.compile(r"[a-z0-9]+(?:-[a-z0-9]+)*")
|
||||
|
||||
|
||||
class ChangelogError(RuntimeError):
|
||||
"""Raised when the official patch feed cannot be parsed safely."""
|
||||
|
||||
|
||||
class ChangelogResponse(Protocol):
|
||||
"""The minimal response interface used by the changelog client."""
|
||||
|
||||
def read(self, size: int = -1) -> bytes: ...
|
||||
|
||||
def __enter__(self) -> ChangelogResponse: ...
|
||||
|
||||
def __exit__(
|
||||
self,
|
||||
exception_type: type[BaseException] | None,
|
||||
exception: BaseException | None,
|
||||
traceback: object | None,
|
||||
) -> bool | None: ...
|
||||
|
||||
|
||||
ChangelogOpener = Callable[..., ChangelogResponse]
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class PatchChange:
|
||||
text: str
|
||||
type_key: str
|
||||
|
||||
@property
|
||||
def tag(self) -> str:
|
||||
return patch_change_tag(self.type_key)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class PatchSection:
|
||||
title: str
|
||||
items: tuple[PatchChange, ...]
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class PatchNote:
|
||||
id: int
|
||||
slug: str
|
||||
title: str
|
||||
version: str | None
|
||||
tags: tuple[str, ...]
|
||||
summary: str
|
||||
sections: tuple[PatchSection, ...]
|
||||
published_at: str | None
|
||||
|
||||
@property
|
||||
def url(self) -> str:
|
||||
return f"{PATCH_NEWS_URL}{self.slug}"
|
||||
|
||||
|
||||
def patch_change_tag(type_key: str) -> str:
|
||||
"""Turn API keys such as ``new-feature`` into compact UI tags."""
|
||||
|
||||
label = " ".join(type_key.replace("_", "-").split("-"))
|
||||
return label.upper() if label else "CHANGE"
|
||||
|
||||
|
||||
def fetch_patch_notes(
|
||||
cache_path: Path,
|
||||
*,
|
||||
timeout_seconds: float = 15.0,
|
||||
opener: ChangelogOpener = urlopen,
|
||||
) -> tuple[PatchNote, ...]:
|
||||
"""Fetch the ten latest patches, falling back to the last valid cache."""
|
||||
|
||||
request = Request(
|
||||
PATCHES_ENDPOINT,
|
||||
headers={"User-Agent": "PT-Linux/0.6 (+https://pristontale.eu/)"},
|
||||
method="GET",
|
||||
)
|
||||
try:
|
||||
response = opener(request, timeout=timeout_seconds)
|
||||
with response:
|
||||
payload = response.read(MAX_PATCH_BYTES + 1)
|
||||
if len(payload) > MAX_PATCH_BYTES:
|
||||
raise ChangelogError("official patch response exceeds the 2 MiB limit")
|
||||
patches = _parse_patch_notes(payload)
|
||||
_write_patch_cache(cache_path, patches)
|
||||
return patches
|
||||
except (OSError, ChangelogError, UnicodeDecodeError, json.JSONDecodeError):
|
||||
return read_cached_patch_notes(cache_path)
|
||||
|
||||
|
||||
def read_cached_patch_notes(cache_path: Path) -> tuple[PatchNote, ...]:
|
||||
"""Read the last valid patch feed without performing network I/O."""
|
||||
|
||||
try:
|
||||
value = json.loads(cache_path.read_text(encoding="utf-8"))
|
||||
except (OSError, UnicodeError, json.JSONDecodeError):
|
||||
return ()
|
||||
if not isinstance(value, list):
|
||||
return ()
|
||||
patches: list[PatchNote] = []
|
||||
for item in value[:MAX_PATCHES]:
|
||||
patch = _patch_note_from_cache(item)
|
||||
if patch is not None:
|
||||
patches.append(patch)
|
||||
return tuple(patches)
|
||||
|
||||
|
||||
def _parse_patch_notes(payload: bytes) -> tuple[PatchNote, ...]:
|
||||
value = json.loads(payload.decode("utf-8-sig"))
|
||||
if isinstance(value, dict):
|
||||
value = value.get("data")
|
||||
if not isinstance(value, list):
|
||||
raise ChangelogError("official patch response is not a patch list")
|
||||
parsed: list[tuple[float, int, PatchNote]] = []
|
||||
for item in value[:_MAX_PATCH_CANDIDATES]:
|
||||
result = _parse_patch_note(item)
|
||||
if result is not None:
|
||||
timestamp, patch = result
|
||||
parsed.append((timestamp, patch.id, patch))
|
||||
parsed.sort(key=lambda entry: (entry[0], entry[1]), reverse=True)
|
||||
patches = tuple(entry[2] for entry in parsed[:MAX_PATCHES])
|
||||
if not patches:
|
||||
raise ChangelogError("official patch response contains no usable patches")
|
||||
return patches
|
||||
|
||||
|
||||
def _parse_patch_note(value: object) -> tuple[float, PatchNote] | None:
|
||||
if not isinstance(value, dict):
|
||||
return None
|
||||
patch_id = value.get("id")
|
||||
slug = _clean_text(value.get("slug"), 180)
|
||||
title = _clean_text(value.get("title"), 200)
|
||||
summary = _clean_text(value.get("summary"), 3000, allow_empty=True)
|
||||
if (
|
||||
isinstance(patch_id, bool)
|
||||
or not isinstance(patch_id, int)
|
||||
or slug is None
|
||||
or _SLUG_PATTERN.fullmatch(slug) is None
|
||||
or title is None
|
||||
or summary is None
|
||||
):
|
||||
return None
|
||||
version = _clean_text(value.get("version"), 64, allow_none=True)
|
||||
if value.get("version") is not None and version is None:
|
||||
return None
|
||||
timestamp = _timestamp(value.get("date"))
|
||||
patch = PatchNote(
|
||||
id=patch_id,
|
||||
slug=slug,
|
||||
title=title,
|
||||
version=version,
|
||||
tags=_parse_tags(value.get("tags")),
|
||||
summary=summary,
|
||||
sections=_parse_sections(value.get("changes")),
|
||||
published_at=_format_timestamp(timestamp),
|
||||
)
|
||||
return (timestamp if timestamp is not None else -1.0, patch)
|
||||
|
||||
|
||||
def _parse_tags(value: object) -> tuple[str, ...]:
|
||||
if not isinstance(value, list):
|
||||
return ()
|
||||
tags: list[str] = []
|
||||
seen: set[str] = set()
|
||||
for item in value[:16]:
|
||||
tag = _clean_text(item, 48)
|
||||
if tag is None or tag.casefold() in seen:
|
||||
continue
|
||||
seen.add(tag.casefold())
|
||||
tags.append(tag)
|
||||
return tuple(tags)
|
||||
|
||||
|
||||
def _parse_sections(value: object) -> tuple[PatchSection, ...]:
|
||||
if not isinstance(value, dict):
|
||||
return ()
|
||||
sections_value = value.get("sections")
|
||||
if not isinstance(sections_value, list):
|
||||
return ()
|
||||
sections: list[PatchSection] = []
|
||||
changes_seen = 0
|
||||
for section_value in sections_value[:MAX_PATCH_SECTIONS]:
|
||||
if not isinstance(section_value, dict):
|
||||
continue
|
||||
title = _clean_text(section_value.get("title"), 100) or "Changes"
|
||||
items_value = section_value.get("items")
|
||||
if not isinstance(items_value, list):
|
||||
continue
|
||||
items: list[PatchChange] = []
|
||||
for item in items_value:
|
||||
if changes_seen >= MAX_PATCH_CHANGES:
|
||||
break
|
||||
if not isinstance(item, dict):
|
||||
continue
|
||||
text = _clean_text(item.get("text"), 4000)
|
||||
type_key = _clean_text(item.get("type_key"), 64)
|
||||
if text is None:
|
||||
continue
|
||||
items.append(
|
||||
PatchChange(
|
||||
text=text,
|
||||
type_key=type_key or _type_key_from_title(title),
|
||||
)
|
||||
)
|
||||
changes_seen += 1
|
||||
if items:
|
||||
sections.append(PatchSection(title=title, items=tuple(items)))
|
||||
if changes_seen >= MAX_PATCH_CHANGES:
|
||||
break
|
||||
return tuple(sections)
|
||||
|
||||
|
||||
def _patch_note_from_cache(value: object) -> PatchNote | None:
|
||||
if not isinstance(value, dict):
|
||||
return None
|
||||
patch_id = value.get("id")
|
||||
slug = _clean_text(value.get("slug"), 180)
|
||||
title = _clean_text(value.get("title"), 200)
|
||||
summary = _clean_text(value.get("summary"), 3000, allow_empty=True)
|
||||
version = _clean_text(value.get("version"), 64, allow_none=True)
|
||||
published_at = _clean_text(value.get("published_at"), 64, allow_none=True)
|
||||
if (
|
||||
isinstance(patch_id, bool)
|
||||
or not isinstance(patch_id, int)
|
||||
or slug is None
|
||||
or _SLUG_PATTERN.fullmatch(slug) is None
|
||||
or title is None
|
||||
or summary is None
|
||||
):
|
||||
return None
|
||||
sections_value = value.get("sections")
|
||||
sections: list[PatchSection] = []
|
||||
if isinstance(sections_value, list):
|
||||
for section_value in sections_value[:MAX_PATCH_SECTIONS]:
|
||||
if not isinstance(section_value, dict):
|
||||
continue
|
||||
section_title = _clean_text(section_value.get("title"), 100)
|
||||
items_value = section_value.get("items")
|
||||
if section_title is None or not isinstance(items_value, list):
|
||||
continue
|
||||
items: list[PatchChange] = []
|
||||
for item in items_value:
|
||||
if not isinstance(item, dict):
|
||||
continue
|
||||
text = _clean_text(item.get("text"), 4000)
|
||||
type_key = _clean_text(item.get("type_key"), 64)
|
||||
if text is not None and type_key is not None:
|
||||
items.append(PatchChange(text=text, type_key=type_key))
|
||||
if items:
|
||||
sections.append(PatchSection(section_title, tuple(items)))
|
||||
return PatchNote(
|
||||
id=patch_id,
|
||||
slug=slug,
|
||||
title=title,
|
||||
version=version,
|
||||
tags=_parse_tags(value.get("tags")),
|
||||
summary=summary,
|
||||
sections=tuple(sections),
|
||||
published_at=published_at,
|
||||
)
|
||||
|
||||
|
||||
def _clean_text(
|
||||
value: object,
|
||||
maximum_length: int,
|
||||
*,
|
||||
allow_empty: bool = False,
|
||||
allow_none: bool = False,
|
||||
) -> str | None:
|
||||
if value is None and allow_none:
|
||||
return None
|
||||
if not isinstance(value, str):
|
||||
return None
|
||||
text = " ".join(value.split())
|
||||
if len(text) > maximum_length or (not text and not allow_empty):
|
||||
return None
|
||||
return text
|
||||
|
||||
|
||||
def _timestamp(value: object) -> float | None:
|
||||
if isinstance(value, bool) or not isinstance(value, (int, float)):
|
||||
return None
|
||||
timestamp = float(value)
|
||||
return timestamp if math.isfinite(timestamp) else None
|
||||
|
||||
|
||||
def _format_timestamp(value: float | None) -> str | None:
|
||||
if value is None:
|
||||
return None
|
||||
try:
|
||||
return datetime.fromtimestamp(value, UTC).astimezone().strftime("%d %b %Y")
|
||||
except (OverflowError, OSError, ValueError):
|
||||
return None
|
||||
|
||||
|
||||
def _type_key_from_title(title: str) -> str:
|
||||
return re.sub(r"[^a-z0-9]+", "-", title.casefold()).strip("-") or "change"
|
||||
|
||||
|
||||
def _write_patch_cache(path: Path, patches: tuple[PatchNote, ...]) -> None:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
temporary = path.with_name(f".{path.name}.{os.getpid()}.tmp")
|
||||
try:
|
||||
temporary.write_text(
|
||||
json.dumps([asdict(patch) for patch in patches], indent=2) + "\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
temporary.replace(path)
|
||||
except OSError:
|
||||
temporary.unlink(missing_ok=True)
|
||||
@@ -10,7 +10,7 @@ import json
|
||||
from pathlib import Path
|
||||
import sys
|
||||
import time
|
||||
from typing import Sequence
|
||||
from typing import Callable, Sequence
|
||||
|
||||
from ptdiag.logs import (
|
||||
LogFileState,
|
||||
@@ -21,7 +21,12 @@ from ptdiag.logs import (
|
||||
|
||||
from . import __version__
|
||||
from .command import CommandError, CommandRunner
|
||||
from .depot import SyncProgress, sync_managed_game
|
||||
from .depot import (
|
||||
SyncProgress,
|
||||
read_cached_depot_launch_catalog,
|
||||
select_depot_start_option,
|
||||
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
|
||||
@@ -33,12 +38,13 @@ from .launcher import (
|
||||
launch_official_updater,
|
||||
prepare_launcher_binary,
|
||||
)
|
||||
from .locking import launcher_operation_lock
|
||||
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,
|
||||
@@ -147,6 +153,10 @@ def build_parser() -> argparse.ArgumentParser:
|
||||
|
||||
play = subparsers.add_parser("play", help="launch the game in the managed Proton window")
|
||||
play.add_argument("--game-dir", type=Path)
|
||||
play.add_argument(
|
||||
"--realm",
|
||||
help="launch an available realm from the last official depot manifest",
|
||||
)
|
||||
play.add_argument("--no-box", action="store_true", help="launch without X11 window styling")
|
||||
play.add_argument(
|
||||
"--publisher-dll",
|
||||
@@ -296,6 +306,7 @@ def main(argv: Sequence[str] | None = None) -> int:
|
||||
arguments.dry_run,
|
||||
arguments.skip_runtime_setup,
|
||||
arguments.json,
|
||||
realm=arguments.realm,
|
||||
)
|
||||
if command == "login-trace":
|
||||
return _login_trace(
|
||||
@@ -687,6 +698,7 @@ def _maybe_import_game(
|
||||
|
||||
|
||||
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; "
|
||||
@@ -706,8 +718,10 @@ def _play(
|
||||
dry_run: bool,
|
||||
skip_runtime_setup: bool,
|
||||
as_json: bool,
|
||||
*,
|
||||
realm: str | None = None,
|
||||
) -> int:
|
||||
with _operation_guard(paths, "play", dry_run):
|
||||
with _play_guard(paths, dry_run) as play_lease:
|
||||
return _play_locked(
|
||||
paths,
|
||||
game_dir,
|
||||
@@ -715,6 +729,9 @@ def _play(
|
||||
dry_run,
|
||||
skip_runtime_setup,
|
||||
as_json,
|
||||
realm=realm,
|
||||
reuse_prepared=not play_lease.preparation_allowed,
|
||||
before_game=play_lease.allow_shared_clients,
|
||||
)
|
||||
|
||||
|
||||
@@ -725,6 +742,10 @@ def _play_locked(
|
||||
dry_run: bool,
|
||||
skip_runtime_setup: bool,
|
||||
as_json: bool,
|
||||
*,
|
||||
realm: str | None = None,
|
||||
reuse_prepared: bool = False,
|
||||
before_game: Callable[[], None] | None = None,
|
||||
) -> int:
|
||||
if game_dir is None and game_import_requires_verification(paths):
|
||||
raise LauncherError(
|
||||
@@ -732,18 +753,55 @@ def _play_locked(
|
||||
"run pristontale-eu update before playing"
|
||||
)
|
||||
settings = _load_or_default_settings(paths, persist=not dry_run)
|
||||
start_option = select_depot_start_option(
|
||||
read_cached_depot_launch_catalog(paths.depot_launch_catalog_cache),
|
||||
realm or settings.realm,
|
||||
)
|
||||
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:
|
||||
_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 not skip_runtime_setup:
|
||||
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(
|
||||
@@ -751,13 +809,17 @@ def _play_locked(
|
||||
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,
|
||||
start_option=start_option,
|
||||
use_gamebox=not no_box,
|
||||
concurrent_prefix_launch=reuse_prepared,
|
||||
)
|
||||
except KeyboardInterrupt:
|
||||
current_game_log = _current_game_log(
|
||||
@@ -784,6 +846,7 @@ def _play_locked(
|
||||
"ok": outcome.returncode == 0,
|
||||
"dry_run": dry_run,
|
||||
"runtime": asdict(selection),
|
||||
"realm": asdict(start_option),
|
||||
"runtime_view": asdict(overlay) if overlay else None,
|
||||
"setup_commands": [asdict(result) for result in setup_results],
|
||||
"game": asdict(outcome),
|
||||
@@ -1000,10 +1063,19 @@ def _load_or_default_settings(paths: AppPaths, *, persist: bool) -> Settings:
|
||||
|
||||
|
||||
def _runner(paths: AppPaths, operation: str, dry_run: bool) -> CommandRunner:
|
||||
timestamp = datetime.now().astimezone().strftime("%Y%m%d-%H%M%S")
|
||||
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,
|
||||
|
||||
213
ptlinux/dashboard_debug.py
Normal file
213
ptlinux/dashboard_debug.py
Normal file
@@ -0,0 +1,213 @@
|
||||
"""Read-only state snapshots for the launcher's hidden debug view."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import asdict, dataclass
|
||||
from datetime import datetime
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Callable
|
||||
|
||||
from .import_state import game_import_requires_verification
|
||||
from .paths import AppPaths
|
||||
from .runtime import (
|
||||
RuntimeErrorPT,
|
||||
RuntimeSelection,
|
||||
resolve_proton_sync_mode,
|
||||
select_runtime,
|
||||
)
|
||||
from .settings import Settings
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class RuntimeDebugInfo:
|
||||
"""Configured, selected, and last-prepared runtime state."""
|
||||
|
||||
requested_kind: str
|
||||
selected_kind: str | None
|
||||
executable: str | None
|
||||
proton_path: str | None
|
||||
prefix: str | None
|
||||
sync_configured: str
|
||||
sync_effective: str | None
|
||||
prepared_kind: str | None
|
||||
prepared_proton_path: str | None
|
||||
prepared_at: str | None
|
||||
marker_matches_selection: bool | None
|
||||
dxvk_d3d9_override: str | None
|
||||
error: str | None
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class VerificationDebugInfo:
|
||||
"""Cached trust evidence from completed depot and import operations."""
|
||||
|
||||
import_state: str
|
||||
import_verified_at: str | None
|
||||
import_build: int | None
|
||||
last_sync_at: str | None
|
||||
last_full_verification_at: str | None
|
||||
build: int | None
|
||||
version: str | None
|
||||
mode: str | None
|
||||
files_total: int | None
|
||||
files_downloaded: int | None
|
||||
files_deleted: int | None
|
||||
files_created: int | None
|
||||
files_quarantined: int | None
|
||||
|
||||
|
||||
RuntimeSelector = Callable[[Settings], RuntimeSelection]
|
||||
SyncResolver = Callable[[str], str]
|
||||
|
||||
|
||||
def inspect_runtime_debug(
|
||||
paths: AppPaths,
|
||||
settings: Settings,
|
||||
*,
|
||||
selector: RuntimeSelector = select_runtime,
|
||||
sync_resolver: SyncResolver = resolve_proton_sync_mode,
|
||||
) -> RuntimeDebugInfo:
|
||||
"""Resolve the next runtime and compare it with the preparation marker."""
|
||||
|
||||
marker = _read_json_object(paths.game_runtime_marker)
|
||||
prepared_runtime: dict[str, object] | None = None
|
||||
if marker is not None:
|
||||
raw_prepared_runtime = marker.get("runtime")
|
||||
if isinstance(raw_prepared_runtime, dict):
|
||||
prepared_runtime = raw_prepared_runtime
|
||||
prepared_kind = _text(prepared_runtime.get("kind")) if prepared_runtime else None
|
||||
prepared_proton_path = (
|
||||
_text(prepared_runtime.get("proton_path")) if prepared_runtime else None
|
||||
)
|
||||
prepared_at = _text(marker.get("configured_at")) if marker else None
|
||||
dxvk_override = _text(marker.get("dxvk_d3d9_override")) if marker else None
|
||||
selection: RuntimeSelection | None = None
|
||||
error: str | None = None
|
||||
try:
|
||||
selection = selector(settings)
|
||||
except (OSError, RuntimeErrorPT, ValueError) as exc:
|
||||
error = str(exc).strip() or type(exc).__name__
|
||||
selected_kind = selection.kind if selection is not None else None
|
||||
prefix_kind = selected_kind or prepared_kind
|
||||
prefix = None
|
||||
if prefix_kind == "umu":
|
||||
prefix = str(paths.umu_prefix)
|
||||
elif prefix_kind in {"proton7", "proton7-container"}:
|
||||
prefix = str(paths.proton_prefix)
|
||||
sync_effective = None
|
||||
if prefix_kind in {"proton7", "proton7-container"}:
|
||||
try:
|
||||
sync_effective = sync_resolver(settings.proton_sync)
|
||||
except ValueError as exc:
|
||||
error = error or (str(exc).strip() or type(exc).__name__)
|
||||
marker_matches = (
|
||||
prepared_runtime == asdict(selection)
|
||||
if prepared_runtime is not None and selection is not None
|
||||
else None
|
||||
)
|
||||
return RuntimeDebugInfo(
|
||||
requested_kind=settings.runtime,
|
||||
selected_kind=selected_kind,
|
||||
executable=selection.executable if selection is not None else None,
|
||||
proton_path=selection.proton_path if selection is not None else None,
|
||||
prefix=prefix,
|
||||
sync_configured=settings.proton_sync,
|
||||
sync_effective=sync_effective,
|
||||
prepared_kind=prepared_kind,
|
||||
prepared_proton_path=prepared_proton_path,
|
||||
prepared_at=prepared_at,
|
||||
marker_matches_selection=marker_matches,
|
||||
dxvk_d3d9_override=dxvk_override,
|
||||
error=error,
|
||||
)
|
||||
|
||||
|
||||
def read_verification_debug(paths: AppPaths) -> VerificationDebugInfo:
|
||||
"""Read cached verification state without hashing or changing game files."""
|
||||
|
||||
depot = _read_json_object(paths.native_depot_state)
|
||||
if depot is not None and depot.get("dry_run") is not False:
|
||||
depot = None
|
||||
import_manifest_exists = paths.game_import_manifest.exists()
|
||||
import_manifest = _read_json_object(paths.game_import_manifest)
|
||||
if not import_manifest_exists:
|
||||
import_state = "not applicable"
|
||||
elif game_import_requires_verification(paths):
|
||||
import_state = "pending"
|
||||
else:
|
||||
import_state = "verified"
|
||||
import_verified_at = (
|
||||
_text(import_manifest.get("verification_completed_at"))
|
||||
if import_state == "verified" and import_manifest is not None
|
||||
else None
|
||||
)
|
||||
import_build = (
|
||||
_non_negative_int(import_manifest.get("verification_build"))
|
||||
if import_state == "verified" and import_manifest is not None
|
||||
else None
|
||||
)
|
||||
last_sync_at = _text(depot.get("completed_at")) if depot else None
|
||||
full_verifications = [
|
||||
value
|
||||
for value in (
|
||||
last_sync_at if depot is not None and depot.get("repair") is True else None,
|
||||
import_verified_at,
|
||||
)
|
||||
if value is not None
|
||||
]
|
||||
last_full_verification_at = (
|
||||
max(full_verifications, key=_iso_timestamp)
|
||||
if full_verifications
|
||||
else None
|
||||
)
|
||||
return VerificationDebugInfo(
|
||||
import_state=import_state,
|
||||
import_verified_at=import_verified_at,
|
||||
import_build=import_build,
|
||||
last_sync_at=last_sync_at,
|
||||
last_full_verification_at=last_full_verification_at,
|
||||
build=_non_negative_int(depot.get("build")) if depot else None,
|
||||
version=_text(depot.get("version")) if depot else None,
|
||||
mode=(
|
||||
"full repair"
|
||||
if depot is not None and depot.get("repair") is True
|
||||
else "update"
|
||||
if depot is not None and depot.get("repair") is False
|
||||
else None
|
||||
),
|
||||
files_total=_non_negative_int(depot.get("files_total")) if depot else None,
|
||||
files_downloaded=(
|
||||
_non_negative_int(depot.get("files_downloaded")) if depot else None
|
||||
),
|
||||
files_deleted=_non_negative_int(depot.get("files_deleted")) if depot else None,
|
||||
files_created=_non_negative_int(depot.get("files_created")) if depot else None,
|
||||
files_quarantined=(
|
||||
_non_negative_int(depot.get("files_quarantined")) if depot else None
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def _read_json_object(path: Path) -> dict[str, object] | None:
|
||||
try:
|
||||
value = json.loads(path.read_text(encoding="utf-8"))
|
||||
except (OSError, UnicodeError, json.JSONDecodeError):
|
||||
return None
|
||||
return value if isinstance(value, dict) else None
|
||||
|
||||
|
||||
def _text(value: object) -> str | None:
|
||||
return value.strip() if isinstance(value, str) and value.strip() else None
|
||||
|
||||
|
||||
def _non_negative_int(value: object) -> int | None:
|
||||
if isinstance(value, bool) or not isinstance(value, int) or value < 0:
|
||||
return None
|
||||
return value
|
||||
|
||||
|
||||
def _iso_timestamp(value: str) -> float:
|
||||
try:
|
||||
return datetime.fromisoformat(value.replace("Z", "+00:00")).timestamp()
|
||||
except (OverflowError, ValueError):
|
||||
return float("-inf")
|
||||
196
ptlinux/depot.py
196
ptlinux/depot.py
@@ -73,6 +73,7 @@ ALWAYS_VERIFY_PATHS = frozenset(
|
||||
)
|
||||
SHA1_PATTERN = re.compile(r"^[0-9a-fA-F]{40}$")
|
||||
CONTENT_RANGE_PATTERN = re.compile(r"^bytes\s+(\d+)-(\d+)/(\d+)$", re.IGNORECASE)
|
||||
DEPOT_LAUNCH_CATALOG_SCHEMA = 1
|
||||
|
||||
|
||||
class DepotError(RuntimeError):
|
||||
@@ -105,6 +106,22 @@ class CreateIfMissing:
|
||||
content_base64: str
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class DepotStartOption:
|
||||
"""One publisher-advertised realm launch target."""
|
||||
|
||||
name: str
|
||||
path: str
|
||||
arguments: str
|
||||
|
||||
|
||||
DEFAULT_DEPOT_START_OPTION = DepotStartOption(
|
||||
name="Babel",
|
||||
path="Game.exe",
|
||||
arguments="PTEU_REBORN",
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class DepotSnapshot:
|
||||
"""The current, verified shape of the public game depot."""
|
||||
@@ -117,6 +134,101 @@ class DepotSnapshot:
|
||||
files: tuple[DepotFile, ...]
|
||||
deleted_files: tuple[str, ...]
|
||||
create_if_missing: tuple[CreateIfMissing, ...]
|
||||
start_options: tuple[DepotStartOption, ...] = (DEFAULT_DEPOT_START_OPTION,)
|
||||
default_start_option: str = DEFAULT_DEPOT_START_OPTION.name
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class DepotLaunchCatalog:
|
||||
"""Small cached subset of a depot manifest needed for offline Play."""
|
||||
|
||||
build: int
|
||||
start_options: tuple[DepotStartOption, ...]
|
||||
default_start_option: str
|
||||
|
||||
|
||||
def launch_catalog_from_snapshot(snapshot: DepotSnapshot) -> DepotLaunchCatalog:
|
||||
return DepotLaunchCatalog(
|
||||
build=snapshot.build,
|
||||
start_options=snapshot.start_options,
|
||||
default_start_option=snapshot.default_start_option,
|
||||
)
|
||||
|
||||
|
||||
def select_depot_start_option(
|
||||
catalog: DepotLaunchCatalog | None,
|
||||
preferred_realm: str | None = None,
|
||||
) -> DepotStartOption:
|
||||
"""Choose a published realm, falling back to Babel when possible."""
|
||||
|
||||
current = catalog or DepotLaunchCatalog(
|
||||
build=0,
|
||||
start_options=(DEFAULT_DEPOT_START_OPTION,),
|
||||
default_start_option=DEFAULT_DEPOT_START_OPTION.name,
|
||||
)
|
||||
if preferred_realm:
|
||||
preferred = preferred_realm.casefold()
|
||||
for option in current.start_options:
|
||||
if option.name.casefold() == preferred:
|
||||
return option
|
||||
for option in current.start_options:
|
||||
if option.name.casefold() == DEFAULT_DEPOT_START_OPTION.name.casefold():
|
||||
return option
|
||||
for option in current.start_options:
|
||||
if option.name.casefold() == current.default_start_option.casefold():
|
||||
return option
|
||||
return current.start_options[0]
|
||||
|
||||
|
||||
def cache_depot_launch_catalog(path: Path, snapshot: DepotSnapshot) -> None:
|
||||
"""Cache only publisher-advertised launch metadata for network-free Play."""
|
||||
|
||||
catalog = launch_catalog_from_snapshot(snapshot)
|
||||
_write_json(
|
||||
path,
|
||||
{
|
||||
"schema": DEPOT_LAUNCH_CATALOG_SCHEMA,
|
||||
"build": catalog.build,
|
||||
"default_start_option": catalog.default_start_option,
|
||||
"start_options": [asdict(option) for option in catalog.start_options],
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
def read_cached_depot_launch_catalog(path: Path) -> DepotLaunchCatalog | None:
|
||||
"""Read the last valid launch catalog without performing network I/O."""
|
||||
|
||||
try:
|
||||
value = json.loads(path.read_text(encoding="utf-8"))
|
||||
if not isinstance(value, dict):
|
||||
return None
|
||||
if value.get("schema") != DEPOT_LAUNCH_CATALOG_SCHEMA:
|
||||
return None
|
||||
build = _positive_int(value.get("build"), "cached depot build")
|
||||
raw_options = value.get("start_options")
|
||||
if not isinstance(raw_options, list):
|
||||
return None
|
||||
manifest_options = [
|
||||
{
|
||||
"Name": item.get("name"),
|
||||
"Path": item.get("path"),
|
||||
"Arguments": item.get("arguments"),
|
||||
}
|
||||
if isinstance(item, dict)
|
||||
else item
|
||||
for item in raw_options
|
||||
]
|
||||
options, default_name = _parse_start_options(
|
||||
manifest_options,
|
||||
value.get("default_start_option"),
|
||||
)
|
||||
except (OSError, UnicodeError, json.JSONDecodeError, DepotError):
|
||||
return None
|
||||
return DepotLaunchCatalog(
|
||||
build=build,
|
||||
start_options=options,
|
||||
default_start_option=default_name,
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
@@ -579,6 +691,7 @@ def sync_managed_game(
|
||||
quarantine=quarantine,
|
||||
)
|
||||
_write_json(paths.native_depot_state, asdict(result))
|
||||
cache_depot_launch_catalog(paths.depot_launch_catalog_cache, snapshot)
|
||||
_emit_progress(
|
||||
progress,
|
||||
"complete",
|
||||
@@ -913,6 +1026,19 @@ def _parse_snapshot(
|
||||
|
||||
deleted = _parse_deleted_paths(build_manifest.get("df"), seen_paths)
|
||||
create_if_missing = _parse_create_if_missing(depot_manifest.get("CreateIfMissing"))
|
||||
start_options, default_start_option = _parse_start_options(
|
||||
depot_manifest.get("StartOptions"),
|
||||
depot_manifest.get("DefaultStartOption"),
|
||||
)
|
||||
canonical_file_paths = {file.path.casefold(): file.path for file in files}
|
||||
start_options = tuple(
|
||||
DepotStartOption(
|
||||
name=option.name,
|
||||
path=canonical_file_paths.get(option.path.casefold(), option.path),
|
||||
arguments=option.arguments,
|
||||
)
|
||||
for option in start_options
|
||||
)
|
||||
return DepotSnapshot(
|
||||
region=region,
|
||||
server=server,
|
||||
@@ -922,6 +1048,8 @@ def _parse_snapshot(
|
||||
files=tuple(files),
|
||||
deleted_files=deleted,
|
||||
create_if_missing=create_if_missing,
|
||||
start_options=start_options,
|
||||
default_start_option=default_start_option,
|
||||
)
|
||||
|
||||
|
||||
@@ -966,6 +1094,74 @@ def _parse_create_if_missing(value: object) -> tuple[CreateIfMissing, ...]:
|
||||
return tuple(entries)
|
||||
|
||||
|
||||
def _parse_start_options(
|
||||
value: object,
|
||||
default_value: object,
|
||||
) -> tuple[tuple[DepotStartOption, ...], str]:
|
||||
"""Validate the launch targets published by the official depot manifest."""
|
||||
|
||||
# Older test fixtures and historical manifests did not expose StartOptions.
|
||||
# Babel remains the launcher-safe compatibility default for that case.
|
||||
if value is None:
|
||||
return (DEFAULT_DEPOT_START_OPTION,), DEFAULT_DEPOT_START_OPTION.name
|
||||
if not isinstance(value, list) or not value:
|
||||
raise DepotError("StartOptions must be a non-empty array")
|
||||
options: list[DepotStartOption] = []
|
||||
seen_names: set[str] = set()
|
||||
for item in value:
|
||||
if not isinstance(item, dict):
|
||||
raise DepotError("StartOptions contains a non-object entry")
|
||||
name = _start_option_text(item.get("Name"), "StartOptions Name", 64)
|
||||
folded_name = name.casefold()
|
||||
if folded_name in seen_names:
|
||||
raise DepotError(f"StartOptions has duplicate realm name: {name}")
|
||||
seen_names.add(folded_name)
|
||||
path = _normalise_relative_path(item.get("Path"), f"StartOptions path for {name}")
|
||||
if PurePosixPath(path).suffix.casefold() != ".exe":
|
||||
raise DepotError(f"StartOptions path is not an executable for {name}: {path}")
|
||||
arguments = _start_option_text(
|
||||
item.get("Arguments"),
|
||||
f"StartOptions arguments for {name}",
|
||||
2048,
|
||||
)
|
||||
options.append(DepotStartOption(name=name, path=path, arguments=arguments))
|
||||
|
||||
if default_value is None:
|
||||
default_name = next(
|
||||
(
|
||||
option.name
|
||||
for option in options
|
||||
if option.name.casefold() == DEFAULT_DEPOT_START_OPTION.name.casefold()
|
||||
),
|
||||
options[0].name,
|
||||
)
|
||||
else:
|
||||
default_name = _start_option_text(
|
||||
default_value,
|
||||
"DefaultStartOption",
|
||||
64,
|
||||
)
|
||||
matching_default = next(
|
||||
(option.name for option in options if option.name.casefold() == default_name.casefold()),
|
||||
None,
|
||||
)
|
||||
if matching_default is None:
|
||||
raise DepotError(
|
||||
f"DefaultStartOption does not match an advertised realm: {default_name}"
|
||||
)
|
||||
default_name = matching_default
|
||||
return tuple(options), default_name
|
||||
|
||||
|
||||
def _start_option_text(value: object, field: str, maximum_length: int) -> str:
|
||||
if not isinstance(value, str):
|
||||
raise DepotError(f"{field} must be a string")
|
||||
text = value.strip()
|
||||
if not text or len(text) > maximum_length or any(ord(character) < 32 for character in text):
|
||||
raise DepotError(f"invalid {field}")
|
||||
return text
|
||||
|
||||
|
||||
def _decode_create_if_missing(entry: CreateIfMissing) -> bytes:
|
||||
try:
|
||||
content = b64decode(entry.content_base64, validate=True)
|
||||
|
||||
59
ptlinux/desktop_identity.py
Normal file
59
ptlinux/desktop_identity.py
Normal file
@@ -0,0 +1,59 @@
|
||||
"""Stable desktop identity shared by the launcher and Wine game window."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from pathlib import Path
|
||||
from typing import Mapping
|
||||
|
||||
|
||||
APPLICATION_NAME = "PristonTale EU"
|
||||
APPLICATION_WM_CLASS = "PristonTaleEU"
|
||||
APPLICATION_WM_INSTANCE = "pristontale-eu"
|
||||
APPLICATION_ICON_FILENAME = "eu.pristontale.PTLinux.png"
|
||||
APPLICATION_ICON_VARIABLE = "PTLINUX_ICON_PATH"
|
||||
|
||||
|
||||
def find_application_icon(
|
||||
environment: Mapping[str, str] | None = None,
|
||||
) -> Path | None:
|
||||
"""Find the installed icon, with source-tree fallbacks for development."""
|
||||
|
||||
env = os.environ if environment is None else environment
|
||||
candidates: list[Path] = []
|
||||
explicit = env.get(APPLICATION_ICON_VARIABLE)
|
||||
if explicit:
|
||||
candidates.append(Path(explicit).expanduser())
|
||||
|
||||
# Some distributions bundle the icon beside this module. The installer
|
||||
# publishes the same asset through the user's XDG data directory.
|
||||
candidates.append(Path(__file__).with_name(APPLICATION_ICON_FILENAME))
|
||||
home_value = env.get("HOME")
|
||||
if home_value:
|
||||
home = Path(home_value).expanduser()
|
||||
data_home_value = env.get("XDG_DATA_HOME")
|
||||
data_home = (
|
||||
Path(data_home_value).expanduser()
|
||||
if data_home_value
|
||||
else home / ".local/share"
|
||||
)
|
||||
candidates.append(data_home / "pixmaps" / APPLICATION_ICON_FILENAME)
|
||||
default_icon = home / ".local/share/pixmaps" / APPLICATION_ICON_FILENAME
|
||||
if default_icon not in candidates:
|
||||
candidates.append(default_icon)
|
||||
|
||||
# The source checkout keeps installer assets under packaging/.
|
||||
candidates.append(
|
||||
Path(__file__).resolve().parent.parent
|
||||
/ "packaging"
|
||||
/ APPLICATION_ICON_FILENAME
|
||||
)
|
||||
seen: set[Path] = set()
|
||||
for candidate in candidates:
|
||||
resolved = candidate.resolve(strict=False)
|
||||
if resolved in seen:
|
||||
continue
|
||||
seen.add(resolved)
|
||||
if resolved.is_file():
|
||||
return resolved
|
||||
return None
|
||||
@@ -11,6 +11,7 @@ import shutil
|
||||
from typing import Mapping
|
||||
|
||||
from .launcher import find_wine_tools
|
||||
from .runtime import find_steam_proton7
|
||||
|
||||
|
||||
ARCH_FAMILY_DISTRIBUTIONS = frozenset(
|
||||
@@ -176,19 +177,7 @@ def _package_hint(distribution: str) -> str | None:
|
||||
|
||||
|
||||
def _has_steam_proton7(environment: Mapping[str, str]) -> bool:
|
||||
home_value = environment.get("HOME")
|
||||
if not home_value:
|
||||
return False
|
||||
home = Path(home_value)
|
||||
for steam_root in (
|
||||
home / ".local/share/Steam",
|
||||
home / ".steam/steam",
|
||||
home / "snap/steam/common/.local/share/Steam",
|
||||
home / ".var/app/com.valvesoftware.Steam/.local/share/Steam",
|
||||
):
|
||||
if (steam_root / "steamapps/common/Proton 7.0/proton").is_file():
|
||||
return True
|
||||
return False
|
||||
return find_steam_proton7(environment) is not None
|
||||
|
||||
|
||||
def _platform_optional_tools(distribution: str, *, architecture: str) -> list[ToolStatus]:
|
||||
|
||||
@@ -10,8 +10,13 @@ import sys
|
||||
import time
|
||||
from collections.abc import Collection, Iterable, Sequence
|
||||
|
||||
from .desktop_identity import (
|
||||
APPLICATION_NAME,
|
||||
APPLICATION_WM_CLASS,
|
||||
APPLICATION_WM_INSTANCE,
|
||||
)
|
||||
|
||||
WINDOW_TITLE = "PristonTale EU"
|
||||
WINDOW_TITLE = APPLICATION_NAME
|
||||
WINDOW_DISCOVERY_TIMEOUT_SECONDS = 60.0
|
||||
WINDOW_READY_DELAY_SECONDS = 0.5
|
||||
_X_ERROR_HANDLER = ctypes.CFUNCTYPE(ctypes.c_int, ctypes.c_void_p, ctypes.c_void_p)
|
||||
@@ -66,6 +71,13 @@ class _MotifWmHints(ctypes.Structure):
|
||||
]
|
||||
|
||||
|
||||
class _XClassHint(ctypes.Structure):
|
||||
_fields_ = [
|
||||
("res_name", ctypes.c_char_p),
|
||||
("res_class", ctypes.c_char_p),
|
||||
]
|
||||
|
||||
|
||||
class _X11:
|
||||
def __init__(self) -> None:
|
||||
library_path = ctypes.util.find_library("X11") or "libX11.so.6"
|
||||
@@ -135,6 +147,33 @@ class _X11:
|
||||
self.library.XMapRaised(self.display, window)
|
||||
self.library.XFlush(self.display)
|
||||
|
||||
def set_window_identity(
|
||||
self,
|
||||
window: int,
|
||||
*,
|
||||
resource_name: str = APPLICATION_WM_INSTANCE,
|
||||
resource_class: str = APPLICATION_WM_CLASS,
|
||||
) -> None:
|
||||
"""Replace Proton's generic steamapp identity before taskbar matching."""
|
||||
|
||||
for label, value in (
|
||||
("resource name", resource_name),
|
||||
("resource class", resource_class),
|
||||
):
|
||||
if not value or "\x00" in value:
|
||||
raise GameboxRuntimeError(f"invalid X11 {label}")
|
||||
hints = _XClassHint(
|
||||
res_name=resource_name.encode("utf-8"),
|
||||
res_class=resource_class.encode("utf-8"),
|
||||
)
|
||||
if self.library.XSetClassHint(
|
||||
self.display,
|
||||
window,
|
||||
ctypes.byref(hints),
|
||||
) == 0:
|
||||
raise GameboxRuntimeError("could not set the game window identity")
|
||||
self.library.XSync(self.display, 0)
|
||||
|
||||
def suppress_wm_ping(self, window: int) -> bool:
|
||||
"""Stop the WM hang prompt for one window, preserving its other protocols."""
|
||||
ping_atom = int(self.library.XInternAtom(self.display, b"_NET_WM_PING", 0))
|
||||
@@ -265,6 +304,12 @@ class _X11:
|
||||
ctypes.c_int,
|
||||
]
|
||||
self.library.XChangeProperty.restype = ctypes.c_int
|
||||
self.library.XSetClassHint.argtypes = [
|
||||
ctypes.c_void_p,
|
||||
ctypes.c_ulong,
|
||||
ctypes.POINTER(_XClassHint),
|
||||
]
|
||||
self.library.XSetClassHint.restype = ctypes.c_int
|
||||
self.library.XGetWMProtocols.argtypes = [
|
||||
ctypes.c_void_p,
|
||||
ctypes.c_ulong,
|
||||
@@ -325,6 +370,18 @@ def _prepare_game_window(
|
||||
mapper.map_window(window, borderless=borderless)
|
||||
|
||||
|
||||
def _apply_game_window_identity(mapper: _X11, window: int) -> None:
|
||||
try:
|
||||
mapper.set_window_identity(window)
|
||||
except Exception as exc:
|
||||
print(
|
||||
f"Warning: could not assign the PristonTale EU taskbar identity ({exc}); "
|
||||
"continuing normally",
|
||||
file=sys.stderr,
|
||||
flush=True,
|
||||
)
|
||||
|
||||
|
||||
def run_game_window(
|
||||
command: Sequence[str],
|
||||
window_title: str = WINDOW_TITLE,
|
||||
@@ -371,6 +428,7 @@ def run_game_window(
|
||||
game_window = window
|
||||
game_title = title
|
||||
game_window_created_at = now
|
||||
_apply_game_window_identity(mapper, window)
|
||||
print(
|
||||
f"Game window created; preparing it: {title}",
|
||||
file=sys.stderr,
|
||||
|
||||
393
ptlinux/identity_probe.py
Normal file
393
ptlinux/identity_probe.py
Normal file
@@ -0,0 +1,393 @@
|
||||
"""Privacy-safe checks for the Windows identity inputs used by the game client."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
import hashlib
|
||||
import json
|
||||
from pathlib import Path
|
||||
import re
|
||||
|
||||
from .command import CommandRunner
|
||||
from .paths import AppPaths
|
||||
from .runtime import RuntimeSelection, run_runtime_prefix_command
|
||||
|
||||
|
||||
IDENTITY_PROBE_SHA256 = "677cb5aa19f8dac65b982462bb30974feed05399fdd759765c3b6e802d4c838b"
|
||||
IDENTITY_PROBE_TIMEOUT_SECONDS = 45.0
|
||||
IDENTITY_PROBE_PRIVACY = (
|
||||
"Only success flags, row counts, value lengths, VARIANT types, adapter type, "
|
||||
"and generic-value quality flags are collected. Computer names, serial numbers, "
|
||||
"MAC addresses, and WMI values never leave the Windows helper."
|
||||
)
|
||||
WMI_PROPERTY_NAMES = (
|
||||
"processor_id",
|
||||
"processor_unique_id",
|
||||
"processor_manufacturer",
|
||||
"processor_max_clock_speed",
|
||||
"bios_release_date",
|
||||
"bios_serial_number",
|
||||
"bios_identification_code",
|
||||
"bios_smbios_version",
|
||||
"bios_manufacturer",
|
||||
"baseboard_serial_number",
|
||||
"baseboard_name",
|
||||
"baseboard_manufacturer",
|
||||
"baseboard_model",
|
||||
"video_name",
|
||||
"video_driver_version",
|
||||
)
|
||||
HRESULT_PATTERN = re.compile(r"0x[0-9A-F]{8}\Z")
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class IdentityProbeResult:
|
||||
helper: str
|
||||
helper_sha256: str | None
|
||||
helper_verified: bool
|
||||
ran: bool
|
||||
ready: bool | None
|
||||
error: str | None
|
||||
report: dict[str, object] | None
|
||||
observations: tuple[str, ...]
|
||||
warnings: tuple[str, ...]
|
||||
|
||||
|
||||
def packaged_identity_probe() -> Path:
|
||||
return Path(__file__).resolve().parent / "native" / "pt-identity-probe.exe"
|
||||
|
||||
|
||||
def probe_client_identity(
|
||||
paths: AppPaths,
|
||||
selection: RuntimeSelection,
|
||||
runner: CommandRunner,
|
||||
*,
|
||||
timeout_seconds: float = IDENTITY_PROBE_TIMEOUT_SECONDS,
|
||||
) -> IdentityProbeResult:
|
||||
"""Check identity API availability without returning the underlying values."""
|
||||
|
||||
helper = packaged_identity_probe()
|
||||
helper_sha256 = _sha256_if_file(helper)
|
||||
helper_verified = helper_sha256 == IDENTITY_PROBE_SHA256
|
||||
if not helper_verified:
|
||||
error = (
|
||||
"the packaged privacy-safe identity helper is missing"
|
||||
if helper_sha256 is None
|
||||
else "the packaged privacy-safe identity helper failed verification"
|
||||
)
|
||||
return IdentityProbeResult(
|
||||
helper=str(helper),
|
||||
helper_sha256=helper_sha256,
|
||||
helper_verified=False,
|
||||
ran=False,
|
||||
ready=None,
|
||||
error=error,
|
||||
report=None,
|
||||
observations=(),
|
||||
warnings=(error,),
|
||||
)
|
||||
try:
|
||||
outcome = run_runtime_prefix_command(
|
||||
paths,
|
||||
selection,
|
||||
runner,
|
||||
[str(helper)],
|
||||
check=False,
|
||||
timeout_seconds=timeout_seconds,
|
||||
)
|
||||
except (OSError, RuntimeError) as exc:
|
||||
error = f"the client-identity helper could not start: {exc}"
|
||||
return IdentityProbeResult(
|
||||
helper=str(helper),
|
||||
helper_sha256=helper_sha256,
|
||||
helper_verified=True,
|
||||
ran=False,
|
||||
ready=None,
|
||||
error=error,
|
||||
report=None,
|
||||
observations=(),
|
||||
warnings=(error,),
|
||||
)
|
||||
if outcome.timed_out:
|
||||
error = "the client-identity helper timed out"
|
||||
return _failed_result(helper, helper_sha256, error, ran=True)
|
||||
if outcome.returncode != 0:
|
||||
status = "unknown" if outcome.returncode is None else str(outcome.returncode)
|
||||
error = f"the client-identity helper exited with status {status}"
|
||||
return _failed_result(helper, helper_sha256, error, ran=True)
|
||||
try:
|
||||
report = _parse_privacy_safe_report(outcome.stdout)
|
||||
except (ValueError, json.JSONDecodeError) as exc:
|
||||
error = f"the client-identity helper returned invalid data: {exc}"
|
||||
return _failed_result(helper, helper_sha256, error, ran=True)
|
||||
warnings = _identity_warnings(report)
|
||||
return IdentityProbeResult(
|
||||
helper=str(helper),
|
||||
helper_sha256=helper_sha256,
|
||||
helper_verified=True,
|
||||
ran=True,
|
||||
ready=not warnings,
|
||||
error=None,
|
||||
report=report,
|
||||
observations=_identity_observations(report),
|
||||
warnings=warnings,
|
||||
)
|
||||
|
||||
|
||||
def _failed_result(
|
||||
helper: Path,
|
||||
helper_sha256: str | None,
|
||||
error: str,
|
||||
*,
|
||||
ran: bool,
|
||||
) -> IdentityProbeResult:
|
||||
return IdentityProbeResult(
|
||||
helper=str(helper),
|
||||
helper_sha256=helper_sha256,
|
||||
helper_verified=True,
|
||||
ran=ran,
|
||||
ready=None,
|
||||
error=error,
|
||||
report=None,
|
||||
observations=(),
|
||||
warnings=(error,),
|
||||
)
|
||||
|
||||
|
||||
def _parse_privacy_safe_report(stdout: str) -> dict[str, object]:
|
||||
lines = [line.strip() for line in stdout.splitlines() if line.strip()]
|
||||
if not lines:
|
||||
raise ValueError("empty output")
|
||||
value = json.loads(lines[-1])
|
||||
root = _object(value, "root")
|
||||
if _integer(root.get("schema"), "schema") != 2:
|
||||
raise ValueError("unsupported schema")
|
||||
computer = _object(root.get("computer_name"), "computer_name")
|
||||
volume = _object(root.get("volume"), "volume")
|
||||
adapters = _object(root.get("adapters"), "adapters")
|
||||
firmware = _object(root.get("firmware"), "firmware")
|
||||
wmi = _object(root.get("wmi"), "wmi")
|
||||
properties = _object(wmi.get("properties"), "wmi.properties")
|
||||
safe_properties: dict[str, object] = {}
|
||||
for name in WMI_PROPERTY_NAMES:
|
||||
entry = _object(properties.get(name), f"wmi.properties.{name}")
|
||||
safe_properties[name] = {
|
||||
"result": _hresult(entry.get("result"), f"{name}.result"),
|
||||
"variant_type": _integer(entry.get("variant_type"), f"{name}.variant_type"),
|
||||
"rows": _integer(entry.get("rows"), f"{name}.rows"),
|
||||
"nonempty_rows": _integer(
|
||||
entry.get("nonempty_rows"),
|
||||
f"{name}.nonempty_rows",
|
||||
),
|
||||
"maximum_length": _integer(
|
||||
entry.get("maximum_length"),
|
||||
f"{name}.maximum_length",
|
||||
),
|
||||
"placeholder_rows": _integer(
|
||||
entry.get("placeholder_rows"),
|
||||
f"{name}.placeholder_rows",
|
||||
),
|
||||
"wine_rows": _integer(
|
||||
entry.get("wine_rows"),
|
||||
f"{name}.wine_rows",
|
||||
),
|
||||
}
|
||||
return {
|
||||
"schema": 2,
|
||||
"computer_name": {
|
||||
"success": _boolean(computer.get("success"), "computer_name.success"),
|
||||
"nonempty": _boolean(computer.get("nonempty"), "computer_name.nonempty"),
|
||||
"length": _integer(computer.get("length"), "computer_name.length"),
|
||||
},
|
||||
"volume": {
|
||||
"success": _boolean(volume.get("success"), "volume.success"),
|
||||
"serial_nonzero": _boolean(
|
||||
volume.get("serial_nonzero"),
|
||||
"volume.serial_nonzero",
|
||||
),
|
||||
},
|
||||
"adapters": {
|
||||
"status": _integer(adapters.get("status"), "adapters.status"),
|
||||
"count": _integer(adapters.get("count"), "adapters.count"),
|
||||
"first_address_length": _integer(
|
||||
adapters.get("first_address_length"),
|
||||
"adapters.first_address_length",
|
||||
),
|
||||
"first_address_nonzero": _boolean(
|
||||
adapters.get("first_address_nonzero"),
|
||||
"adapters.first_address_nonzero",
|
||||
),
|
||||
"first_address_locally_administered": _boolean(
|
||||
adapters.get("first_address_locally_administered"),
|
||||
"adapters.first_address_locally_administered",
|
||||
),
|
||||
"first_address_multicast": _boolean(
|
||||
adapters.get("first_address_multicast"),
|
||||
"adapters.first_address_multicast",
|
||||
),
|
||||
"first_type": _integer(adapters.get("first_type"), "adapters.first_type"),
|
||||
},
|
||||
"firmware": {
|
||||
"raw_smbios_bytes": _integer(
|
||||
firmware.get("raw_smbios_bytes"),
|
||||
"firmware.raw_smbios_bytes",
|
||||
),
|
||||
},
|
||||
"wmi": {
|
||||
"com_initialize": _hresult(wmi.get("com_initialize"), "wmi.com_initialize"),
|
||||
"com_security": _hresult(wmi.get("com_security"), "wmi.com_security"),
|
||||
"connect": _hresult(wmi.get("connect"), "wmi.connect"),
|
||||
"properties": safe_properties,
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def _identity_warnings(report: dict[str, object]) -> tuple[str, ...]:
|
||||
warnings: list[str] = []
|
||||
computer = _object(report["computer_name"], "computer_name")
|
||||
volume = _object(report["volume"], "volume")
|
||||
adapters = _object(report["adapters"], "adapters")
|
||||
wmi = _object(report["wmi"], "wmi")
|
||||
properties = _object(wmi["properties"], "wmi.properties")
|
||||
computer_success = _boolean(computer["success"], "computer_name.success")
|
||||
computer_nonempty = _boolean(computer["nonempty"], "computer_name.nonempty")
|
||||
volume_success = _boolean(volume["success"], "volume.success")
|
||||
volume_serial_nonzero = _boolean(volume["serial_nonzero"], "volume.serial_nonzero")
|
||||
adapter_status = _integer(adapters["status"], "adapters.status")
|
||||
adapter_count = _integer(adapters["count"], "adapters.count")
|
||||
first_address_length = _integer(
|
||||
adapters["first_address_length"],
|
||||
"adapters.first_address_length",
|
||||
)
|
||||
first_address_nonzero = _boolean(
|
||||
adapters["first_address_nonzero"],
|
||||
"adapters.first_address_nonzero",
|
||||
)
|
||||
if not computer_success or not computer_nonempty:
|
||||
warnings.append("Wine did not provide the computer name used by the login request.")
|
||||
if not volume_success or not volume_serial_nonzero:
|
||||
warnings.append("Wine did not provide a usable C: volume identity.")
|
||||
if (
|
||||
adapter_status != 0
|
||||
or adapter_count == 0
|
||||
or first_address_length < 6
|
||||
or not first_address_nonzero
|
||||
):
|
||||
warnings.append(
|
||||
"Wine did not provide a usable first network-adapter identity; the game "
|
||||
"would send its UUID fallback instead."
|
||||
)
|
||||
if _hresult(wmi["connect"], "wmi.connect") != "0x00000000":
|
||||
warnings.append("Wine could not connect to ROOT\\CIMV2 for the login identity query.")
|
||||
else:
|
||||
if not _any_nonempty(properties, ("processor_id", "processor_unique_id")):
|
||||
warnings.append("Wine returned no processor identifier through WMI.")
|
||||
if not _any_nonempty(
|
||||
properties,
|
||||
(
|
||||
"bios_release_date",
|
||||
"bios_serial_number",
|
||||
"bios_identification_code",
|
||||
"bios_smbios_version",
|
||||
"bios_manufacturer",
|
||||
),
|
||||
):
|
||||
warnings.append("Wine returned no BIOS identity fields through WMI.")
|
||||
if not _any_nonempty(
|
||||
properties,
|
||||
(
|
||||
"baseboard_serial_number",
|
||||
"baseboard_name",
|
||||
"baseboard_manufacturer",
|
||||
"baseboard_model",
|
||||
),
|
||||
):
|
||||
warnings.append("Wine returned no baseboard identity fields through WMI.")
|
||||
if not _any_nonempty(properties, ("video_name",)):
|
||||
warnings.append("Wine returned no video-controller name through WMI.")
|
||||
return tuple(warnings)
|
||||
|
||||
|
||||
def _identity_observations(report: dict[str, object]) -> tuple[str, ...]:
|
||||
observations: list[str] = []
|
||||
adapters = _object(report["adapters"], "adapters")
|
||||
if _boolean(
|
||||
adapters["first_address_locally_administered"],
|
||||
"adapters.first_address_locally_administered",
|
||||
):
|
||||
observations.append(
|
||||
"The first network-adapter address is locally administered."
|
||||
)
|
||||
if _boolean(
|
||||
adapters["first_address_multicast"],
|
||||
"adapters.first_address_multicast",
|
||||
):
|
||||
observations.append("The first network-adapter address has the multicast bit set.")
|
||||
wmi = _object(report["wmi"], "wmi")
|
||||
properties = _object(wmi["properties"], "wmi.properties")
|
||||
placeholder_names: list[str] = []
|
||||
wine_names: list[str] = []
|
||||
for name in WMI_PROPERTY_NAMES:
|
||||
entry = _object(properties[name], name)
|
||||
if _integer(entry["placeholder_rows"], f"{name}.placeholder_rows") > 0:
|
||||
placeholder_names.append(name)
|
||||
if _integer(entry["wine_rows"], f"{name}.wine_rows") > 0:
|
||||
wine_names.append(name)
|
||||
if placeholder_names:
|
||||
observations.append(
|
||||
"Common placeholder WMI values were observed in: "
|
||||
+ ", ".join(placeholder_names)
|
||||
)
|
||||
if wine_names:
|
||||
observations.append(
|
||||
"Wine-labelled WMI values were observed in: " + ", ".join(wine_names)
|
||||
)
|
||||
return tuple(observations)
|
||||
|
||||
|
||||
def _any_nonempty(properties: dict[str, object], names: tuple[str, ...]) -> bool:
|
||||
for name in names:
|
||||
entry = _object(properties[name], name)
|
||||
if (
|
||||
_hresult(entry["result"], f"{name}.result") == "0x00000000"
|
||||
and _integer(entry["nonempty_rows"], f"{name}.nonempty_rows") > 0
|
||||
):
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _object(value: object, label: str) -> dict[str, object]:
|
||||
if not isinstance(value, dict) or not all(isinstance(key, str) for key in value):
|
||||
raise ValueError(f"{label} must be an object")
|
||||
return value
|
||||
|
||||
|
||||
def _boolean(value: object, label: str) -> bool:
|
||||
if not isinstance(value, bool):
|
||||
raise ValueError(f"{label} must be a boolean")
|
||||
return value
|
||||
|
||||
|
||||
def _integer(value: object, label: str) -> int:
|
||||
if isinstance(value, bool) or not isinstance(value, int) or value < 0 or value > 0xFFFFFFFF:
|
||||
raise ValueError(f"{label} must be an unsigned 32-bit integer")
|
||||
return value
|
||||
|
||||
|
||||
def _hresult(value: object, label: str) -> str:
|
||||
if not isinstance(value, str) or HRESULT_PATTERN.fullmatch(value) is None:
|
||||
raise ValueError(f"{label} must be a hexadecimal HRESULT")
|
||||
return value
|
||||
|
||||
|
||||
def _sha256_if_file(path: Path) -> str | None:
|
||||
try:
|
||||
if not path.is_file():
|
||||
return None
|
||||
digest = hashlib.sha256()
|
||||
with path.open("rb") as handle:
|
||||
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
|
||||
digest.update(chunk)
|
||||
return digest.hexdigest()
|
||||
except OSError:
|
||||
return None
|
||||
85
ptlinux/import_state.py
Normal file
85
ptlinux/import_state.py
Normal file
@@ -0,0 +1,85 @@
|
||||
"""Persistent trust state for game trees seeded from another installation."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
from uuid import uuid4
|
||||
|
||||
from .paths import AppPaths
|
||||
|
||||
|
||||
IMPORT_STATE_SCHEMA = 2
|
||||
|
||||
|
||||
def game_import_requires_verification(paths: AppPaths) -> bool:
|
||||
"""Return whether a managed imported tree still needs a full depot repair."""
|
||||
|
||||
manifest = paths.game_import_manifest
|
||||
if not manifest.exists():
|
||||
return False
|
||||
try:
|
||||
value = json.loads(manifest.read_text(encoding="utf-8"))
|
||||
except (OSError, UnicodeError, json.JSONDecodeError):
|
||||
return True
|
||||
if not isinstance(value, dict):
|
||||
return True
|
||||
build = value.get("verification_build")
|
||||
completed_at = value.get("verification_completed_at")
|
||||
return not (
|
||||
value.get("verification_required") is False
|
||||
and isinstance(build, int)
|
||||
and not isinstance(build, bool)
|
||||
and build > 0
|
||||
and isinstance(completed_at, str)
|
||||
and bool(completed_at.strip())
|
||||
)
|
||||
|
||||
|
||||
def mark_game_import_verified(
|
||||
paths: AppPaths,
|
||||
*,
|
||||
build: int,
|
||||
files_quarantined: int,
|
||||
quarantine_directory: str | None,
|
||||
) -> None:
|
||||
"""Record a completed full-manifest verification of an imported tree."""
|
||||
|
||||
if isinstance(build, bool) or build <= 0:
|
||||
raise ValueError("verified import build must be positive")
|
||||
if files_quarantined < 0:
|
||||
raise ValueError("quarantined file count cannot be negative")
|
||||
try:
|
||||
value = json.loads(paths.game_import_manifest.read_text(encoding="utf-8"))
|
||||
except (OSError, UnicodeError, json.JSONDecodeError):
|
||||
value = {}
|
||||
if not isinstance(value, dict):
|
||||
value = {}
|
||||
value.update(
|
||||
{
|
||||
"schema": IMPORT_STATE_SCHEMA,
|
||||
"verification_required": False,
|
||||
"verification_build": build,
|
||||
"verification_completed_at": datetime.now().astimezone().isoformat(),
|
||||
"files_quarantined": files_quarantined,
|
||||
"quarantine_directory": quarantine_directory,
|
||||
}
|
||||
)
|
||||
_write_json(paths.game_import_manifest, value)
|
||||
|
||||
|
||||
def _write_json(path: Path, value: object) -> None:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
temporary = path.with_name(f".{path.name}.{os.getpid()}.{uuid4().hex}.tmp")
|
||||
try:
|
||||
with temporary.open("x", encoding="utf-8") as handle:
|
||||
json.dump(value, handle, indent=2, sort_keys=True)
|
||||
handle.write("\n")
|
||||
handle.flush()
|
||||
os.fsync(handle.fileno())
|
||||
temporary.replace(path)
|
||||
except BaseException:
|
||||
temporary.unlink(missing_ok=True)
|
||||
raise
|
||||
@@ -7,13 +7,51 @@ from datetime import datetime
|
||||
import fcntl
|
||||
import os
|
||||
from pathlib import Path
|
||||
from typing import Iterator
|
||||
from typing import Iterator, TextIO
|
||||
|
||||
|
||||
class OperationBusyError(RuntimeError):
|
||||
"""Raised when another launcher-prefix operation is still active."""
|
||||
|
||||
|
||||
class LauncherPlayLease:
|
||||
"""Exclusive preparation lease that can become a shared game lease."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
handle: TextIO | None,
|
||||
*,
|
||||
preparation_allowed: bool,
|
||||
record_owned: bool,
|
||||
) -> None:
|
||||
self._handle = handle
|
||||
self.preparation_allowed = preparation_allowed
|
||||
self._record_owned = record_owned
|
||||
self._shared = not preparation_allowed
|
||||
|
||||
@classmethod
|
||||
def dry_run(cls) -> LauncherPlayLease:
|
||||
return cls(None, preparation_allowed=True, record_owned=False)
|
||||
|
||||
@property
|
||||
def shared(self) -> bool:
|
||||
return self._shared
|
||||
|
||||
def allow_shared_clients(self) -> None:
|
||||
"""Downgrade after all runtime mutations have completed."""
|
||||
|
||||
if self._shared:
|
||||
return
|
||||
if self._handle is not None:
|
||||
_write_owner(self._handle, "play")
|
||||
fcntl.flock(self._handle.fileno(), fcntl.LOCK_SH)
|
||||
self._shared = True
|
||||
|
||||
@property
|
||||
def record_owned(self) -> bool:
|
||||
return self._record_owned
|
||||
|
||||
|
||||
@contextmanager
|
||||
def launcher_operation_lock(lock_path: Path, operation: str) -> Iterator[None]:
|
||||
lock_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
@@ -26,13 +64,7 @@ def launcher_operation_lock(lock_path: Path, operation: str) -> Iterator[None]:
|
||||
raise OperationBusyError(
|
||||
f"launcher files are busy ({owner}); wait for it to finish and try again"
|
||||
) from exc
|
||||
handle.seek(0)
|
||||
handle.truncate()
|
||||
handle.write(
|
||||
f"{operation}, pid={os.getpid()}, "
|
||||
f"started={datetime.now().astimezone().isoformat()}\n"
|
||||
)
|
||||
handle.flush()
|
||||
_write_owner(handle, operation)
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
@@ -40,3 +72,67 @@ def launcher_operation_lock(lock_path: Path, operation: str) -> Iterator[None]:
|
||||
handle.truncate()
|
||||
handle.flush()
|
||||
fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
|
||||
|
||||
|
||||
@contextmanager
|
||||
def launcher_play_lock(
|
||||
lock_path: Path,
|
||||
operation: str = "play",
|
||||
) -> Iterator[LauncherPlayLease]:
|
||||
"""Serialize preparation, then allow multiple active game clients.
|
||||
|
||||
The first play process acquires an exclusive lease so it may safely rebuild
|
||||
the runtime view or prefix. It downgrades that lease immediately before
|
||||
starting Game.exe. Later play processes then receive shared, read-only leases,
|
||||
while updates and repairs remain excluded until every client exits.
|
||||
"""
|
||||
|
||||
lock_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
with lock_path.open("a+", encoding="utf-8") as handle:
|
||||
try:
|
||||
fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
|
||||
except BlockingIOError:
|
||||
try:
|
||||
fcntl.flock(handle.fileno(), fcntl.LOCK_SH | fcntl.LOCK_NB)
|
||||
except BlockingIOError as exc:
|
||||
owner = _read_owner(handle)
|
||||
raise OperationBusyError(
|
||||
f"launcher files are busy ({owner}); wait for it to finish and try again"
|
||||
) from exc
|
||||
lease = LauncherPlayLease(
|
||||
handle,
|
||||
preparation_allowed=False,
|
||||
record_owned=False,
|
||||
)
|
||||
else:
|
||||
_write_owner(handle, f"{operation} preparation")
|
||||
lease = LauncherPlayLease(
|
||||
handle,
|
||||
preparation_allowed=True,
|
||||
record_owned=True,
|
||||
)
|
||||
try:
|
||||
yield lease
|
||||
finally:
|
||||
# A downgraded owner record remains useful while any secondary shared
|
||||
# lease is active. The next exclusive operation atomically replaces it.
|
||||
if lease.record_owned and not lease.shared:
|
||||
handle.seek(0)
|
||||
handle.truncate()
|
||||
handle.flush()
|
||||
fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
|
||||
|
||||
|
||||
def _read_owner(handle: TextIO) -> str:
|
||||
handle.seek(0)
|
||||
return handle.read().strip() or "unknown operation"
|
||||
|
||||
|
||||
def _write_owner(handle: TextIO, operation: str) -> None:
|
||||
handle.seek(0)
|
||||
handle.truncate()
|
||||
handle.write(
|
||||
f"{operation}, pid={os.getpid()}, "
|
||||
f"started={datetime.now().astimezone().isoformat()}\n"
|
||||
)
|
||||
handle.flush()
|
||||
|
||||
BIN
ptlinux/native/pt-identity-probe.exe
Normal file
BIN
ptlinux/native/pt-identity-probe.exe
Normal file
Binary file not shown.
@@ -20,6 +20,8 @@ OFFICIAL_LOGO_URL = "https://pristontale.eu/public/img/pristontale/logo/logo_sma
|
||||
BROADCAST_ENDPOINT = "http://5.135.35.188:90/broadcasts"
|
||||
MAX_NEWS_BYTES = 2 * 1024 * 1024
|
||||
MAX_NEWS_IMAGE_BYTES = 12 * 1024 * 1024
|
||||
MAX_BROADCAST_AUTHOR_CHARACTERS = 100
|
||||
MAX_BROADCAST_CONTENT_CHARACTERS = 2000
|
||||
PNG_SIGNATURE = b"\x89PNG\r\n\x1a\n"
|
||||
|
||||
|
||||
@@ -241,26 +243,41 @@ def _parse_latest_broadcast(payload: bytes) -> Broadcast:
|
||||
value = json.loads(payload.decode("utf-8-sig"))
|
||||
if not isinstance(value, list) or not value or not isinstance(value[0], dict):
|
||||
raise NewsError("broadcast response does not contain a latest broadcast")
|
||||
latest = value[0]
|
||||
author = latest.get("author")
|
||||
content = latest.get("content")
|
||||
return _broadcast_from_mapping(value[0])
|
||||
|
||||
|
||||
def _broadcast_from_mapping(latest: dict[object, object]) -> Broadcast:
|
||||
author = _bounded_text(latest.get("author"), MAX_BROADCAST_AUTHOR_CHARACTERS)
|
||||
content = _bounded_text(latest.get("content"), MAX_BROADCAST_CONTENT_CHARACTERS)
|
||||
url = _broadcast_url(latest.get("url"))
|
||||
if (
|
||||
not isinstance(author, str)
|
||||
or not author.strip()
|
||||
or not isinstance(content, str)
|
||||
or not content.strip()
|
||||
author is None
|
||||
or content is None
|
||||
or url is None
|
||||
):
|
||||
raise NewsError("latest broadcast is incomplete")
|
||||
published_at = (
|
||||
_format_timestamp(latest.get("timestamp"))
|
||||
if "timestamp" in latest
|
||||
else _bounded_text(latest.get("published_at"), 64)
|
||||
)
|
||||
return Broadcast(
|
||||
author=author.strip(),
|
||||
content=content.strip(),
|
||||
published_at=_format_timestamp(latest.get("timestamp")),
|
||||
author=author,
|
||||
content=content,
|
||||
published_at=published_at,
|
||||
url=url,
|
||||
)
|
||||
|
||||
|
||||
def _bounded_text(value: object, maximum_characters: int) -> str | None:
|
||||
if not isinstance(value, str):
|
||||
return None
|
||||
text = value.strip()
|
||||
if not text or len(text) > maximum_characters:
|
||||
return None
|
||||
return text
|
||||
|
||||
|
||||
def _format_timestamp(value: object) -> str | None:
|
||||
if isinstance(value, bool) or not isinstance(value, (int, float)):
|
||||
return None
|
||||
@@ -288,10 +305,16 @@ def _image_url(value: object) -> str | None:
|
||||
def _broadcast_url(value: object) -> str | None:
|
||||
if not isinstance(value, str) or not value.strip():
|
||||
return None
|
||||
parsed = urlparse(value)
|
||||
if parsed.scheme not in {"http", "https"} or not parsed.netloc:
|
||||
url = value.strip()
|
||||
parsed = urlparse(url)
|
||||
if (
|
||||
parsed.scheme.casefold() != "https"
|
||||
or not parsed.netloc
|
||||
or parsed.username is not None
|
||||
or parsed.password is not None
|
||||
):
|
||||
return None
|
||||
return value.strip()
|
||||
return url
|
||||
|
||||
|
||||
def _read_cache(path: Path) -> tuple[NewsArticle, ...]:
|
||||
@@ -320,8 +343,8 @@ def _write_cache(path: Path, articles: tuple[NewsArticle, ...]) -> None:
|
||||
def _read_broadcast_cache(path: Path) -> Broadcast | None:
|
||||
try:
|
||||
value = json.loads(path.read_text(encoding="utf-8"))
|
||||
return Broadcast(**value) if isinstance(value, dict) else None
|
||||
except (OSError, UnicodeError, TypeError, json.JSONDecodeError):
|
||||
return _broadcast_from_mapping(value) if isinstance(value, dict) else None
|
||||
except (OSError, UnicodeError, NewsError, json.JSONDecodeError):
|
||||
return None
|
||||
|
||||
|
||||
|
||||
@@ -156,10 +156,22 @@ class AppPaths:
|
||||
|
||||
return self.state / "native-depot.json"
|
||||
|
||||
@property
|
||||
def depot_launch_catalog_cache(self) -> Path:
|
||||
"""Last official realm choices used for network-free game launches."""
|
||||
|
||||
return self.cache / "depot-launch-catalog.json"
|
||||
|
||||
@property
|
||||
def news_cache(self) -> Path:
|
||||
return self.cache / "news.json"
|
||||
|
||||
@property
|
||||
def patches_cache(self) -> Path:
|
||||
"""Cached official patch changelog shown beside the news feed."""
|
||||
|
||||
return self.cache / "patches.json"
|
||||
|
||||
@property
|
||||
def broadcast_cache(self) -> Path:
|
||||
"""Cached latest announcement from the broadcast service."""
|
||||
|
||||
@@ -7,19 +7,23 @@ from datetime import datetime
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
from pathlib import Path, PurePosixPath
|
||||
import platform
|
||||
import re
|
||||
import shlex
|
||||
import shutil
|
||||
import sys
|
||||
from typing import Mapping, Sequence
|
||||
|
||||
from .command import CommandOutcome, CommandRunner, OutputLineFilter
|
||||
from .depot import DEFAULT_DEPOT_START_OPTION, DepotStartOption
|
||||
from .gamebox import build_gamebox_command
|
||||
from .paths import AppPaths
|
||||
from .settings import Settings
|
||||
|
||||
|
||||
WINDOWS_GAME_EXE = r"C:\Program Files (x86)\Pristontale EU\Game.exe"
|
||||
WINDOWS_GAME_DIRECTORY = r"C:\Program Files (x86)\Pristontale EU"
|
||||
WINDOWS_GAME_EXE = rf"{WINDOWS_GAME_DIRECTORY}\Game.exe"
|
||||
NATIVE_D3D43_DLLS = ("d3dx9_43", "d3dcompiler_43")
|
||||
DXVK_D3D9_OVERRIDE = "native,builtin"
|
||||
DLL_OVERRIDES_KEY = r"HKCU\Software\Wine\DllOverrides"
|
||||
@@ -33,6 +37,12 @@ CURRENT_UMU_DISTRIBUTIONS = frozenset(
|
||||
{"arch", "cachyos", "endeavouros", "garuda", "manjaro"}
|
||||
)
|
||||
GAME_RUNTIME_CONFIGURATION_VERSION = 1
|
||||
PROTON7_RELATIVE_PATH = Path("steamapps/common/Proton 7.0/proton")
|
||||
STEAM_LIBRARY_FOLDERS_RELATIVE_PATH = Path("steamapps/libraryfolders.vdf")
|
||||
STEAM_LIBRARY_FOLDERS_MAX_BYTES = 4 * 1024 * 1024
|
||||
_VDF_QUOTED_PAIR = re.compile(
|
||||
r'"((?:\\.|[^"\\])*)"\s*"((?:\\.|[^"\\])*)"'
|
||||
)
|
||||
|
||||
|
||||
class RuntimeErrorPT(RuntimeError):
|
||||
@@ -77,7 +87,7 @@ def select_runtime(
|
||||
proton = Path(settings.proton_path).expanduser().resolve()
|
||||
if not proton.is_file():
|
||||
raise RuntimeErrorPT(f"configured Proton script does not exist: {proton}")
|
||||
steam_root = _steam_root_for_proton(proton)
|
||||
steam_root = _steam_root_for_proton(proton, env)
|
||||
if settings.runtime == "proton7-container":
|
||||
if not umu:
|
||||
raise RuntimeErrorPT(
|
||||
@@ -134,33 +144,108 @@ def select_runtime(
|
||||
def find_steam_proton7(
|
||||
environment: Mapping[str, str] | None = None,
|
||||
) -> tuple[Path, Path] | None:
|
||||
"""Find Proton 7 in a Steam client root or one of its registered libraries.
|
||||
|
||||
The returned Steam root is always the client installation, even when the
|
||||
Proton tool itself lives in a secondary library. Proton 7 uses that client
|
||||
root for Steam's bundled media compatibility libraries.
|
||||
"""
|
||||
|
||||
env = os.environ if environment is None else environment
|
||||
home_value = env.get("HOME")
|
||||
if not home_value:
|
||||
for steam_root in _steam_client_roots(env):
|
||||
for library_root in _steam_library_roots(steam_root):
|
||||
proton = library_root / PROTON7_RELATIVE_PATH
|
||||
if proton.is_file():
|
||||
return proton.resolve(), steam_root
|
||||
return None
|
||||
home = Path(home_value).expanduser()
|
||||
roots: list[Path] = []
|
||||
explicit = env.get("STEAM_COMPAT_CLIENT_INSTALL_PATH")
|
||||
|
||||
|
||||
def _steam_client_roots(environment: Mapping[str, str]) -> tuple[Path, ...]:
|
||||
candidates: list[Path] = []
|
||||
explicit = environment.get("STEAM_COMPAT_CLIENT_INSTALL_PATH")
|
||||
if explicit:
|
||||
roots.append(Path(explicit).expanduser())
|
||||
roots.extend(
|
||||
candidates.append(Path(explicit).expanduser())
|
||||
home_value = environment.get("HOME")
|
||||
if home_value:
|
||||
home = Path(home_value).expanduser()
|
||||
candidates.extend(
|
||||
(
|
||||
home / ".local/share/Steam",
|
||||
home / ".steam/steam",
|
||||
home / ".steam/root",
|
||||
home / ".steam/debian-installation",
|
||||
home / "snap/steam/common/.local/share/Steam",
|
||||
home / "snap/steam/common/.steam/steam",
|
||||
home / ".var/app/com.valvesoftware.Steam/.local/share/Steam",
|
||||
)
|
||||
)
|
||||
return _unique_resolved_paths(candidates)
|
||||
|
||||
|
||||
def _steam_library_roots(steam_root: Path) -> tuple[Path, ...]:
|
||||
registered = _read_steam_library_folders(
|
||||
steam_root / STEAM_LIBRARY_FOLDERS_RELATIVE_PATH
|
||||
)
|
||||
return _unique_resolved_paths((steam_root, *registered))
|
||||
|
||||
|
||||
def _read_steam_library_folders(path: Path) -> tuple[Path, ...]:
|
||||
"""Read current and legacy Steam libraryfolders.vdf path records."""
|
||||
|
||||
try:
|
||||
with path.open("rb") as handle:
|
||||
content = handle.read(STEAM_LIBRARY_FOLDERS_MAX_BYTES + 1)
|
||||
except OSError:
|
||||
return ()
|
||||
if len(content) > STEAM_LIBRARY_FOLDERS_MAX_BYTES:
|
||||
return ()
|
||||
text = content.decode("utf-8-sig", errors="replace")
|
||||
result: list[Path] = []
|
||||
for raw_key, raw_value in _VDF_QUOTED_PAIR.findall(text):
|
||||
key = _decode_vdf_string(raw_key)
|
||||
value = _decode_vdf_string(raw_value)
|
||||
# Current files use a named "path" property. Older Steam clients used
|
||||
# numeric top-level keys, which are distinguishable from app-size pairs
|
||||
# because a Linux library path must be absolute.
|
||||
if key.casefold() != "path" and not key.isdecimal():
|
||||
continue
|
||||
if not value or "\x00" in value:
|
||||
continue
|
||||
library = Path(value).expanduser()
|
||||
if library.is_absolute():
|
||||
result.append(library)
|
||||
return tuple(result)
|
||||
|
||||
|
||||
def _decode_vdf_string(value: str) -> str:
|
||||
result: list[str] = []
|
||||
index = 0
|
||||
while index < len(value):
|
||||
character = value[index]
|
||||
if character == "\\" and index + 1 < len(value):
|
||||
next_character = value[index + 1]
|
||||
if next_character in {'"', "\\"}:
|
||||
result.append(next_character)
|
||||
index += 2
|
||||
continue
|
||||
result.append(character)
|
||||
index += 1
|
||||
return "".join(result)
|
||||
|
||||
|
||||
def _unique_resolved_paths(paths: Sequence[Path]) -> tuple[Path, ...]:
|
||||
result: list[Path] = []
|
||||
seen: set[Path] = set()
|
||||
for root in roots:
|
||||
resolved = root.resolve()
|
||||
for path in paths:
|
||||
try:
|
||||
resolved = path.resolve()
|
||||
except (OSError, RuntimeError):
|
||||
continue
|
||||
if resolved in seen:
|
||||
continue
|
||||
seen.add(resolved)
|
||||
proton = resolved / "steamapps/common/Proton 7.0/proton"
|
||||
if proton.is_file():
|
||||
return proton, resolved
|
||||
return None
|
||||
result.append(resolved)
|
||||
return tuple(result)
|
||||
|
||||
|
||||
def prepare_game_runtime(
|
||||
@@ -277,13 +362,25 @@ def prepare_game_runtime(
|
||||
return results
|
||||
|
||||
|
||||
def game_runtime_ready(paths: AppPaths, selection: RuntimeSelection) -> bool:
|
||||
"""Return whether another client may reuse the prepared runtime read-only."""
|
||||
|
||||
if selection.kind == "umu":
|
||||
return native_d3d_ready(paths.umu_prefix)
|
||||
if selection.kind in {"proton7", "proton7-container"}:
|
||||
return _game_runtime_ready(paths, selection)
|
||||
return False
|
||||
|
||||
|
||||
def run_game(
|
||||
paths: AppPaths,
|
||||
settings: Settings,
|
||||
selection: RuntimeSelection,
|
||||
runner: CommandRunner,
|
||||
*,
|
||||
start_option: DepotStartOption = DEFAULT_DEPOT_START_OPTION,
|
||||
use_gamebox: bool = True,
|
||||
concurrent_prefix_launch: bool = False,
|
||||
timeout_seconds: float = 12 * 60 * 60,
|
||||
check: bool = True,
|
||||
extra_environment: Mapping[str, str] | None = None,
|
||||
@@ -291,11 +388,15 @@ def run_game(
|
||||
) -> CommandOutcome:
|
||||
if not paths.runtime_view.is_dir() and not runner.dry_run:
|
||||
raise RuntimeErrorPT("runtime view is missing; prepare the game first")
|
||||
native_executable, windows_executable, game_arguments = _game_start_command(
|
||||
paths,
|
||||
start_option,
|
||||
)
|
||||
if selection.kind == "umu":
|
||||
game_command = [
|
||||
selection.executable,
|
||||
str(paths.runtime_view / "Game.exe"),
|
||||
"PTEU_REBORN",
|
||||
native_executable,
|
||||
*game_arguments,
|
||||
]
|
||||
environment = _umu_environment(paths)
|
||||
elif selection.kind == "proton7-container":
|
||||
@@ -303,8 +404,8 @@ def run_game(
|
||||
raise RuntimeErrorPT("Proton script path is missing")
|
||||
game_command = [
|
||||
selection.executable,
|
||||
str(paths.runtime_view / "Game.exe"),
|
||||
"PTEU_REBORN",
|
||||
native_executable,
|
||||
*game_arguments,
|
||||
]
|
||||
environment = _umu_proton7_environment(
|
||||
paths,
|
||||
@@ -320,14 +421,21 @@ def run_game(
|
||||
"ptlinux.proton_compat",
|
||||
selection.proton_path,
|
||||
"runinprefix",
|
||||
WINDOWS_GAME_EXE,
|
||||
"PTEU_REBORN",
|
||||
windows_executable,
|
||||
*game_arguments,
|
||||
]
|
||||
environment = _proton_environment(
|
||||
paths,
|
||||
selection,
|
||||
synchronization=resolve_proton_sync_mode(settings.proton_sync),
|
||||
)
|
||||
if selection.kind in {"umu", "proton7-container"}:
|
||||
# Keep every UMU launch identical except for the verb, as required when
|
||||
# several games share one prefix. The first client owns prefix startup;
|
||||
# every later shared-lease client joins the active wineserver directly.
|
||||
environment["PROTON_VERB"] = (
|
||||
"runinprefix" if concurrent_prefix_launch else "waitforexitandrun"
|
||||
)
|
||||
if use_gamebox:
|
||||
game_command = build_gamebox_command(game_command, settings)
|
||||
if extra_environment:
|
||||
@@ -346,6 +454,57 @@ def run_game(
|
||||
)
|
||||
|
||||
|
||||
def _game_start_command(
|
||||
paths: AppPaths,
|
||||
option: DepotStartOption,
|
||||
) -> tuple[str, str, tuple[str, ...]]:
|
||||
relative = PurePosixPath(option.path)
|
||||
if (
|
||||
relative.is_absolute()
|
||||
or not relative.parts
|
||||
or any(part in {"", ".", ".."} for part in relative.parts)
|
||||
or relative.suffix.casefold() != ".exe"
|
||||
):
|
||||
raise RuntimeErrorPT(f"invalid realm executable path: {option.path!r}")
|
||||
try:
|
||||
arguments = tuple(shlex.split(option.arguments))
|
||||
except ValueError as exc:
|
||||
raise RuntimeErrorPT(
|
||||
f"invalid launch arguments for realm {option.name}: {exc}"
|
||||
) from exc
|
||||
if not arguments:
|
||||
raise RuntimeErrorPT(f"realm {option.name} has no launch arguments")
|
||||
native = str(_resolve_runtime_path_case(paths.runtime_view, relative))
|
||||
windows_relative = "\\".join(relative.parts)
|
||||
return native, rf"{WINDOWS_GAME_DIRECTORY}\{windows_relative}", arguments
|
||||
|
||||
|
||||
def _resolve_runtime_path_case(root: Path, relative: PurePosixPath) -> Path:
|
||||
"""Match Windows manifest casing against the case-sensitive runtime view."""
|
||||
|
||||
direct = root.joinpath(*relative.parts)
|
||||
if direct.exists():
|
||||
return direct
|
||||
cursor = root
|
||||
for part in relative.parts:
|
||||
if not cursor.is_dir():
|
||||
return direct
|
||||
try:
|
||||
matches = tuple(
|
||||
entry for entry in cursor.iterdir() if entry.name.casefold() == part.casefold()
|
||||
)
|
||||
except OSError:
|
||||
return direct
|
||||
if not matches:
|
||||
return direct
|
||||
if len(matches) > 1:
|
||||
raise RuntimeErrorPT(
|
||||
f"realm executable path has ambiguous casing: {relative.as_posix()}"
|
||||
)
|
||||
cursor = matches[0]
|
||||
return cursor
|
||||
|
||||
|
||||
def run_proton_prefix_command(
|
||||
paths: AppPaths,
|
||||
selection: RuntimeSelection,
|
||||
@@ -584,15 +743,27 @@ def _ensure_proton_game_link(paths: AppPaths) -> None:
|
||||
link.symlink_to(paths.runtime_view, target_is_directory=True)
|
||||
|
||||
|
||||
def _steam_root_for_proton(proton: Path) -> Path:
|
||||
parts = proton.parts
|
||||
try:
|
||||
index = parts.index("steamapps")
|
||||
except ValueError as exc:
|
||||
def _steam_root_for_proton(
|
||||
proton: Path,
|
||||
environment: Mapping[str, str] | None = None,
|
||||
) -> Path:
|
||||
steamapps = next(
|
||||
(parent for parent in proton.parents if parent.name.casefold() == "steamapps"),
|
||||
None,
|
||||
)
|
||||
if steamapps is None:
|
||||
raise RuntimeErrorPT(
|
||||
"configured Proton path must be inside a Steam steamapps directory"
|
||||
) from exc
|
||||
return Path(*parts[:index])
|
||||
)
|
||||
library_root = steamapps.parent.resolve()
|
||||
env = os.environ if environment is None else environment
|
||||
for steam_root in _steam_client_roots(env):
|
||||
if library_root in _steam_library_roots(steam_root):
|
||||
return steam_root
|
||||
# Preserve support for an otherwise valid, manually managed Steam layout.
|
||||
# Without a registered client root, the library root is the best available
|
||||
# compatibility value and matches the launcher's historical behaviour.
|
||||
return library_root
|
||||
|
||||
|
||||
def native_d3d_ready(prefix: Path) -> bool:
|
||||
|
||||
157
ptlinux/server_status.py
Normal file
157
ptlinux/server_status.py
Normal file
@@ -0,0 +1,157 @@
|
||||
"""Official realm health used by the native launcher dashboard."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
import json
|
||||
from typing import Callable, Mapping, Protocol
|
||||
from urllib.request import Request, urlopen
|
||||
|
||||
from . import __version__
|
||||
|
||||
|
||||
# This is the same publisher endpoint embedded in Windows launcher 1.0.0.64.
|
||||
SERVER_STATUS_ENDPOINT = (
|
||||
"https://pristontale.eu/api/pristontale/developer"
|
||||
"?api_key=d9OlDSvworitU4ajDiH6dU6cxXApc4M0"
|
||||
)
|
||||
MAX_SERVER_STATUS_BYTES = 1024 * 1024
|
||||
|
||||
|
||||
class ServerStatusError(RuntimeError):
|
||||
"""Raised when the official status response cannot be interpreted safely."""
|
||||
|
||||
|
||||
class ServerStatusResponse(Protocol):
|
||||
def read(self, size: int = -1) -> bytes: ...
|
||||
|
||||
def __enter__(self) -> ServerStatusResponse: ...
|
||||
|
||||
def __exit__(
|
||||
self,
|
||||
exception_type: type[BaseException] | None,
|
||||
exception: BaseException | None,
|
||||
traceback: object | None,
|
||||
) -> bool | None: ...
|
||||
|
||||
|
||||
ServerStatusOpener = Callable[..., ServerStatusResponse]
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class RealmServerStatus:
|
||||
name: str
|
||||
online: bool | None
|
||||
players_online: int | None
|
||||
login_online: bool | None = None
|
||||
game_online: bool | None = None
|
||||
|
||||
|
||||
def fetch_server_status(
|
||||
*,
|
||||
timeout_seconds: float = 10.0,
|
||||
opener: ServerStatusOpener = urlopen,
|
||||
) -> tuple[RealmServerStatus, ...]:
|
||||
request = Request(
|
||||
SERVER_STATUS_ENDPOINT,
|
||||
headers={"User-Agent": f"PT-Linux/{__version__} (+https://pristontale.eu/)"},
|
||||
method="GET",
|
||||
)
|
||||
try:
|
||||
response = opener(request, timeout=timeout_seconds)
|
||||
with response:
|
||||
payload = response.read(MAX_SERVER_STATUS_BYTES + 1)
|
||||
except OSError as exc:
|
||||
raise ServerStatusError("official server status is unavailable") from exc
|
||||
if len(payload) > MAX_SERVER_STATUS_BYTES:
|
||||
raise ServerStatusError("official server status exceeds the 1 MiB limit")
|
||||
return parse_server_status(payload)
|
||||
|
||||
|
||||
def parse_server_status(payload: bytes) -> tuple[RealmServerStatus, ...]:
|
||||
try:
|
||||
value = json.loads(payload.decode("utf-8-sig"))
|
||||
except (UnicodeError, json.JSONDecodeError) as exc:
|
||||
raise ServerStatusError("official server status is not valid JSON") from exc
|
||||
# Historical versions returned an array containing the status object. The
|
||||
# current endpoint returns that object directly.
|
||||
if isinstance(value, list) and value:
|
||||
value = value[0]
|
||||
if not isinstance(value, dict):
|
||||
raise ServerStatusError("official server status is not a realm object")
|
||||
|
||||
statuses: list[RealmServerStatus] = []
|
||||
for raw_name, raw_components in value.items():
|
||||
if not isinstance(raw_name, str) or raw_name.casefold() == "website":
|
||||
continue
|
||||
name = raw_name.strip()
|
||||
if not name or len(name) > 64 or not isinstance(raw_components, Mapping):
|
||||
continue
|
||||
status = _parse_realm(name, raw_components)
|
||||
if status is not None:
|
||||
statuses.append(status)
|
||||
if not statuses:
|
||||
raise ServerStatusError("official server status contains no realms")
|
||||
return tuple(statuses)
|
||||
|
||||
|
||||
def _parse_realm(
|
||||
name: str,
|
||||
components: Mapping[object, object],
|
||||
) -> RealmServerStatus | None:
|
||||
service_states: list[bool] = []
|
||||
login_online: bool | None = None
|
||||
game_online: bool | None = None
|
||||
players_online: int | None = None
|
||||
for raw_key, raw_value in components.items():
|
||||
if not isinstance(raw_key, str):
|
||||
continue
|
||||
key = raw_key.casefold()
|
||||
if key == "users.online":
|
||||
players_online = _non_negative_integer(raw_value)
|
||||
continue
|
||||
if not key.startswith("server."):
|
||||
continue
|
||||
state = _online_state(raw_value)
|
||||
if state is None:
|
||||
continue
|
||||
service_states.append(state)
|
||||
if "login" in key:
|
||||
login_online = state
|
||||
elif "game" in key:
|
||||
game_online = state
|
||||
if not service_states and players_online is None:
|
||||
return None
|
||||
online = all(service_states) if service_states else None
|
||||
display_name = name[0].upper() + name[1:]
|
||||
return RealmServerStatus(
|
||||
name=display_name,
|
||||
online=online,
|
||||
players_online=players_online,
|
||||
login_online=login_online,
|
||||
game_online=game_online,
|
||||
)
|
||||
|
||||
|
||||
def _online_state(value: object) -> bool | None:
|
||||
if isinstance(value, bool):
|
||||
return value
|
||||
if isinstance(value, (int, float)):
|
||||
return value > 0
|
||||
if isinstance(value, str):
|
||||
text = value.strip().casefold()
|
||||
if text in {"online", "up", "true", "1"}:
|
||||
return True
|
||||
if text in {"offline", "down", "false", "0"}:
|
||||
return False
|
||||
return None
|
||||
|
||||
|
||||
def _non_negative_integer(value: object) -> int | None:
|
||||
if isinstance(value, bool):
|
||||
return None
|
||||
if isinstance(value, int) and value >= 0:
|
||||
return value
|
||||
if isinstance(value, str) and value.isascii() and value.isdecimal():
|
||||
return int(value)
|
||||
return None
|
||||
@@ -19,6 +19,7 @@ SETTINGS_SCHEMA = 1
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Settings:
|
||||
realm: str = "Babel"
|
||||
width: int = 1600
|
||||
height: int = 900
|
||||
output_width: int = 1600
|
||||
@@ -133,6 +134,7 @@ def _read_settings_object(path: Path) -> dict[str, object] | None:
|
||||
|
||||
def _settings_from_object(value: dict[str, object]) -> Settings:
|
||||
settings = Settings(
|
||||
realm=_string(value, "realm", "Babel").strip(),
|
||||
width=_integer(value, "width", 1600),
|
||||
height=_integer(value, "height", 900),
|
||||
output_width=_integer(value, "output_width", 1600),
|
||||
@@ -215,6 +217,12 @@ def _fsync_directory(path: Path) -> None:
|
||||
|
||||
|
||||
def _validate(settings: Settings) -> None:
|
||||
if (
|
||||
len(settings.realm) > 64
|
||||
or not settings.realm.strip()
|
||||
or any(ord(character) < 32 for character in settings.realm)
|
||||
):
|
||||
raise ValueError("realm must be a non-empty display name")
|
||||
for name, value in (
|
||||
("width", settings.width),
|
||||
("height", settings.height),
|
||||
|
||||
705
ptlinux/tk_chrome.py
Normal file
705
ptlinux/tk_chrome.py
Normal file
@@ -0,0 +1,705 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Any, Callable, Literal
|
||||
|
||||
from .desktop_identity import APPLICATION_WM_CLASS, APPLICATION_WM_INSTANCE
|
||||
|
||||
|
||||
DialogKind = Literal["info", "success", "warning", "error", "question"]
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class DialogPalette:
|
||||
"""Colors and glyph used by one launcher dialog intent."""
|
||||
|
||||
accent: str
|
||||
accent_soft: str
|
||||
glyph: str
|
||||
|
||||
|
||||
_DIALOG_PALETTES: dict[str, DialogPalette] = {
|
||||
"info": DialogPalette("#d98927", "#3a2819", "i"),
|
||||
"success": DialogPalette("#55b878", "#173122", "✓"),
|
||||
"warning": DialogPalette("#e0a248", "#3b2a18", "!"),
|
||||
"error": DialogPalette("#cf604c", "#3a1d19", "×"),
|
||||
"question": DialogPalette("#d98927", "#3a2819", "?"),
|
||||
}
|
||||
|
||||
|
||||
def dialog_palette(kind: DialogKind) -> DialogPalette:
|
||||
"""Return the stable visual palette for a themed launcher dialog."""
|
||||
|
||||
try:
|
||||
return _DIALOG_PALETTES[kind]
|
||||
except KeyError as error:
|
||||
raise ValueError(f"unsupported launcher dialog kind: {kind}") from error
|
||||
|
||||
|
||||
def square_icon_layout(width: int, height: int) -> tuple[int, int, int]:
|
||||
"""Return square canvas size and centered offsets for a window icon."""
|
||||
|
||||
if width <= 0 or height <= 0:
|
||||
raise ValueError("icon dimensions must be positive")
|
||||
size = max(width, height)
|
||||
return size, (size - width) // 2, (size - height) // 2
|
||||
|
||||
|
||||
def icon_subsample_factor(width: int, height: int, max_dimension: int) -> int:
|
||||
"""Return the integer Tk subsample needed to fit an image within a square."""
|
||||
|
||||
if width <= 0 or height <= 0:
|
||||
raise ValueError("icon dimensions must be positive")
|
||||
if max_dimension <= 0:
|
||||
raise ValueError("maximum icon dimension must be positive")
|
||||
largest_dimension = max(width, height)
|
||||
return max(1, (largest_dimension + max_dimension - 1) // max_dimension)
|
||||
|
||||
|
||||
def load_square_window_icon(master: Any, path: Path) -> Any:
|
||||
"""Load a PNG into a square transparent Tk image for desktop shells."""
|
||||
|
||||
import tkinter as tk
|
||||
|
||||
source = tk.PhotoImage(master=master, file=str(path))
|
||||
size, offset_x, offset_y = square_icon_layout(source.width(), source.height())
|
||||
if source.width() == source.height():
|
||||
return source
|
||||
icon = tk.PhotoImage(master=master, width=size, height=size)
|
||||
icon.tk.call(
|
||||
str(icon),
|
||||
"copy",
|
||||
str(source),
|
||||
"-to",
|
||||
offset_x,
|
||||
offset_y,
|
||||
)
|
||||
return icon
|
||||
|
||||
|
||||
def centered_window_geometry(
|
||||
width: int,
|
||||
height: int,
|
||||
*,
|
||||
parent_x: int,
|
||||
parent_y: int,
|
||||
parent_width: int,
|
||||
parent_height: int,
|
||||
) -> str:
|
||||
"""Return Tk geometry centered over a parent window."""
|
||||
|
||||
if width <= 0 or height <= 0:
|
||||
raise ValueError("window dimensions must be positive")
|
||||
x = parent_x + (max(1, parent_width) - width) // 2
|
||||
y = parent_y + (max(1, parent_height) - height) // 2
|
||||
return f"{width}x{height}{_geometry_offset(x)}{_geometry_offset(y)}"
|
||||
|
||||
|
||||
def dragged_window_geometry(
|
||||
pointer_x: int,
|
||||
pointer_y: int,
|
||||
*,
|
||||
offset_x: int,
|
||||
offset_y: int,
|
||||
) -> str:
|
||||
"""Return the Tk position for a pointer-driven custom title bar."""
|
||||
|
||||
return (
|
||||
f"{_geometry_offset(pointer_x - offset_x)}"
|
||||
f"{_geometry_offset(pointer_y - offset_y)}"
|
||||
)
|
||||
|
||||
|
||||
def _geometry_offset(value: int) -> str:
|
||||
return f"+{value}" if value >= 0 else str(value)
|
||||
|
||||
|
||||
def _bind_window_drag(handle: Any, *widgets: Any) -> None:
|
||||
position = [0, 0]
|
||||
|
||||
def begin(event: Any) -> None:
|
||||
position[0] = int(event.x_root) - int(handle.winfo_x())
|
||||
position[1] = int(event.y_root) - int(handle.winfo_y())
|
||||
|
||||
def move(event: Any) -> None:
|
||||
handle.geometry(
|
||||
dragged_window_geometry(
|
||||
int(event.x_root),
|
||||
int(event.y_root),
|
||||
offset_x=position[0],
|
||||
offset_y=position[1],
|
||||
)
|
||||
)
|
||||
|
||||
for widget in widgets:
|
||||
widget.bind("<ButtonPress-1>", begin, add="+")
|
||||
widget.bind("<B1-Motion>", move, add="+")
|
||||
|
||||
|
||||
class DashboardChrome:
|
||||
"""Controller for one frameless launcher window and its controls."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
root: Any,
|
||||
content: Any,
|
||||
close_button: Any,
|
||||
minimize_button: Any,
|
||||
*,
|
||||
managed: bool,
|
||||
identity_configured: bool,
|
||||
allow_close: bool,
|
||||
) -> None:
|
||||
self.root = root
|
||||
self.content = content
|
||||
self.managed = managed
|
||||
self.identity_configured = identity_configured
|
||||
self._close_button = close_button
|
||||
self._minimize_button = minimize_button
|
||||
self._allow_close = allow_close
|
||||
self._close_command: Callable[[], None] = root.destroy
|
||||
if close_button is not None:
|
||||
close_button.configure(command=self.close)
|
||||
if minimize_button is not None:
|
||||
minimize_button.configure(command=self.minimize)
|
||||
root.bind("<Alt-F4>", self._close_from_event, add="+")
|
||||
|
||||
def set_close_command(self, command: Callable[[], None]) -> None:
|
||||
self._close_command = command
|
||||
|
||||
def close(self) -> None:
|
||||
if self._allow_close:
|
||||
self._close_command()
|
||||
|
||||
def minimize(self) -> None:
|
||||
"""Ask the window manager to iconify the managed dashboard."""
|
||||
|
||||
import tkinter as tk
|
||||
|
||||
try:
|
||||
self.root.iconify()
|
||||
except tk.TclError:
|
||||
pass
|
||||
|
||||
def _close_from_event(self, _event: Any) -> str:
|
||||
self.close()
|
||||
return "break"
|
||||
|
||||
|
||||
|
||||
def _configure_x11_window(
|
||||
root: Any,
|
||||
*,
|
||||
resource_name: str,
|
||||
resource_class: str,
|
||||
) -> tuple[bool, bool]:
|
||||
"""Hide the X11 frame and give desktop shells a stable window identity."""
|
||||
|
||||
import ctypes.util
|
||||
|
||||
library_path = ctypes.util.find_library("X11")
|
||||
if library_path is None:
|
||||
return False, False
|
||||
try:
|
||||
x11 = ctypes.CDLL(library_path)
|
||||
except OSError:
|
||||
return False, False
|
||||
|
||||
class XClassHint(ctypes.Structure):
|
||||
_fields_ = [
|
||||
("res_name", ctypes.c_char_p),
|
||||
("res_class", ctypes.c_char_p),
|
||||
]
|
||||
|
||||
x11.XOpenDisplay.argtypes = [ctypes.c_char_p]
|
||||
x11.XOpenDisplay.restype = ctypes.c_void_p
|
||||
x11.XQueryTree.argtypes = [
|
||||
ctypes.c_void_p,
|
||||
ctypes.c_ulong,
|
||||
ctypes.POINTER(ctypes.c_ulong),
|
||||
ctypes.POINTER(ctypes.c_ulong),
|
||||
ctypes.POINTER(ctypes.POINTER(ctypes.c_ulong)),
|
||||
ctypes.POINTER(ctypes.c_uint),
|
||||
]
|
||||
x11.XQueryTree.restype = ctypes.c_int
|
||||
x11.XInternAtom.argtypes = [ctypes.c_void_p, ctypes.c_char_p, ctypes.c_int]
|
||||
x11.XInternAtom.restype = ctypes.c_ulong
|
||||
x11.XChangeProperty.argtypes = [
|
||||
ctypes.c_void_p,
|
||||
ctypes.c_ulong,
|
||||
ctypes.c_ulong,
|
||||
ctypes.c_ulong,
|
||||
ctypes.c_int,
|
||||
ctypes.c_int,
|
||||
ctypes.POINTER(ctypes.c_ubyte),
|
||||
ctypes.c_int,
|
||||
]
|
||||
x11.XChangeProperty.restype = ctypes.c_int
|
||||
x11.XSetClassHint.argtypes = [
|
||||
ctypes.c_void_p,
|
||||
ctypes.c_ulong,
|
||||
ctypes.POINTER(XClassHint),
|
||||
]
|
||||
x11.XSetClassHint.restype = ctypes.c_int
|
||||
x11.XFlush.argtypes = [ctypes.c_void_p]
|
||||
x11.XFlush.restype = ctypes.c_int
|
||||
x11.XFree.argtypes = [ctypes.c_void_p]
|
||||
x11.XFree.restype = ctypes.c_int
|
||||
x11.XCloseDisplay.argtypes = [ctypes.c_void_p]
|
||||
x11.XCloseDisplay.restype = ctypes.c_int
|
||||
root.update_idletasks()
|
||||
display = x11.XOpenDisplay(None)
|
||||
if display is None:
|
||||
return False, False
|
||||
children = ctypes.POINTER(ctypes.c_ulong)()
|
||||
try:
|
||||
root_window = ctypes.c_ulong()
|
||||
wrapper_window = ctypes.c_ulong()
|
||||
child_count = ctypes.c_uint()
|
||||
queried = x11.XQueryTree(
|
||||
display,
|
||||
root.winfo_id(),
|
||||
ctypes.byref(root_window),
|
||||
ctypes.byref(wrapper_window),
|
||||
ctypes.byref(children),
|
||||
ctypes.byref(child_count),
|
||||
)
|
||||
if not queried or not wrapper_window.value:
|
||||
return False, False
|
||||
atom = x11.XInternAtom(display, b"_MOTIF_WM_HINTS", 0)
|
||||
if not atom:
|
||||
return False, False
|
||||
motif_hints_decorations = 1 << 1
|
||||
hints = (ctypes.c_ulong * 5)(motif_hints_decorations, 0, 0, 0, 0)
|
||||
changed = x11.XChangeProperty(
|
||||
display,
|
||||
wrapper_window.value,
|
||||
atom,
|
||||
atom,
|
||||
32,
|
||||
0,
|
||||
ctypes.cast(hints, ctypes.POINTER(ctypes.c_ubyte)),
|
||||
len(hints),
|
||||
)
|
||||
identity = XClassHint(
|
||||
res_name=resource_name.encode("utf-8"),
|
||||
res_class=resource_class.encode("utf-8"),
|
||||
)
|
||||
identity_changed = x11.XSetClassHint(
|
||||
display,
|
||||
wrapper_window.value,
|
||||
ctypes.byref(identity),
|
||||
)
|
||||
x11.XFlush(display)
|
||||
return bool(changed), bool(identity_changed)
|
||||
finally:
|
||||
if children:
|
||||
x11.XFree(children)
|
||||
x11.XCloseDisplay(display)
|
||||
|
||||
|
||||
def create_window_chrome(
|
||||
root: Any,
|
||||
*,
|
||||
title: str,
|
||||
subtitle: str,
|
||||
window_icon: Any | None = None,
|
||||
show_title_bar: bool = True,
|
||||
show_close: bool = True,
|
||||
show_minimize: bool = False,
|
||||
allow_close: bool = True,
|
||||
) -> DashboardChrome:
|
||||
"""Install polished managed chrome around one launcher-owned Tk window."""
|
||||
|
||||
import tkinter as tk
|
||||
|
||||
managed, identity_configured = _configure_x11_window(
|
||||
root,
|
||||
resource_name=APPLICATION_WM_INSTANCE,
|
||||
resource_class=APPLICATION_WM_CLASS,
|
||||
)
|
||||
if not managed:
|
||||
root.overrideredirect(True)
|
||||
border = tk.Frame(
|
||||
root,
|
||||
background="#6a482d",
|
||||
borderwidth=0,
|
||||
highlightthickness=0,
|
||||
)
|
||||
border.pack(fill="both", expand=True)
|
||||
surface = tk.Frame(
|
||||
border,
|
||||
background="#0c0908",
|
||||
borderwidth=0,
|
||||
highlightthickness=0,
|
||||
)
|
||||
surface.pack(fill="both", expand=True, padx=1, pady=1)
|
||||
title_bar = tk.Frame(
|
||||
surface,
|
||||
background="#17100d",
|
||||
height=32,
|
||||
borderwidth=0,
|
||||
highlightthickness=0,
|
||||
cursor="fleur",
|
||||
)
|
||||
if show_title_bar:
|
||||
title_bar.pack(fill="x")
|
||||
title_bar.pack_propagate(False)
|
||||
tk_accent = tk.Frame(title_bar, background="#d98927", width=3)
|
||||
tk_accent.pack(side="left", fill="y")
|
||||
title_icon: Any | None = None
|
||||
if window_icon is not None:
|
||||
try:
|
||||
factor = icon_subsample_factor(
|
||||
window_icon.width(),
|
||||
window_icon.height(),
|
||||
20,
|
||||
)
|
||||
title_icon = window_icon.subsample(factor, factor)
|
||||
except (tk.TclError, ValueError):
|
||||
title_icon = None
|
||||
if title_icon is not None:
|
||||
mark = tk.Label(
|
||||
title_bar,
|
||||
image=title_icon,
|
||||
width=22,
|
||||
height=22,
|
||||
background="#17100d",
|
||||
borderwidth=0,
|
||||
highlightthickness=0,
|
||||
cursor="fleur",
|
||||
)
|
||||
mark.image = title_icon
|
||||
else:
|
||||
mark = tk.Frame(
|
||||
title_bar,
|
||||
width=22,
|
||||
height=22,
|
||||
background="#17100d",
|
||||
borderwidth=0,
|
||||
highlightthickness=0,
|
||||
cursor="fleur",
|
||||
)
|
||||
mark.pack(side="left", padx=(9, 7))
|
||||
identity = tk.Frame(title_bar, background="#17100d", cursor="fleur")
|
||||
identity.pack(side="left", fill="y")
|
||||
title_label = tk.Label(
|
||||
identity,
|
||||
text=title,
|
||||
background="#17100d",
|
||||
foreground="#f2dfbd",
|
||||
font=("Sans", 8, "bold"),
|
||||
cursor="fleur",
|
||||
)
|
||||
title_label.pack(side="left", pady=(0, 1))
|
||||
subtitle_label = tk.Label(
|
||||
identity,
|
||||
text=f" · {subtitle}",
|
||||
background="#17100d",
|
||||
foreground="#8f7b69",
|
||||
font=("Sans", 7),
|
||||
cursor="fleur",
|
||||
)
|
||||
subtitle_label.pack(side="left", pady=(0, 1))
|
||||
close_button: Any = None
|
||||
if show_close:
|
||||
close_button = tk.Button(
|
||||
title_bar,
|
||||
text="×",
|
||||
background="#17100d",
|
||||
foreground="#c9b7a4",
|
||||
activebackground="#a63f2f",
|
||||
activeforeground="#ffffff",
|
||||
font=("Sans", 12),
|
||||
width=4,
|
||||
borderwidth=0,
|
||||
highlightthickness=0,
|
||||
relief="flat",
|
||||
cursor="hand2",
|
||||
takefocus=False,
|
||||
)
|
||||
close_button.pack(side="right", fill="y")
|
||||
minimize_button: Any = None
|
||||
if show_minimize:
|
||||
minimize_button = tk.Button(
|
||||
title_bar,
|
||||
text="—",
|
||||
background="#17100d",
|
||||
foreground="#c9b7a4",
|
||||
activebackground="#312219",
|
||||
activeforeground="#ffffff",
|
||||
font=("Sans", 10, "bold"),
|
||||
width=4,
|
||||
borderwidth=0,
|
||||
highlightthickness=0,
|
||||
relief="flat",
|
||||
cursor="hand2",
|
||||
takefocus=False,
|
||||
)
|
||||
minimize_button.pack(side="right", fill="y")
|
||||
content = tk.Frame(
|
||||
surface,
|
||||
background="#0c0908",
|
||||
borderwidth=0,
|
||||
highlightthickness=0,
|
||||
)
|
||||
content.pack(fill="both", expand=True)
|
||||
if show_title_bar:
|
||||
_bind_window_drag(root, title_bar, mark, identity, title_label, subtitle_label)
|
||||
return DashboardChrome(
|
||||
root,
|
||||
content,
|
||||
close_button,
|
||||
minimize_button,
|
||||
managed=managed,
|
||||
identity_configured=identity_configured,
|
||||
allow_close=allow_close,
|
||||
)
|
||||
|
||||
|
||||
def create_dashboard_chrome(
|
||||
root: Any,
|
||||
*,
|
||||
title: str,
|
||||
subtitle: str,
|
||||
window_icon: Any | None = None,
|
||||
) -> DashboardChrome:
|
||||
"""Install the draggable frame and full controls used by the dashboard."""
|
||||
|
||||
return create_window_chrome(
|
||||
root,
|
||||
title=title,
|
||||
subtitle=subtitle,
|
||||
window_icon=window_icon,
|
||||
show_close=True,
|
||||
show_minimize=True,
|
||||
allow_close=True,
|
||||
)
|
||||
|
||||
|
||||
def show_themed_dialog(
|
||||
parent: Any,
|
||||
*,
|
||||
title: str,
|
||||
message: str,
|
||||
kind: DialogKind = "info",
|
||||
confirm: bool = False,
|
||||
accept_text: str = "OK",
|
||||
cancel_text: str = "CANCEL",
|
||||
window_icon: Any | None = None,
|
||||
) -> bool:
|
||||
"""Show one frameless modal dialog matching the native launcher palette."""
|
||||
|
||||
import tkinter as tk
|
||||
|
||||
palette = dialog_palette(kind)
|
||||
dialog = tk.Toplevel(parent)
|
||||
dialog.withdraw()
|
||||
dialog.title(title)
|
||||
dialog.configure(background=palette.accent)
|
||||
dialog.resizable(False, False)
|
||||
dialog.overrideredirect(True)
|
||||
if window_icon is not None:
|
||||
try:
|
||||
dialog.iconphoto(False, window_icon)
|
||||
except tk.TclError:
|
||||
pass
|
||||
try:
|
||||
dialog.transient(parent)
|
||||
except tk.TclError:
|
||||
pass
|
||||
result = False
|
||||
|
||||
def finish(value: bool) -> None:
|
||||
nonlocal result
|
||||
result = value
|
||||
try:
|
||||
dialog.grab_release()
|
||||
except tk.TclError:
|
||||
pass
|
||||
dialog.destroy()
|
||||
|
||||
border = tk.Frame(dialog, background=palette.accent)
|
||||
border.pack(fill="both", expand=True)
|
||||
surface = tk.Frame(border, background="#191311")
|
||||
surface.pack(fill="both", expand=True, padx=1, pady=1)
|
||||
header = tk.Frame(surface, background="#24170f", height=66)
|
||||
header.pack(fill="x")
|
||||
header.pack_propagate(False)
|
||||
tk_accent = tk.Frame(header, background=palette.accent, width=5)
|
||||
tk_accent.pack(side="left", fill="y")
|
||||
header_copy = tk.Frame(header, background="#24170f")
|
||||
header_copy.pack(side="left", fill="both", expand=True, padx=16, pady=10)
|
||||
eyebrow = tk.Label(
|
||||
header_copy,
|
||||
text="PRISTONTALE EU · LINUX LAUNCHER",
|
||||
background="#24170f",
|
||||
foreground=palette.accent,
|
||||
font=("Sans", 7, "bold"),
|
||||
anchor="w",
|
||||
)
|
||||
eyebrow.pack(fill="x")
|
||||
title_label = tk.Label(
|
||||
header_copy,
|
||||
text=title,
|
||||
background="#24170f",
|
||||
foreground="#fff0d2",
|
||||
font=("Georgia", 15, "bold"),
|
||||
anchor="w",
|
||||
)
|
||||
title_label.pack(fill="x", pady=(1, 0))
|
||||
close_button = tk.Button(
|
||||
header,
|
||||
text="×",
|
||||
command=lambda: finish(False),
|
||||
background="#24170f",
|
||||
foreground="#c9b7a4",
|
||||
activebackground="#a63f2f",
|
||||
activeforeground="#ffffff",
|
||||
font=("Sans", 12),
|
||||
width=4,
|
||||
borderwidth=0,
|
||||
highlightthickness=0,
|
||||
relief="flat",
|
||||
cursor="hand2",
|
||||
takefocus=False,
|
||||
)
|
||||
close_button.pack(side="right", fill="y")
|
||||
body = tk.Frame(surface, background="#191311", padx=20, pady=20)
|
||||
body.pack(fill="both", expand=True)
|
||||
modal_icon: Any | None = None
|
||||
if window_icon is not None:
|
||||
try:
|
||||
factor = icon_subsample_factor(
|
||||
window_icon.width(),
|
||||
window_icon.height(),
|
||||
46,
|
||||
)
|
||||
modal_icon = window_icon.subsample(factor, factor)
|
||||
except (tk.TclError, ValueError):
|
||||
modal_icon = None
|
||||
if modal_icon is not None:
|
||||
icon_host = tk.Frame(
|
||||
body,
|
||||
width=46,
|
||||
height=46,
|
||||
background="#191311",
|
||||
borderwidth=0,
|
||||
highlightthickness=0,
|
||||
)
|
||||
icon_host.pack(side="left", anchor="n", padx=(0, 16))
|
||||
icon_host.pack_propagate(False)
|
||||
icon = tk.Label(
|
||||
icon_host,
|
||||
image=modal_icon,
|
||||
background="#191311",
|
||||
borderwidth=0,
|
||||
highlightthickness=0,
|
||||
)
|
||||
icon.place(relx=0.5, rely=0.5, anchor="center")
|
||||
else:
|
||||
icon = tk.Canvas(
|
||||
body,
|
||||
width=46,
|
||||
height=46,
|
||||
background="#191311",
|
||||
borderwidth=0,
|
||||
highlightthickness=0,
|
||||
)
|
||||
icon.create_oval(
|
||||
3,
|
||||
3,
|
||||
43,
|
||||
43,
|
||||
fill=palette.accent_soft,
|
||||
outline=palette.accent,
|
||||
width=2,
|
||||
)
|
||||
icon.create_text(
|
||||
23,
|
||||
23,
|
||||
text=palette.glyph,
|
||||
fill=palette.accent,
|
||||
font=("Sans", 17, "bold"),
|
||||
)
|
||||
icon.pack(side="left", anchor="n", padx=(0, 16))
|
||||
message_label = tk.Label(
|
||||
body,
|
||||
text=message,
|
||||
background="#191311",
|
||||
foreground="#ded1c4",
|
||||
font=("Sans", 10),
|
||||
justify="left",
|
||||
anchor="w",
|
||||
wraplength=420,
|
||||
)
|
||||
message_label.pack(side="left", fill="both", expand=True, anchor="n")
|
||||
footer = tk.Frame(surface, background="#100d0c", padx=18, pady=12)
|
||||
footer.pack(fill="x")
|
||||
|
||||
def button(
|
||||
container: Any,
|
||||
*,
|
||||
label: str,
|
||||
command: Callable[[], None],
|
||||
primary: bool,
|
||||
) -> Any:
|
||||
background = palette.accent if primary else "#2a201c"
|
||||
foreground = "#17100d" if primary else "#ead8c2"
|
||||
active_background = "#efae57" if primary else "#3a2b25"
|
||||
return tk.Button(
|
||||
container,
|
||||
text=label,
|
||||
command=command,
|
||||
background=background,
|
||||
foreground=foreground,
|
||||
activebackground=active_background,
|
||||
activeforeground="#ffffff" if not primary else "#17100d",
|
||||
font=("Sans", 8, "bold"),
|
||||
padx=16,
|
||||
pady=7,
|
||||
borderwidth=0,
|
||||
highlightthickness=0,
|
||||
relief="flat",
|
||||
cursor="hand2",
|
||||
)
|
||||
|
||||
accept_button = button(
|
||||
footer,
|
||||
label=accept_text,
|
||||
command=lambda: finish(True),
|
||||
primary=True,
|
||||
)
|
||||
accept_button.pack(side="right")
|
||||
if confirm:
|
||||
cancel_button = button(
|
||||
footer,
|
||||
label=cancel_text,
|
||||
command=lambda: finish(False),
|
||||
primary=False,
|
||||
)
|
||||
cancel_button.pack(side="right", padx=(0, 8))
|
||||
_bind_window_drag(dialog, header, header_copy, eyebrow, title_label)
|
||||
dialog.bind("<Escape>", lambda _event: finish(False))
|
||||
dialog.bind("<Return>", lambda _event: finish(True))
|
||||
parent.update_idletasks()
|
||||
dialog.update_idletasks()
|
||||
width = 560
|
||||
height = max(218, dialog.winfo_reqheight())
|
||||
dialog.geometry(
|
||||
centered_window_geometry(
|
||||
width,
|
||||
height,
|
||||
parent_x=parent.winfo_rootx(),
|
||||
parent_y=parent.winfo_rooty(),
|
||||
parent_width=parent.winfo_width(),
|
||||
parent_height=parent.winfo_height(),
|
||||
)
|
||||
)
|
||||
dialog.deiconify()
|
||||
dialog.lift()
|
||||
dialog.grab_set()
|
||||
accept_button.focus_set()
|
||||
parent.wait_window(dialog)
|
||||
return result
|
||||
2808
ptlinux/ui.py
2808
ptlinux/ui.py
File diff suppressed because it is too large
Load Diff
@@ -1,6 +1,6 @@
|
||||
[project]
|
||||
name = "PTEU-Linux-Launcher"
|
||||
version = "0.6.5"
|
||||
version = "0.8.1"
|
||||
description = "PristonTale EU Linux installer, clean launcher, and diagnostic harness"
|
||||
requires-python = ">=3.11"
|
||||
dependencies = []
|
||||
|
||||
@@ -1,11 +1,11 @@
|
||||
{
|
||||
"archive_sha256": "c1132f2e2d74579ea29fa59d69cb1df881800445be11d17e0fff7bc5d34c12fa",
|
||||
"archive_size": 226431,
|
||||
"archive_url": "https://git.m4ko.net/m4ko/pristontale-eu-linux/releases/download/v0.6.0/pristontale-eu-linux-update-v0.6.0.tar.gz",
|
||||
"archive_sha256": "fac790b8bdca1689999de0562c360f299df6a78f542f86ada86b997b0091eafb",
|
||||
"archive_size": 254243,
|
||||
"archive_url": "https://git.m4ko.net/m4ko/pristontale-eu-linux/releases/download/v0.8.0/pristontale-eu-linux-update-v0.8.0.tar.gz",
|
||||
"channel": "stable",
|
||||
"minimum_updater_version": "0.5.0",
|
||||
"notes_url": "https://git.m4ko.net/m4ko/pristontale-eu-linux/releases/tag/v0.6.0",
|
||||
"published_at": "2026-08-29T21:16:21Z",
|
||||
"notes_url": "https://git.m4ko.net/m4ko/pristontale-eu-linux/releases/tag/v0.8.0",
|
||||
"published_at": "2026-08-31T21:27:27Z",
|
||||
"schema": 1,
|
||||
"version": "0.6.0"
|
||||
"version": "0.8.0"
|
||||
}
|
||||
|
||||
Binary file not shown.
Reference in New Issue
Block a user