Kiwy-Signage/windows/webview2_runtime.py

"""webview2_runtime.py — make sure the WebView2 Runtime is present.

Why this exists
---------------
WebView2 splits into two parts:

* the **SDK** (the ``Microsoft.Web.WebView2.Core.dll`` + ``WebView2Loader.dll``
  we bundle in ``windows/webview2_sdk``), which is just the API surface, and
* the **Runtime** (``msedgewebview2.exe`` etc.), the actual Chromium engine.

The SDK is useless without the Runtime. The Runtime ships with Windows 11 and
is present on the vast majority of Windows 10 machines, but Microsoft still
recommends checking for it and installing it when missing — so that is what
this module does.

Deployment notes (per Microsoft's distribution guidance):

* If the Runtime is missing we run an installer with ``/silent /install``.
* Run it **without elevation** → per-user install, which never shows a UAC
  prompt. That matters for an unattended signage player: a UAC dialog on a
  kiosk screen is a failure, not a prompt.
* Two installers are supported, in order of preference:
  1. ``MicrosoftEdgeWebView2RuntimeInstallerX64.exe`` — the ~203 MB offline
     *standalone* installer. Works with no internet (drop it in
     ``windows/webview2_runtime/`` to have it bundled).
  2. ``MicrosoftEdgeWebview2Setup.exe`` — the ~1.7 MB *bootstrapper*, which
     downloads the Runtime from Microsoft. Bundled by default.

Nothing here ever raises: a failure just means web links fall back to the
Chrome/Edge subprocess engine, which is far better than the player crashing.
"""

from __future__ import annotations

import os
import subprocess
import sys
import threading
import time
from pathlib import Path

# Per Microsoft, the Runtime's presence/version lives in this registry value.
# (Edge Update client GUID for the Evergreen WebView2 Runtime.)
_CLIENT_GUID = '{F3017226-FE2A-4295-8BDF-00C3A9A7E4C5}'

_STANDALONE_NAME = 'MicrosoftEdgeWebView2RuntimeInstallerX64.exe'
_BOOTSTRAPPER_NAME = 'MicrosoftEdgeWebview2Setup.exe'

#: Don't re-attempt a failing install on every single start-up.
_RETRY_COOLDOWN_SECONDS = 6 * 60 * 60

_install_lock = threading.Lock()
_install_state = {
    'attempted': False,
    'installing': False,
    'installed': None,      # bool once known
    'version': '',
    'error': '',
}

#: Set once a background install has finished, so a weblink can wait for it.
_install_done = threading.Event()


# ── Detection ────────────────────────────────────────────────────────
def _parse_version(text):
    """Return a comparable tuple from a version string like '152.0.4191.66'."""
    parts = []
    for chunk in str(text or '').split('.'):
        digits = ''.join(c for c in chunk if c.isdigit())
        parts.append(int(digits) if digits else 0)
    while len(parts) < 4:
        parts.append(0)
    return tuple(parts[:4])


def _read_registry_version():
    """Read the installed Runtime version from the registry, or ''.

    Checks both install scopes: HKLM (per-machine) and HKCU (per-user). On
    64-bit Windows the per-machine value lives under WOW6432Node because the
    Edge Updater is a 32-bit component.
    """
    if sys.platform != 'win32':
        return ''
    try:
        import winreg
    except Exception:
        return ''

    candidates = [
        # (hive, subkey, access flag)
        (winreg.HKEY_LOCAL_MACHINE,
         rf'SOFTWARE\WOW6432Node\Microsoft\EdgeUpdate\Clients\{_CLIENT_GUID}', 0),
        (winreg.HKEY_LOCAL_MACHINE,
         rf'SOFTWARE\Microsoft\EdgeUpdate\Clients\{_CLIENT_GUID}',
         getattr(winreg, 'KEY_WOW64_32KEY', 0)),
        (winreg.HKEY_CURRENT_USER,
         rf'SOFTWARE\Microsoft\EdgeUpdate\Clients\{_CLIENT_GUID}', 0),
    ]
    for hive, subkey, access in candidates:
        try:
            with winreg.OpenKey(hive, subkey, 0,
                                winreg.KEY_READ | access) as key:
                value, _ = winreg.QueryValueEx(key, 'pv')
            value = str(value or '').strip()
            if value and _parse_version(value) > (0, 0, 0, 0):
                return value
        except Exception:
            continue
    return ''


def get_runtime_version():
    """Version of the installed Evergreen Runtime, or '' when absent."""
    version = _read_registry_version()
    if version:
        return version
    # Fallback: ask the SDK itself (also covers preview channels).
    try:
        from webview2_browser import _find_sdk_dir, _load_webview2_types

        sdk = _find_sdk_dir()
        if sdk is not None:
            types = _load_webview2_types(sdk)
            reported = types['env'].GetAvailableBrowserVersionString()
            return str(reported).strip() if reported else ''
    except Exception:
        pass
    return ''


def is_runtime_installed():
    """True when a usable WebView2 Runtime is present."""
    return bool(get_runtime_version())


# ── Installer discovery ──────────────────────────────────────────────
def _search_dirs():
    """Folders that may hold an installer, best (standalone) first."""
    here = Path(__file__).resolve().parent
    dirs = [here / 'webview2_runtime', here]
    meipass = getattr(sys, '_MEIPASS', None)
    if meipass:
        dirs.append(Path(meipass) / 'webview2_runtime')
    # Next to the .exe, so an operator can drop the offline installer in
    # without rebuilding.
    data_dir = os.environ.get('KIWY_DATA_DIR')
    if data_dir:
        dirs.append(Path(data_dir) / 'webview2_runtime')
        dirs.append(Path(data_dir))
    env = os.environ.get('KIWY_WEBVIEW2_INSTALLER')
    if env:
        dirs.insert(0, Path(env).parent)
    return dirs


def find_installer():
    """Locate a usable installer. Returns ``(path, kind)`` or ``(None, None)``.

    The standalone (offline) installer is preferred: it does not depend on the
    target machine having internet access, which is the normal case for a
    signage player on an isolated LAN.
    """
    env = os.environ.get('KIWY_WEBVIEW2_INSTALLER')
    if env and Path(env).is_file():
        return Path(env), 'explicit'

    found = {'standalone': None, 'bootstrapper': None}
    for directory in _search_dirs():
        try:
            if found['standalone'] is None:
                candidate = directory / _STANDALONE_NAME
                if candidate.is_file():
                    found['standalone'] = candidate
            if found['bootstrapper'] is None:
                candidate = directory / _BOOTSTRAPPER_NAME
                if candidate.is_file():
                    found['bootstrapper'] = candidate
        except OSError:
            continue

    if found['standalone'] is not None:
        return found['standalone'], 'standalone'
    if found['bootstrapper'] is not None:
        return found['bootstrapper'], 'bootstrapper'
    return None, None


def _has_internet(timeout=4.0):
    """Quick reachability probe. A closed network returns False fast."""
    try:
        import socket

        with socket.create_connection(('www.msftconnecttest.com', 80),
                                      timeout=timeout):
            return True
    except Exception:
        return False


# ── Cooldown bookkeeping ─────────────────────────────────────────────
def _marker_path():
    data_dir = os.environ.get('KIWY_DATA_DIR') or os.getcwd()
    return Path(data_dir) / 'logs' / '.webview2_install_attempt'


def _recent_failed_attempt():
    try:
        marker = _marker_path()
        if not marker.is_file():
            return False
        age = time.time() - marker.stat().st_mtime
        return age < _RETRY_COOLDOWN_SECONDS
    except Exception:
        return False


def _record_attempt():
    try:
        marker = _marker_path()
        marker.parent.mkdir(parents=True, exist_ok=True)
        marker.write_text(str(int(time.time())))
    except Exception:
        pass


def _clear_attempt_marker():
    try:
        marker = _marker_path()
        if marker.is_file():
            marker.unlink()
    except Exception:
        pass


# ── Install ──────────────────────────────────────────────────────────
def _create_no_window():
    """Keep the installer from flashing a console window on the signage."""
    try:
        return subprocess.CREATE_NO_WINDOW
    except AttributeError:
        return 0x08000000


def _run_installer(path, timeout):
    """Run the installer silently. Returns (ok, detail)."""
    # `/silent /install` is the documented silent invocation. Deliberately NOT
    # elevated: a non-elevated run performs a per-user install, which never
    # raises a UAC prompt on the kiosk display.
    args = [str(path), '/silent', '/install']
    _log(f'WebView2: running installer {Path(path).name} /silent /install '
         f'(per-user, no elevation)')
    try:
        result = subprocess.run(
            args,
            timeout=timeout,
            creationflags=_create_no_window(),
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
        )
    except subprocess.TimeoutExpired:
        return False, f'installer timed out after {int(timeout)}s'
    except Exception as exc:
        return False, f'could not run installer: {exc}'

    code = result.returncode
    detail = (result.stdout or b'').decode('utf-8', 'replace').strip()
    # Edge Update installers commonly report 0 (success) or 3010 (reboot
    # required). They also return non-zero HRESULTs when the Runtime is already
    # installed at an equal/newer version — which is why the caller decides
    # success by re-reading the installed version rather than trusting this
    # code. We only use it to explain a failure.
    if code in (0, 3010):
        return True, f'installer exit code {code}'
    return False, f'installer exit code {code}{": " + detail if detail else ""}'


def ensure_runtime(timeout=600):
    """Install the Runtime when missing. Blocking; never raises.

    Returns a dict describing the outcome (``installed``, ``version``,
    ``error``, ``action``).
    """
    with _install_lock:
        if is_runtime_installed():
            version = get_runtime_version()
            _install_state.update(
                attempted=True, installing=False, installed=True,
                version=version, error='',
            )
            return dict(_install_state, action='already-present')

        if _recent_failed_attempt():
            _install_state.update(
                attempted=True, installing=False, installed=False, error='',
            )
            return dict(_install_state, action='skipped-recent-failure')

        path, kind = find_installer()
        if path is None:
            message = ('no WebView2 installer found (expected '
                       f'{_STANDALONE_NAME} or {_BOOTSTRAPPER_NAME} in '
                       'windows/webview2_runtime/)')
            _log(f'WebView2: {message}')
            _install_state.update(
                attempted=True, installing=False, installed=False,
                error=message,
            )
            return dict(_install_state, action='installer-missing')

        if kind == 'bootstrapper':
            # The bootstrapper downloads the Runtime from Microsoft. On a closed
            # network that can never succeed, so fail fast with an actionable
            # message instead of hanging for the whole timeout.
            if not _has_internet():
                message = ('the WebView2 Runtime is missing and this machine has '
                           f'no internet access; only the ONLINE bootstrapper '
                           f'({_BOOTSTRAPPER_NAME}) is available. Bundle the '
                           f'offline installer ({_STANDALONE_NAME}, run '
                           'webview2_runtime/download_runtime_installers.ps1 '
                           '-Offline) to run on a closed network.')
                _log(f'WebView2: {message}')
                _install_state.update(
                    attempted=True, installing=False, installed=False,
                    error=message,
                )
                _install_done.set()
                return dict(_install_state, action='offline-no-installer')
            _log('WebView2: Runtime missing — using the ONLINE bootstrapper '
                 '(downloads ~150 MB). Add the offline standalone installer to '
                 'avoid needing internet.')

        _install_state.update(attempted=True, installing=True, error='')
        _record_attempt()
        ok, detail = _run_installer(path, timeout)
        if not ok and 'already installed' not in detail.lower():
            _log(f'WebView2: installer reported {detail}')

        # Decide success by RE-READING the installed version, not by the exit
        # code: a non-zero HRESULT can simply mean "nothing to do".
        version = ''
        deadline = time.monotonic() + 30
        while time.monotonic() < deadline:
            version = get_runtime_version()
            if version:
                break
            time.sleep(1.0)

        if version:
            _clear_attempt_marker()
            _log(f'WebView2: Runtime available (v{version}) [{detail}]')
            _install_state.update(
                installing=False, installed=True, version=version,
                error='', action=f'installed-{kind}',
            )
        else:
            message = detail or 'Runtime still not detected after install'
            _log(f'WebView2: install did not take effect ({message})')
            _install_state.update(
                installing=False, installed=False, version='', error=message,
            )
        _install_done.set()
        return dict(_install_state, action=f'attempted-{kind}')


def ensure_runtime_async(timeout=600):
    """Kick off :func:`ensure_runtime` on a background thread.

    Called at start-up so a missing Runtime installs while the player is still
    syncing its playlist, instead of freezing the UI.
    """
    if is_runtime_installed():
        _install_done.set()
        _install_state.update(installed=True, version=get_runtime_version())
        return None

    def _worker():
        try:
            ensure_runtime(timeout=timeout)
        except Exception as exc:  # defensive: never kill the player
            _log(f'WebView2: background install failed: {exc}')
            _install_state.update(installing=False, installed=False, error=str(exc))
            _install_done.set()

    thread = threading.Thread(target=_worker, name='webview2-install', daemon=True)
    thread.start()
    return thread


def wait_for_install(timeout):
    """Wait (briefly, on the watcher thread) for a pending install.

    Returns True when a Runtime is available afterwards.
    """
    if is_runtime_installed():
        return True
    if not _install_state.get('installing'):
        return False
    _install_done.wait(timeout=max(0.0, float(timeout)))
    return is_runtime_installed()


def get_state():
    """Snapshot of the installer state, for logging/diagnostics."""
    state = dict(_install_state)
    if state.get('installed') is None:
        state['installed'] = is_runtime_installed()
        state['version'] = state['version'] or get_runtime_version()
    return state


def describe():
    """One-line status for the startup log."""
    version = get_runtime_version()
    if version:
        return f'WebView2 Runtime present (v{version})'
    path, kind = find_installer()
    if path is None:
        return 'WebView2 Runtime MISSING and no bundled installer found'
    if kind == 'standalone':
        return (f'WebView2 Runtime MISSING (will install OFFLINE via '
                f'{path.name} — no internet needed)')
    return (f'WebView2 Runtime MISSING (will install via the ONLINE '
            f'bootstrapper {path.name}; needs internet)')


def is_offline_ready():
    """True when a Runtime is present, or can be installed without internet.

    This is the property that matters for a closed-network deployment: web
    links will work on first start with no outbound connectivity.
    """
    if is_runtime_installed():
        return True
    path, kind = find_installer()
    return path is not None and kind in ('standalone', 'explicit')


def _log(message):
    try:
        from kivy.logger import Logger

        Logger.info(f'[WebView2] {message}')
    except Exception:
        print(f'[WebView2] {message}')