123 lines
3.8 KiB
Python
123 lines
3.8 KiB
Python
"""Extracted from octo_updater.py. Keep this module focused on its named responsibility."""
|
|
import json
|
|
import hashlib
|
|
import os
|
|
import sys
|
|
import ssl
|
|
import re
|
|
import subprocess
|
|
import urllib.request
|
|
from urllib.parse import urlsplit
|
|
import shutil
|
|
import stat
|
|
import struct
|
|
import time
|
|
import math
|
|
import threading
|
|
import queue
|
|
from functools import cache
|
|
from pathlib import Path
|
|
|
|
UPDATER_VERSION = "1.3.1"
|
|
SERVER = "https://octowow.st"
|
|
UA = f"OctoUpdater/{UPDATER_VERSION}"
|
|
DOWNLOAD_RETRY = 5
|
|
DOWNLOAD_TIMEOUT = 10 # seconds without any data before a transfer aborts
|
|
|
|
# Where the app lives: next to the .exe when frozen (PyInstaller), otherwise
|
|
# next to this script — never the current working directory, which varies with
|
|
# how the app was launched. This anchors the default game folder.
|
|
if getattr(sys, "frozen", False):
|
|
APP_DIR = os.path.dirname(os.path.abspath(sys.executable))
|
|
else:
|
|
APP_DIR = os.path.dirname(os.path.abspath(__file__))
|
|
|
|
|
|
def _default_app_data_dir() -> str:
|
|
base = os.environ.get("LOCALAPPDATA")
|
|
if base:
|
|
path = os.path.join(base, "OctoUpdater")
|
|
try:
|
|
os.makedirs(path, exist_ok=True)
|
|
return path
|
|
except OSError:
|
|
pass
|
|
return APP_DIR
|
|
|
|
|
|
APP_DATA_DIR = _default_app_data_dir()
|
|
|
|
CONFIG_FILE = os.path.join(APP_DATA_DIR, "config.json")
|
|
|
|
# First-run default game folder, anchored to the app dir (not the CWD).
|
|
DEFAULT_GAME_DIR = os.path.join(APP_DIR, "OctoWoW")
|
|
|
|
|
|
def _relocate_legacy_data():
|
|
# Old config (<1.3: octo_updater_config.json beside the app) becomes
|
|
# config.json in APP_DATA_DIR. Rename + relocate, only when the old name
|
|
# exists and the new path doesn't (idempotent). The legacy game-hash cache
|
|
# is no longer used and is deleted by _migrate_1_3.
|
|
old_name, new = "octo_updater_config.json", CONFIG_FILE
|
|
for old in (os.path.join(APP_DIR, old_name),
|
|
os.path.join(APP_DATA_DIR, old_name)):
|
|
if old == new or not os.path.exists(old) or os.path.exists(new):
|
|
continue
|
|
try:
|
|
shutil.move(old, new)
|
|
break
|
|
except OSError:
|
|
pass
|
|
|
|
_CONFIG_LOCK = threading.RLock()
|
|
|
|
def load_config() -> dict:
|
|
try:
|
|
with open(CONFIG_FILE) as f:
|
|
return json.load(f)
|
|
except FileNotFoundError:
|
|
return {}
|
|
except Exception:
|
|
return {}
|
|
|
|
|
|
def _atomic_write(path: str, text: str):
|
|
"""Write via a temp file + atomic rename so a crash mid-write can never
|
|
leave a truncated/corrupt file at `path`."""
|
|
tmp = path + ".tmp"
|
|
with open(tmp, "w") as f:
|
|
f.write(text)
|
|
f.flush()
|
|
os.fsync(f.fileno())
|
|
os.replace(tmp, path)
|
|
|
|
|
|
def save_config(data: dict):
|
|
with _CONFIG_LOCK:
|
|
try:
|
|
_atomic_write(CONFIG_FILE, json.dumps(data, indent=2))
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def update_config(mutator):
|
|
"""Load the current on-disk config under the lock, apply `mutator(cfg)`,
|
|
save atomically, and return the result. Every config change — main thread
|
|
or worker — should go through this so no stale in-memory snapshot can
|
|
overwrite keys another thread just persisted."""
|
|
with _CONFIG_LOCK:
|
|
cfg = load_config()
|
|
mutator(cfg)
|
|
save_config(cfg)
|
|
return cfg
|
|
|
|
|
|
def ensure_dir(path):
|
|
Path(path).mkdir(parents=True, exist_ok=True)
|
|
|
|
|
|
# ── logging ─────────────────────────────────────────────────────────────────
|
|
# One thread-safe log sink for the whole app. Any function — worker thread or
|
|
# main — calls log(); the GUI drains _LOG_Q on the main thread (see
|
|
# OctoUpdaterApp._poll) and renders each line. This keeps all Tk access on the
|