gow3_launcher/scripts/mods.py

#!/usr/bin/env python3
# SPDX-License-Identifier: GPL-2.0-or-later
"""Merge loose-file mods through links, preserving the original game and mod files."""
import argparse
import json
import os
from pathlib import Path
import shutil
import sys
import tempfile

def child(folder, name):
    """`name` inside `folder`, matching an existing entry case-insensitively (mods made on
    Windows often differ from the game's lower-case names)."""
    folder = Path(folder)
    for entry in (folder.iterdir() if folder.is_dir() else ()):
        if entry.name.casefold() == name.casefold():
            return entry
    return folder / name


KNOWN_GOW3_DIRS = {
    'exec', 'dutch', 'english', 'french', 'german',
    'italian', 'polish', 'portuguese', 'russian', 'spanish'
}


def content_root(folder):
    """(root, prefix): where a mod's files are and the game path they go to, or None.

    Accepted layouts: <mod>/dvdroot_ps4, <mod>/app0/dvdroot_ps4, <mod>/EXEC,
    <mod>/Game/EXEC, XML patches, and wrapper folders around them."""
    folder = Path(folder)
    if not folder.is_dir():
        return None
    if folder.name.casefold() in ('gow3-windows', 'gow3-linux') or (folder / 'run.py').is_file():
        return None

    candidates = [folder]
    for _ in range(3):
        next_candidates = []
        for cand in candidates:
            if not cand.is_dir():
                continue
            if child(cand, 'dvdroot_ps4').is_dir():
                return cand, ''
            for d in KNOWN_GOW3_DIRS:
                if child(cand, d).is_dir():
                    return cand, ''
            if list(cand.glob('*.xml')) or child(cand, 'patches').is_dir():
                return cand, ''
            for sub in cand.iterdir():
                if sub.is_dir() and not sub.name.startswith('.'):
                    next_candidates.append(sub)
        candidates = next_candidates
    return None


def discover(root):
    """Named mods: folders of `root` with an accepted layout (content_root)."""
    root = Path(root)
    if not root.is_dir():
        return []
    return sorted((p.name for p in root.iterdir() if p.is_dir() and content_root(p)),
                  key=lambda name: (name.casefold(), name))


def selected(root, config):
    available = discover(root)
    if not config or not Path(config).is_file():
        return available
    settings = json.loads(Path(config).read_text())
    disabled_names = settings.get('disabled', [])
    if not isinstance(disabled_names, list) or not all(isinstance(n, str) for n in disabled_names):
        raise ValueError('Disabled mods must be a list of folder names')
    disabled = set(disabled_names)
    order = settings.get('order', [])
    if not isinstance(order, list) or not all(isinstance(n, str) for n in order):
        raise ValueError('Mod order must be a list of folder names')
    # New folders are enabled automatically and appended in alphabetical order.
    return list(dict.fromkeys(n for n in [*order, *available]
                             if n in available and n not in disabled))


def mod_files(folder):
    """(game path, source) of every file a mod replaces or adds under the game root."""
    layout = content_root(folder)
    if not layout:
        if list(Path(folder).rglob('*.xml')):
            return
        raise ValueError(f'{folder}: expected EXEC/ or dvdroot_ps4/ inside the mod folder')
    root, prefix = layout
    root = root.resolve()
    for directory, folders, files in os.walk(root, followlinks=False):
        directory = Path(directory)
        for name in [*folders, *files]:
            if (directory / name).is_symlink():
                raise ValueError(f'Mod symlinks are unsupported: {directory / name}')
        for name in sorted(files):
            source = directory / name
            relative = Path(prefix) / source.relative_to(root) if prefix else source.relative_to(root)
            # Skip documentation / metadata at the mod root
            if len(relative.parts) == 1:
                if source.suffix.casefold() in ('.txt', '.md', '.json', '.jpg', '.png', '.ini'):
                    continue
            # Strip packaging wrapper prefixes (app0, dvdroot_ps4, game) so files align with the game root
            while len(relative.parts) > 1 and relative.parts[0].casefold() in ('app0', 'dvdroot_ps4', 'game'):
                relative = Path(*relative.parts[1:])

            # Skip patch XML files in mod folder (handled by patches.py)
            if source.suffix.casefold() == '.xml':
                continue

            if relative.parts[0].casefold() in ('eboot.bin', 'sce_module', 'sce_sys'):
                if relative.parts[0].casefold() in ('eboot.bin', 'sce_module'):
                    raise ValueError(f'Executable/system replacement is unsupported: {source}')
                continue
            if not source.is_file():
                continue
            yield relative, source


def conflicts(root, names):
    """Files more than one of `names` (in load order) replaces: [(game path, [mods])], the last
    mod of each list wins. Paths match case-insensitively, as in build_overlay; unreadable mods
    are skipped."""
    owners = {}
    for name in names:
        try:
            for relative, _source in mod_files(Path(root) / name):
                key = relative.as_posix().casefold()
                owners.setdefault(key, (relative.as_posix(), []))[1].append(name)
        except (OSError, ValueError):
            continue
    return sorted((path, mods) for path, mods in owners.values() if len(mods) > 1)


def apply_preset(preset, available):
    """(order, disabled) of a saved preset for the mods present now: missing ones are dropped,
    new ones go last and enabled."""
    order = preset.get('order') if isinstance(preset.get('order'), list) else []
    disabled = preset.get('disabled') if isinstance(preset.get('disabled'), list) else []
    order = list(dict.fromkeys(n for n in [*order, *available] if n in available))
    return order, [n for n in disabled if n in available]


# Windows without the symlink privilege (Developer Mode off): directories become junctions,
# files hard links (copies across volumes). Those are not symlinks, so the overlay remembers
# what it linked and where to.
LINKED = {}


def make_link(link, target):
    if link.is_symlink() or link.is_file():
        try:
            link.unlink()
        except OSError:
            pass
    elif link.is_dir():
        try:
            remove_link(link)
        except OSError:
            pass
    try:
        link.symlink_to(target, target_is_directory=target.is_dir())
        return
    except OSError:
        if os.name != 'nt':
            raise
    if target.is_dir():
        import _winapi
        try:
            _winapi.CreateJunction(str(target), str(link))
        except OSError:
            # Fallback if junction creation fails across volumes or permissions
            shutil.copytree(target, link, dirs_exist_ok=True)
    else:
        try:
            os.link(target, link)
        except OSError:
            shutil.copy2(target, link)
    p_str = str(link)
    LINKED[p_str] = target
    LINKED[p_str.casefold()] = target


def symlinks_work(directory):
    probe = Path(tempfile.mkdtemp(prefix='link-test-', dir=directory))
    try:
        (probe / 'link').symlink_to(probe, target_is_directory=True)
        return True
    except OSError:
        return False
    finally:
        shutil.rmtree(probe, ignore_errors=True)


def is_link(path):
    p_str = str(path)
    if path.is_symlink():
        return True
    if p_str in LINKED or p_str.casefold() in LINKED:
        return True
    if hasattr(path, 'is_junction') and path.is_junction():
        return True
    if os.name == 'nt':
        try:
            import stat
            mode = os.lstat(path).st_mode
            if stat.S_ISLNK(mode):
                return True
        except OSError:
            pass
    return False


def link_target(path):
    p_str = str(path)
    return LINKED.get(p_str) or LINKED.get(p_str.casefold()) or path.resolve(strict=True)


def remove_link(path):
    p_str = str(path)
    try:
        if path.is_symlink():
            path.unlink()
        elif hasattr(path, 'is_junction') and path.is_junction():
            os.rmdir(path)
        elif path.is_dir():
            os.rmdir(path)
        else:
            path.unlink()
    except OSError:
        try:
            if path.is_symlink():
                path.unlink()
            elif os.path.isdir(path):
                os.rmdir(path)
            else:
                path.unlink()
        except OSError:
            pass
    LINKED.pop(p_str, None)
    LINKED.pop(p_str.casefold(), None)


def expand(directory):
    """Materialize one directory level; never write through a directory link."""
    if is_link(directory):
        target = link_target(directory)
        if not target.is_dir():
            raise ValueError(f'File/directory conflict at {directory.name}')
        remove_link(directory)
        directory.mkdir(exist_ok=True)
        for entry in target.iterdir():
            make_link(directory / entry.name, entry)
    elif directory.exists() and not directory.is_dir():
        raise ValueError(f'File/directory conflict at {directory.name}')
    else:
        directory.mkdir(exist_ok=True)


def on_rm_error(func, path, exc_info):
    try:
        os.chmod(path, 0o777)
        if os.path.isdir(path):
            os.rmdir(path)
        else:
            os.unlink(path)
    except OSError:
        pass


def build_overlay(game, out, mods):
    game = Path(game).resolve(strict=True)
    replacements = []
    owners = {}
    for name, root in mods:
        count = 0
        for relative, source in mod_files(root):
            key = relative.as_posix().casefold()
            if key in owners:
                print(f'Mods: {relative}: {owners[key]} -> {name}', file=sys.stderr)
            owners[key] = name
            replacements.append((relative, source))
            count += 1
        print(f'Mods: {name}: {count} files', file=sys.stderr)
    if not replacements:
        return game
    out = Path(out).resolve()
    out.mkdir(parents=True, exist_ok=True)
    for old_dir in out.glob('mod-game-*'):
        if old_dir.is_dir():
            shutil.rmtree(old_dir, onerror=on_rm_error)
    if os.name == 'nt' and not symlinks_work(out) and out.drive.casefold() != game.drive.casefold():
        # Hard links only reach files on the same volume: the overlay goes beside the game.
        out = game.parent
    overlay = Path(tempfile.mkdtemp(prefix='mod-game-', dir=out))
    try:
        for entry in game.iterdir():
            make_link(overlay / entry.name, entry)
        replaced = added = 0
        for relative, source in replacements:
            # Each component takes the game's spelling when it exists in another case.
            parent = overlay
            for part in relative.parts[:-1]:
                parent = child(parent, part)
                expand(parent)
            destination = child(parent, relative.parts[-1])
            if is_link(destination):
                replaced += link_target(destination).is_relative_to(game)
                remove_link(destination)
            elif destination.exists():
                raise ValueError(f'File/directory conflict: {relative}')
            else:
                added += 1
            make_link(destination, source)
        print(f'Mods: {replaced} game files replaced, {added} added', file=sys.stderr)
        return overlay
    except BaseException:
        shutil.rmtree(overlay, onerror=on_rm_error)
        raise


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('game', type=Path)
    parser.add_argument('--out', required=True, type=Path)
    parser.add_argument('--mods-dir', required=True, type=Path)
    parser.add_argument('--config', type=Path)
    parser.add_argument('--enabled', choices=('0', '1'), default='1')
    args = parser.parse_args()
    layers = []
    if args.enabled == '1':
        # shadPS4's loose overlay convention, beside the original game.
        legacy = Path(str(args.game.resolve()) + '-mods')
        if legacy.is_dir():
            layers.append((legacy.name, legacy))
        # A directly selected loose overlay is also accepted.
        if child(args.mods_dir, 'dvdroot_ps4').is_dir():
            layers.append((args.mods_dir.name, args.mods_dir))
        else:
            for name in selected(args.mods_dir, args.config):
                layers.append((name, args.mods_dir / name))
    print(build_overlay(args.game, args.out, layers))


if __name__ == '__main__':
    try:
        main()
    except (OSError, ValueError, TypeError) as error:
        print(f'Mods: {error}', file=sys.stderr)
        sys.exit(1)