searchi/tools/media-runtime/provision.py

"""Verify and stage standalone FFmpeg tools. No user-wide installation or GPU use."""
import argparse
import hashlib
import json
import os
from pathlib import Path
import platform
import shutil
import tarfile
import tempfile
import urllib.request
import zipfile

ROOT = Path(__file__).resolve().parents[2]
MAX_MEMBER = 256 * 1024 * 1024


def verified(path, artifact):
    if not path.is_file() or path.stat().st_size != artifact["bytes"]:
        return False
    with path.open("rb") as stream:
        return hashlib.file_digest(stream, "sha256").hexdigest() == artifact["sha256"]


def install(stream, target, expected):
    if not 0 < expected <= MAX_MEMBER:
        raise ValueError("Archive member exceeds size ceiling")
    target.parent.mkdir(parents=True, exist_ok=True)
    temporary = target.with_suffix(target.suffix + ".partial")
    try:
        with temporary.open("wb") as output:
            copied = 0
            while block := stream.read(65536):
                copied += len(block)
                if copied > expected:
                    raise ValueError("Archive member exceeds declared length")
                output.write(block)
            if copied != expected:
                raise ValueError("Archive member is truncated")
        if target.name in {"ffmpeg", "ffprobe"}:
            temporary.chmod(0o755)
        os.replace(temporary, target)
    finally:
        temporary.unlink(missing_ok=True)


def extract(archive, destination, windows):
    required = {"ffmpeg.exe", "ffprobe.exe"} if windows else {"ffmpeg", "ffprobe"}
    found = set()
    def target_for(name):
        # Never extract arbitrary paths: selected basenames map to fixed targets.
        basename = name.replace("\\", "/").split("/")[-1]
        if basename in required:
            if basename in found:
                raise ValueError("Duplicate decoder executable")
            found.add(basename)
            return destination / "bin" / basename
        if basename.lower().startswith(("license", "copying")) or basename.lower() in {"readme.txt", "readme.md"}:
            parts = name.replace("\\", "/").split("/")
            if any(part in {".", "..", ""} or ":" in part for part in parts):
                raise ValueError("Unsafe notice path")
            return destination / "licenses" / Path(*parts)
        return None
    if archive.suffix == ".zip":
        with zipfile.ZipFile(archive) as package:
            for member in package.infolist():
                target = target_for(member.filename)
                if target is not None and not member.is_dir():
                    with package.open(member) as stream:
                        install(stream, target, member.file_size)
    else:
        with tarfile.open(archive, "r:xz") as package:
            for member in package:
                target = target_for(member.name)
                if target is not None:
                    if not member.isfile():
                        raise ValueError("Selected archive member is not a regular file")
                    with package.extractfile(member) as stream:
                        install(stream, target, member.size)
    if found != required:
        raise ValueError("Archive is missing required decoder tools")


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--stage", action="store_true", help="Copy verified decoder tools and notices into Tauri resources")
    args = parser.parse_args()
    manifest = json.loads(Path(__file__).with_name("manifest.json").read_text())
    machine = platform.machine().lower()
    architecture = "aarch64" if machine in {"arm64", "aarch64"} else "x86_64" if machine in {"amd64", "x86_64"} else machine
    system = platform.system().lower()
    key = f"{system}-{architecture}"
    if key not in manifest["platforms"]:
        raise RuntimeError(f"No verified decoder artifact for {key}; macOS packaging remains a release gate")
    artifact = manifest["platforms"][key]
    directory = ROOT / ".searchi/media-runtime"
    directory.mkdir(parents=True, exist_ok=True)
    archive = directory / artifact["archive"]
    if not verified(archive, artifact):
        url = f"https://github.com/{manifest['provider']}/releases/download/{manifest['release']}/{artifact['archive']}"
        with tempfile.NamedTemporaryFile(dir=directory, delete=False) as output:
            temporary = Path(output.name)
            try:
                with urllib.request.urlopen(urllib.request.Request(url, headers={"User-Agent": "searchi-runtime-setup"}), timeout=120) as response:
                    copied = 0
                    while block := response.read(65536):
                        copied += len(block)
                        if copied > artifact["bytes"]:
                            raise ValueError("Decoder download exceeds manifest size")
                        output.write(block)
                output.close()
                if not verified(temporary, artifact):
                    raise ValueError("Decoder size/SHA-256 mismatch")
                os.replace(temporary, archive)
            finally:
                output.close()
                temporary.unlink(missing_ok=True)
    extract(archive, directory, system == "windows")
    if args.stage:
        destination = ROOT / "apps/desktop/src-tauri/resources/media-runtime"
        shutil.copytree(directory / "bin", destination / "bin", dirs_exist_ok=True)
        shutil.copytree(directory / "licenses", destination / "licenses", dirs_exist_ok=True)
        shutil.copy2(Path(__file__).with_name("manifest.json"), destination / "manifest.json")
    print(f"Verified media tools available at {directory / 'bin'}")


if __name__ == "__main__":
    main()