pristontale-eu-linux/ptdiag/logs.py

"""Native game-log discovery, freshness validation, and parsing."""

from __future__ import annotations

from dataclasses import asdict, dataclass, field
from pathlib import Path
import re
from typing import Iterable, Mapping


DEFAULT_NATIVE_LOG_PATTERNS = ("log_*.txt", "*.log")

_SCENE_RE = re.compile(
	r"\[S1\]\s+after import\s+(?P<scene>[^:]+):\s*"
	r"materials=(?P<materials>\d+)\s+textures=(?P<textures>\d+)",
	re.IGNORECASE,
)
_FAILED_FILE_RE = re.compile(
	r"Failed to (?P<operation>find|load|read) file\s*:\s*(?P<path>.*)$",
	re.IGNORECASE,
)
_FAILED_IMPORT_RE = re.compile(
	r"Failed to import scene:\s*(?P<scene>.*)$",
	re.IGNORECASE,
)
_MODULE_CODE_SYNC_RE = re.compile(
	r"Module Code Sync Error\s*\(\[(?P<left>-?\d+)\]\s*!=\s*"
	r"\[(?P<right>-?\d+)\]\)",
	re.IGNORECASE,
)
_TIMESTAMP_RE = re.compile(
	r"^\s*(?:\[)?(?P<timestamp>"
	r"\d{4}-\d{2}-\d{2}[T ][0-9:.+Z-]+|\d{2}:\d{2}:\d{2}(?:\.\d+)?)"
	r"(?:\])?"
)


@dataclass(frozen=True)
class LogFileState:
	path: Path
	mtime_ns: int
	size: int
	inode: int


@dataclass(frozen=True)
class LogEvent:
	kind: str
	line_number: int
	raw: str
	timestamp_text: str | None = None
	value: str | None = None

	def to_dict(self) -> dict[str, str | int | None]:
		return asdict(self)


@dataclass(frozen=True)
class SceneImport:
	name: str
	materials: int
	textures: int
	line_number: int
	timestamp_text: str | None = None

	def to_dict(self) -> dict[str, str | int | None]:
		return asdict(self)


@dataclass
class NativeLogAnalysis:
	scenes: dict[str, list[SceneImport]] = field(default_factory=dict)
	events: list[LogEvent] = field(default_factory=list)
	failed_files: list[LogEvent] = field(default_factory=list)
	failed_import_scenes: list[str] = field(default_factory=list)
	load_texture_to_material_failed: bool = False
	import_scene_threw: bool = False
	stage_without_new_stage: bool = False
	tex_reclaim_count: int = 0
	module_code_sync_error: bool = False
	module_code_sync_left: int | None = None
	module_code_sync_right: int | None = None
	module_code_sync_delta: int | None = None
	banned_code_1: bool = False

	def latest_scene(self, name: str) -> SceneImport | None:
		wanted = _scene_key(name)
		for scene_name, imports in self.scenes.items():
			if _scene_key(scene_name) == wanted and imports:
				return imports[-1]
		return None

	def first_event(self, kind: str) -> LogEvent | None:
		return next((event for event in self.events if event.kind == kind), None)

	def to_summary(self) -> dict[str, object]:
		summary: dict[str, object] = {
			"scenes": {
				name: [item.to_dict() for item in imports]
				for name, imports in sorted(self.scenes.items())
			},
			"load_texture_to_material_failed": self.load_texture_to_material_failed,
			"import_scene_threw": self.import_scene_threw,
			"failed_import_scenes": self.failed_import_scenes,
			"failed_files": [event.to_dict() for event in self.failed_files],
			"stage_without_new_stage": self.stage_without_new_stage,
			"tex_reclaim_count": self.tex_reclaim_count,
			"module_code_sync_error": self.module_code_sync_error,
			"module_code_sync_left": self.module_code_sync_left,
			"module_code_sync_right": self.module_code_sync_right,
			"module_code_sync_delta": self.module_code_sync_delta,
			"banned_code_1": self.banned_code_1,
		}
		for output_key, scene_name in (
			("whispering_vale", "WhisperingVale"),
			("ricarten", "Ricarten"),
		):
			scene = self.latest_scene(scene_name)
			if scene is not None:
				summary[output_key] = {
					"materials": scene.materials,
					"textures": scene.textures,
				}
		if "whispering_vale" in summary:
			whispering_vale = summary["whispering_vale"]
			assert isinstance(whispering_vale, dict)
			whispering_vale["load_texture_to_material_failed"] = (
				self.load_texture_to_material_failed
			)
			whispering_vale["import_scene_failed"] = self.import_scene_threw or any(
				"whisperingvale" in _scene_key(item)
				for item in self.failed_import_scenes
			)
		return summary


def snapshot_native_logs(
	log_directory: Path,
	patterns: Iterable[str] = DEFAULT_NATIVE_LOG_PATTERNS,
) -> dict[str, LogFileState]:
	states: dict[str, LogFileState] = {}
	if not log_directory.is_dir():
		return states
	for pattern in patterns:
		for path in log_directory.glob(pattern):
			if not path.is_file():
				continue
			try:
				state = _state(path)
			except FileNotFoundError:
				continue
			states[_path_key(path)] = state
	return states


def is_current_run_log(
	state: LogFileState,
	baseline: Mapping[str, LogFileState],
	test_start_ns: int,
) -> bool:
	"""Return true only for a new log or a changed post-start existing log.

	A path absent from the pre-launch baseline is new for this run. An existing
	path must both change and have an mtime at or after the recorded test start.
	The additional change check prevents a stale future-dated file from passing.
	"""
	previous = baseline.get(_path_key(state.path))
	if previous is None:
		return True
	changed = (
		state.mtime_ns != previous.mtime_ns
		or state.size != previous.size
		or state.inode != previous.inode
	)
	return changed and state.mtime_ns >= test_start_ns


def discover_current_native_logs(
	log_directory: Path,
	baseline: Mapping[str, LogFileState],
	test_start_ns: int,
	patterns: Iterable[str] = DEFAULT_NATIVE_LOG_PATTERNS,
) -> list[LogFileState]:
	current = snapshot_native_logs(log_directory, patterns)
	eligible = [
		state
		for state in current.values()
		if is_current_run_log(state, baseline, test_start_ns)
	]
	return sorted(eligible, key=lambda item: (item.mtime_ns, str(item.path)))


def newest_current_native_log(
	log_directory: Path,
	baseline: Mapping[str, LogFileState],
	test_start_ns: int,
	patterns: Iterable[str] = DEFAULT_NATIVE_LOG_PATTERNS,
) -> LogFileState | None:
	logs = discover_current_native_logs(log_directory, baseline, test_start_ns, patterns)
	return logs[-1] if logs else None


def parse_native_log(path: Path) -> NativeLogAnalysis:
	with path.open("r", encoding="utf-8", errors="replace") as handle:
		return parse_native_log_lines(handle)


def read_current_run_log_text(
	state: LogFileState,
	baseline: Mapping[str, LogFileState],
) -> str:
	"""Read only bytes attributable to this run when a log path is reused."""
	previous = baseline.get(_path_key(state.path))
	offset = 0
	if (
		previous is not None
		and state.inode == previous.inode
		and state.size >= previous.size
	):
		offset = previous.size
	with state.path.open("rb") as handle:
		handle.seek(offset)
		return handle.read().decode("utf-8", errors="replace")


def parse_current_run_log(
	state: LogFileState,
	baseline: Mapping[str, LogFileState],
) -> NativeLogAnalysis:
	return parse_native_log_lines(read_current_run_log_text(state, baseline).splitlines())


def parse_native_log_lines(lines: Iterable[str]) -> NativeLogAnalysis:
	analysis = NativeLogAnalysis()
	for line_number, raw_line in enumerate(lines, start=1):
		raw = raw_line.rstrip("\r\n")
		timestamp = _timestamp(raw)
		scene_match = _SCENE_RE.search(raw)
		if scene_match is not None:
			scene = SceneImport(
				name=scene_match.group("scene").strip(),
				materials=int(scene_match.group("materials")),
				textures=int(scene_match.group("textures")),
				line_number=line_number,
				timestamp_text=timestamp,
			)
			analysis.scenes.setdefault(scene.name, []).append(scene)
			analysis.events.append(
				LogEvent("scene_import", line_number, raw, timestamp, scene.name)
			)

		if "GRX::SurfaceManager::LoadTextureToMaterial" in raw:
			analysis.load_texture_to_material_failed = True
			analysis.events.append(
				LogEvent("load_texture_to_material", line_number, raw, timestamp)
			)
		if "ImportScene threw" in raw:
			analysis.import_scene_threw = True
			analysis.events.append(LogEvent("import_scene_threw", line_number, raw, timestamp))

		module_sync = _MODULE_CODE_SYNC_RE.search(raw)
		if module_sync is not None:
			left = int(module_sync.group("left"))
			right = int(module_sync.group("right"))
			analysis.module_code_sync_error = True
			analysis.module_code_sync_left = left
			analysis.module_code_sync_right = right
			analysis.module_code_sync_delta = abs(left - right)
			analysis.events.append(
				LogEvent(
					"module_code_sync_error",
					line_number,
					raw,
					timestamp,
					f"{left} != {right}",
				)
			)
		if "banned code 1" in raw.casefold():
			analysis.banned_code_1 = True
			analysis.events.append(LogEvent("banned_code_1", line_number, raw, timestamp))

		failed_import = _FAILED_IMPORT_RE.search(raw)
		if failed_import is not None:
			value = failed_import.group("scene").strip()
			analysis.failed_import_scenes.append(value)
			analysis.events.append(
				LogEvent("failed_import_scene", line_number, raw, timestamp, value)
			)

		failed_file = _FAILED_FILE_RE.search(raw)
		if failed_file is not None:
			operation = failed_file.group("operation").casefold()
			value = failed_file.group("path").strip()
			event = LogEvent(f"failed_{operation}_file", line_number, raw, timestamp, value)
			analysis.failed_files.append(event)
			analysis.events.append(event)

		if "Stage without new stage" in raw:
			analysis.stage_without_new_stage = True
			analysis.events.append(
				LogEvent("stage_without_new_stage", line_number, raw, timestamp)
			)
		if "TexReclaim" in raw:
			analysis.tex_reclaim_count += 1
			analysis.events.append(LogEvent("tex_reclaim", line_number, raw, timestamp))
	return analysis


def _state(path: Path) -> LogFileState:
	stat = path.stat()
	return LogFileState(path.absolute(), stat.st_mtime_ns, stat.st_size, stat.st_ino)


def _path_key(path: Path) -> str:
	return str(path.absolute())


def _scene_key(value: str) -> str:
	return re.sub(r"[^a-z0-9]", "", value.casefold())


def _timestamp(line: str) -> str | None:
	match = _TIMESTAMP_RE.match(line)
	return match.group("timestamp") if match is not None else None