209 lines
7.7 KiB
Python
209 lines
7.7 KiB
Python
"""Extracted from octo_updater.py. Keep this module focused on its named responsibility."""
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
import hashlib
|
|
import platform
|
|
#from .config import APP_DATA_DIR, DOWNLOAD_TIMEOUT, ensure_dir
|
|
from .app_log import log
|
|
from .net import secure_urlopen
|
|
import urllib.request
|
|
|
|
|
|
# NOTE: Creates dir and verifies it was made
|
|
def ensure_dir(path):
|
|
Path(path).mkdir(parents=True, exist_ok=True)
|
|
|
|
# Windows: %LOCALAPPDATA%\octo-updater\
|
|
# Linux: $XDG_DATA_HOME/octo-updater or $HOME/.local/share/octo-updater
|
|
def user_data_dir() -> Path:
|
|
"""Alters which directory is used for application data. Returns Path object to directory"""
|
|
if sys.platform == "win32":
|
|
local_app_data = os.environ.get("LOCALAPPDATA")
|
|
if not local_app_data:
|
|
raise RuntimeError("LOCALAPPDATA is not set")
|
|
|
|
return Path(local_app_data)
|
|
|
|
# For linux , will first check for XDG_DATA_HOME"
|
|
local_app_data = os.environ.get("XDG_DATA_HOME")
|
|
if local_app_data:
|
|
return Path(local_app_data) / "OctoUpdater"
|
|
|
|
return Path.home() / ".local" / "share" / "OctoUpdater"
|
|
|
|
|
|
# Windows: %APPDATA%\octo-updater
|
|
# Linux: $XDG_CONFIG_HOME/octo-updater or $HOME/.config/octo-updater
|
|
def user_config_dir() -> Path:
|
|
"""Alters which directory is used for application data. Returns Path object to directory"""
|
|
|
|
if sys.platform == "win32":
|
|
local_config_data = os.environ.get("APPDATA")
|
|
if not local_config_data:
|
|
raise RuntimeError("APPDATA is not set")
|
|
|
|
return Path(local_config_data)
|
|
|
|
# For linux , will first check for XDG_CONFIG_HOME" , else set to ~/.config/octo-updater
|
|
local_config_data = os.environ.get("XDG_CONFIG_HOME")
|
|
if local_config_data:
|
|
return Path(local_config_data) / "OctoUpdater"
|
|
|
|
return Path.home() / ".config" / "OctoUpdater"
|
|
|
|
# NOTE: I think this is for opening links in the launcher ?
|
|
def open_directory(path: str) -> None:
|
|
if sys.platform == "win32":
|
|
subprocess.Popen(["explorer.exe", path])
|
|
else:
|
|
opener = shutil.which("xdg-open") # Point this to default linux browser ?
|
|
if not opener:
|
|
raise RuntimeError("xdg-open is required to open directories")
|
|
subprocess.Popen([opener, path])
|
|
|
|
# NOTE: If windows, creates ntfs junction , else , creates symlink
|
|
def create_directory_link(link: str, target: str) -> None:
|
|
if sys.platform == "win32":
|
|
flags = getattr(subprocess, "CREATE_NO_WINDOW", 0)
|
|
result = subprocess.run(["cmd", "/c", "mklink", "/J", link, target], capture_output=True, text=True, creationflags=flags)
|
|
if result.returncode != 0:
|
|
raise RuntimeError(result.stderr.strip() or result.stdout.strip() or "mklink failed")
|
|
return
|
|
os.symlink(target, link, target_is_directory=True)
|
|
|
|
# NOTE: allows dynamic scaling of UI window ?
|
|
def enable_dpi_awareness() -> None:
|
|
if sys.platform != "win32":
|
|
return
|
|
import ctypes
|
|
try:
|
|
user32 = ctypes.windll.user32
|
|
fn = getattr(user32, "SetProcessDpiAwarenessContext", None)
|
|
if fn:
|
|
fn(ctypes.c_void_p(-4))
|
|
return
|
|
except Exception:
|
|
pass
|
|
try:
|
|
ctypes.windll.shcore.SetProcessDpiAwareness(2)
|
|
return
|
|
except Exception:
|
|
pass
|
|
try:
|
|
ctypes.windll.user32.SetProcessDPIAware()
|
|
except Exception:
|
|
pass
|
|
|
|
def request_defender_exclusion(path: str) -> bool:
|
|
"""Request a Windows Defender exclusion. Returns False off Windows."""
|
|
if sys.platform != "win32":
|
|
return False
|
|
import ctypes
|
|
escaped = path.replace("'", "''")
|
|
command = f"Add-MpPreference -ExclusionPath '{escaped}'"
|
|
result = ctypes.windll.shell32.ShellExecuteW(
|
|
None, "runas", "powershell.exe",
|
|
f'-NoProfile -WindowStyle Hidden -Command "{command}"', None, 0)
|
|
return result > 32
|
|
|
|
|
|
# TODO: Find solution to launch client natively on linux , either using wine or bottles ?
|
|
def launch_executable(executable: str, cwd: str) -> None:
|
|
"""Launch the managed executable using native process semantics."""
|
|
if sys.platform != "win32":
|
|
# TODO: implement solution to run game on linux ....
|
|
raise RuntimeError("native Linux game launching is not configured")
|
|
flags = (getattr(subprocess, "DETACHED_PROCESS", 0)
|
|
| getattr(subprocess, "CREATE_BREAKAWAY_FROM_JOB", 0))
|
|
try:
|
|
subprocess.Popen([executable], cwd=cwd, creationflags=flags, close_fds=True)
|
|
except OSError:
|
|
flags &= ~getattr(subprocess, "CREATE_BREAKAWAY_FROM_JOB", 0)
|
|
subprocess.Popen([executable], cwd=cwd, creationflags=flags, close_fds=True)
|
|
|
|
|
|
def display_info() -> dict:
|
|
"""Return primary display dimensions and refresh rate on Windows."""
|
|
if sys.platform != "win32":
|
|
raise RuntimeError("display information is not implemented for this platform")
|
|
import ctypes
|
|
class DEVMODE(ctypes.Structure):
|
|
_fields_ = [
|
|
("dmDeviceName", ctypes.c_wchar * 32), ("dmSpecVersion", ctypes.c_ushort),
|
|
("dmDriverVersion", ctypes.c_ushort), ("dmSize", ctypes.c_ushort),
|
|
("dmDriverExtra", ctypes.c_ushort), ("dmFields", ctypes.c_ulong),
|
|
("dmPositionX", ctypes.c_long), ("dmPositionY", ctypes.c_long),
|
|
("dmDisplayOrientation", ctypes.c_ulong), ("dmDisplayFixedOutput", ctypes.c_ulong),
|
|
("dmColor", ctypes.c_short), ("dmDuplex", ctypes.c_short),
|
|
("dmYResolution", ctypes.c_short), ("dmTTOption", ctypes.c_short),
|
|
("dmCollate", ctypes.c_short), ("dmFormName", ctypes.c_wchar * 32),
|
|
("dmLogPixels", ctypes.c_ushort), ("dmBitsPerPel", ctypes.c_ulong),
|
|
("dmPelsWidth", ctypes.c_ulong), ("dmPelsHeight", ctypes.c_ulong),
|
|
("dmDisplayFlags", ctypes.c_ulong), ("dmDisplayFrequency", ctypes.c_ulong),
|
|
]
|
|
dm = DEVMODE()
|
|
dm.dmSize = ctypes.sizeof(DEVMODE)
|
|
ctypes.windll.user32.EnumDisplaySettingsW(None, -1, ctypes.byref(dm))
|
|
return {"width": dm.dmPelsWidth, "height": dm.dmPelsHeight,
|
|
"refresh_rate": dm.dmDisplayFrequency}
|
|
|
|
|
|
|
|
# NOTE: Returns a path to an aria2c executable , otherwise returns none
|
|
def find_aria2c() -> str | None:
|
|
return shutil.which("aria2c")
|
|
|
|
|
|
# NOTE: Function that installs the exe for aria2c on windows devices, takes path as input , returns
|
|
# path to aria2c exe
|
|
def install_aria2c_windows( APP_DATA_DIR,
|
|
ALLOWED_DOWNLOAD_HOSTS,
|
|
log_fn) -> Path:
|
|
|
|
ARIA2C_PATH = os.path.join(APP_DATA_DIR, "aria2c.exe")
|
|
|
|
ARIA2_ZIP_SHA256 = "35f6514cc5dd7e98a87b3c4c2d25a0754b9b063dbe59bc0f22d483464f61e5b6"
|
|
|
|
ARIA2_ZIP_URL = (
|
|
"https://github.com/aria2/aria2/releases/download/"
|
|
"release-1.37.0/aria2-1.37.0-win-32bit-build1.zip"
|
|
)
|
|
|
|
#req = urllib.request.Request(ARIA2_ZIP_URL, headers={"User-Agent": UA})
|
|
req = urllib.request.Request(ARIA2_ZIP_URL)
|
|
log_fn("Fetching aria2c (one-time, ~2.5 MB)…", "acct")
|
|
|
|
with secure_urlopen(req, timeout=60, allowed_hosts=ALLOWED_DOWNLOAD_HOSTS) as r:
|
|
data = r.read()
|
|
|
|
digest = hashlib.sha256(data).hexdigest()
|
|
|
|
if digest != ARIA2_ZIP_SHA256:
|
|
raise RuntimeError(
|
|
f"aria2 checksum mismatch (got {digest[:12]}…); refusing to run it")
|
|
|
|
import zipfile, io
|
|
with zipfile.ZipFile(io.BytesIO(data)) as zf:
|
|
name = next((n for n in zf.namelist()
|
|
if n.lower().endswith("aria2c.exe")), None)
|
|
if not name:
|
|
raise RuntimeError("aria2c.exe not found in the aria2 archive")
|
|
exe = zf.read(name)
|
|
|
|
ensure_dir(APP_DATA_DIR)
|
|
tmp = ARIA2C_PATH + ".part"
|
|
with open(tmp, "wb") as f:
|
|
f.write(exe)
|
|
os.replace(tmp, ARIA2C_PATH)
|
|
log_fn("aria2c ready.", "ok")
|
|
|
|
return ARIA2C_PATH
|
|
|
|
|
|
|
|
|
|
|