pristontale-eu-linux/scripts/prepare_channel_metadata.py

#!/usr/bin/env python3
"""Build one signed channel view from authenticated release history metadata."""

from __future__ import annotations

import argparse
import json
import os
from pathlib import Path
import sys

if __package__:
	from .repository_paths import ensure_project_imports
else:
	from repository_paths import ensure_project_imports

ensure_project_imports()
from ptlinux.self_update import LauncherUpdateError, parse_update_manifest  # noqa: E402
from ptlinux.update_resolver import (  # noqa: E402
	ReleaseModelError,
	ReleaseRecord,
	UpdateResolver,
	validate_channel,
)
from scripts.release_lib import ReleaseToolError  # noqa: E402


def prepare_channel_metadata(
	history: object,
	channel: str,
	manifest_bytes: bytes,
) -> tuple[dict[str, object], dict[str, object]]:
	"""Return the selected history record and a status-aware manifest copy."""

	try:
		validate_channel(channel)
	except ReleaseModelError as exc:
		raise ReleaseToolError(str(exc)) from exc
	if not isinstance(history, dict) or history.get("schema") != 1:
		raise ReleaseToolError("release history must use schema 1")
	releases = history.get("releases")
	if not isinstance(releases, list):
		raise ReleaseToolError("release history releases must be an array")
	records: list[tuple[dict[str, object], ReleaseRecord]] = []
	for value in releases:
		if not isinstance(value, dict) or value.get("channel") != channel:
			continue
		try:
			record = ReleaseRecord(
				version=_string(value, "version"),
				channel=_string(value, "channel"),
				status=_string(value, "status"),
				tag=_string(value, "tag"),
				manifest_url=_string(value, "manifest_url"),
			).validate()
		except ReleaseModelError as exc:
			raise ReleaseToolError(f"invalid release history record: {exc}") from exc
		records.append((value, record))
	if not records:
		raise ReleaseToolError(f"release history has no {channel} releases")
	try:
		resolver = UpdateResolver(record for _value, record in records)
		selected_record = resolver.latest_eligible(channel)
		if selected_record is None:
			selected_record = resolver.eligible_releases(channel, include_inactive=True)[0]
	except ReleaseModelError as exc:
		raise ReleaseToolError(f"invalid release history: {exc}") from exc
	selected = next(
		value for value, record in records if record.version == selected_record.version
	)
	try:
		manifest = parse_update_manifest(manifest_bytes)
	except LauncherUpdateError as exc:
		raise ReleaseToolError(f"selected channel manifest is invalid: {exc}") from exc
	for field, actual, expected in (
		("version", manifest.version, selected_record.version),
		("channel", manifest.channel, selected_record.channel),
		("tag", manifest.tag, selected_record.tag),
	):
		if actual != expected:
			raise ReleaseToolError(
				f"selected manifest {field} {actual!r} does not match history {expected!r}"
			)
	try:
		manifest_value = json.loads(manifest_bytes.decode("utf-8"))
	except (UnicodeError, json.JSONDecodeError) as exc:
		raise ReleaseToolError(f"selected channel manifest is invalid JSON: {exc}") from exc
	if not isinstance(manifest_value, dict):
		raise ReleaseToolError("selected channel manifest must be an object")
	manifest_value["status"] = selected_record.status
	history_compatibility = selected.get("compatibility")
	if not isinstance(history_compatibility, dict):
		raise ReleaseToolError("selected release compatibility must be an object")
	config_schema = _positive_integer(history_compatibility, "config_schema")
	minimum_config_schema = _positive_integer(
		history_compatibility,
		"minimum_config_schema",
	)
	if minimum_config_schema > config_schema:
		raise ReleaseToolError(
			"selected release minimum_config_schema exceeds config_schema"
		)
	manifest_compatibility = manifest_value.get("compatibility")
	if not isinstance(manifest_compatibility, dict):
		raise ReleaseToolError("selected channel manifest compatibility is invalid")
	manifest_compatibility["config_schema"] = config_schema
	manifest_compatibility["minimum_config_schema"] = minimum_config_schema
	return selected, manifest_value


def _string(value: dict[str, object], key: str) -> str:
	item = value.get(key)
	if not isinstance(item, str) or not item:
		raise ReleaseToolError(f"release history field {key!r} must be a string")
	return item


def _positive_integer(value: dict[str, object], key: str) -> int:
	item = value.get(key)
	if isinstance(item, bool) or not isinstance(item, int) or item <= 0:
		raise ReleaseToolError(
			f"selected release compatibility {key!r} must be positive"
		)
	return item


def _write_json(path: Path, value: object) -> None:
	path.parent.mkdir(parents=True, exist_ok=True)
	temporary = path.with_name(f".{path.name}.{os.getpid()}.tmp")
	try:
		with temporary.open("x", encoding="utf-8", newline="\n") as handle:
			json.dump(value, handle, indent=2, sort_keys=True)
			handle.write("\n")
			handle.flush()
			os.fsync(handle.fileno())
		temporary.chmod(0o644)
		os.replace(temporary, path)
	finally:
		temporary.unlink(missing_ok=True)


def main() -> int:
	parser = argparse.ArgumentParser(description=__doc__)
	parser.add_argument("history", type=Path)
	parser.add_argument("channel", choices=("stable", "prerelease"))
	parser.add_argument("manifest", type=Path)
	parser.add_argument("output", type=Path)
	arguments = parser.parse_args()
	try:
		history = json.loads(arguments.history.read_text(encoding="utf-8"))
		selected, manifest = prepare_channel_metadata(
			history,
			arguments.channel,
			arguments.manifest.read_bytes(),
		)
		_write_json(arguments.output, manifest)
		print(_string(selected, "version"))
	except (OSError, UnicodeError, json.JSONDecodeError, ReleaseToolError) as exc:
		print(f"Channel metadata preparation failed: {exc}", file=sys.stderr)
		return 1
	return 0


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