kam-mobile/tools/resource_staging.py

"""Load the permanent runtime-resource selection."""
from __future__ import annotations

import json
import os
import shutil
import urllib.parse
from pathlib import Path

from common import ROOT, ProbeError, safe_relative, sha256, write_json


RUNTIME_RESOURCE_ROOT = ROOT / 'private-assets'
RUNTIME_RESOURCE_MANIFEST = ROOT / 'config/runtime-resources.json'
_SCHEMA = 'kam-runtime-resources/1'
_KEYS = {'schema', 'entries'}
_ENTRY_KEYS = {'kind', 'path'}


def _load_runtime_resource_manifest() -> tuple[dict[str, str], ...]:
    """Return the validated authored resource selection, without resolving it."""
    try:
        value = json.loads(RUNTIME_RESOURCE_MANIFEST.read_text(encoding='utf-8'))
    except (OSError, UnicodeError, json.JSONDecodeError) as error:
        raise ProbeError(f'Cannot read runtime-resource manifest: {error}') from error
    if not isinstance(value, dict) or set(value) != _KEYS:
        raise ProbeError('Invalid runtime-resource manifest fields')
    if value['schema'] != _SCHEMA:
        raise ProbeError('Invalid runtime-resource manifest schema')
    if not isinstance(value['entries'], list):
        raise ProbeError('Runtime-resource entries must be an array')

    seen_paths: set[str] = set()
    entries = []
    for entry in value['entries']:
        if not isinstance(entry, dict) or set(entry) != _ENTRY_KEYS:
            raise ProbeError('Invalid runtime-resource entry fields')
        kind, path = entry['kind'], entry['path']
        if not isinstance(kind, str) or kind not in {'file', 'directory'} or not isinstance(path, str) or not path:
            raise ProbeError('Invalid runtime-resource entry')
        safe_relative(path)
        if path in seen_paths:
            raise ProbeError(f'Duplicate runtime-resource path: {path!r}')
        seen_paths.add(path)
        entries.append({'kind': kind, 'path': path})
    return tuple(entries)


def _resolve_runtime_resource_files(entries: tuple[dict[str, str], ...]) -> tuple[str, ...]:
    """Expand parsed resource entries into exact, safe file identities."""
    root = RUNTIME_RESOURCE_ROOT
    if root.is_symlink() or not root.is_dir():
        raise ProbeError(f'Runtime-resource root is not a real directory: {root}')
    try:
        fixed_root = root.resolve(strict=True)
    except OSError as error:
        raise ProbeError(f'Cannot resolve runtime-resource root: {error}') from error

    def checked(path):
        if path.is_symlink():
            raise ProbeError(f'Runtime-resource symlink is not allowed: {path}')
        try:
            resolved = path.resolve(strict=True)
        except OSError as error:
            raise ProbeError(f'Cannot resolve runtime resource: {path}') from error
        if not resolved.is_relative_to(fixed_root):
            raise ProbeError(f'Runtime resource escapes fixed root: {path}')
        return resolved

    def exact(path: str):
        current = root
        for component in path.split('/'):
            if not current.is_dir():
                raise ProbeError(f'Runtime-resource path component is not a directory: {current}')
            try:
                with os.scandir(current) as scanned:
                    child = next((Path(entry.path) for entry in scanned
                                  if entry.name == component), None)
            except OSError as error:
                raise ProbeError(f'Cannot inspect runtime-resource path: {current}') from error
            if child is None:
                raise ProbeError(f'Runtime-resource path not found: {path}')
            checked(child)
            current = child
        return current

    selected: set[str] = set()
    for entry in entries:
        kind, authored = entry['kind'], entry['path']
        target = exact(authored)
        if kind == 'file':
            if target.name == '.gitkeep' or not target.is_file() or target.is_symlink():
                raise ProbeError(f'Runtime-resource file is not a regular selectable file: {authored}')
            selected.add(authored)
            continue
        if not target.is_dir() or target.is_symlink():
            raise ProbeError(f'Runtime-resource directory is not a real directory: {authored}')
        produced = False
        pending = [target]
        while pending:
            directory = pending.pop()
            try:
                children = list(os.scandir(directory))
            except OSError as error:
                raise ProbeError(f'Cannot inspect runtime-resource directory: {directory}') from error
            for child_entry in children:
                child = Path(child_entry.path)
                checked(child)
                if child_entry.is_symlink():
                    raise ProbeError(f'Runtime-resource symlink is not allowed: {child}')
                if child_entry.is_dir(follow_symlinks=False):
                    pending.append(child)
                elif child_entry.is_file(follow_symlinks=False):
                    if child.name != '.gitkeep':
                        produced = True
                        selected.add(child.relative_to(root).as_posix())
                else:
                    raise ProbeError(f'Runtime-resource node is not a regular file or directory: {child}')
        if not produced:
            raise ProbeError(f'Runtime-resource directory is empty: {authored}')

    folded: dict[str, str] = {}
    for path in selected:
        prior = folded.setdefault(path.casefold(), path)
        if prior != path:
            raise ProbeError(f'Case-colliding runtime-resource paths: {prior!r}, {path!r}')
    return tuple(sorted(selected, key=lambda path: (path.casefold(), path)))


def stage_runtime_resources(castle_data_root: Path) -> tuple[dict[str, object], ...]:
    """Stage the selected runtime resources into a Castle data directory."""
    paths = _resolve_runtime_resource_files(_load_runtime_resource_manifest())
    records = []
    for path in paths:
        source = RUNTIME_RESOURCE_ROOT / path
        if source.is_symlink() or not source.is_file():
            raise ProbeError(f'Runtime-resource file disappeared or changed type: {path}')
        try:
            digest = sha256(source)
            records.append({'path': path, 'size': source.stat().st_size, 'sha256': digest,
                            'url': 'castle-data:/game/' + urllib.parse.quote(path, safe='/')})
        except OSError as error:
            raise ProbeError(f'Cannot inventory runtime resource: {path}: {error}') from error

    game = castle_data_root / 'game'
    manifest = castle_data_root / 'assets-manifest.json'
    try:
        if game.is_symlink() or game.is_file():
            game.unlink()
        elif game.exists():
            shutil.rmtree(game)
        if manifest.exists() or manifest.is_symlink():
            manifest.unlink()
        for record in records:
            path = record['path']
            source = RUNTIME_RESOURCE_ROOT / path
            if source.is_symlink() or not source.is_file():
                raise ProbeError(f'Runtime-resource file disappeared or changed type: {path}')
            destination = game / path
            destination.parent.mkdir(parents=True, exist_ok=True)
            shutil.copyfile(source, destination)
            if sha256(destination) != record['sha256']:
                raise ProbeError(f'Runtime-resource copy hash mismatch: {path}')
        write_json(manifest, {'schema': 'kam-runtime-assets/1', 'files': records})
    except ProbeError:
        shutil.rmtree(game, ignore_errors=True)
        try:
            manifest.unlink(missing_ok=True)
        except OSError:
            pass
        raise
    except OSError as error:
        shutil.rmtree(game, ignore_errors=True)
        try:
            manifest.unlink(missing_ok=True)
        except OSError:
            pass
        raise ProbeError(f'Cannot stage runtime resources: {error}') from error
    return tuple(records)