"""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