Compare commits
13 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 6900c20734 | |||
| f96aba9d99 | |||
| 78dade0d86 | |||
| d819809197 | |||
| dadae8f60a | |||
| 99982b12d9 | |||
| 75b21acf87 | |||
| b16540099d | |||
| b0855ccd9b | |||
| 87dee93b86 | |||
| fe883047e0 | |||
| 20a0bbcd75 | |||
| ed0c278a8b |
23
.gitignore
vendored
23
.gitignore
vendored
@@ -3,8 +3,25 @@ __pycache__/
|
|||||||
*.py[cod]
|
*.py[cod]
|
||||||
.pytest_cache/
|
.pytest_cache/
|
||||||
.idea/
|
.idea/
|
||||||
dist/
|
.run/
|
||||||
release-keys/
|
release-keys/
|
||||||
config/local.toml
|
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
|
# PristonTale EU for Linux
|
||||||
|
|
||||||
Version 0.6.7 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
|
native Linux dashboard. Game files come directly from the publisher and are not
|
||||||
included in this archive.
|
included in this archive.
|
||||||
|
|
||||||
@@ -15,11 +15,11 @@ Then run:
|
|||||||
```text
|
```text
|
||||||
openssl pkey -pubin -in launcher-update-public.pem -outform DER | sha256sum
|
openssl pkey -pubin -in launcher-update-public.pem -outform DER | sha256sum
|
||||||
openssl dgst -sha256 -verify launcher-update-public.pem \
|
openssl dgst -sha256 -verify launcher-update-public.pem \
|
||||||
-signature pristontale-eu-linux-installer-v0-6-7.tar.gz.sig \
|
-signature pristontale-eu-linux-installer-v0-8-1.tar.gz.sig \
|
||||||
pristontale-eu-linux-installer-v0-6-7.tar.gz
|
pristontale-eu-linux-installer-v0-8-1.tar.gz
|
||||||
sha256sum -c pristontale-eu-linux-installer-v0-6-7.tar.gz.sha256
|
sha256sum -c pristontale-eu-linux-installer-v0-8-1.tar.gz.sha256
|
||||||
tar -xzf pristontale-eu-linux-installer-v0-6-7.tar.gz
|
tar -xzf pristontale-eu-linux-installer-v0-8-1.tar.gz
|
||||||
cd pristontale-eu-linux-installer-v0-6-7
|
cd pristontale-eu-linux-installer-v0-8-1
|
||||||
./install.sh
|
./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
|
upgrades. The installer adds **PristonTale EU** to the application menu and
|
||||||
installs the `pristontale-eu` command in `~/.local/bin`.
|
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
|
## Use
|
||||||
|
|
||||||
Open **PristonTale EU** from the application menu or run:
|
Open **PristonTale EU** from the application menu or run:
|
||||||
|
|||||||
@@ -4,4 +4,4 @@ from .paths import AppPaths
|
|||||||
|
|
||||||
__all__ = ["AppPaths", "__version__"]
|
__all__ = ["AppPaths", "__version__"]
|
||||||
|
|
||||||
__version__ = "0.6.7"
|
__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)
|
||||||
@@ -819,6 +819,7 @@ def _play_locked(
|
|||||||
runner,
|
runner,
|
||||||
start_option=start_option,
|
start_option=start_option,
|
||||||
use_gamebox=not no_box,
|
use_gamebox=not no_box,
|
||||||
|
concurrent_prefix_launch=reuse_prepared,
|
||||||
)
|
)
|
||||||
except KeyboardInterrupt:
|
except KeyboardInterrupt:
|
||||||
current_game_log = _current_game_log(
|
current_game_log = _current_game_log(
|
||||||
|
|||||||
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")
|
||||||
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 typing import Mapping
|
||||||
|
|
||||||
from .launcher import find_wine_tools
|
from .launcher import find_wine_tools
|
||||||
|
from .runtime import find_steam_proton7
|
||||||
|
|
||||||
|
|
||||||
ARCH_FAMILY_DISTRIBUTIONS = frozenset(
|
ARCH_FAMILY_DISTRIBUTIONS = frozenset(
|
||||||
@@ -176,19 +177,7 @@ def _package_hint(distribution: str) -> str | None:
|
|||||||
|
|
||||||
|
|
||||||
def _has_steam_proton7(environment: Mapping[str, str]) -> bool:
|
def _has_steam_proton7(environment: Mapping[str, str]) -> bool:
|
||||||
home_value = environment.get("HOME")
|
return find_steam_proton7(environment) is not None
|
||||||
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
|
|
||||||
|
|
||||||
|
|
||||||
def _platform_optional_tools(distribution: str, *, architecture: str) -> list[ToolStatus]:
|
def _platform_optional_tools(distribution: str, *, architecture: str) -> list[ToolStatus]:
|
||||||
|
|||||||
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
|
||||||
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"
|
BROADCAST_ENDPOINT = "http://5.135.35.188:90/broadcasts"
|
||||||
MAX_NEWS_BYTES = 2 * 1024 * 1024
|
MAX_NEWS_BYTES = 2 * 1024 * 1024
|
||||||
MAX_NEWS_IMAGE_BYTES = 12 * 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"
|
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"))
|
value = json.loads(payload.decode("utf-8-sig"))
|
||||||
if not isinstance(value, list) or not value or not isinstance(value[0], dict):
|
if not isinstance(value, list) or not value or not isinstance(value[0], dict):
|
||||||
raise NewsError("broadcast response does not contain a latest broadcast")
|
raise NewsError("broadcast response does not contain a latest broadcast")
|
||||||
latest = value[0]
|
return _broadcast_from_mapping(value[0])
|
||||||
author = latest.get("author")
|
|
||||||
content = latest.get("content")
|
|
||||||
|
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"))
|
url = _broadcast_url(latest.get("url"))
|
||||||
if (
|
if (
|
||||||
not isinstance(author, str)
|
author is None
|
||||||
or not author.strip()
|
or content is None
|
||||||
or not isinstance(content, str)
|
|
||||||
or not content.strip()
|
|
||||||
or url is None
|
or url is None
|
||||||
):
|
):
|
||||||
raise NewsError("latest broadcast is incomplete")
|
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(
|
return Broadcast(
|
||||||
author=author.strip(),
|
author=author,
|
||||||
content=content.strip(),
|
content=content,
|
||||||
published_at=_format_timestamp(latest.get("timestamp")),
|
published_at=published_at,
|
||||||
url=url,
|
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:
|
def _format_timestamp(value: object) -> str | None:
|
||||||
if isinstance(value, bool) or not isinstance(value, (int, float)):
|
if isinstance(value, bool) or not isinstance(value, (int, float)):
|
||||||
return None
|
return None
|
||||||
@@ -288,10 +305,16 @@ def _image_url(value: object) -> str | None:
|
|||||||
def _broadcast_url(value: object) -> str | None:
|
def _broadcast_url(value: object) -> str | None:
|
||||||
if not isinstance(value, str) or not value.strip():
|
if not isinstance(value, str) or not value.strip():
|
||||||
return None
|
return None
|
||||||
parsed = urlparse(value)
|
url = value.strip()
|
||||||
if parsed.scheme not in {"http", "https"} or not parsed.netloc:
|
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 None
|
||||||
return value.strip()
|
return url
|
||||||
|
|
||||||
|
|
||||||
def _read_cache(path: Path) -> tuple[NewsArticle, ...]:
|
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:
|
def _read_broadcast_cache(path: Path) -> Broadcast | None:
|
||||||
try:
|
try:
|
||||||
value = json.loads(path.read_text(encoding="utf-8"))
|
value = json.loads(path.read_text(encoding="utf-8"))
|
||||||
return Broadcast(**value) if isinstance(value, dict) else None
|
return _broadcast_from_mapping(value) if isinstance(value, dict) else None
|
||||||
except (OSError, UnicodeError, TypeError, json.JSONDecodeError):
|
except (OSError, UnicodeError, NewsError, json.JSONDecodeError):
|
||||||
return None
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -9,6 +9,7 @@ import json
|
|||||||
import os
|
import os
|
||||||
from pathlib import Path, PurePosixPath
|
from pathlib import Path, PurePosixPath
|
||||||
import platform
|
import platform
|
||||||
|
import re
|
||||||
import shlex
|
import shlex
|
||||||
import shutil
|
import shutil
|
||||||
import sys
|
import sys
|
||||||
@@ -36,6 +37,12 @@ CURRENT_UMU_DISTRIBUTIONS = frozenset(
|
|||||||
{"arch", "cachyos", "endeavouros", "garuda", "manjaro"}
|
{"arch", "cachyos", "endeavouros", "garuda", "manjaro"}
|
||||||
)
|
)
|
||||||
GAME_RUNTIME_CONFIGURATION_VERSION = 1
|
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):
|
class RuntimeErrorPT(RuntimeError):
|
||||||
@@ -80,7 +87,7 @@ def select_runtime(
|
|||||||
proton = Path(settings.proton_path).expanduser().resolve()
|
proton = Path(settings.proton_path).expanduser().resolve()
|
||||||
if not proton.is_file():
|
if not proton.is_file():
|
||||||
raise RuntimeErrorPT(f"configured Proton script does not exist: {proton}")
|
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 settings.runtime == "proton7-container":
|
||||||
if not umu:
|
if not umu:
|
||||||
raise RuntimeErrorPT(
|
raise RuntimeErrorPT(
|
||||||
@@ -137,33 +144,108 @@ def select_runtime(
|
|||||||
def find_steam_proton7(
|
def find_steam_proton7(
|
||||||
environment: Mapping[str, str] | None = None,
|
environment: Mapping[str, str] | None = None,
|
||||||
) -> tuple[Path, Path] | 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
|
env = os.environ if environment is None else environment
|
||||||
home_value = env.get("HOME")
|
for steam_root in _steam_client_roots(env):
|
||||||
if not home_value:
|
for library_root in _steam_library_roots(steam_root):
|
||||||
return None
|
proton = library_root / PROTON7_RELATIVE_PATH
|
||||||
home = Path(home_value).expanduser()
|
if proton.is_file():
|
||||||
roots: list[Path] = []
|
return proton.resolve(), steam_root
|
||||||
explicit = env.get("STEAM_COMPAT_CLIENT_INSTALL_PATH")
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
def _steam_client_roots(environment: Mapping[str, str]) -> tuple[Path, ...]:
|
||||||
|
candidates: list[Path] = []
|
||||||
|
explicit = environment.get("STEAM_COMPAT_CLIENT_INSTALL_PATH")
|
||||||
if explicit:
|
if explicit:
|
||||||
roots.append(Path(explicit).expanduser())
|
candidates.append(Path(explicit).expanduser())
|
||||||
roots.extend(
|
home_value = environment.get("HOME")
|
||||||
(
|
if home_value:
|
||||||
home / ".local/share/Steam",
|
home = Path(home_value).expanduser()
|
||||||
home / ".steam/steam",
|
candidates.extend(
|
||||||
home / "snap/steam/common/.local/share/Steam",
|
(
|
||||||
home / ".var/app/com.valvesoftware.Steam/.local/share/Steam",
|
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()
|
seen: set[Path] = set()
|
||||||
for root in roots:
|
for path in paths:
|
||||||
resolved = root.resolve()
|
try:
|
||||||
|
resolved = path.resolve()
|
||||||
|
except (OSError, RuntimeError):
|
||||||
|
continue
|
||||||
if resolved in seen:
|
if resolved in seen:
|
||||||
continue
|
continue
|
||||||
seen.add(resolved)
|
seen.add(resolved)
|
||||||
proton = resolved / "steamapps/common/Proton 7.0/proton"
|
result.append(resolved)
|
||||||
if proton.is_file():
|
return tuple(result)
|
||||||
return proton, resolved
|
|
||||||
return None
|
|
||||||
|
|
||||||
|
|
||||||
def prepare_game_runtime(
|
def prepare_game_runtime(
|
||||||
@@ -298,6 +380,7 @@ def run_game(
|
|||||||
*,
|
*,
|
||||||
start_option: DepotStartOption = DEFAULT_DEPOT_START_OPTION,
|
start_option: DepotStartOption = DEFAULT_DEPOT_START_OPTION,
|
||||||
use_gamebox: bool = True,
|
use_gamebox: bool = True,
|
||||||
|
concurrent_prefix_launch: bool = False,
|
||||||
timeout_seconds: float = 12 * 60 * 60,
|
timeout_seconds: float = 12 * 60 * 60,
|
||||||
check: bool = True,
|
check: bool = True,
|
||||||
extra_environment: Mapping[str, str] | None = None,
|
extra_environment: Mapping[str, str] | None = None,
|
||||||
@@ -346,6 +429,13 @@ def run_game(
|
|||||||
selection,
|
selection,
|
||||||
synchronization=resolve_proton_sync_mode(settings.proton_sync),
|
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:
|
if use_gamebox:
|
||||||
game_command = build_gamebox_command(game_command, settings)
|
game_command = build_gamebox_command(game_command, settings)
|
||||||
if extra_environment:
|
if extra_environment:
|
||||||
@@ -653,15 +743,27 @@ def _ensure_proton_game_link(paths: AppPaths) -> None:
|
|||||||
link.symlink_to(paths.runtime_view, target_is_directory=True)
|
link.symlink_to(paths.runtime_view, target_is_directory=True)
|
||||||
|
|
||||||
|
|
||||||
def _steam_root_for_proton(proton: Path) -> Path:
|
def _steam_root_for_proton(
|
||||||
parts = proton.parts
|
proton: Path,
|
||||||
try:
|
environment: Mapping[str, str] | None = None,
|
||||||
index = parts.index("steamapps")
|
) -> Path:
|
||||||
except ValueError as exc:
|
steamapps = next(
|
||||||
|
(parent for parent in proton.parents if parent.name.casefold() == "steamapps"),
|
||||||
|
None,
|
||||||
|
)
|
||||||
|
if steamapps is None:
|
||||||
raise RuntimeErrorPT(
|
raise RuntimeErrorPT(
|
||||||
"configured Proton path must be inside a Steam steamapps directory"
|
"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:
|
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
|
||||||
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
|
||||||
1648
ptlinux/ui.py
1648
ptlinux/ui.py
File diff suppressed because it is too large
Load Diff
@@ -1,6 +1,6 @@
|
|||||||
[project]
|
[project]
|
||||||
name = "PTEU-Linux-Launcher"
|
name = "PTEU-Linux-Launcher"
|
||||||
version = "0.6.7"
|
version = "0.8.1"
|
||||||
description = "PristonTale EU Linux installer, clean launcher, and diagnostic harness"
|
description = "PristonTale EU Linux installer, clean launcher, and diagnostic harness"
|
||||||
requires-python = ">=3.11"
|
requires-python = ">=3.11"
|
||||||
dependencies = []
|
dependencies = []
|
||||||
|
|||||||
@@ -1,11 +1,11 @@
|
|||||||
{
|
{
|
||||||
"archive_sha256": "58d1161b60b5169f39244b0e01dc470eeb1b1978100da30e454c18ce44006d0c",
|
"archive_sha256": "fac790b8bdca1689999de0562c360f299df6a78f542f86ada86b997b0091eafb",
|
||||||
"archive_size": 230390,
|
"archive_size": 254243,
|
||||||
"archive_url": "https://git.m4ko.net/m4ko/pristontale-eu-linux/releases/download/v0.6.7/pristontale-eu-linux-update-v0.6.7.tar.gz",
|
"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",
|
"channel": "stable",
|
||||||
"minimum_updater_version": "0.5.0",
|
"minimum_updater_version": "0.5.0",
|
||||||
"notes_url": "https://git.m4ko.net/m4ko/pristontale-eu-linux/releases/tag/v0.6.7",
|
"notes_url": "https://git.m4ko.net/m4ko/pristontale-eu-linux/releases/tag/v0.8.0",
|
||||||
"published_at": "2026-08-30T10:35:20Z",
|
"published_at": "2026-08-31T21:27:27Z",
|
||||||
"schema": 1,
|
"schema": 1,
|
||||||
"version": "0.6.7"
|
"version": "0.8.0"
|
||||||
}
|
}
|
||||||
|
|||||||
Binary file not shown.
Reference in New Issue
Block a user