clan-app: renamed deps folder to 'backend'
This commit is contained in:
Qubasa
0
pkgs/clan-app/clan_app/backends/webview/__init__.py
Normal file
0
pkgs/clan-app/clan_app/backends/webview/__init__.py
Normal file
126
pkgs/clan-app/clan_app/backends/webview/_webview_ffi.py
Normal file
126
pkgs/clan-app/clan_app/backends/webview/_webview_ffi.py
Normal file
@@ -0,0 +1,126 @@
|
||||
import ctypes
|
||||
import ctypes.util
|
||||
import os
|
||||
import platform
|
||||
from ctypes import CFUNCTYPE, c_char_p, c_int, c_void_p
|
||||
from pathlib import Path
|
||||
|
||||
# Native handle kinds
|
||||
WEBVIEW_NATIVE_HANDLE_KIND_UI_WINDOW = 0
|
||||
WEBVIEW_NATIVE_HANDLE_KIND_UI_WIDGET = 1
|
||||
WEBVIEW_NATIVE_HANDLE_KIND_BROWSER_CONTROLLER = 2
|
||||
|
||||
|
||||
def _encode_c_string(s: str) -> bytes:
|
||||
return s.encode("utf-8")
|
||||
|
||||
|
||||
def _get_webview_version() -> str:
|
||||
"""Get webview version from environment variable or use default"""
|
||||
return os.getenv("WEBVIEW_VERSION", "0.8.1")
|
||||
|
||||
|
||||
def _get_lib_names() -> list[str]:
|
||||
"""Get platform-specific library names."""
|
||||
system = platform.system().lower()
|
||||
machine = platform.machine().lower()
|
||||
|
||||
if system == "windows":
|
||||
if machine in {"amd64", "x86_64"}:
|
||||
return ["webview.dll", "WebView2Loader.dll"]
|
||||
if machine == "arm64":
|
||||
msg = "arm64 is not supported on Windows"
|
||||
raise RuntimeError(msg)
|
||||
msg = f"Unsupported architecture: {machine}"
|
||||
raise RuntimeError(msg)
|
||||
if system == "darwin":
|
||||
return ["libwebview.dylib"]
|
||||
# linux
|
||||
return ["libwebview.so"]
|
||||
|
||||
|
||||
def _be_sure_libraries() -> list[Path] | None:
|
||||
"""Ensure libraries exist and return paths."""
|
||||
lib_dir = os.environ.get("WEBVIEW_LIB_DIR")
|
||||
if not lib_dir:
|
||||
msg = "WEBVIEW_LIB_DIR environment variable is not set"
|
||||
raise RuntimeError(msg)
|
||||
lib_dir_p = Path(lib_dir)
|
||||
lib_names = _get_lib_names()
|
||||
lib_paths = [lib_dir_p / lib_name for lib_name in lib_names]
|
||||
|
||||
# Check if any library is missing
|
||||
missing_libs = [path for path in lib_paths if not path.exists()]
|
||||
if not missing_libs:
|
||||
return lib_paths
|
||||
return None
|
||||
|
||||
|
||||
class _WebviewLibrary:
|
||||
def __init__(self) -> None:
|
||||
lib_names = _get_lib_names()
|
||||
|
||||
library_path = ctypes.util.find_library(lib_names[0])
|
||||
if not library_path:
|
||||
library_paths = _be_sure_libraries()
|
||||
if not library_paths:
|
||||
msg = f"Failed to find required library: {lib_names}"
|
||||
raise RuntimeError(msg)
|
||||
try:
|
||||
self.lib = ctypes.cdll.LoadLibrary(str(library_paths[0]))
|
||||
except Exception as e:
|
||||
print(f"Failed to load webview library: {e}")
|
||||
raise
|
||||
|
||||
# Define FFI functions
|
||||
self.webview_create = self.lib.webview_create
|
||||
self.webview_create.argtypes = [c_int, c_void_p]
|
||||
self.webview_create.restype = c_void_p
|
||||
|
||||
self.webview_create_with_app_id = self.lib.webview_create_with_app_id
|
||||
self.webview_create_with_app_id.argtypes = [c_int, c_void_p, c_char_p]
|
||||
self.webview_create_with_app_id.restype = c_void_p
|
||||
|
||||
self.webview_destroy = self.lib.webview_destroy
|
||||
self.webview_destroy.argtypes = [c_void_p]
|
||||
|
||||
self.webview_run = self.lib.webview_run
|
||||
self.webview_run.argtypes = [c_void_p]
|
||||
|
||||
self.webview_terminate = self.lib.webview_terminate
|
||||
self.webview_terminate.argtypes = [c_void_p]
|
||||
|
||||
self.webview_set_title = self.lib.webview_set_title
|
||||
self.webview_set_title.argtypes = [c_void_p, c_char_p]
|
||||
|
||||
self.webview_set_size = self.lib.webview_set_size
|
||||
self.webview_set_size.argtypes = [c_void_p, c_int, c_int, c_int]
|
||||
|
||||
self.webview_navigate = self.lib.webview_navigate
|
||||
self.webview_navigate.argtypes = [c_void_p, c_char_p]
|
||||
|
||||
self.webview_init = self.lib.webview_init
|
||||
self.webview_init.argtypes = [c_void_p, c_char_p]
|
||||
|
||||
self.webview_eval = self.lib.webview_eval
|
||||
self.webview_eval.argtypes = [c_void_p, c_char_p]
|
||||
|
||||
self.webview_bind = self.lib.webview_bind
|
||||
self.webview_bind.argtypes = [c_void_p, c_char_p, c_void_p, c_void_p]
|
||||
|
||||
self.webview_unbind = self.lib.webview_unbind
|
||||
self.webview_unbind.argtypes = [c_void_p, c_char_p]
|
||||
|
||||
self.webview_return = self.lib.webview_return
|
||||
self.webview_return.argtypes = [c_void_p, c_char_p, c_int, c_char_p]
|
||||
|
||||
self.webview_get_native_handle = self.lib.webview_get_native_handle
|
||||
self.webview_get_native_handle.argtypes = [c_void_p, c_int]
|
||||
self.webview_get_native_handle.restype = c_void_p
|
||||
|
||||
self.binding_callback_t = CFUNCTYPE(None, c_char_p, c_char_p, c_void_p)
|
||||
|
||||
self.CFUNCTYPE = CFUNCTYPE
|
||||
|
||||
|
||||
_webview_lib = _WebviewLibrary()
|
||||
300
pkgs/clan-app/clan_app/backends/webview/webview.py
Normal file
300
pkgs/clan-app/clan_app/backends/webview/webview.py
Normal file
@@ -0,0 +1,300 @@
|
||||
import functools
|
||||
import json
|
||||
import logging
|
||||
import platform
|
||||
import threading
|
||||
from collections.abc import Callable
|
||||
from dataclasses import dataclass, field
|
||||
from enum import IntEnum
|
||||
from time import sleep
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
from clan_lib.api import MethodRegistry, message_queue
|
||||
from clan_lib.api.tasks import WebThread
|
||||
|
||||
from ._webview_ffi import (
|
||||
_encode_c_string,
|
||||
_webview_lib,
|
||||
)
|
||||
from .webview_bridge import WebviewBridge
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from clan_app.middleware.base import Middleware
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class SizeHint(IntEnum):
|
||||
NONE = 0
|
||||
MIN = 1
|
||||
MAX = 2
|
||||
FIXED = 3
|
||||
|
||||
|
||||
class FuncStatus(IntEnum):
|
||||
SUCCESS = 0
|
||||
FAILURE = 1
|
||||
|
||||
|
||||
class NativeHandleKind(IntEnum):
|
||||
# Top-level window. @c GtkWindow pointer (GTK), @c NSWindow pointer (Cocoa)
|
||||
# or @c HWND (Win32)
|
||||
UI_WINDOW = 0
|
||||
|
||||
# Browser widget. @c GtkWidget pointer (GTK), @c NSView pointer (Cocoa) or
|
||||
# @c HWND (Win32).
|
||||
UI_WIDGET = 1
|
||||
|
||||
# Browser controller. @c WebKitWebView pointer (WebKitGTK), @c WKWebView
|
||||
# pointer (Cocoa/WebKit) or @c ICoreWebView2Controller pointer
|
||||
# (Win32/WebView2).
|
||||
BROWSER_CONTROLLER = 2
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Size:
|
||||
width: int
|
||||
height: int
|
||||
hint: SizeHint
|
||||
|
||||
|
||||
@dataclass
|
||||
class Webview:
|
||||
title: str
|
||||
debug: bool = False
|
||||
size: Size | None = None
|
||||
window: int | None = None
|
||||
shared_threads: dict[str, WebThread] | None = None
|
||||
app_id: str | None = None
|
||||
|
||||
# initialized later
|
||||
_bridge: WebviewBridge | None = None
|
||||
_handle: Any | None = None
|
||||
__callbacks: dict[str, Callable[..., Any]] = field(default_factory=dict)
|
||||
_middleware: list["Middleware"] = field(default_factory=list)
|
||||
|
||||
@property
|
||||
def callbacks(self) -> dict[str, Callable[..., Any]]:
|
||||
return self.__callbacks
|
||||
|
||||
@callbacks.setter
|
||||
def callbacks(self, value: dict[str, Callable[..., Any]]) -> None:
|
||||
del value # Unused
|
||||
msg = "Cannot set callbacks directly"
|
||||
raise AttributeError(msg)
|
||||
|
||||
def delete_callback(self, name: str) -> None:
|
||||
if name in self.callbacks:
|
||||
del self.__callbacks[name]
|
||||
else:
|
||||
msg = f"Callback {name} does not exist. Cannot delete."
|
||||
raise RuntimeError(msg)
|
||||
|
||||
def add_callback(self, name: str, callback: Callable[..., Any]) -> None:
|
||||
if name in self.callbacks:
|
||||
msg = f"Callback {name} already exists. Cannot add."
|
||||
raise RuntimeError(msg)
|
||||
self.__callbacks[name] = callback
|
||||
|
||||
def _create_handle(self) -> None:
|
||||
# Initialize the webview handle
|
||||
with_debugger = True
|
||||
|
||||
# Use webview_create_with_app_id only on Linux if app_id is provided
|
||||
if self.app_id and platform.system() == "Linux":
|
||||
handle = _webview_lib.webview_create_with_app_id(
|
||||
int(with_debugger), self.window, _encode_c_string(self.app_id)
|
||||
)
|
||||
else:
|
||||
handle = _webview_lib.webview_create(int(with_debugger), self.window)
|
||||
|
||||
# Since we can't use object.__setattr__, we'll initialize differently
|
||||
# by storing in __dict__ directly (this works for init=False fields)
|
||||
self._handle = handle
|
||||
|
||||
if self.title:
|
||||
self.set_title(self.title)
|
||||
|
||||
if self.size:
|
||||
self.set_size(self.size)
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
self.setup_notify() # Start the notification loop
|
||||
|
||||
def setup_notify(self) -> None:
|
||||
def loop() -> None:
|
||||
while True:
|
||||
try:
|
||||
msg = message_queue.get() # Blocks until available
|
||||
js_code = f"window.notifyBus({json.dumps(msg)});"
|
||||
self.eval(js_code)
|
||||
except (json.JSONDecodeError, RuntimeError, AttributeError) as e:
|
||||
print("Bridge notify error:", e)
|
||||
sleep(0.01) # avoid busy loop
|
||||
|
||||
threading.Thread(target=loop, daemon=True).start()
|
||||
|
||||
@property
|
||||
def handle(self) -> Any:
|
||||
"""Get the webview handle, creating it if necessary."""
|
||||
if self._handle is None:
|
||||
self._create_handle()
|
||||
return self._handle
|
||||
|
||||
@property
|
||||
def bridge(self) -> "WebviewBridge":
|
||||
"""Get the bridge, creating it if necessary."""
|
||||
if self._bridge is None:
|
||||
self.create_bridge()
|
||||
if self._bridge is None:
|
||||
msg = "Bridge should be created"
|
||||
raise RuntimeError(msg)
|
||||
return self._bridge
|
||||
|
||||
def api_wrapper(
|
||||
self,
|
||||
method_name: str,
|
||||
op_key_bytes: bytes,
|
||||
request_data: bytes,
|
||||
arg: int,
|
||||
) -> None:
|
||||
"""Legacy API wrapper - delegates to the bridge."""
|
||||
del arg # Unused but required for C callback signature
|
||||
self.bridge.handle_webview_call(
|
||||
method_name=method_name,
|
||||
op_key_bytes=op_key_bytes,
|
||||
request_data=request_data,
|
||||
)
|
||||
|
||||
@property
|
||||
def threads(self) -> dict[str, WebThread]:
|
||||
"""Access threads from the bridge for compatibility."""
|
||||
return self.bridge.threads
|
||||
|
||||
def add_middleware(self, middleware: "Middleware") -> None:
|
||||
"""Add middleware to the middleware chain."""
|
||||
if self._bridge is not None:
|
||||
msg = "Cannot add middleware after bridge creation."
|
||||
raise RuntimeError(msg)
|
||||
|
||||
self._middleware.append(middleware)
|
||||
|
||||
def create_bridge(self) -> WebviewBridge:
|
||||
"""Create and initialize the WebviewBridge with current middleware."""
|
||||
# Use shared_threads if provided, otherwise let WebviewBridge use its default
|
||||
if self.shared_threads is not None:
|
||||
bridge = WebviewBridge(
|
||||
webview=self,
|
||||
middleware_chain=tuple(self._middleware),
|
||||
threads=self.shared_threads,
|
||||
)
|
||||
else:
|
||||
bridge = WebviewBridge(
|
||||
webview=self,
|
||||
middleware_chain=tuple(self._middleware),
|
||||
threads={},
|
||||
)
|
||||
self._bridge = bridge
|
||||
|
||||
return bridge
|
||||
|
||||
# Legacy methods for compatibility
|
||||
def set_size(self, value: Size) -> None:
|
||||
"""Set the webview size (legacy compatibility)."""
|
||||
_webview_lib.webview_set_size(
|
||||
self.handle,
|
||||
value.width,
|
||||
value.height,
|
||||
value.hint,
|
||||
)
|
||||
|
||||
def set_title(self, value: str) -> None:
|
||||
"""Set the webview title (legacy compatibility)."""
|
||||
_webview_lib.webview_set_title(self.handle, _encode_c_string(value))
|
||||
|
||||
def destroy(self) -> None:
|
||||
"""Destroy the webview."""
|
||||
for name in list(self.callbacks.keys()):
|
||||
self.unbind(name)
|
||||
_webview_lib.webview_terminate(self.handle)
|
||||
_webview_lib.webview_destroy(self.handle)
|
||||
# Can't set _handle to None on frozen dataclass
|
||||
|
||||
def navigate(self, url: str) -> None:
|
||||
"""Navigate to a URL."""
|
||||
_webview_lib.webview_navigate(self.handle, _encode_c_string(url))
|
||||
|
||||
def run(self) -> None:
|
||||
"""Run the webview."""
|
||||
_webview_lib.webview_run(self.handle)
|
||||
log.info("Shutting down webview...")
|
||||
self.destroy()
|
||||
|
||||
def bind_jsonschema_api(self, api: MethodRegistry) -> None:
|
||||
for name in api.functions:
|
||||
wrapper = functools.partial(
|
||||
self.api_wrapper,
|
||||
name,
|
||||
)
|
||||
c_callback = _webview_lib.binding_callback_t(wrapper)
|
||||
|
||||
self.add_callback(name, c_callback)
|
||||
|
||||
_webview_lib.webview_bind(
|
||||
self.handle,
|
||||
_encode_c_string(name),
|
||||
c_callback,
|
||||
None,
|
||||
)
|
||||
|
||||
def bind(self, name: str, callback: Callable[..., Any]) -> None:
|
||||
def wrapper(seq: bytes, req: bytes, _arg: int) -> None:
|
||||
args = json.loads(req.decode())
|
||||
try:
|
||||
result = callback(*args)
|
||||
success = True
|
||||
except Exception as e: # noqa: BLE001
|
||||
result = str(e)
|
||||
success = False
|
||||
self.return_(seq.decode(), 0 if success else 1, json.dumps(result))
|
||||
|
||||
c_callback = _webview_lib.binding_callback_t(wrapper)
|
||||
self.add_callback(name, c_callback)
|
||||
_webview_lib.webview_bind(self.handle, _encode_c_string(name), c_callback, None)
|
||||
|
||||
def get_native_handle(
|
||||
self, kind: NativeHandleKind = NativeHandleKind.UI_WINDOW
|
||||
) -> int | None:
|
||||
"""Get the native handle (platform-dependent).
|
||||
|
||||
Args:
|
||||
kind: Handle kind - NativeHandleKind enum value
|
||||
|
||||
Returns:
|
||||
Native handle as integer, or None if failed
|
||||
|
||||
"""
|
||||
handle = _webview_lib.webview_get_native_handle(self.handle, kind.value)
|
||||
return handle if handle else None
|
||||
|
||||
def unbind(self, name: str) -> None:
|
||||
if name in self.callbacks:
|
||||
_webview_lib.webview_unbind(self.handle, _encode_c_string(name))
|
||||
self.delete_callback(name)
|
||||
|
||||
def return_(self, seq: str, status: int, result: str) -> None:
|
||||
_webview_lib.webview_return(
|
||||
self.handle,
|
||||
_encode_c_string(seq),
|
||||
status,
|
||||
_encode_c_string(result),
|
||||
)
|
||||
|
||||
def eval(self, source: str) -> None:
|
||||
_webview_lib.webview_eval(self.handle, _encode_c_string(source))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
wv = Webview(title="Hello, World!")
|
||||
wv.navigate("https://www.google.com")
|
||||
wv.run()
|
||||
94
pkgs/clan-app/clan_app/backends/webview/webview_bridge.py
Normal file
94
pkgs/clan-app/clan_app/backends/webview/webview_bridge.py
Normal file
@@ -0,0 +1,94 @@
|
||||
import json
|
||||
import logging
|
||||
from dataclasses import dataclass, field
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from clan_lib.api import dataclass_to_dict
|
||||
from clan_lib.api.tasks import WebThread
|
||||
|
||||
from clan_app.api.api_bridge import ApiBridge, BackendRequest, BackendResponse
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from clan_app.middleware.base import Middleware
|
||||
|
||||
from .webview import Webview
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@dataclass
|
||||
class WebviewBridge(ApiBridge):
|
||||
"""Webview-specific implementation of the API bridge."""
|
||||
|
||||
webview: "Webview"
|
||||
middleware_chain: tuple["Middleware", ...]
|
||||
threads: dict[str, WebThread] = field(default_factory=dict)
|
||||
|
||||
def send_api_response(self, response: BackendResponse) -> None:
|
||||
"""Send response back to the webview client."""
|
||||
serialized = json.dumps(
|
||||
dataclass_to_dict(response),
|
||||
indent=4,
|
||||
ensure_ascii=False,
|
||||
)
|
||||
|
||||
log.debug(f"Sending response: {serialized}")
|
||||
# Import FuncStatus locally to avoid circular import
|
||||
from .webview import FuncStatus # noqa: PLC0415
|
||||
|
||||
self.webview.return_(response._op_key, FuncStatus.SUCCESS, serialized) # noqa: SLF001
|
||||
|
||||
def handle_webview_call(
|
||||
self,
|
||||
method_name: str,
|
||||
op_key_bytes: bytes,
|
||||
request_data: bytes,
|
||||
) -> None:
|
||||
"""Handle a call from webview's JavaScript bridge."""
|
||||
try:
|
||||
webview_op_key = op_key_bytes.decode()
|
||||
raw_args = json.loads(request_data.decode())
|
||||
# Parse the webview-specific request format
|
||||
header = {}
|
||||
args = {}
|
||||
if len(raw_args) == 1:
|
||||
request = raw_args[0]
|
||||
header = request.get("header", {})
|
||||
if not isinstance(header, dict):
|
||||
msg = f"Expected header to be a dict, got {type(header)}"
|
||||
raise TypeError(msg) # noqa: TRY301
|
||||
|
||||
body = request.get("body", {})
|
||||
if not isinstance(body, dict):
|
||||
msg = f"Expected body to be a dict, got {type(body)}"
|
||||
raise TypeError(msg) # noqa: TRY301
|
||||
|
||||
args = body
|
||||
elif len(raw_args) > 1:
|
||||
msg = "Expected a single argument, got multiple arguments"
|
||||
raise ValueError(msg) # noqa: TRY301
|
||||
|
||||
# Create API request
|
||||
api_request = BackendRequest(
|
||||
method_name=method_name,
|
||||
args=args,
|
||||
header=header,
|
||||
op_key=webview_op_key,
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
msg = f"Error while handling webview call {method_name} with op_key {webview_op_key}"
|
||||
log.exception(msg)
|
||||
self.send_api_error_response(
|
||||
webview_op_key,
|
||||
str(e),
|
||||
["webview_bridge", method_name],
|
||||
)
|
||||
return
|
||||
|
||||
# Process in a separate thread using the inherited method
|
||||
self.process_request_in_thread(
|
||||
api_request,
|
||||
thread_name=f"WebviewThread-{method_name}",
|
||||
wait_for_completion=False,
|
||||
)
|
||||
Reference in New Issue
Block a user