kosta2014p
ec6f13ddcc
Офлайн-лаунчер Minecraft: аккаунты без Microsoft, установка версий и лоадеров (Fabric, Quilt, Forge, NeoForge), каталог модпаков Modrinth, поиск и обновление модов, разбор вылетов, системный трей, автозагрузка, плагины, прокси и зеркала.
391 lines
12 KiB
Python
391 lines
12 KiB
Python
"""Плагины 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)
|