PixelLauncher/launcher/java_fetch.py

"""Скачивание и установка Java с Eclipse Adoptium, если она не найдена.

Ставит JRE в `.launcher/runtime/java-<major>` и возвращает путь к java.
"""

import os
import shutil
import sys
import tarfile
import zipfile

from . import http
from .i18n import T

API = ("https://api.adoptium.net/v3/binary/latest/{feature}/ga/"
       "{platform}/{arch}/jre/hotspot/normal/eclipse")
LTS = (8, 11, 17, 21, 25)


def _platform():
    if sys.platform.startswith("win"):
        return "windows"
    if sys.platform == "darwin":
        return "mac"
    return "linux"


def _arch():
    try:
        import platform
        m = platform.machine().lower()
    except Exception:  # noqa: BLE001
        return "x64"
    if m in ("arm64", "aarch64"):
        return "aarch64"
    if m in ("x86_64", "amd64"):
        return "x64"
    return "x86"


def _exe():
    return "java.exe" if os.name == "nt" else "java"


def pick_feature(required):
    """Ближайшая LTS-версия не ниже требуемой."""
    for f in LTS:
        if f >= required:
            return f
    return LTS[-1]


def runtime_dir(runtime_root, feature):
    return os.path.join(runtime_root, "java-" + str(feature))


def find_runtime_java(runtime_root, feature):
    """Путь к bin/java установленной рантайм-Java, "" — не найдено."""
    base = runtime_dir(runtime_root, feature)
    exe = _exe()
    direct = os.path.join(base, "bin", exe)
    if os.path.isfile(direct):
        return direct
    # архив Adoptium распаковывается во вложенную папку (jdk-21.0.x+y)
    try:
        for root, dirs, files in os.walk(base):
            if root.count(os.sep) - base.count(os.sep) >= 3:
                dirs[:] = []
                continue
            if exe in files and os.path.basename(root) == "bin":
                return os.path.join(root, exe)
    except OSError:
        pass
    return ""


def find_installed(runtime_root, required):
    """Лучшая уже скачанная Java не ниже required (путь, "" — нет).

    Название каталога (java-21) хранит major-версию — не запускаем
    `java -version` повторно.
    """
    best, best_major = "", 999
    try:
        names = os.listdir(runtime_root)
    except OSError:
        return ""
    for name in names:
        if not name.startswith("java-"):
            continue
        try:
            major = int(name.split("-", 1)[1])
        except ValueError:
            continue
        if not (required <= major < best_major):
            continue
        exe = find_runtime_java(runtime_root, major)
        if exe:
            best, best_major = exe, major
    return best


def ensure_java(required, runtime_root, progress=None, log=None, cancel=None):
    """Возвращает путь к Java не ниже required; при необходимости скачивает.

    progress(done_bytes, total_bytes|None) — прогресс скачивания архива.
    """
    feature = pick_feature(required)
    exe = find_runtime_java(runtime_root, feature)
    if exe:
        return exe

    url = API.format(feature=feature, platform=_platform(), arch=_arch())
    dest = runtime_dir(runtime_root, feature)
    os.makedirs(dest, exist_ok=True)
    archive = os.path.join(runtime_root, "java-%s.download" % feature)
    if os.path.exists(archive):
        try:
            os.remove(archive)
        except OSError:
            pass

    if log:
        log(T("Скачивание Java {v} (Adoptium)...").format(v=feature))
    total = http.head_size(url)
    try:
        http.download_file(url, archive, size=total, progress=progress,
                           cancel=cancel)
        if log:
            log(T("Распаковка Java..."))
        _extract(archive, dest)
    finally:
        try:
            os.remove(archive)
        except OSError:
            pass

    exe = find_runtime_java(runtime_root, feature)
    if not exe:
        raise RuntimeError(T("Не удалось распаковать Java"))
    if log:
        log(T("Java {v} установлена").format(v=feature), "ok")
    return exe


def _safe_join(base, name):
    """Путь внутри base; None — попытка выйти за пределы (path traversal)."""
    parts = [p for p in name.replace("\\", "/").split("/")
             if p not in ("", ".", "..")]
    if not parts:
        return None
    return os.path.join(base, *parts)


def _extract(archive, dest):
    # формат определяем по сигнатуре: имя файла служебное (java-XX.download)
    with open(archive, "rb") as f:
        magic = f.read(4)
    if magic[:2] == b"PK":  # ZIP
        with zipfile.ZipFile(archive) as z:
            for info in z.infolist():
                target = _safe_join(dest, info.filename)
                if target is None:
                    continue
                if info.is_dir():
                    os.makedirs(target, exist_ok=True)
                    continue
                os.makedirs(os.path.dirname(target), exist_ok=True)
                with z.open(info) as src, open(target, "wb") as dst:
                    shutil.copyfileobj(src, dst)
    else:
        with tarfile.open(archive, "r:*") as t:
            for m in t.getmembers():
                if not (m.isfile() or m.isdir()):
                    continue
                target = _safe_join(dest, m.name)
                if target is None:
                    continue
                if m.isdir():
                    os.makedirs(target, exist_ok=True)
                    continue
                os.makedirs(os.path.dirname(target), exist_ok=True)
                src = t.extractfile(m)
                if src is None:
                    continue
                with src, open(target, "wb") as dst:
                    shutil.copyfileobj(src, dst)

    # права на исполнение для содержимого bin/ (linux/mac)
    if os.name != "nt":
        for root, dirs, files in os.walk(dest):
            if os.path.basename(root) != "bin":
                continue
            for f in files:
                try:
                    os.chmod(os.path.join(root, f), 0o755)
                except OSError:
                    pass