diff --git a/gui/wolf_workflow_tab.py b/gui/wolf_workflow_tab.py index f4ac291..04eb4ea 100644 --- a/gui/wolf_workflow_tab.py +++ b/gui/wolf_workflow_tab.py @@ -59,6 +59,7 @@ from PyQt5.QtWidgets import ( QListWidget, QListWidgetItem, QMessageBox, + QProgressBar, QPushButton, QScrollArea, QSizePolicy, @@ -243,11 +244,13 @@ _WOLF_SPEAKER_PROMPT = ( class _WolfTaskWorker(QThread): """Run a blocking WolfDawn task callable off the UI thread. - The task receives a ``log`` callable and returns ``(success, message)``. + The task receives ``log`` and ``progress`` callables and returns + ``(success, message)``. """ done = pyqtSignal(bool, str) log = pyqtSignal(str) + progress = pyqtSignal(int, int, str) def __init__(self, task): super().__init__() @@ -255,7 +258,10 @@ class _WolfTaskWorker(QThread): def run(self): try: - ok, msg = self._task(self.log.emit) + ok, msg = self._task( + self.log.emit, + lambda current, total, label: self.progress.emit(current, total, label), + ) self.done.emit(bool(ok), str(msg)) except Exception as exc: # pragma: no cover - defensive import traceback @@ -391,7 +397,7 @@ class WolfWorkflowTab(QWidget): except Exception as exc: log(f" ⚠ could not snapshot original {src.name}: {exc}") - def _ensure_originals(self, manifest: dict, log) -> None: + def _ensure_originals(self, manifest: dict, log, progress=None) -> None: """Make sure a pristine snapshot exists; if entries are missing it and the game still has its .wolf archives (packaging renames them to .wolf.bak), rebuild the snapshot by unpacking those archives into originals/.""" @@ -435,7 +441,13 @@ class WolfWorkflowTab(QWidget): inputs.append(str(staged)) else: inputs.append(str(arc)) - res = wolfdawn.unpack_all(inputs, str(originals), log_fn=log) + res = wolfdawn.unpack_all( + inputs, + str(originals), + log_fn=log, + progress_fn=progress, + progress_total=len(inputs), + ) if not res.ok: log(f" ⚠ could not rebuild originals (unpack exit {res.returncode}).") @@ -587,6 +599,34 @@ class WolfWorkflowTab(QWidget): ) lp_layout.addWidget(log_header) + self.task_progress_row = QWidget() + progress_layout = QVBoxLayout(self.task_progress_row) + progress_layout.setContentsMargins(10, 8, 10, 4) + progress_layout.setSpacing(4) + self.task_progress_label = QLabel("") + self.task_progress_label.setStyleSheet("color:#9d9d9d;font-size:11px;background:transparent;") + progress_layout.addWidget(self.task_progress_label) + self.task_progress_bar = QProgressBar() + self.task_progress_bar.setFixedHeight(16) + self.task_progress_bar.setTextVisible(True) + self.task_progress_bar.setStyleSheet(""" + QProgressBar { + border: 1px solid #3c3c3c; + border-radius: 3px; + text-align: center; + background-color: #252526; + color: #cccccc; + font-size: 10px; + } + QProgressBar::chunk { + background-color: #007acc; + border-radius: 2px; + } + """) + progress_layout.addWidget(self.task_progress_bar) + self.task_progress_row.setVisible(False) + lp_layout.addWidget(self.task_progress_row) + self.log_area = QTextEdit() self.log_area.setReadOnly(True) self.log_area.setFont(QFont("Consolas", 9)) @@ -670,6 +710,24 @@ class WolfWorkflowTab(QWidget): for btn in self._buttons: btn.setEnabled(not busy) + def _show_task_progress(self, current: int, total: int, label: str): + self.task_progress_row.setVisible(True) + self.task_progress_label.setText(label) + if total <= 0: + self.task_progress_bar.setRange(0, 0) + self.task_progress_bar.setFormat("") + else: + self.task_progress_bar.setRange(0, total) + self.task_progress_bar.setValue(max(0, min(current, total))) + self.task_progress_bar.setFormat(f"{current}/{total} %p%") + + def _hide_task_progress(self): + self.task_progress_row.setVisible(False) + self.task_progress_label.clear() + self.task_progress_bar.setRange(0, 100) + self.task_progress_bar.setValue(0) + self.task_progress_bar.setFormat("") + def _run_task(self, task, on_done=None): """Run *task* in a worker thread; disables buttons until it finishes.""" if self._worker is not None and self._worker.isRunning(): @@ -679,11 +737,14 @@ class WolfWorkflowTab(QWidget): QMessageBox.information(self, "Busy", "An import is already running. Please wait.") return self._set_busy(True) + self._show_task_progress(0, 0, "Working …") worker = _WolfTaskWorker(task) self._worker = worker worker.log.connect(self._log) + worker.progress.connect(self._show_task_progress) def _finished(ok: bool, msg: str): + self._hide_task_progress() self._set_busy(False) if msg: self._log(("✅ " if ok else "❌ ") + msg) @@ -1241,7 +1302,7 @@ class WolfWorkflowTab(QWidget): work_dir = self._work_dir() files_dir = Path("files") - def task(log): + def task(log, progress=None): from util.dazedformat import format_json_files total = 0 @@ -1289,7 +1350,7 @@ class WolfWorkflowTab(QWidget): self._log("⚠ No game folder set. Complete Step 0 first.") return - def task(log): + def task(log, progress=None): from util.dazedformat import format_json_files work_dir = self._work_dir() @@ -1347,35 +1408,62 @@ class WolfWorkflowTab(QWidget): return out_dir = root / "Data" - def task(log): + def task(log, progress=None): from util import wolfdawn log(f"Unpacking {len(archives)} archive(s) into {out_dir} …") - res = wolfdawn.unpack_all([str(a) for a in archives], str(out_dir), log_fn=log) + res = wolfdawn.unpack_all( + [str(a) for a in archives], + str(out_dir), + log_fn=log, + progress_fn=progress, + progress_total=len(archives), + ) if not res.ok: return False, f"unpack-all exited {res.returncode}" return True, "Unpacked archives into Data/." self._run_task(task, on_done=on_done) - def _ensure_text_archives_unpacked(self, data_dir: Path, log) -> bool: + def _ensure_text_archives_unpacked(self, data_dir: Path, log, progress=None) -> bool: """Unpack BasicData/MapData archives when their binaries are not loose yet.""" from util import wolfdawn archives = find_wolf_text_archives(self._game_root, data_dir) - ok = True + pending: list[Path] = [] basic = data_dir / "BasicData" if "BasicData" in archives and not (basic / "CommonEvent.dat").is_file(): - log(f"Unpacking {archives['BasicData'].name} …") - res = wolfdawn.unpack_all([str(archives["BasicData"])], str(data_dir), log_fn=log) - if not res.ok: - log(f" ⚠ unpack failed (exit {res.returncode})") - ok = False + pending.append(archives["BasicData"]) if "MapData" in archives and not wolf_has_maps(data_dir): - log(f"Unpacking {archives['MapData'].name} …") - res = wolfdawn.unpack_all([str(archives["MapData"])], str(data_dir), log_fn=log) + pending.append(archives["MapData"]) + if not pending: + return True + + ok = True + total = len(pending) + for idx, arc in enumerate(pending): + if progress: + progress(idx, total, f"Unpacking {arc.name} ({idx + 1}/{total}) …") + log(f"Unpacking {arc.name} …") + + def scoped_progress(current, _total, label, base=idx, archive=arc): + if not progress: + return + step = min(current, 1) + detail = label or archive.name + progress(base + step, total, detail) + + res = wolfdawn.unpack_all( + [str(arc)], + str(data_dir), + log_fn=log, + progress_fn=scoped_progress if progress else None, + progress_total=1, + ) if not res.ok: log(f" ⚠ unpack failed (exit {res.returncode})") ok = False + elif progress: + progress(idx + 1, total, f"Unpacked {arc.name}") return ok def _extract_to_work_dir( @@ -1399,10 +1487,10 @@ class WolfWorkflowTab(QWidget): work_dir = self._work_dir() game_root = self._game_root - def task(log): + def task(log, progress=None): from util import wolfdawn - if not self._ensure_text_archives_unpacked(Path(data_dir), log): + if not self._ensure_text_archives_unpacked(Path(data_dir), log, progress): return False, "Could not unpack text archives (see log)." layout = detect_wolf_layout(self._game_root) @@ -2104,7 +2192,7 @@ class WolfWorkflowTab(QWidget): return manifest = self._read_manifest() - def task(log): + def task(log, progress=None): from util import wolfdawn json_files = [] @@ -2155,12 +2243,12 @@ class WolfWorkflowTab(QWidget): except Exception as e: log(f" ⚠ could not update {json_name} in {WORK_DIR_NAME}/: {e}") - def task(log): + def task(log, progress=None): from util import wolfdawn # Guarantee a pristine snapshot to inject from (rebuild from archives if # an older extraction predates snapshotting). - self._ensure_originals(manifest, log) + self._ensure_originals(manifest, log, progress) entries = manifest["entries"] data_dir = manifest["data_dir"] @@ -2343,7 +2431,7 @@ class WolfWorkflowTab(QWidget): QMessageBox.information(self, "Package", "No .wolf archives to back up — already loose.") return - def task(log): + def task(log, progress=None): renamed = 0 for arc in archives: bak = arc.with_suffix(arc.suffix + ".bak") @@ -2373,7 +2461,7 @@ class WolfWorkflowTab(QWidget): like = cand break - def task(log): + def task(log, progress=None): from util import wolfdawn log(f"Repacking {data_dir} → {output} …") @@ -2423,7 +2511,7 @@ class WolfWorkflowTab(QWidget): basic = info.get("basic_data") game_dat = Path(basic) / "Game.dat" if basic else None - def task(log): + def task(log, progress=None): from util import wolfdawn if not game_dat or not game_dat.is_file(): diff --git a/tests/test_wolf_unpack_progress.py b/tests/test_wolf_unpack_progress.py new file mode 100644 index 0000000..1eb31f5 --- /dev/null +++ b/tests/test_wolf_unpack_progress.py @@ -0,0 +1,91 @@ +"""Tests for WolfDawn unpack progress helpers.""" + +from __future__ import annotations + +import unittest +from pathlib import Path +from unittest.mock import patch + +from util.wolfdawn import ( + _UNPACK_ARCHIVE_DONE_RE, + _UNPACK_SUMMARY_RE, + _run_streaming, + count_unpack_archives, +) + + +class WolfUnpackProgressTests(unittest.TestCase): + def test_unpack_archive_line_regex(self): + line = " MapData.wolf -> /tmp/out/MapData (74 files)" + match = _UNPACK_ARCHIVE_DONE_RE.match(line) + self.assertIsNotNone(match) + self.assertEqual(match.group("name"), "MapData.wolf") + self.assertEqual(match.group("files"), "74") + + def test_unpack_summary_regex(self): + line = "unpack-all: 24 archive(s), 60265 files" + match = _UNPACK_SUMMARY_RE.match(line) + self.assertIsNotNone(match) + self.assertEqual(match.group("archives"), "24") + self.assertEqual(match.group("files"), "60265") + + def test_count_unpack_archives_explicit_list(self): + import tempfile + + with tempfile.TemporaryDirectory() as raw: + base = Path(raw) + (base / "A.wolf").write_bytes(b"") + (base / "B.wolf").write_bytes(b"") + self.assertEqual(count_unpack_archives([base / "A.wolf", base / "B.wolf"]), 2) + + def test_count_unpack_archives_directory(self): + import tempfile + + with tempfile.TemporaryDirectory() as raw: + base = Path(raw) + (base / "MapData.wolf").write_bytes(b"") + (base / "BasicData.wolf").write_bytes(b"") + self.assertEqual(count_unpack_archives([base]), 2) + + def test_run_streaming_emits_unpack_progress(self): + events: list[tuple[int, int, str]] = [] + + def fake_popen(argv, **kwargs): + class FakeProc: + stdout = [ + " BasicData.wolf -> /tmp/BasicData (10 files)\n", + " MapData.wolf -> /tmp/MapData (5 files)\n", + "unpack-all: 2 archive(s), 15 files\n", + ] + + def __init__(self): + self._idx = 0 + + def readline(self): + if self._idx >= len(self.stdout): + return "" + line = self.stdout[self._idx] + self._idx += 1 + return line + + def wait(self): + return 0 + + return FakeProc() + + with patch("util.wolfdawn.ensure_wolf_binary", return_value=Path("/wolf")): + with patch("util.wolfdawn.subprocess.Popen", side_effect=fake_popen): + result = _run_streaming( + ["unpack-all", "A.wolf", "-o", "/tmp/out"], + progress_fn=lambda c, t, l: events.append((c, t, l)), + progress_total=2, + ) + + self.assertTrue(result.ok) + self.assertEqual(events[0], (0, 2, "Unpacking archive 1 of 2 …")) + self.assertEqual(events[-1][0], 2) + self.assertEqual(events[-1][1], 2) + + +if __name__ == "__main__": + unittest.main() diff --git a/util/wolfdawn/__init__.py b/util/wolfdawn/__init__.py index 7b63697..ecae330 100644 --- a/util/wolfdawn/__init__.py +++ b/util/wolfdawn/__init__.py @@ -28,7 +28,7 @@ import urllib.request import zipfile from dataclasses import dataclass from pathlib import Path -from typing import Iterable, Optional, Sequence, Union +from typing import Callable, Iterable, Optional, Sequence, Union __all__ = [ "WolfDawnError", @@ -58,6 +58,15 @@ _NAMES_INJECT_COUNTS_RE = re.compile( r"applied\s+(\d+)\s+name change.*?(\d+)\s+drifted", re.IGNORECASE | re.DOTALL ) +ProgressFn = Callable[[int, int, str], None] + +_UNPACK_ARCHIVE_DONE_RE = re.compile( + r"^\s+(?P\S+\.wolf)\s+->\s+(?P.+?)\s+\((?P\d+) files\)\s*$" +) +_UNPACK_SUMMARY_RE = re.compile( + r"^unpack-all:\s+(?P\d+) archive\(s\),\s+(?P\d+) files\s*$" +) + def parse_strings_inject_counts(stdout: str) -> tuple[int | None, int | None]: """Return (applied, drifted) from wolf strings-inject output, or (None, None).""" @@ -308,18 +317,135 @@ def _run(args: Sequence[str], log_fn=None) -> WolfResult: return result +def _emit_unpack_progress( + progress_fn: ProgressFn | None, + current: int, + total: int, + label: str, +) -> None: + if progress_fn: + progress_fn(current, total, label) + + +def _run_streaming( + args: Sequence[str], + log_fn=None, + progress_fn: ProgressFn | None = None, + progress_total: int | None = None, +) -> WolfResult: + """Run ``wolf`` and stream merged stdout/stderr line-by-line.""" + binary = ensure_wolf_binary(log_fn=log_fn) + argv = [str(binary), *[str(a) for a in args]] + stdout_lines: list[str] = [] + stderr_lines: list[str] = [] + is_unpack = bool(args) and args[0] == "unpack-all" + total = progress_total or 0 + current = 0 + + if is_unpack and progress_fn and total > 0: + _emit_unpack_progress( + progress_fn, + 0, + total, + f"Unpacking archive 1 of {total} …", + ) + + proc = subprocess.Popen( + argv, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + encoding="utf-8", + errors="replace", + ) + assert proc.stdout is not None + for raw_line in proc.stdout: + line = raw_line.rstrip("\r\n") + stdout_lines.append(line) + if log_fn: + log_fn(line) + if not is_unpack or not progress_fn: + continue + summary = _UNPACK_SUMMARY_RE.match(line) + if summary: + total = int(summary.group("archives")) + continue + match = _UNPACK_ARCHIVE_DONE_RE.match(line) + if not match: + continue + current += 1 + if total <= 0: + total = current + name = match.group("name") + files = match.group("files") + if current < total: + label = f"Unpacked {name} ({files} files) — archive {current + 1} of {total} …" + else: + label = f"Unpacked {name} ({files} files)" + _emit_unpack_progress(progress_fn, current, total, label) + + returncode = proc.wait() + stdout = "\n".join(stdout_lines) + return WolfResult( + returncode=returncode, + stdout=stdout, + stderr=stdout, + argv=argv, + ) + + +def count_unpack_archives(inputs: Union[PathLike, Iterable[PathLike]]) -> int | None: + """Return how many ``.wolf`` archives *inputs* will unpack, or None if unknown.""" + if isinstance(inputs, (str, Path)): + inputs = [inputs] + total = 0 + for item in inputs: + path = Path(item) + if path.is_file(): + lower = path.name.lower() + if lower.endswith(".wolf") or lower.endswith(".wolf.bak"): + total += 1 + else: + return None + elif path.is_dir(): + total += len(list(path.glob("*.wolf"))) + else: + return None + return total + + # --------------------------------------------------------------------------- # Command wrappers # --------------------------------------------------------------------------- -def unpack_all(inputs: Union[PathLike, Iterable[PathLike]], out_dir: PathLike, log_fn=None) -> WolfResult: +def unpack_all( + inputs: Union[PathLike, Iterable[PathLike]], + out_dir: PathLike, + log_fn=None, + progress_fn: ProgressFn | None = None, + progress_total: int | None = None, +) -> WolfResult: """``wolf unpack-all ... -o ``. Each ``Name.wolf`` is unpacked to ``/Name/``. + + When *progress_fn* is set, it receives ``(current, total, label)`` as each + archive finishes. Pass *progress_total* when known; otherwise the archive + count is inferred from ``unpack-all: N archive(s), …`` in the CLI output. """ if isinstance(inputs, (str, Path)): inputs = [inputs] - args = ["unpack-all", *[_str(p) for p in inputs], "-o", _str(out_dir)] + input_list = [_str(p) for p in inputs] + if progress_total is None: + progress_total = count_unpack_archives(inputs) + args = ["unpack-all", *input_list, "-o", _str(out_dir)] + if progress_fn is not None: + return _run_streaming( + args, + log_fn=log_fn, + progress_fn=progress_fn, + progress_total=progress_total, + ) return _run(args, log_fn=log_fn)