PixelLauncher/launcher/protocol.py

"""Протокол pixellauncher:// — ссылки с сайта открывает лаунчер.

Схемы:
    pixellauncher://plugin/<файл.py>  — установить плагин из магазина
    pixellauncher://mrpack/<url>      — скачать и импортировать модпак

Реестр Windows: HKCU\\Software\\Classes\\pixellauncher (пишет установщик).

Одна живая копия лаунчера: .proto/instance.json с PID. Если лаунчер уже
запущен, новый экземпляр кладёт ссылку в очередь (.proto/q-*.json) и
выходит — работающая копия сама подхватывает очередь.
"""

import json
import os
import re
import time

from . import config

PROTO_DIR = os.path.join(config.APP_DIR, ".proto")
_INSTANCE = os.path.join(PROTO_DIR, "instance.json")


def parse(url):
    """Разбирает ссылку. Возвращает (kind, target) или None."""
    if not isinstance(url, str):
        return None
    prefix = "pixellauncher://"
    if not url.lower().startswith(prefix):
        return None
    rest = url[len(prefix):]
    low = rest.lower()
    if low.startswith("plugin/"):
        name = rest[len("plugin/"):]
        name = name.split("?")[0].split("#")[0].strip()
        try:
            from urllib.parse import unquote
            name = unquote(name)
        except Exception:  # noqa: BLE001
            pass
        if (not name or len(name) > 100 or "/" in name or "\\" in name
                or name.startswith(".") or not name.endswith(".py")):
            return None
        return ("plugin", name)
    if low.startswith("mrpack/"):
        target = rest[len("mrpack/"):]
        try:
            from urllib.parse import unquote
            target = unquote(target)
        except Exception:  # noqa: BLE001
            pass
        if target.startswith("https://") or target.startswith("http://"):
            return ("mrpack", target)
        return None
    return None


# ------------------------------------------------------------------ инстанс
def _pid_alive(pid):
    if pid <= 0:
        return False
    if os.name == "nt":
        import ctypes
        k32 = ctypes.windll.kernel32
        PROCESS_QUERY_LIMITED_INFORMATION = 0x1000
        STILL_ACTIVE = 259
        h = k32.OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, 0, pid)
        if not h:
            return False
        try:
            code = ctypes.c_ulong()
            if (k32.GetExitCodeProcess(h, ctypes.byref(code))
                    and code.value == STILL_ACTIVE):
                return True
            return False
        finally:
            k32.CloseHandle(h)
    try:
        os.kill(pid, 0)
        return True
    except OSError:
        return False


def other_instance_running():
    """True, если лаунчер с этим PID уже живёт (не мы сами)."""
    try:
        with open(_INSTANCE, encoding="utf-8") as f:
            pid = int(json.load(f).get("pid", 0))
    except (OSError, ValueError, TypeError, AttributeError):
        return False
    if pid == os.getpid():
        return False
    return _pid_alive(pid)


def mark_instance():
    try:
        os.makedirs(PROTO_DIR, exist_ok=True)
        with open(_INSTANCE, "w", encoding="utf-8") as f:
            json.dump({"pid": os.getpid(), "ts": int(time.time())}, f)
    except OSError:
        pass


def clear_instance():
    """Снимает свою отметку (чужую — не трогает)."""
    try:
        with open(_INSTANCE, encoding="utf-8") as f:
            if int(json.load(f).get("pid", 0)) != os.getpid():
                return
    except (OSError, ValueError, TypeError, AttributeError):
        pass
    try:
        os.remove(_INSTANCE)
    except OSError:
        pass


# ------------------------------------------------------------------ очередь
def queue_url(url):
    """Кладёт ссылку в очередь для уже запущенного лаунчера."""
    try:
        os.makedirs(PROTO_DIR, exist_ok=True)
        name = "q-%d-%d.json" % (int(time.time() * 1000), os.getpid())
        with open(os.path.join(PROTO_DIR, name), "w",
                  encoding="utf-8") as f:
            json.dump({"url": url, "ts": int(time.time())}, f)
        return True
    except OSError:
        return False


def queue_files():
    """Читает и удаляет ссылки из очереди: [(имя, url), ...]."""
    out = []
    try:
        names = sorted(f for f in os.listdir(PROTO_DIR)
                       if f.startswith("q-") and f.endswith(".json"))
    except OSError:
        return out
    for name in names:
        path = os.path.join(PROTO_DIR, name)
        url = ""
        try:
            with open(path, encoding="utf-8") as f:
                url = str(json.load(f).get("url") or "")
        except (OSError, ValueError, TypeError, AttributeError):
            pass
        try:
            os.remove(path)
        except OSError:
            pass
        if url:
            out.append((name, url))
    return out


# ------------------------------------------------------------------ загрузка
def fetch_mrpack(url, timeout=120):
    """Скачивает .mrpack в папку очереди. Возвращает локальный путь."""
    from . import http
    os.makedirs(PROTO_DIR, exist_ok=True)
    base = url.rstrip("/").split("/")[-1] or "pack.mrpack"
    base = re.sub(r"[^\w.\-]", "_", base)
    if not base.endswith(".mrpack"):
        base += ".mrpack"
    path = os.path.join(PROTO_DIR, "incoming-" + base)
    data = http.get_bytes(url, timeout=timeout)
    with open(path, "wb") as f:
        f.write(data)
    return path