pristontale-eu-linux/ptlinux/changelog.py

"""Official patch changelog support for the native Linux dashboard."""

from __future__ import annotations

from dataclasses import asdict, dataclass
from datetime import UTC, datetime
import json
import math
import os
from pathlib import Path
import re
from typing import Callable, Protocol
from urllib.request import Request, urlopen


PATCHES_ENDPOINT = "https://pristontale.eu/api/changelog/patches"
PATCH_NEWS_URL = "https://pristontale.eu/news#"
MAX_PATCHES = 10
MAX_PATCH_BYTES = 2 * 1024 * 1024
MAX_PATCH_SECTIONS = 32
MAX_PATCH_CHANGES = 200
_MAX_PATCH_CANDIDATES = 100
_SLUG_PATTERN = re.compile(r"[a-z0-9]+(?:-[a-z0-9]+)*")


class ChangelogError(RuntimeError):
	"""Raised when the official patch feed cannot be parsed safely."""


class ChangelogResponse(Protocol):
	"""The minimal response interface used by the changelog client."""

	def read(self, size: int = -1) -> bytes: ...

	def __enter__(self) -> ChangelogResponse: ...

	def __exit__(
		self,
		exception_type: type[BaseException] | None,
		exception: BaseException | None,
		traceback: object | None,
	) -> bool | None: ...


ChangelogOpener = Callable[..., ChangelogResponse]


@dataclass(frozen=True)
class PatchChange:
	text: str
	type_key: str

	@property
	def tag(self) -> str:
		return patch_change_tag(self.type_key)


@dataclass(frozen=True)
class PatchSection:
	title: str
	items: tuple[PatchChange, ...]


@dataclass(frozen=True)
class PatchNote:
	id: int
	slug: str
	title: str
	version: str | None
	tags: tuple[str, ...]
	summary: str
	sections: tuple[PatchSection, ...]
	published_at: str | None

	@property
	def url(self) -> str:
		return f"{PATCH_NEWS_URL}{self.slug}"


def patch_change_tag(type_key: str) -> str:
	"""Turn API keys such as ``new-feature`` into compact UI tags."""

	label = " ".join(type_key.replace("_", "-").split("-"))
	return label.upper() if label else "CHANGE"


def fetch_patch_notes(
	cache_path: Path,
	*,
	timeout_seconds: float = 15.0,
	opener: ChangelogOpener = urlopen,
) -> tuple[PatchNote, ...]:
	"""Fetch the ten latest patches, falling back to the last valid cache."""

	request = Request(
		PATCHES_ENDPOINT,
		headers={"User-Agent": "PT-Linux/0.6 (+https://pristontale.eu/)"},
		method="GET",
	)
	try:
		response = opener(request, timeout=timeout_seconds)
		with response:
			payload = response.read(MAX_PATCH_BYTES + 1)
		if len(payload) > MAX_PATCH_BYTES:
			raise ChangelogError("official patch response exceeds the 2 MiB limit")
		patches = _parse_patch_notes(payload)
		_write_patch_cache(cache_path, patches)
		return patches
	except (OSError, ChangelogError, UnicodeDecodeError, json.JSONDecodeError):
		return read_cached_patch_notes(cache_path)


def read_cached_patch_notes(cache_path: Path) -> tuple[PatchNote, ...]:
	"""Read the last valid patch feed without performing network I/O."""

	try:
		value = json.loads(cache_path.read_text(encoding="utf-8"))
	except (OSError, UnicodeError, json.JSONDecodeError):
		return ()
	if not isinstance(value, list):
		return ()
	patches: list[PatchNote] = []
	for item in value[:MAX_PATCHES]:
		patch = _patch_note_from_cache(item)
		if patch is not None:
			patches.append(patch)
	return tuple(patches)


def _parse_patch_notes(payload: bytes) -> tuple[PatchNote, ...]:
	value = json.loads(payload.decode("utf-8-sig"))
	if isinstance(value, dict):
		value = value.get("data")
	if not isinstance(value, list):
		raise ChangelogError("official patch response is not a patch list")
	parsed: list[tuple[float, int, PatchNote]] = []
	for item in value[:_MAX_PATCH_CANDIDATES]:
		result = _parse_patch_note(item)
		if result is not None:
			timestamp, patch = result
			parsed.append((timestamp, patch.id, patch))
	parsed.sort(key=lambda entry: (entry[0], entry[1]), reverse=True)
	patches = tuple(entry[2] for entry in parsed[:MAX_PATCHES])
	if not patches:
		raise ChangelogError("official patch response contains no usable patches")
	return patches


def _parse_patch_note(value: object) -> tuple[float, PatchNote] | None:
	if not isinstance(value, dict):
		return None
	patch_id = value.get("id")
	slug = _clean_text(value.get("slug"), 180)
	title = _clean_text(value.get("title"), 200)
	summary = _clean_text(value.get("summary"), 3000, allow_empty=True)
	if (
		isinstance(patch_id, bool)
		or not isinstance(patch_id, int)
		or slug is None
		or _SLUG_PATTERN.fullmatch(slug) is None
		or title is None
		or summary is None
	):
		return None
	version = _clean_text(value.get("version"), 64, allow_none=True)
	if value.get("version") is not None and version is None:
		return None
	timestamp = _timestamp(value.get("date"))
	patch = PatchNote(
		id=patch_id,
		slug=slug,
		title=title,
		version=version,
		tags=_parse_tags(value.get("tags")),
		summary=summary,
		sections=_parse_sections(value.get("changes")),
		published_at=_format_timestamp(timestamp),
	)
	return (timestamp if timestamp is not None else -1.0, patch)


def _parse_tags(value: object) -> tuple[str, ...]:
	if not isinstance(value, list):
		return ()
	tags: list[str] = []
	seen: set[str] = set()
	for item in value[:16]:
		tag = _clean_text(item, 48)
		if tag is None or tag.casefold() in seen:
			continue
		seen.add(tag.casefold())
		tags.append(tag)
	return tuple(tags)


def _parse_sections(value: object) -> tuple[PatchSection, ...]:
	if not isinstance(value, dict):
		return ()
	sections_value = value.get("sections")
	if not isinstance(sections_value, list):
		return ()
	sections: list[PatchSection] = []
	changes_seen = 0
	for section_value in sections_value[:MAX_PATCH_SECTIONS]:
		if not isinstance(section_value, dict):
			continue
		title = _clean_text(section_value.get("title"), 100) or "Changes"
		items_value = section_value.get("items")
		if not isinstance(items_value, list):
			continue
		items: list[PatchChange] = []
		for item in items_value:
			if changes_seen >= MAX_PATCH_CHANGES:
				break
			if not isinstance(item, dict):
				continue
			text = _clean_text(item.get("text"), 4000)
			type_key = _clean_text(item.get("type_key"), 64)
			if text is None:
				continue
			items.append(
				PatchChange(
					text=text,
					type_key=type_key or _type_key_from_title(title),
				)
			)
			changes_seen += 1
		if items:
			sections.append(PatchSection(title=title, items=tuple(items)))
		if changes_seen >= MAX_PATCH_CHANGES:
			break
	return tuple(sections)


def _patch_note_from_cache(value: object) -> PatchNote | None:
	if not isinstance(value, dict):
		return None
	patch_id = value.get("id")
	slug = _clean_text(value.get("slug"), 180)
	title = _clean_text(value.get("title"), 200)
	summary = _clean_text(value.get("summary"), 3000, allow_empty=True)
	version = _clean_text(value.get("version"), 64, allow_none=True)
	published_at = _clean_text(value.get("published_at"), 64, allow_none=True)
	if (
		isinstance(patch_id, bool)
		or not isinstance(patch_id, int)
		or slug is None
		or _SLUG_PATTERN.fullmatch(slug) is None
		or title is None
		or summary is None
	):
		return None
	sections_value = value.get("sections")
	sections: list[PatchSection] = []
	if isinstance(sections_value, list):
		for section_value in sections_value[:MAX_PATCH_SECTIONS]:
			if not isinstance(section_value, dict):
				continue
			section_title = _clean_text(section_value.get("title"), 100)
			items_value = section_value.get("items")
			if section_title is None or not isinstance(items_value, list):
				continue
			items: list[PatchChange] = []
			for item in items_value:
				if not isinstance(item, dict):
					continue
				text = _clean_text(item.get("text"), 4000)
				type_key = _clean_text(item.get("type_key"), 64)
				if text is not None and type_key is not None:
					items.append(PatchChange(text=text, type_key=type_key))
			if items:
				sections.append(PatchSection(section_title, tuple(items)))
	return PatchNote(
		id=patch_id,
		slug=slug,
		title=title,
		version=version,
		tags=_parse_tags(value.get("tags")),
		summary=summary,
		sections=tuple(sections),
		published_at=published_at,
	)


def _clean_text(
	value: object,
	maximum_length: int,
	*,
	allow_empty: bool = False,
	allow_none: bool = False,
) -> str | None:
	if value is None and allow_none:
		return None
	if not isinstance(value, str):
		return None
	text = " ".join(value.split())
	if len(text) > maximum_length or (not text and not allow_empty):
		return None
	return text


def _timestamp(value: object) -> float | None:
	if isinstance(value, bool) or not isinstance(value, (int, float)):
		return None
	timestamp = float(value)
	return timestamp if math.isfinite(timestamp) else None


def _format_timestamp(value: float | None) -> str | None:
	if value is None:
		return None
	try:
		return datetime.fromtimestamp(value, UTC).astimezone().strftime("%d %b %Y")
	except (OverflowError, OSError, ValueError):
		return None


def _type_key_from_title(title: str) -> str:
	return re.sub(r"[^a-z0-9]+", "-", title.casefold()).strip("-") or "change"


def _write_patch_cache(path: Path, patches: tuple[PatchNote, ...]) -> None:
	path.parent.mkdir(parents=True, exist_ok=True)
	temporary = path.with_name(f".{path.name}.{os.getpid()}.tmp")
	try:
		temporary.write_text(
			json.dumps([asdict(patch) for patch in patches], indent=2) + "\n",
			encoding="utf-8",
		)
		temporary.replace(path)
	except OSError:
		temporary.unlink(missing_ok=True)