PixelLauncher 2.1.0 — стабильная сборка

Офлайн-лаунчер Minecraft: аккаунты без Microsoft, установка версий и лоадеров
(Fabric, Quilt, Forge, NeoForge), каталог модпаков Modrinth, поиск и обновление
модов, разбор вылетов, системный трей, автозагрузка, плагины, прокси и зеркала.

launcher/plugins.py

@@ -0,0 +1,390 @@
+"""Плагины PixelLauncher: загрузка .py и событийная модель.
+
+Формат файла (точно совпадает с верстаком на pixellauncher.ru):
+
+    # Плагин для PixelLauncher.
+    PLUGIN_NAME = "Мой плагин"
+    PLUGIN_VERSION = "1.0"
+    PLUGIN_AUTHOR = "Nik"
+    PLUGIN_DESCRIPTION = (
+        "Что делает."
+    )
+
+    def on_load(api):
+        api.log("привет")
+
+        def on_game_exit(profile_name, crashed, exit_code):
+            ...
+
+        api.on("game_exit", on_game_exit)
+        api.add_action("Моя кнопка", action)
+
+    def on_unload():          # необязательно
+        ...
+
+События:
+    launcher_ready()
+    before_launch(profile_name, version)
+    after_launch(profile_name, version, pid)
+    game_exit(profile_name, crashed, exit_code)
+    profile_switched(profile_name)
+    account_switched(username)
+    launcher_closing()
+
+Плагины лежат в папке plugins рядом с main.py (APP_DIR/plugins);
+json-хранилище каждого плагина — в plugins/storage/<файл>.py.json.
+"""
+
+import importlib.util
+import json
+import os
+import re
+import threading
+import traceback
+
+from . import config
+
+PLUGIN_DIR = os.path.join(config.APP_DIR, "plugins")
+STORAGE_DIR = os.path.join(PLUGIN_DIR, "storage")
+
+# каталог плагинов на сайте: магазин и pixellauncher://plugin/<файл>
+STORE_BASE = "https://pixellauncher.ru/plugins"
+CATALOG_URL = STORE_BASE + "/catalog.json"
+
+EVENTS = ("launcher_ready", "before_launch", "after_launch", "game_exit",
+          "profile_switched", "account_switched", "launcher_closing")
+
+# --- привязка к UI (App). Без неё api.log печатает в stdout,
+# --- а окна не показываются (удобно для тестов).
+_hooks = {"log": None, "message": None, "ask_text": None,
+          "settings": None, "game_dir": None, "game_running": None}
+_hooks_lock = threading.Lock()
+
+
+def bind_ui(log=None, message=None, ask_text=None, settings=None,
+            game_dir=None, game_running=None):
+    """Подключает обработчики UI. Вызывается из App один раз."""
+    with _hooks_lock:
+        for k, v in (("log", log), ("message", message),
+                     ("ask_text", ask_text), ("settings", settings),
+                     ("game_dir", game_dir),
+                     ("game_running", game_running)):
+            if v is not None:
+                _hooks[k] = v
+
+
+def _hook(name, *args):
+    fn = _hooks.get(name)
+    if fn is None:
+        return None
+    try:
+        return fn(*args)
+    except Exception:  # noqa: BLE001
+        return None
+
+
+def _call_log(msg):
+    if _hook("log", str(msg)) is None:
+        try:
+            print(msg)
+        except Exception:  # noqa: BLE001
+            pass
+
+
+def _ui_message(level, title, text):
+    if _hooks.get("message") is None:
+        _call_log("%s: %s" % (title, text))
+        return
+    _hook("message", level, str(title), str(text))
+
+
+# ------------------------------------------------------------------ хранилище
+_st_lock = threading.Lock()
+
+
+def _storage_path(plugin_file):
+    safe = re.sub(r"[^\w.\-]", "_", plugin_file, flags=re.ASCII)
+    return os.path.join(STORAGE_DIR, safe + ".json")
+
+
+def _storage_read(plugin_file):
+    path = _storage_path(plugin_file)
+    with _st_lock:
+        try:
+            with open(path, encoding="utf-8") as f:
+                data = json.load(f)
+            if isinstance(data, dict):
+                return data
+        except (OSError, ValueError):
+            pass
+    return {}
+
+
+def _storage_write(plugin_file, data):
+    if not isinstance(data, dict):
+        return
+    path = _storage_path(plugin_file)
+    with _st_lock:
+        try:
+            os.makedirs(STORAGE_DIR, exist_ok=True)
+            tmp = path + ".tmp"
+            with open(tmp, "w", encoding="utf-8") as f:
+                json.dump(data, f, ensure_ascii=False, indent=2)
+            os.replace(tmp, path)
+        except OSError:
+            pass
+
+
+# ------------------------------------------------------------------ Api
+class _Api:
+    """API, которое видит плагин внутри on_load(api)."""
+
+    def __init__(self, plugin):
+        self._p = plugin
+
+    # --- события и кнопки
+    def on(self, event, fn):
+        self._p.on(event, fn)
+
+    def add_action(self, label, fn):
+        self._p.add_action(label, fn)
+
+    # --- журнал и окна
+    def log(self, message):
+        _call_log("[%s] %s" % (self._p.name, message))
+
+    def show_message(self, title, text):
+        _ui_message("info", title, text)
+
+    def show_warning(self, title, text):
+        _ui_message("warn", title, text)
+
+    def ask_text(self, title, label, default=""):
+        res = _hook("ask_text", str(title), str(label), str(default))
+        return res if isinstance(res, str) or res is None else None
+
+    # --- состояние лаунчера
+    def get_minecraft_dir(self):
+        v = _hook("game_dir")
+        return v if isinstance(v, str) else config.DEFAULT_GAME_DIR
+
+    def get_current_profile(self):
+        s = _hook("settings") or {}
+        v = s.get("active_profile")
+        return str(v) if v else None
+
+    def get_active_account(self):
+        s = _hook("settings") or {}
+        return str(s.get("active_account") or "")
+
+    def get_profiles(self):
+        s = _hook("settings") or {}
+        out = []
+        for p in (s.get("profiles") or []):
+            name = str((p or {}).get("name") or "")
+            if name:
+                out.append(name)
+        return out
+
+    def is_game_running(self):
+        return bool(_hook("game_running"))
+
+    # --- своё хранилище
+    def get_storage(self):
+        return _storage_read(self._p.file)
+
+    def save_storage(self, data):
+        _storage_write(self._p.file, data)
+
+
+# ------------------------------------------------------------------ Plugin
+class Plugin:
+    """Один загруженный плагин."""
+
+    def __init__(self, path):
+        self.path = path
+        self.file = os.path.basename(path)
+        self.name = self.file[:-3] if self.file.endswith(".py") else self.file
+        self.version = ""
+        self.author = ""
+        self.description = ""
+        self.module = None
+        self.handlers = {}     # event -> [callable]
+        self.actions = []      # [(label, callable)]
+
+    def load(self):
+        """Импортирует файл и вызывает on_load(api)."""
+        modname = "pixelplugin_" + re.sub(r"\W", "_", self.file[:-3])
+        spec = importlib.util.spec_from_file_location(modname, self.path)
+        if spec is None or spec.loader is None:
+            raise ImportError("нельзя построить спецификацию модуля")
+        mod = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(mod)
+        self.module = mod
+        self.name = str(getattr(mod, "PLUGIN_NAME", "") or self.name)
+        self.version = str(getattr(mod, "PLUGIN_VERSION", "") or "")
+        self.author = str(getattr(mod, "PLUGIN_AUTHOR", "") or "")
+        self.description = str(getattr(mod, "PLUGIN_DESCRIPTION", "") or "")
+        on_load = getattr(mod, "on_load", None)
+        if callable(on_load):
+            on_load(_Api(self))
+        return self
+
+    def on(self, event, fn):
+        if event not in EVENTS:
+            _call_log("Плагин %s: неизвестное событие «%s»"
+                      % (self.file, event))
+            return
+        if callable(fn):
+            self.handlers.setdefault(event, []).append(fn)
+
+    def add_action(self, label, fn):
+        if callable(fn):
+            self.actions.append((str(label), fn))
+
+    def emit(self, event, **kw):
+        for fn in list(self.handlers.get(event, ())):
+            try:
+                fn(**kw)
+            except Exception as e:  # noqa: BLE001
+                _call_log("Плагин %s: ошибка события %s: %s"
+                          % (self.file, event, e))
+
+    def unload(self):
+        fn = getattr(self.module, "on_unload", None) if self.module else None
+        if callable(fn):
+            try:
+                fn()
+            except Exception as e:  # noqa: BLE001
+                _call_log("Плагин %s: on_unload: %s" % (self.file, e))
+        self.handlers.clear()
+        self.actions.clear()
+
+
+# ------------------------------------------------------------------ менеджер
+_loaded = []
+_failures = []
+_mgr_lock = threading.Lock()
+_scan_lock = threading.Lock()
+
+
+def plugin_files():
+    try:
+        return sorted(f for f in os.listdir(PLUGIN_DIR)
+                      if f.endswith(".py") and not f.startswith(("_", ".")))
+    except OSError:
+        return []
+
+
+def scan():
+    """(Пере)загружает все плагины. Возвращает (загружено, ошибок)."""
+    with _scan_lock:
+        unload_all()
+        loaded, failures = [], []
+        try:
+            os.makedirs(PLUGIN_DIR, exist_ok=True)
+        except OSError:
+            pass
+        for fname in plugin_files():
+            p = Plugin(os.path.join(PLUGIN_DIR, fname))
+            try:
+                p.load()
+                loaded.append(p)
+            except Exception as e:  # noqa: BLE001
+                short = "%s: %s" % (type(e).__name__, e)
+                failures.append((fname, short))
+                _call_log("Плагин %s не загрузился: %s" % (fname, short))
+                try:
+                    traceback.print_exc()
+                except Exception:  # noqa: BLE001
+                    pass
+        with _mgr_lock:
+            _loaded[:] = loaded
+            _failures[:] = failures
+    if loaded:
+        _call_log("Плагины: загружено %d%s"
+                  % (len(loaded),
+                     ", ошибок %d" % len(failures) if failures else ""))
+    return len(loaded), len(failures)
+
+
+def unload_all():
+    with _mgr_lock:
+        old = list(_loaded)
+        _loaded.clear()
+    for p in old:
+        p.unload()
+
+
+def loaded():
+    with _mgr_lock:
+        return list(_loaded)
+
+
+def failures():
+    with _mgr_lock:
+        return list(_failures)
+
+
+def emit(event, **kw):
+    """Рассылает событие всем загруженным плагинам."""
+    for p in loaded():
+        p.emit(event, **kw)
+
+
+def actions():
+    """Кнопки плагинов: [(label, fn, plugin_name)]."""
+    out = []
+    for p in loaded():
+        for label, fn in list(p.actions):
+            out.append((label, fn, p.name))
+    return out
+
+
+# ------------------------------------------------------------------ установка
+def install_file(filename, data):
+    """Сохраняет файл плагина в папку plugins. Возвращает путь."""
+    safe = os.path.basename(str(filename).replace("\\", "/"))
+    if (not safe.endswith(".py") or len(safe) > 100
+            or safe.startswith(("_", ".")) or "/" in safe):
+        raise ValueError("недопустимое имя файла плагина: %r" % filename)
+    if isinstance(data, str):
+        data = data.encode("utf-8")
+    os.makedirs(PLUGIN_DIR, exist_ok=True)
+    path = os.path.join(PLUGIN_DIR, safe)
+    tmp = path + ".tmp"
+    with open(tmp, "wb") as f:
+        f.write(data)
+    os.replace(tmp, path)
+    return path
+
+
+def remove_file(filename):
+    """Удаляет установленный плагин из локальной папки."""
+    safe = os.path.basename(str(filename).replace("\\", "/"))
+    path = os.path.join(PLUGIN_DIR, safe)
+    if os.path.isfile(path):
+        os.remove(path)
+        return True
+    return False
+
+
+def fetch_catalog(timeout=15):
+    """Каталог с сайта: [{file, title, desc, ...}, ...]."""
+    from . import http
+    raw = http.get_bytes(CATALOG_URL, timeout=timeout)
+    data = json.loads(raw.decode("utf-8"))
+    if not isinstance(data, list):
+        raise ValueError("пустой каталог плагинов")
+    return [x for x in data
+            if isinstance(x, dict) and isinstance(x.get("file"), str)
+            and x["file"]]
+
+
+def download_plugin(filename, timeout=30):
+    """Скачивает плагин из магазина и ставит локально. Возвращает путь."""
+    from urllib.parse import quote
+    from . import http
+    safe = os.path.basename(str(filename).replace("\\", "/"))
+    data = http.get_bytes(STORE_BASE + "/" + quote(safe), timeout=timeout)
+    return install_file(safe, data)