Kiwy-Signage/windows/webview2_browser.py

"""webview2_browser.py — Embedded WebView2 (Edge/Chromium) INSIDE Kivy's window.

Why this exists
---------------
The old weblink engines launched a *separate* browser process (Chrome/Edge
kiosk subprocess, or the dormant `cef_browser.py`). That model caused every
weblink bug in the tracker: the browser opening behind the Kivy window, being
handed off to an existing instance and exiting instantly, fighting for
foreground/z-order, and leaking `msedge.exe`/`chrome.exe` processes that were
never closed.

WebView2 renders as a **child HWND of Kivy's own SDL window**, so:

  * no separate top-level window → nothing can open "in the background",
  * nothing to hand the URL off to → no instant-exit hand-off,
  * no z-order/foreground fight → it is literally a child of our window,
  * teardown is ours → no leaked browser processes,
  * the page renders at exactly the rectangle we give it (1920x1080 or
    whatever the Kivy window currently is).

Licensing / distribution: the WebView2 **runtime** is a free, evergreen,
Microsoft-shipped component (already present on this host as
``152.0.4191.66``). We only ship the small managed SDK + native loader DLLs.

Implementation notes
--------------------
* We talk to the .NET SDK through **pythonnet** (``clr``).
* Every WebView2 API is async (returns a .NET ``Task``). We must NOT call
  ``.GetAwaiter().GetResult()``: the continuation needs the *same* thread's
  message pump, so blocking would deadlock. Instead each task is **polled from
  Kivy's Clock** (the SDL thread, which pumps Win32 messages) and consumed when
  ``IsCompleted``. This mirrors how the old CEF code pumped via the Clock.
* All public methods are safe to call from the Kivy main thread.
"""

from __future__ import annotations

import ctypes
import os
import sys
import threading
from pathlib import Path

# ── SDK discovery ────────────────────────────────────────────────────
# The managed Microsoft.Web.WebView2.Core.dll and the native
# WebView2Loader.dll must sit in a folder we can find both in development and
# inside the PyInstaller bundle.
_SDK_ENV_VAR = 'KIWY_WEBVIEW2_SDK'


def _sdk_candidates():
    here = Path(__file__).resolve().parent
    yield here / 'webview2_sdk'
    # PyInstaller one-folder layout: bundled data lands next to the exe
    # (sys._MEIPASS points at the temporary _MEIxxx dir).
    meipass = getattr(sys, '_MEIPASS', None)
    if meipass:
        yield Path(meipass) / 'webview2_sdk'
    yield here


def _find_sdk_dir():
    env = os.environ.get(_SDK_ENV_VAR)
    if env and (Path(env) / 'Microsoft.Web.WebView2.Core.dll').is_file():
        return Path(env)
    for candidate in _sdk_candidates():
        try:
            if (candidate / 'Microsoft.Web.WebView2.Core.dll').is_file():
                return candidate
        except OSError:
            continue
    return None


# ── Win32 helpers ────────────────────────────────────────────────────
_SW_HIDE = 0
_SW_SHOWNORMAL = 1


class WebView2Browser:
    """One embedded WebView2 instance parented to the Kivy (SDL) window.

    Lifecycle::

        show(url) -> is_showing() (True once painted) -> resize(w,h) -> hide()
        -> shutdown()

    ``hide()`` only hides the controller (it stays alive), so switching back to
    a weblink later is instant. ``shutdown()`` disposes it for good.
    """

    #: Set True by the integration layer when the SDK + runtime are usable.
    _import_error = None

    def __init__(self, hwnd_provider=None, user_data_dir=None):
        self._hwnd_provider = hwnd_provider
        self._user_data_dir = user_data_dir or os.path.join(
            os.environ.get('KIWY_DATA_DIR', os.getcwd()), '.webview2-profile'
        )
        self._env = None
        self._controller = None
        self._webview = None
        self._hwnd = None
        self._showing = False
        self._stage = 'idle'          # idle | env | controller | ready
        self._pending_url = None
        self._failed_reason = ''
        self._poll_event = None
        self._lock = threading.RLock()
        self._task = None
        self._task_kind = None
        self._size = (0, 0)
        # Navigation outcome. `_showing` only means "the controller was told to
        # be visible", which happens the instant Navigate() is called — it says
        # nothing about whether the page actually loaded. On a closed network
        # that distinction is the whole point: an unreachable host paints a
        # Chromium error page, so without this the player would show a blank
        # error for the full slot instead of skipping the item.
        self._navigation_ok = None       # None = pending/unknown
        self._navigation_status = ''
        self._navigation_handlers = []   # keep refs: .NET must not GC these

    # ── Availability ─────────────────────────────────────────────────
    @staticmethod
    def is_available():
        """True when pythonnet + the SDK DLLs + a runtime are all present."""
        if sys.platform != 'win32':
            return False
        sdk = _find_sdk_dir()
        if sdk is None:
            return False
        try:
            import clr  # noqa: F401  (pythonnet)
        except Exception as exc:
            WebView2Browser._import_error = f'pythonnet unavailable: {exc}'
            return False
        try:
            cls = _load_webview2_types(sdk)
            version = cls['env'].GetAvailableBrowserVersionString()
            return bool(version)
        except Exception as exc:
            WebView2Browser._import_error = f'WebView2 unavailable: {exc}'
            return False

    # ── Public API (Kivy main thread) ────────────────────────────────
    def show(self, url):
        """Begin displaying ``url``. Returns True once the request is accepted.

        Rendering is asynchronous: the caller should poll :meth:`is_showing`
        (the session's ``wait_visible`` does this on the watcher thread).
        """
        with self._lock:
            self._failed_reason = ''
            self._pending_url = url

            if self._stage == 'ready' and self._controller is not None:
                return self._navigate(url)

            if self._stage in ('env', 'controller'):
                return True  # already starting up; URL is queued

            # Start-up order: environment → controller → navigate.
            self._stage = 'env'
            if not self._start_environment():
                self._stage = 'idle'
                return False
            if self._stage != 'env':
                # The environment resolved synchronously (fast path).
                return self._after_environment()
            return True

    def hide(self):
        """Hide the page without destroying the controller (fast re-show)."""
        with self._lock:
            self._showing = False
            self._pending_url = None
            controller = self._controller
        if controller is not None:
            try:
                controller.IsVisible = False
            except Exception:
                pass

    def is_showing(self):
        """True while the page is actually on screen."""
        with self._lock:
            if self._failed_reason:
                return False
            if self._controller is None:
                return False
            return self._showing

    def is_starting(self):
        """True while the environment/controller is still being created.

        WebView2 start-up is asynchronous. A controller that does not exist yet
        is NOT the same as a browser that has gone away, and conflating the two
        made the *first* weblink after a cold start be skipped instantly (the
        watcher saw "not alive" and advanced). Callers should treat
        ``is_starting()`` as "still alive, not yet painted".
        """
        with self._lock:
            if self._failed_reason:
                return False
            return self._stage in ('env', 'controller')

    def is_alive(self):
        """True when the browser is starting up or showing. False only on failure."""
        return self.is_showing() or self.is_starting()

    @property
    def failed_reason(self):
        return self._failed_reason

    def resize(self, width, height):
        """Fit the page to ``width`` x ``height`` physical pixels."""
        width, height = int(width), int(height)
        if width <= 0 or height <= 0:
            return
        with self._lock:
            self._size = (width, height)
            controller = self._controller
        if controller is None:
            return
        try:
            from System.Drawing import Rectangle

            controller.Bounds = Rectangle(0, 0, width, height)
        except Exception as exc:
            _log(f'WebView2 resize failed (non-fatal): {exc}')

    def shutdown(self):
        """Dispose the controller and environment. Never raises."""
        with self._lock:
            self._showing = False
            self._stop_poll_locked()
            controller, self._controller = self._controller, None
            webview, self._webview = self._webview, None
            env, self._env = self._env, None
            self._stage = 'idle'
        for obj, label in ((webview, 'webview'), (controller, 'controller')):
            if obj is None:
                continue
            try:
                dispose = getattr(obj, 'Dispose', None)
                if dispose is not None:
                    dispose()
            except Exception as exc:
                _log(f'WebView2 {label} dispose failed (non-fatal): {exc}')
        if ctypes is not None:
            try:
                ctypes.windll.ole32.CoUninitialize()
            except Exception:
                pass
        del env

    # ── Start-up ─────────────────────────────────────────────────────
    def _start_environment(self):
        sdk = _find_sdk_dir()
        if sdk is None:
            self._failed_reason = 'WebView2 SDK not found'
            _log('WebView2: SDK DLLs not found (expected Microsoft.Web.WebView2.Core.dll)')
            return False
        try:
            types = _load_webview2_types(sdk)
        except Exception as exc:
            self._failed_reason = f'WebView2 SDK load failed: {exc}'
            _log(f'WebView2: SDK load failed: {exc}')
            return False

        # The controller must live on a thread with a message pump; Kivy's SDL
        # thread qualifies, and COM must be initialised on it first.
        try:
            ctypes.windll.ole32.CoInitializeEx(None, 0x2)  # STA
        except Exception:
            pass

        try:
            os.makedirs(self._user_data_dir, exist_ok=True)
        except Exception as exc:
            _log(f'WebView2: could not create profile dir ({exc}); using temp')
            import tempfile

            self._user_data_dir = tempfile.mkdtemp(prefix='kiwy-wv2-')

        _log(f'WebView2: creating environment (profile={self._user_data_dir})')
        try:
            options = _build_environment_options()
            task = _create_environment_async(types, self._user_data_dir, options)
        except Exception as exc:
            self._failed_reason = f'CreateAsync failed: {exc}'
            _log(f'WebView2: environment creation failed: {exc}')
            return False

        self._task = task
        self._task_kind = 'env'
        self._start_poll()
        return True

    def _start_controller(self):
        hwnd = 0
        if self._hwnd_provider is not None:
            try:
                hwnd = self._hwnd_provider() or 0
            except Exception as exc:
                _log(f'WebView2: hwnd provider failed: {exc}')
        if not hwnd:
            self._failed_reason = 'Kivy window handle not found'
            _log('WebView2: could not locate the Kivy SDL window handle')
            return False
        self._hwnd = int(hwnd)

        _log(f'WebView2: creating controller inside hwnd=0x{self._hwnd:x}')
        try:
            # The parent window must be a .NET IntPtr; a plain Python int does
            # not match the overload and pythonnet raises "No method matches
            # given arguments".
            from System import IntPtr

            parent = IntPtr(self._hwnd)
            # HWND hosting: WebView2 creates its own child window in `parent`.
            task = self._env.CreateCoreWebView2ControllerAsync(parent)
        except Exception as exc:
            self._failed_reason = f'controller creation failed: {exc}'
            _log(f'WebView2: controller creation failed: {exc}')
            return False

        self._task = task
        self._task_kind = 'controller'
        self._stage = 'controller'
        self._start_poll()
        return True

    def _after_environment(self):
        """Called once the environment resolved."""
        if self._env is None:
            return False
        started = self._start_controller()
        if not started and self._stage == 'controller':
            return True  # still coming up asynchronously
        return started

    # ── Async task polling (Kivy Clock) ──────────────────────────────
    def _start_poll(self):
        try:
            from kivy.clock import Clock

            if self._poll_event is None:
                self._poll_event = Clock.schedule_interval(self._tick, 0.05)
        except Exception:
            # No Kivy (or called off-thread): poll from a plain timer instead.
            if self._poll_event is None:
                self._poll_event = _ThreadTimer(0.05, self._tick, None)

    def _stop_poll_locked(self):
        event, self._poll_event = self._poll_event, None
        if event is None:
            return
        try:
            cancel = getattr(event, 'cancel', None)
            if cancel is not None:
                cancel()
            else:
                event.stop()
        except Exception:
            pass

    def _tick(self, _dt):
        """Consume the in-flight Task once it completes."""
        with self._lock:
            task, kind = self._task, self._task_kind
            if task is None:
                self._stop_poll_locked()
                return False
            try:
                done = bool(task.IsCompleted)
            except Exception as exc:
                self._failed_reason = f'task poll failed: {exc}'
                self._task = None
                self._stop_poll_locked()
                return False
            if not done:
                return True
            self._task, self._task_kind = None, None
            self._stop_poll_locked()

        try:
            if task.IsFaulted:
                exc = task.Exception
                detail = ''
                try:
                    detail = exc.GetBaseException().Message
                except Exception:
                    detail = str(exc)
                self._failed_reason = f'{kind} failed: {detail}'
                _log(f'WebView2: {kind} task faulted: {detail}')
                return False
            result = task.Result
        except Exception as exc:
            self._failed_reason = f'{kind} task error: {exc}'
            _log(f'WebView2: {kind} task error: {exc}')
            return False

        if kind == 'env':
            self._env = result
            _log('WebView2: environment ready')
            if not self._start_controller():
                self._stage = 'idle'
            return False

        if kind == 'controller':
            self._controller = result
            self._on_controller_ready()
            return False

        return False

    def _on_controller_ready(self):
        """Wire up the page: bounds, settings, first navigation."""
        controller = self._controller
        try:
            controller.IsVisible = False       # stay hidden until navigated
        except Exception:
            pass

        webview = None
        try:
            webview = controller.CoreWebView2
        except Exception as exc:
            _log(f'WebView2: CoreWebView2 unavailable: {exc}')
        if webview is None:
            self._failed_reason = 'CoreWebView2 was not created'
            return
        self._webview = webview

        # Chrome-less, kiosk-like surface: no context menu, no devtools,
        # no accelerators that could let an operator escape the signage.
        try:
            settings = webview.Settings
            settings.AreDefaultContextMenusEnabled = False
            settings.AreDevToolsEnabled = False
            settings.IsStatusBarEnabled = False
            settings.AreBrowserAcceleratorKeysEnabled = False
            settings.IsZoomControlEnabled = False
            settings.AreDefaultScriptDialogsEnabled = False
        except Exception as exc:
            _log(f'WebView2: settings tweak failed (non-fatal): {exc}')

        self._hook_navigation_events(webview)

        width, height = self._size
        if width > 0 and height > 0:
            self.resize(width, height)

        self._stage = 'ready'
        _log('WebView2: controller ready')

        with self._lock:
            url, self._pending_url = self._pending_url, None
        if url:
            self._navigate(url)

    def _hook_navigation_events(self, webview):
        """Track whether the page actually loaded.

        ``is_showing()`` alone is misleading: it becomes True the moment
        ``Navigate()`` is called, before anything has been fetched. On a closed
        network the weblink host is often unreachable, and Chromium then paints
        an error page — which the player must treat as a failure so the item is
        skipped rather than shown as a broken screen for its whole slot.

        Handlers are stored on the instance: if the delegate were only a local,
        the .NET GC would collect it and the event would silently stop firing.
        """
        try:
            handler = _NavigationCompletedHandler(self)
            webview.NavigationCompleted += handler
            self._navigation_handlers.append(handler)
            _log('WebView2: navigation tracking enabled')
        except Exception as exc:
            # Not fatal: without it we simply cannot distinguish a loaded page
            # from an error page, and fall back to "visible means OK".
            _log(f'WebView2: could not hook NavigationCompleted ({exc})')

    def navigation_succeeded(self):
        """True / False once navigation finished, None while still pending."""
        with self._lock:
            return self._navigation_ok

    def navigation_status(self):
        with self._lock:
            return self._navigation_status

    def _navigate(self, url):
        webview = self._webview
        if webview is None:
            return False
        with self._lock:
            self._navigation_ok = None
            self._navigation_status = ''
        try:
            webview.Navigate(url)
        except Exception as exc:
            self._failed_reason = f'navigate failed: {exc}'
            _log(f'WebView2: navigate failed: {exc}')
            return False
        width, height = self._size
        if width > 0 and height > 0:
            self.resize(width, height)
        try:
            self._controller.IsVisible = True
        except Exception as exc:
            _log(f'WebView2: could not show controller: {exc}')
            return False
        with self._lock:
            self._showing = True
        _log(f'WebView2: navigated to {url[:80]}')
        return True


# ── Module helpers ───────────────────────────────────────────────────
_TYPES_CACHE = {}


def _load_webview2_types(sdk_dir):
    """Import the managed SDK and return the types we need (cached)."""
    key = str(sdk_dir)
    cached = _TYPES_CACHE.get(key)
    if cached:
        return cached

    if hasattr(os, 'add_dll_directory'):
        try:
            os.add_dll_directory(str(sdk_dir))  # let the loader find WebView2Loader.dll
        except Exception:
            pass
    if key not in sys.path:
        sys.path.insert(0, key)

    import clr

    # Framework assemblies we rely on (Rectangle for Bounds).
    try:
        clr.AddReference('System.Drawing')
    except Exception:
        pass
    clr.AddReference(str(sdk_dir / 'Microsoft.Web.WebView2.Core.dll'))

    from Microsoft.Web.WebView2.Core import CoreWebView2Environment

    types = {'env': CoreWebView2Environment}
    _TYPES_CACHE[key] = types
    return types


def _create_environment_async(types, user_data_dir, options=None):
    """Call CreateAsync with the options object.

    The SDK exposes exactly one overload:
    ``CreateAsync(string browserExecutableFolder, string userDataFolder,
    CoreWebView2EnvironmentOptions options)``.
    """
    env_type = types['env']
    last = None
    # Preferred: explicit options (used to pass offline browser arguments).
    if options is not None:
        try:
            return env_type.CreateAsync(None, user_data_dir, options)
        except Exception as exc:
            last = exc
    attempts = (
        (None, user_data_dir, None),
        (None, user_data_dir),
    )
    for args in attempts:
        try:
            return env_type.CreateAsync(*args)
        except Exception as exc:
            last = exc
    raise last if last is not None else RuntimeError('CreateAsync failed')


def _offline_browser_arguments():
    """Chromium flags that stop internet chatter on a closed network.

    A signage player normally lives on an isolated LAN. By default Chromium
    still tries to reach the internet for component updates, field trials,
    safe-browsing lists, translate, and Google services. On a closed network
    every one of those attempts has to time out, which costs start-up latency
    (and, if DNS resolves but routes black-hole, can stall for many seconds).

    These flags disable that background traffic. They do NOT affect loading
    actual pages — a weblink pointing at the local server still works, and one
    pointing at the public internet simply fails fast with a normal
    ERR_INTERNET_DISCONNECTED instead of hanging.
    """
    return ' '.join([
        '--disable-background-networking',
        '--disable-component-update',
        '--disable-domain-reliability',
        '--disable-features=Translate,OptimizationHints,MediaRouter,'
        'CalculateNativeWinOcclusion',
        '--disable-sync',
        '--no-first-run',
        '--no-default-browser-check',
        '--no-pings',
        '--disable-breakpad',
        '--metrics-recording-only',
        '--disable-client-side-phishing-detection',
    ])


def _build_environment_options():
    """Create a CoreWebView2EnvironmentOptions with offline flags applied."""
    try:
        from Microsoft.Web.WebView2.Core import CoreWebView2EnvironmentOptions

        options = CoreWebView2EnvironmentOptions()
        options.AdditionalBrowserArguments = _offline_browser_arguments()
        # Don't phone home with crash reports.
        try:
            options.IsCustomCrashReportingEnabled = False
        except Exception:
            pass
        _log('WebView2: offline browser arguments applied')
        return options
    except Exception as exc:
        _log(f'WebView2: could not build environment options ({exc}); '
             'continuing with defaults')
        return None


class _ThreadTimer:
    """Minimal fallback timer used only when Kivy's Clock is unavailable."""

    def __init__(self, interval, func, _unused):
        self._interval = float(interval)
        self._func = func
        self._stop = threading.Event()
        self._thread = threading.Thread(target=self._run, daemon=True)
        self._thread.start()

    def _run(self):
        while not self._stop.wait(self._interval):
            try:
                if self._func(None) is False:
                    return
            except Exception:
                return

    def cancel(self):
        self._stop.set()


class _NavigationCompletedHandler:
    """Adapter for WebView2's ``NavigationCompleted`` event.

    The event is ``System.EventHandler<CoreWebView2NavigationCompletedEventArgs>``
    — there is no ``CoreWebView2NavigationCompletedEventHandler`` type to import
    (attempting to import one fails). pythonnet converts a plain Python callable
    to the generic delegate automatically, so that is what we pass.

    The callable is kept on the browser instance: a delegate referenced only by
    a local would be collected by the .NET GC, after which the event silently
    stops firing.
    """

    def __init__(self, browser):
        self._browser = browser

    def __call__(self, sender, args):
        """Fires on the WebView2 thread that owns the message loop."""
        try:
            success = bool(args.IsSuccess)
            status = _describe_navigation_error(args, success) if not success else ''
            with self._browser._lock:
                self._browser._navigation_ok = success
                self._browser._navigation_status = status
            if success:
                _log('WebView2: page loaded')
            else:
                _log(f'WebView2: page failed to load ({status or "unknown"})')
        except Exception as exc:
            _log(f'WebView2: navigation handler error ({exc})')


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

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


def _describe_navigation_error(args, success):
    """Human-readable reason for a failed navigation.

    ``WebErrorStatus`` is an enum whose numeric value is not useful on its own;
    when it reports ``Unknown`` (common for connection-level failures) the HTTP
    status is more informative, so prefer whichever actually says something.
    """
    parts = []
    try:
        error_status = str(args.WebErrorStatus)
        if error_status and error_status.lower() != 'unknown':
            parts.append(error_status)
    except Exception:
        pass
    try:
        http_status = int(args.HttpStatusCode)
        if http_status > 0:
            parts.append(f'HTTP {http_status}')
    except Exception:
        pass
    if parts:
        return ', '.join(parts)
    return 'connection failed (host unreachable or DNS failure)'