218 lines
7.5 KiB
Python
218 lines
7.5 KiB
Python
"""Vanilla PAK baseline, mod slot allocation and title-update detection.
|
|
|
|
Wilds has two parallel patch chains. Mods must sit *above* the vanilla ceiling of
|
|
their chain, and the ceiling rises with every title update, so slots are derived
|
|
rather than fixed.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import re
|
|
import struct
|
|
import time
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
|
|
from .layout import PAK, Entry
|
|
from .util import atomic_write_json, read_json
|
|
|
|
PAK_RE = re.compile(r"^re_chunk_000\.pak(?P<sub>\.sub_000\.pak)?\.patch_(?P<n>\d{3})\.pak$")
|
|
CHAINS = ("main", "sub")
|
|
_BASES = {"main": "re_chunk_000.pak", "sub": "re_chunk_000.pak.sub_000.pak"}
|
|
|
|
|
|
def pak_name(chain: str, n: int) -> str:
|
|
return f"{_BASES[chain]}.patch_{n:03d}.pak"
|
|
|
|
|
|
def parse_pak_name(name: str) -> tuple[str, int] | None:
|
|
m = PAK_RE.match(name)
|
|
if not m:
|
|
return None
|
|
return ("sub" if m.group("sub") else "main"), int(m.group("n"))
|
|
|
|
|
|
def scan_game_paks(game_dir: Path) -> dict[str, dict[int, dict]]:
|
|
"""chain -> patch number -> {name,size,mtime_ns}. Only real patch files are considered."""
|
|
out: dict[str, dict[int, dict]] = {c: {} for c in CHAINS}
|
|
try:
|
|
it = list(os.scandir(game_dir))
|
|
except OSError:
|
|
return out
|
|
for e in it:
|
|
parsed = parse_pak_name(e.name)
|
|
if parsed is None:
|
|
continue
|
|
try:
|
|
st = e.stat()
|
|
except OSError:
|
|
continue
|
|
out[parsed[0]][parsed[1]] = {"name": e.name, "size": st.st_size, "mtime_ns": st.st_mtime_ns}
|
|
return out
|
|
|
|
|
|
# -- baseline ---------------------------------------------------------------
|
|
|
|
@dataclass
|
|
class Baseline:
|
|
chains: dict[str, dict] = field(default_factory=dict) # chain -> {"ceiling": int, "files": {name: {...}}}
|
|
captured: float = 0.0
|
|
|
|
def ceiling(self, chain: str) -> int:
|
|
return int(self.chains.get(chain, {}).get("ceiling", 0))
|
|
|
|
def to_json(self) -> dict:
|
|
return {"version": 1, "captured": self.captured, "chains": self.chains}
|
|
|
|
@classmethod
|
|
def from_json(cls, raw) -> "Baseline | None":
|
|
if not isinstance(raw, dict) or not isinstance(raw.get("chains"), dict):
|
|
return None
|
|
return cls(raw["chains"], float(raw.get("captured", 0)))
|
|
|
|
|
|
def capture_baseline(game_dir: Path, owned: set[str] = frozenset()) -> Baseline:
|
|
"""Snapshot vanilla PAKs. `owned` = file names we created; they are not vanilla."""
|
|
scan = scan_game_paks(game_dir)
|
|
chains = {}
|
|
for chain in CHAINS:
|
|
files = {i["name"]: {"size": i["size"], "mtime_ns": i["mtime_ns"]}
|
|
for n, i in scan[chain].items() if i["name"] not in owned}
|
|
nums = [n for n, i in scan[chain].items() if i["name"] not in owned]
|
|
chains[chain] = {"ceiling": max(nums, default=0), "files": files}
|
|
return Baseline(chains, time.time())
|
|
|
|
|
|
def load_baseline(path: Path) -> Baseline | None:
|
|
return Baseline.from_json(read_json(path, None))
|
|
|
|
|
|
def save_baseline(path: Path, baseline: Baseline) -> None:
|
|
atomic_write_json(path, baseline.to_json())
|
|
|
|
|
|
# -- slot allocation --------------------------------------------------------
|
|
|
|
@dataclass(frozen=True)
|
|
class Slot:
|
|
mod_id: str
|
|
src: Path
|
|
chain: str
|
|
number: int
|
|
|
|
@property
|
|
def dest(self) -> str:
|
|
# Deploy to pak_mods/ so REFramework's IntegrityCheckBypass can load them
|
|
# natively (fixing Linux/Proton issues and silent patch failures).
|
|
return f"pak_mods/{self.chain}_{self.number:03d}_{self.mod_id}.pak"
|
|
|
|
|
|
def _pak_sort_key(p: Path):
|
|
parsed = parse_pak_name(p.name)
|
|
return (parsed[1] if parsed else 10 ** 6, p.name.lower())
|
|
|
|
|
|
def allocate(mods: list[tuple[str, list[Entry]]], baseline: Baseline) -> list[Slot]:
|
|
"""Assign contiguous slots above each chain's ceiling, following load order."""
|
|
nxt = {c: baseline.ceiling(c) + 1 for c in CHAINS}
|
|
slots: list[Slot] = []
|
|
for mod_id, entries in mods:
|
|
for chain in CHAINS:
|
|
paks = sorted((e.src for e in entries if e.kind == PAK and e.chain == chain), key=_pak_sort_key)
|
|
for src in paks:
|
|
slots.append(Slot(mod_id, src, chain, nxt[chain]))
|
|
nxt[chain] += 1
|
|
return slots
|
|
|
|
|
|
# -- title update detection -------------------------------------------------
|
|
|
|
@dataclass
|
|
class UpdateReport:
|
|
updated: bool = False
|
|
reasons: list[str] = field(default_factory=list)
|
|
|
|
|
|
def detect_update(game_dir: Path, baseline: Baseline, owned: set[str]) -> UpdateReport:
|
|
"""A PAK we do not own above the ceiling, or a changed/missing vanilla PAK, means Steam patched."""
|
|
rep = UpdateReport()
|
|
scan = scan_game_paks(game_dir)
|
|
for chain in CHAINS:
|
|
ceiling = baseline.ceiling(chain)
|
|
for n, info in sorted(scan[chain].items()):
|
|
if n > ceiling and info["name"] not in owned:
|
|
rep.reasons.append(f"new PAK {info['name']}")
|
|
present = {i["name"]: i for i in scan[chain].values()}
|
|
for name, rec in baseline.chains.get(chain, {}).get("files", {}).items():
|
|
cur = present.get(name)
|
|
if cur is None:
|
|
rep.reasons.append(f"vanilla PAK {name} disappeared")
|
|
elif cur["size"] != rec["size"]:
|
|
rep.reasons.append(f"vanilla PAK {name} changed size")
|
|
rep.updated = bool(rep.reasons)
|
|
return rep
|
|
|
|
|
|
# -- PAK content hashes (best-effort conflict detection) ---------------------
|
|
|
|
def murmur3_32(data: bytes, seed: int = 0) -> int:
|
|
c1, c2 = 0xCC9E2D51, 0x1B873593
|
|
h = seed & 0xFFFFFFFF
|
|
n = len(data) // 4
|
|
for (k,) in struct.iter_unpack("<I", data[: n * 4]):
|
|
k = (k * c1) & 0xFFFFFFFF
|
|
k = ((k << 15) | (k >> 17)) & 0xFFFFFFFF
|
|
k = (k * c2) & 0xFFFFFFFF
|
|
h ^= k
|
|
h = ((h << 13) | (h >> 19)) & 0xFFFFFFFF
|
|
h = (h * 5 + 0xE6546B64) & 0xFFFFFFFF
|
|
tail = data[n * 4:]
|
|
k = 0
|
|
for i, b in enumerate(tail):
|
|
k |= b << (8 * i)
|
|
if tail:
|
|
k = (k * c1) & 0xFFFFFFFF
|
|
k = ((k << 15) | (k >> 17)) & 0xFFFFFFFF
|
|
k = (k * c2) & 0xFFFFFFFF
|
|
h ^= k
|
|
h ^= len(data)
|
|
h ^= h >> 16
|
|
h = (h * 0x85EBCA6B) & 0xFFFFFFFF
|
|
h ^= h >> 13
|
|
h = (h * 0xC2B2AE35) & 0xFFFFFFFF
|
|
h ^= h >> 16
|
|
return h
|
|
|
|
|
|
def path_hash(game_path: str) -> tuple[int, int]:
|
|
"""RE Engine KPKA entry key: murmur3 of the lower/upper-cased UTF-16LE path, seed 0xFFFFFFFF."""
|
|
p = game_path.replace("\\", "/")
|
|
return (murmur3_32(p.lower().encode("utf-16-le"), 0xFFFFFFFF),
|
|
murmur3_32(p.upper().encode("utf-16-le"), 0xFFFFFFFF))
|
|
|
|
|
|
_ENTRY = struct.Struct("<IIQQQQQ") # hash_lower, hash_upper, offset, csize, usize, attrs, checksum (48 B)
|
|
_pak_cache: dict[tuple[str, int, int], frozenset] = {}
|
|
|
|
|
|
def pak_hashes(path: Path) -> frozenset[tuple[int, int]]:
|
|
"""Set of (lower, upper) path hashes inside a KPKA archive; empty if unreadable."""
|
|
try:
|
|
st = os.stat(path)
|
|
key = (str(path), st.st_size, st.st_mtime_ns)
|
|
if key in _pak_cache:
|
|
return _pak_cache[key]
|
|
with open(path, "rb") as fh:
|
|
head = fh.read(16)
|
|
if len(head) < 16 or head[:4] != b"KPKA":
|
|
return frozenset()
|
|
total = struct.unpack_from("<I", head, 8)[0]
|
|
if total > 5_000_000 or 16 + total * _ENTRY.size > st.st_size:
|
|
return frozenset()
|
|
table = fh.read(total * _ENTRY.size)
|
|
result = frozenset((lo, up) for lo, up, *_ in _ENTRY.iter_unpack(table))
|
|
_pak_cache[key] = result
|
|
return result
|
|
except (OSError, struct.error):
|
|
return frozenset()
|