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")
_ASSET_MISSING_FILE_THRESHOLD = 200

_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_file_find_count: int = 0
	failed_file_load_count: int = 0
	failed_file_read_count: int = 0
	failed_asset_file_count: int = 0
	failed_texture_file_count: int = 0
	failed_effect_file_count: int = 0
	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

	@property
	def asset_pack_health_failed(self) -> bool:
		return (
			self.failed_asset_file_count >= _ASSET_MISSING_FILE_THRESHOLD
			or self.failed_texture_file_count >= _ASSET_MISSING_FILE_THRESHOLD // 3
			or self.failed_effect_file_count >= max(25, _ASSET_MISSING_FILE_THRESHOLD // 8)
		)

	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 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_file_find_count": self.failed_file_find_count,
			"failed_file_load_count": self.failed_file_load_count,
			"failed_file_read_count": self.failed_file_read_count,
			"failed_asset_file_count": self.failed_asset_file_count,
			"failed_texture_file_count": self.failed_texture_file_count,
			"failed_effect_file_count": self.failed_effect_file_count,
			"asset_pack_health_failed": self.asset_pack_health_failed,
			"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 operation == "find":
				analysis.failed_file_find_count += 1
			elif operation == "load":
				analysis.failed_file_load_count += 1
			else:
				analysis.failed_file_read_count += 1
			if _is_game_asset_path(value):
				analysis.failed_asset_file_count += 1
				if _is_texture_path(value):
					analysis.failed_texture_file_count += 1
				if _is_effect_path(value):
					analysis.failed_effect_file_count += 1

		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


def _is_game_asset_path(path: str) -> bool:
	lower = path.casefold()
	if not lower:
		return False
	if lower.startswith("save/") or lower.startswith("save\\"):
		return False
	if lower.startswith(("game\\", "game/")):
		return True
	if lower.startswith(("field\\", "field/")):
		return True
	if lower.startswith(("image\\", "image/")):
		return True
	if lower.startswith(("effect\\", "effect/")):
		return True
	if lower.startswith(("startimage\\", "startimage/")):
		return True
	return (
		"\\effect\\" in lower
		or "\\image\\" in lower
		or "\\field\\" in lower
		or "\\weapons\\" in lower
		or "\\char\\" in lower
		or "\\sinimage\\" in lower
		or "\\skill\\" in lower
		or lower.endswith(".bmp")
		or lower.endswith(".tga")
		or lower.endswith(".png")
		or lower.endswith(".jpg")
		or lower.endswith(".jpeg")
		or lower.endswith(".gif")
		or lower.endswith(".ase")
		or lower.endswith(".part")
	)


def _is_texture_path(path: str) -> bool:
	lower = path.casefold()
	return (
		lower.endswith(".bmp")
		or lower.endswith(".tga")
		or lower.endswith(".png")
		or lower.endswith(".jpg")
		or lower.endswith(".jpeg")
		or lower.endswith(".gif")
		or lower.endswith(".dds")
	)


def _is_effect_path(path: str) -> bool:
	lower = path.casefold()
	return (
		"\\effect\\" in lower
		or lower.startswith("effect\\")
		or lower.startswith("effect/")
		or lower.endswith(".part")
	)