Launcher/pages/process_log.py

import subprocess
import threading
from collections import deque
from collections.abc import Callable
from datetime import datetime

from PySide6.QtCore import QObject, Signal
from PySide6.QtGui import QCloseEvent
from PySide6.QtWidgets import (
    QHBoxLayout,
    QLabel,
    QMainWindow,
    QPlainTextEdit,
    QPushButton,
    QVBoxLayout,
    QFileDialog,
    QSizePolicy,
    QWidget,
)

from core_lib.qt import messagebox


class _ProcessSignals(QObject):
    output = Signal(str, str)
    # Windows may return an unsigned 32-bit exit code (for example
    # 0xFFFFFFFF), which does not fit in Qt's signed Signal(int).
    finished = Signal(object)
    read_failed = Signal(str)


class ProcessLogWindow(QMainWindow):
    """
    A Simple client log view
    """

    MAX_LOG_LINES = 100_000
    LOG_TRIM_LINES = 10_000

    def __init__(self, app, process: subprocess.Popen, profile_id: str, title: str = "Game Log",
                 parent: QWidget | None = None, finished_callback: Callable[[str, int], None]=None) -> None:
        super().__init__(parent)
        self.app = app
        self.process = process
        self.profile_id = profile_id
        self._signals = _ProcessSignals(self)
        self._last_error_message: str | None = None
        self.lines: list[tuple[str, str]] = []  # type, message
        self._trimmed_line_count = 0
        self._stderr_lines: deque[str] = deque(maxlen=500)

        self.setWindowTitle(self.app.tr_text(title))
        self.resize(700, 720)

        container = QWidget(self)
        layout = QVBoxLayout(container)
        controls = QHBoxLayout()

        layout.setContentsMargins(0, 0, 0, 0)
        controls.setContentsMargins(8, 8, 8, 0)

        self.status_label = QLabel(self.app.tr_text("Running (PID {pid})", pid=process.pid))
        self.kill_button = QPushButton(self.app.tr_text("Kill Process"))
        self.save_log_button = QPushButton(self.app.tr_text("Save Log"))
        self.kill_button.clicked.connect(self.kill_process)
        self.save_log_button.clicked.connect(self.save_log)

        self.log_output = QPlainTextEdit()
        self.log_output.setReadOnly(True)
        self.log_output.setLineWrapMode(QPlainTextEdit.LineWrapMode.NoWrap)
        self.log_output.setMaximumBlockCount(self.MAX_LOG_LINES)
        self.log_output.setSizePolicy(QSizePolicy.Expanding, QSizePolicy.Expanding)

        controls.addWidget(self.status_label)
        controls.addStretch()
        controls.addWidget(self.save_log_button)
        controls.addWidget(self.kill_button)
        layout.addLayout(controls)
        layout.addWidget(self.log_output)
        self.setCentralWidget(container)

        self.setStyleSheet(
            """
            QMainWindow {
                spacing: 0;
            }
            
            QPlainTextEdit {
                background-color: #0f0f10;
                color: "#e5e5e5";
                font-family: Consolas;
                font-size: 10pt;
            }
            
            QScrollBar:vertical {
                background-color: #0f0f10;
                width: 5px;
                margin: 0;
            }
            
            QScrollBar::handle:vertical {
                background-color: #3a3a3a;
                min-height: 24px;
                border-radius: 5px;
            }
            
            QScrollBar::handle:vertical:hover {
                background-color: #4a4a4a;
            }
            
            QScrollBar::add-line:vertical,
            QScrollBar::sub-line:vertical {
                height: 0;
            }
            
            QScrollBar::add-page:vertical,
            QScrollBar::sub-page:vertical {
                background: none;
            }
            
            
            /* Horizontal scrollbar */
            
            QScrollBar:horizontal {
                background-color: #0f0f10;
                height: 5px;
                margin: 0;
            }
            
            QScrollBar::handle:horizontal {
                background-color: #3a3a3a;
                min-width: 24px;
                border-radius: 5px;
            }
            
            QScrollBar::handle:horizontal:hover {
                background-color: #4a4a4a;
            }
            
            QScrollBar::add-line:horizontal,
            QScrollBar::sub-line:horizontal {
                width: 0;
            }
            
            QScrollBar::add-page:horizontal,
            QScrollBar::sub-page:horizontal {
                background: none;
            }
            """
        )

        self._signals.output.connect(self._append_output)
        self._signals.finished.connect(self._process_finished)
        self._signals.read_failed.connect(self._read_failed)

        self.finished_callback = finished_callback

        self._stdout_reader = threading.Thread(
            target=self._read_stdout,
            name=f"process-stdout-{process.pid}",
            daemon=True,
        )
        self._stderr_reader = threading.Thread(
            target=self._read_stderr,
            name=f"process-stderr-{process.pid}",
            daemon=True,
        )
        self._waiter = threading.Thread(
            target=self._wait_for_process,
            name=f"process-wait-{process.pid}",
            daemon=True,
        )
        self._stdout_reader.start()
        self._stderr_reader.start()
        self._waiter.start()

    def append_message(self, message: str) -> None:
        self._append_output("launcher", message)

    def _append_output(self, line_type: str, message: str) -> None:
        """Store and display one line, trimming old output in bounded chunks."""
        if len(self.lines) >= self.MAX_LOG_LINES:
            trim_count = min(self.LOG_TRIM_LINES, len(self.lines))

            # Replace the previous truncation marker; it is not process output.
            if self.lines and self.lines[0][0] == "truncated":
                del self.lines[0]
                trim_count = min(trim_count - 1, len(self.lines))

            del self.lines[:trim_count]
            self._trimmed_line_count += trim_count

            trimmed_at = datetime.now().astimezone().isoformat(timespec="seconds")
            marker = "[launcher] " + self.app.tr_text(
                "Log truncated at {time}; {count} older lines removed.",
                time=trimmed_at,
                count=self._trimmed_line_count,
            )
            self.lines.insert(0, ("truncated", marker))

            # Keep the widget and the saved-log buffer identical after trimming.
            self.log_output.setPlainText(
                "\n".join(stored_message for _, stored_message in self.lines)
            )

        self.lines.append((line_type, message))
        self.log_output.appendPlainText(message)

    @property
    def last_error_message(self) -> str | None:
        return self._last_error_message

    def _read_stdout(self) -> None:
        try:
            if self.process.stdout is not None:
                for line in self.process.stdout:
                    self._signals.output.emit("normal", line.rstrip("\r\n"))
        except Exception as exc:
            self._signals.read_failed.emit(str(exc))

    def _read_stderr(self) -> None:
        try:
            if self.process.stderr is not None:
                for line in self.process.stderr:
                    message = line.rstrip("\r\n")
                    if message.strip():
                        self._last_error_message = message.strip()
                        self._stderr_lines.append(message)
                    self._signals.output.emit("error", message)
        except Exception as exc:
            self._signals.read_failed.emit(str(exc))

    def _wait_for_process(self) -> None:
        exit_code = self._normalize_exit_code(self.process.wait())
        self._stdout_reader.join()
        self._stderr_reader.join()
        self._signals.finished.emit(exit_code)

    @staticmethod
    def _normalize_exit_code(exit_code: int) -> int:
        # Avoid some system (WINDOWS) give exit code that overflow in 32bit integer
        if exit_code > 0x7FFFFFFF:
            return exit_code - 0x100000000
        return exit_code

    def kill_process(self) -> None:
        if self.process.poll() is not None:
            self._process_finished(
                self._normalize_exit_code(self.process.returncode or 0)
            )
            return

        self.kill_button.setEnabled(False)
        self.status_label.setText(self.app.tr_text("Killing process..."))

        try:
            self.process.kill()
        except Exception as exc:
            self.kill_button.setEnabled(True)
            self.status_label.setText(self.app.tr_text("Unable to kill process"))
            self.append_message(f"[launcher] Unable to kill process: {exc}")

    def save_log(self) -> None:
        file_name, _ = QFileDialog.getSaveFileName(
            None,
            self.app.tr_text("Select where to save the log file"),
            "game-output.txt",
            self.app.tr_text("Text Files (*.txt);;All Files (*)"),
        )

        # cancel
        if not file_name:
            return

        if not self.lines:
            messagebox.warning(
                self,
                self.app.tr_text("Warning"),
                self.app.tr_text("Log output is empty. No need to save it."),
            )
            return

        try:
            with open(file_name, "w", encoding="utf-8", newline="\n") as log_file:
                for _, message in self.lines:
                    log_file.write(message + "\n")

            messagebox.info(
                self,
                self.app.tr_text("Log Viewer"),
                self.app.tr_text("Log output saved to {file_name}", file_name=file_name),
            )
        except Exception as exc:
            messagebox.error(
                self,
                self.app.tr_text("Error"),
                self.app.tr_text("Unable to save log file: {error}", error=exc),
            )

    def _process_finished(self, exit_code: int) -> None:
        self.kill_button.setEnabled(False)
        self.status_label.setText(self.app.tr_text(
            "Process exited with code {exit_code}", exit_code=exit_code
        ))
        self.append_message(f"[launcher] Process exited with code {exit_code}.")

        if exit_code != 0 and self._last_error_message:
            self.append_message(
                f"[launcher] Last error: {self._last_error_message}"
            )

        if exit_code != 0:
            error_content = "\n".join(self._stderr_lines)
            if not error_content:
                error_content = (
                    "The client did not write an error message to stderr.\n"
                    f"Process exit code: {exit_code}"
                )

            messagebox.error_with_plain_text(
                self,
                title=self.app.tr_text("Minecraft Client Error"),
                message=self.app.tr_text(
                    "Minecraft exited with code {exit_code}.", exit_code=exit_code
                ),
                content=error_content,
            )

        if callable(self.finished_callback):
            self.finished_callback(self.profile_id, exit_code)

    def _read_failed(self, message: str) -> None:
        self.append_message(f"[launcher] Unable to read process output: {message}")
        if self.process.poll() is not None:
            self._process_finished(
                self._normalize_exit_code(self.process.returncode or 0)
            )

    def closeEvent(self, event: QCloseEvent) -> None:
        # Closing the log window must not stop the game. The process remains
        # available through this window instance and can be shown again.
        event.accept()