From b16540099d80dc4ffcb8f81b8ff2d91328ecb15a Mon Sep 17 00:00:00 2001 From: m4ko Date: Mon, 31 Aug 2026 23:11:21 +0200 Subject: [PATCH] Added changelog class --- ptlinux/changelog.py | 332 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 332 insertions(+) create mode 100644 ptlinux/changelog.py diff --git a/ptlinux/changelog.py b/ptlinux/changelog.py new file mode 100644 index 0000000..0058bdd --- /dev/null +++ b/ptlinux/changelog.py @@ -0,0 +1,332 @@ +"""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)