pristontale-eu-linux/scripts/generate_history.py

#!/usr/bin/env python3
"""Generate signed-history source data from Gitea release API results."""

from __future__ import annotations

import argparse
import json
import os
from pathlib import Path
import sys
from typing import Mapping
from urllib.parse import urlsplit

PROJECT_ROOT = Path(__file__).resolve().parents[1]
if str(PROJECT_ROOT) not in sys.path:
	sys.path.insert(0, str(PROJECT_ROOT))

from scripts.release_lib import (  # noqa: E402
	ReleaseToolError,
	_parse_timestamp,
	load_config,
)
from ptlinux.update_resolver import (
	SUPPORTED_RELEASE_STATUSES,
	ReleaseModelError,
	SemanticVersion,
)


def release_history(
	releases: object,
	generated_at: str,
	statuses: Mapping[str, object] | None = None,
) -> dict[str, object]:
	_parse_timestamp(generated_at)
	if not isinstance(releases, list):
		raise ReleaseToolError("Gitea releases response must be an array")
	records: list[tuple[SemanticVersion, dict[str, object]]] = []
	seen_versions: set[SemanticVersion] = set()
	included_tags: set[str] = set()
	for item in releases:
		if not isinstance(item, dict) or item.get("draft") is True:
			continue
		tag = item.get("tag_name")
		if not isinstance(tag, str):
			continue
		try:
			version = SemanticVersion.from_tag(tag)
		except ReleaseModelError:
			continue
		if version in seen_versions:
			raise ReleaseToolError(f"Gitea returned duplicate release tag {tag}")
		seen_versions.add(version)
		if bool(item.get("prerelease")) != version.is_prerelease:
			continue
		assets = item.get("assets")
		if not isinstance(assets, list):
			continue
		asset_urls: dict[str, str] = {}
		for asset in assets:
			if not isinstance(asset, dict):
				continue
			name = asset.get("name")
			url = asset.get("browser_download_url")
			if isinstance(name, str) and isinstance(url, str):
				asset_urls[name] = url
		manifest_url = asset_urls.get("manifest.json")
		signature_url = asset_urls.get("manifest.json.sig")
		release_date = item.get("published_at") or item.get("created_at")
		release_notes_url = item.get("html_url")
		if (
			manifest_url is None
			or signature_url is None
			or not isinstance(release_date, str)
			or not isinstance(release_notes_url, str)
			or not all(
				_https_url(url)
				for url in (manifest_url, signature_url, release_notes_url)
			)
		):
			continue
		try:
			_parse_timestamp(release_date)
		except ReleaseToolError:
			continue
		release_id = item.get("id")
		override = (statuses or {}).get(tag, "active")
		if isinstance(override, dict):
			status = override.get("status", "active")
			config_schema = _positive_integer(
				override.get("config_schema", 1),
				f"{tag} config_schema",
			)
			minimum_config_schema = _positive_integer(
				override.get("minimum_config_schema", 1),
				f"{tag} minimum_config_schema",
			)
		else:
			status = override
			config_schema = 1
			minimum_config_schema = 1
		if status not in SUPPORTED_RELEASE_STATUSES:
			raise ReleaseToolError(f"invalid status for {tag}: {status!r}")
		if minimum_config_schema > config_schema:
			raise ReleaseToolError(
				f"minimum_config_schema exceeds config_schema for {tag}"
			)
		title = item.get("name")
		if not isinstance(title, str) or not title:
			title = f"PristonTale EU Launcher {version}"
		records.append(
			(
				version,
				{
					"channel": version.channel,
					"compatibility": {
						"config_schema": config_schema,
						"minimum_config_schema": minimum_config_schema,
					},
					"manifest_signature_url": signature_url,
					"manifest_url": manifest_url,
					"release_date": release_date,
					"release_id": (
						release_id
						if isinstance(release_id, int)
						and not isinstance(release_id, bool)
						and release_id > 0
						else None
					),
					"release_notes_url": release_notes_url,
					"status": status,
					"tag": tag,
					"title": title,
					"version": str(version),
				},
			)
		)
		included_tags.add(tag)
	unused_statuses = set(statuses or {}).difference(included_tags)
	if unused_statuses:
		raise ReleaseToolError(
			"release status overrides do not match published, manifest-backed tags: "
			+ ", ".join(sorted(unused_statuses))
		)
	records.sort(key=lambda entry: entry[0], reverse=True)
	config = load_config()
	return {
		"application": config["application"],
		"generated_at": generated_at,
		"releases": [record for _version, record in records],
		"schema": 1,
	}


def load_release_statuses(path: Path) -> dict[str, object]:
	try:
		value = json.loads(path.read_text(encoding="utf-8"))
	except (OSError, UnicodeError, json.JSONDecodeError) as exc:
		raise ReleaseToolError(f"invalid release status file {path}: {exc}") from exc
	if not isinstance(value, dict) or value.get("schema") != 1:
		raise ReleaseToolError("release status file must use schema 1")
	releases = value.get("releases")
	if not isinstance(releases, dict):
		raise ReleaseToolError("release status file releases must be an object")
	statuses: dict[str, object] = {}
	for tag, status_value in releases.items():
		if not isinstance(tag, str):
			raise ReleaseToolError("release status tags must be strings")
		try:
			SemanticVersion.from_tag(tag)
		except ReleaseModelError as exc:
			raise ReleaseToolError(f"invalid release status tag {tag!r}") from exc
		status = status_value.get("status", "active") if isinstance(status_value, dict) else status_value
		if not isinstance(status, str) or status not in SUPPORTED_RELEASE_STATUSES:
			raise ReleaseToolError(f"invalid status for {tag}: {status!r}")
		if isinstance(status_value, dict):
			unknown = set(status_value).difference(
				{"status", "config_schema", "minimum_config_schema"}
			)
			if unknown:
				raise ReleaseToolError(
					f"unknown release override fields for {tag}: {', '.join(sorted(unknown))}"
				)
			config_schema = _positive_integer(
				status_value.get("config_schema", 1),
				f"{tag} config_schema",
			)
			minimum_config_schema = _positive_integer(
				status_value.get("minimum_config_schema", 1),
				f"{tag} minimum_config_schema",
			)
			if minimum_config_schema > config_schema:
				raise ReleaseToolError(
					f"minimum_config_schema exceeds config_schema for {tag}"
				)
			statuses[tag] = {
				"status": status,
				"config_schema": config_schema,
				"minimum_config_schema": minimum_config_schema,
			}
		else:
			statuses[tag] = status
	return statuses


def _positive_integer(value: object, label: str) -> int:
	if isinstance(value, bool) or not isinstance(value, int) or value <= 0:
		raise ReleaseToolError(f"{label} must be a positive integer")
	return value


def _https_url(value: str) -> bool:
	parts = urlsplit(value)
	return (
		parts.scheme.casefold() == "https"
		and bool(parts.netloc)
		and parts.username is None
		and parts.password is None
	)


def main() -> int:
	parser = argparse.ArgumentParser(description=__doc__)
	parser.add_argument("input", type=Path)
	parser.add_argument("output", type=Path)
	parser.add_argument("--generated-at", required=True)
	parser.add_argument("--status-file", type=Path)
	arguments = parser.parse_args()
	try:
		releases = json.loads(arguments.input.read_text(encoding="utf-8"))
		config = load_config()
		status_file = arguments.status_file or (
			PROJECT_ROOT / str(config["release_status_file"])
		)
		history = release_history(
			releases,
			arguments.generated_at,
			load_release_statuses(status_file),
		)
		arguments.output.parent.mkdir(parents=True, exist_ok=True)
		temporary = arguments.output.with_name(
			f".{arguments.output.name}.{os.getpid()}.tmp"
		)
		try:
			with temporary.open("x", encoding="utf-8", newline="\n") as handle:
				json.dump(history, handle, indent=2, sort_keys=True)
				handle.write("\n")
				handle.flush()
				os.fsync(handle.fileno())
			temporary.chmod(0o644)
			os.replace(temporary, arguments.output)
		finally:
			temporary.unlink(missing_ok=True)
	except (OSError, UnicodeError, json.JSONDecodeError, ReleaseToolError) as exc:
		print(f"Release history generation failed: {exc}", file=sys.stderr)
		return 1
	return 0


if __name__ == "__main__":
	raise SystemExit(main())