Stock-Tool/updater.py

"""
Auto-Updater
============
On every launch (after a successful license check), checks the Verimund API
for a newer version. If found, downloads the new exe via the API and
self-replaces via a helper batch script.

Called after a successful license check on every launch.
"""

import os
import shutil
import subprocess
import sys
import tempfile

import requests

# ---------------------------------------------------------------------------
# Current app version — kept in sync by push_update.py before each build.
# ---------------------------------------------------------------------------
APP_VERSION = "1.4.3"


def _exe_dir() -> str:
    if getattr(sys, "frozen", False):
        return os.path.dirname(sys.executable)
    return os.path.dirname(os.path.abspath(__file__))


def _parse_version(v: str) -> tuple:
    try:
        return tuple(int(x) for x in v.strip().split("."))
    except Exception:
        return (0, 0, 0)


def _log_update_error(msg: str) -> None:
    """Log update errors silently — never let this crash the app."""
    try:
        from license_check import get_api_headers, _API_URL
        from hwid import get_hwid
        hwid = get_hwid()
        import license_check
        channel = license_check.get_channel()
        requests.post(
            f"{_API_URL}/log",
            headers=get_api_headers(),
            json={
                "hwid": hwid,
                "app_version": APP_VERSION,
                "channel": channel,
                "message": msg,
            },
            timeout=5,
        )
    except Exception:
        pass


def _self_replace_exe(jwt_token: str, channel: str) -> None:
    """
    Download the new exe from the API, then launch a helper batch script that
    waits for this process to exit, moves the file into place, and restarts.
    Never returns.
    """
    if not getattr(sys, "frozen", False):
        return

    from license_check import _API_URL

    current_exe = os.path.abspath(sys.executable)
    tmp_dir = tempfile.gettempdir()
    pending_exe = os.path.join(tmp_dir, "StockScreener_pending.exe")

    try:
        with requests.post(
            f"{_API_URL}/download",
            headers={"Authorization": f"Bearer {jwt_token}", "Content-Type": "application/json"},
            stream=True,
            timeout=120,
        ) as r:
            r.raise_for_status()
            with open(pending_exe, "wb") as f:
                for chunk in r.iter_content(chunk_size=65536):
                    f.write(chunk)
    except Exception as exc:
        _log_update_error(f"_self_replace_exe download failed: {exc}")
        return

    helper = os.path.join(tmp_dir, "uit_update.bat")
    with open(helper, "w") as f:
        f.write(
            f"@echo off\n"
            f":retry\n"
            f"timeout /t 1 /nobreak >nul\n"
            f"move /Y \"{pending_exe}\" \"{current_exe}\"\n"
            f"if errorlevel 1 goto retry\n"
            f"start \"\" \"{current_exe}\"\n"
            f"del \"%~f0\"\n"
        )

    subprocess.Popen(
        ["cmd", "/c", helper],
        creationflags=subprocess.CREATE_NO_WINDOW | subprocess.DETACHED_PROCESS,
        close_fds=True,
    )
    sys.exit(0)


def get_release_notes() -> tuple:
    """
    Returns (version, release_notes) for the current channel from the API.
    Falls back gracefully on any error.
    """
    try:
        from license_check import get_api_headers, get_jwt, _API_URL
        resp = requests.get(
            f"{_API_URL}/update",
            headers=get_api_headers(),
            timeout=8,
        )
        resp.raise_for_status()
        data = resp.json()
        version = data.get("version", APP_VERSION)
        notes = data.get("release_notes") or "No release notes available."
        return version, notes
    except Exception:
        return APP_VERSION, "Unable to fetch release notes."


def check_and_apply_update() -> None:
    """
    Silently check for updates and apply them. Errors are swallowed so a
    failed update check never blocks the app from launching.
    """
    try:
        from license_check import get_api_headers, get_jwt, get_channel, _API_URL

        resp = requests.get(
            f"{_API_URL}/update",
            headers=get_api_headers(),
            timeout=8,
        )
        resp.raise_for_status()
        data = resp.json()

        latest = data.get("version", APP_VERSION)
        channel = data.get("channel", "stable")

        if _parse_version(latest) <= _parse_version(APP_VERSION):
            return  # already up to date

        jwt_token = get_jwt()
        if not jwt_token:
            _log_update_error("No JWT token available for update download")
            return

        _self_replace_exe(jwt_token, channel)

    except Exception as exc:
        _log_update_error(f"check_and_apply_update failed: {exc}")