sorrow404null
156f94e840
Assembly.LoadFrom can fail on volumes .NET misclassifies as remote; fall back to Assembly.Load(byte[]). Broaden exception handling to BaseException for pythonnet, and skip WebView2 warmup/taskkill when the instance lock is held by another process.
418 lines
17 KiB
Python
418 lines
17 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
WebView2 后端（Windows 首选浏览器内核，失败自动回落 QtWebEngine）。
|
||
|
||
设计要点（来自 P1 实验 tests/_tmp_wv2_demo.py 的实测结论）：
|
||
1. pythonnet + WebView2 .NET SDK（vendor/webview2/ 内 net462 Core.dll + webview2loader_x64.dll）
|
||
2. 必须 OleInitialize（STA）后才能 CreateAsync
|
||
3. 本机的 SDK 怪癖：传任何非空 user_data_folder 都报 RuntimeNotFound → 一律用默认 profile（ud=None）
|
||
4. 残留 msedgewebview2.exe 会锁默认 profile（0x800700AA）→ 初始化前 taskkill
|
||
5. .NET 版 CoreWebView2 不暴露子窗口 HWND → EnumChildWindows 找 Chrome_WidgetWin_* 类
|
||
6. 子窗口天然是 Qt 顶层窗口的子 HWND，不需要 QWindow.fromWinId 包装（对子窗口会失败），
|
||
由 Qt 布局算 slot 矩形后用 SetBoundsAndZoomFactor 驱动（父窗客户区物理像素）
|
||
7. JS→Python 用 WebMessageReceived（JSON），Python→JS 用 ExecuteScriptAsync（与现有
|
||
ChatBridge.run_js 生成的 JS 调用文本完全同构，前端零改动；仅 index.html 的
|
||
window.bridge bootstrap 走双协议）
|
||
"""
|
||
import os
|
||
import sys
|
||
import time
|
||
import shutil
|
||
import subprocess
|
||
|
||
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
WV2_DIR = os.path.join(ROOT, "vendor", "webview2")
|
||
CORE_DLL = os.path.join(WV2_DIR, "net462_Microsoft.Web.WebView2.Core.dll")
|
||
LOADER = os.path.join(WV2_DIR, "webview2loader_x64.dll")
|
||
|
||
_env = None
|
||
_System = None # pythonnet 加载后缓存
|
||
_INSTANCE_LOCK = {"fh": None}
|
||
|
||
|
||
def _instance_lock_path():
|
||
# 可覆盖（测试隔离用：HAOCODE_INSTANCE_LOCK_FILE 指向临时文件，
|
||
# 否则测试会与正在运行的 app 争同一把锁 → 断言依赖环境）
|
||
p = os.environ.get("HAOCODE_INSTANCE_LOCK_FILE")
|
||
if p:
|
||
return p
|
||
d = os.path.join(ROOT, "data")
|
||
try:
|
||
os.makedirs(d, exist_ok=True)
|
||
except Exception:
|
||
pass
|
||
return os.path.join(d, "app_instance.lock")
|
||
|
||
|
||
def acquire_instance_lock():
|
||
"""独占 data/app_instance.lock（非阻塞）。
|
||
|
||
返回 True = 本机唯一实例 → 可安全 taskkill 残留 msedgewebview2.exe
|
||
False = 已有实例在跑 → 必须回落 QtWebEngine
|
||
None = 平台不支持/异常（保守当“不确定”处理）
|
||
|
||
🐛 T0 根因：本函数存在之前，每个 MainWindow()（包括 offscreen 测试）都会走到
|
||
_get_environment() 里的 `taskkill /F /IM msedgewebview2.exe`，
|
||
把【当时正在运行的生产 app】的 WebView2 浏览器进程一并杀掉
|
||
→ 它的 controller 变 disposed（set_bounds 报 0x8007139F）
|
||
→ DOM 照渲染但视觉层永久空白（= “选中会话不渲染核心内容”）。
|
||
"""
|
||
if sys.platform != "win32":
|
||
return None
|
||
if _INSTANCE_LOCK["fh"] is not None:
|
||
return True
|
||
fh = None
|
||
try:
|
||
import msvcrt
|
||
fh = open(_instance_lock_path(), "a+b")
|
||
fh.seek(0) # 固定锁位置（锁定前 1 字节），不依赖 append 模式的当前位置
|
||
msvcrt.locking(fh.fileno(), msvcrt.LK_NBLCK, 1)
|
||
_INSTANCE_LOCK["fh"] = fh
|
||
return True
|
||
except Exception:
|
||
try:
|
||
if fh is not None:
|
||
fh.close()
|
||
except Exception:
|
||
pass
|
||
return False
|
||
|
||
|
||
def _wv2_allowed_here() -> bool:
|
||
"""无头/测试环境一律不得碰 WebView2（共享默认 profile，一碰就可能误杀在生产实例）"""
|
||
if os.environ.get("HAOCODE_FORCE_QTWEBENGINE", "") in ("1", "true", "True"):
|
||
return False
|
||
plat = (os.environ.get("QT_QPA_PLATFORM") or "").strip().lower()
|
||
if plat and plat != "windows":
|
||
return False
|
||
return True
|
||
|
||
|
||
def _pump_wait(op, app, timeout=15.0):
|
||
"""在 Qt 事件循环里等 .NET Task 完成（WebView2 初始化依赖消息泵）"""
|
||
t0 = time.time()
|
||
while not op.IsCompleted and time.time() - t0 < timeout:
|
||
app.processEvents()
|
||
time.sleep(0.005)
|
||
if not op.IsCompleted:
|
||
raise TimeoutError("WebView2 operation timed out")
|
||
if op.IsFaulted:
|
||
raise RuntimeError(str(op.Exception))
|
||
return op.Result
|
||
|
||
|
||
def _ts():
|
||
"""启动链时间戳（WebView2 冷启动耗时诊断，保留）"""
|
||
t = time.time()
|
||
return time.strftime("%H:%M:%S", time.localtime(t)) + f".{int(t*1000) % 1000:03d}"
|
||
|
||
|
||
def get_environment(app):
|
||
"""初始化并返回 CoreWebView2Environment 单例；任何失败返回 None（调用方回落 QtWebEngine）"""
|
||
global _env
|
||
if _env is not None:
|
||
return _env
|
||
if sys.platform != "win32":
|
||
return None
|
||
if not (os.path.exists(CORE_DLL) and os.path.exists(LOADER)):
|
||
return None
|
||
# 🛡 守卫 1：无头/测试环境（QT_QPA_PLATFORM=offscreen 等）绝不启用 WebView2
|
||
if not _wv2_allowed_here():
|
||
print(f"[WV2] QT_QPA_PLATFORM={os.environ.get('QT_QPA_PLATFORM')!r} → 跳过 WebView2，"
|
||
f"回落 QtWebEngine（无头环境不得触碰共享 profile）")
|
||
return None
|
||
# 🛡 守卫 2：本机已有实例在跑 → 不启用 WebView2、更不 taskkill（否则会把它的
|
||
# 浏览器进程杀掉 → 对方 controller disposed → 聊天区永久空白）
|
||
try:
|
||
_lock_ok = acquire_instance_lock()
|
||
except Exception:
|
||
_lock_ok = None
|
||
if _lock_ok is not True:
|
||
print("[WV2] 检测到已有 haocode 实例在运行（instance lock 被占）→ "
|
||
"本实例回落 QtWebEngine；已跳过 taskkill，不会影响对方渲染")
|
||
return None
|
||
try:
|
||
os.environ["PATH"] = WV2_DIR + ";" + os.environ.get("PATH", "")
|
||
try:
|
||
os.add_dll_directory(WV2_DIR)
|
||
except Exception:
|
||
pass
|
||
# .NET Core 的 P/Invoke 默认不查 CWD → 把 loader 复制到 CWD 一份
|
||
try:
|
||
cwd_loader = os.path.join(os.getcwd(), "WebView2Loader.dll")
|
||
if not os.path.exists(cwd_loader):
|
||
shutil.copyfile(LOADER, cwd_loader)
|
||
except Exception:
|
||
pass
|
||
import ctypes
|
||
ctypes.windll.ole32.OleInitialize(None) # STA（COM 初始化要求）
|
||
# 清残留浏览器进程（锁默认 profile 会导致 0x800700AA）
|
||
# 🛡 只有【本机唯一实例】才会走到这里（已在上面用 instance lock 保证），
|
||
# 否则会把兄弟实例的浏览器进程杀掉 → 对方控制器 disposed → 聊天区空白
|
||
print(f"[WV2] {_ts()} warmup: taskkill 残留进程（唯一实例，安全）...")
|
||
try:
|
||
subprocess.run(["taskkill", "/F", "/IM", "msedgewebview2.exe"],
|
||
capture_output=True, timeout=10)
|
||
time.sleep(1.0)
|
||
except Exception:
|
||
pass
|
||
import clr
|
||
global _System
|
||
# 🐛 部分机器（本机 D: 卷）.NET Framework 把工程所在本地卷误判为
|
||
# “网络位置”，clr.AddReference（= Assembly.LoadFrom）报 0x80131515。
|
||
# 快路径保持 LoadFrom（正常机器行为不变）；失败回落 Assembly.Load(byte[])，
|
||
# 字节加载不做路径信任分类（本机实测：D: 字节加载 OK，类型解析 OK）。
|
||
try:
|
||
clr.AddReference(CORE_DLL)
|
||
except BaseException as _ex: # pythonnet 的 .NET 异常未必是 Python Exception 子类
|
||
_msg = str(_ex).splitlines()[0][:120] if str(_ex) else type(_ex).__name__
|
||
print(f"[WV2] AddReference 路径加载失败（{_msg}）→ 回落 Assembly.Load(byte[])")
|
||
from System.Reflection import Assembly as _Assembly
|
||
with open(CORE_DLL, "rb") as _f:
|
||
_Assembly.Load(_f.read())
|
||
import System
|
||
_System = System
|
||
from Microsoft.Web.WebView2.Core import CoreWebView2Environment
|
||
# ⚠️ ud=None：本机 SDK 对非空 user_data 路径报 RuntimeNotFound（实测怪癖）
|
||
_env = _pump_wait(CoreWebView2Environment.CreateAsync(None, None, None), app)
|
||
print(f"[WV2] {_ts()} Runtime ready: {_env.BrowserVersionString}")
|
||
return _env
|
||
except BaseException as ex: # pythonnet/.NET 异常未必是 Exception 子类，必须 BaseException
|
||
_env = None
|
||
print(f"[WV2] init failed → fallback to QtWebEngine: {ex}")
|
||
return None
|
||
|
||
|
||
class Wv2Session:
|
||
"""一个 WebView2 实例：controller + 子窗口 + 消息泵 + bounds 驱动"""
|
||
|
||
def __init__(self, env, parent_hwnd, app):
|
||
self.app = app
|
||
print(f"[WV2] {_ts()} controller create begin (hwnd={parent_hwnd:#x})")
|
||
self.controller = _pump_wait(
|
||
env.CreateCoreWebView2ControllerAsync(_System.IntPtr(int(parent_hwnd))), app)
|
||
print(f"[WV2] {_ts()} controller ready")
|
||
self.core = self.controller.CoreWebView2
|
||
try:
|
||
self.core.Settings.AreDefaultContextMenusEnabled = False
|
||
self.core.Settings.IsZoomControlEnabled = False
|
||
self.core.Settings.IsStatusBarEnabled = False
|
||
except Exception:
|
||
pass
|
||
self.child_hwnd = 0
|
||
self.on_message = None # callable(dict) —— main_window 绑定到 ChatBridge
|
||
self.on_load_finished = None # callable() —— WebView2View 绑定 loadFinished 信号
|
||
self._msg_handler = self._make_msg_handler()
|
||
self.core.add_WebMessageReceived(self._msg_handler)
|
||
self._nw_handler = self._make_navigated_handler()
|
||
self.core.add_NavigationCompleted(self._nw_handler)
|
||
# 🆕 真异步 JS 执行队列：单一定时器轮询完成，回调从主线程定时器发出。
|
||
# （旧版 execute_js_async 是阻塞忙等 + processEvents → 从定时器/事件回调里调用
|
||
# 时产生重入嵌套事件循环 → COM 事件分发崩溃 → 启动卡死/拖动时 webview 不刷新）
|
||
self._js_pending = [] # [task, cb, t0]
|
||
from PyQt6.QtCore import QTimer
|
||
self._js_pump = QTimer()
|
||
self._js_pump.setInterval(25)
|
||
self._js_pump.timeout.connect(self._js_pump_tick)
|
||
self._js_pump.start()
|
||
# 🆕 预热：立即导航 about:blank，让 msedgewebview2 进程/GPU 在 UI 构建期间冷启动
|
||
# （实测本机首次真实页面导航需 12-15s，预热后降到 ~1s）
|
||
try:
|
||
print(f"[WV2] {_ts()} warmup: Navigate about:blank")
|
||
self.core.Navigate("about:blank")
|
||
except Exception as ex:
|
||
print("[WV2] warmup navigate error:", ex)
|
||
|
||
# ---------- 事件 ----------
|
||
def _make_msg_handler(self):
|
||
import json
|
||
from System import EventHandler
|
||
from Microsoft.Web.WebView2.Core import CoreWebView2WebMessageReceivedEventArgs
|
||
|
||
def handler(sender, args):
|
||
try:
|
||
data = json.loads(args.WebMessageAsJson)
|
||
if self.on_message:
|
||
self.on_message(data)
|
||
except Exception as ex:
|
||
print("[WV2] message error:", ex)
|
||
return EventHandler[CoreWebView2WebMessageReceivedEventArgs](handler)
|
||
|
||
def _make_navigated_handler(self):
|
||
from System import EventHandler
|
||
from Microsoft.Web.WebView2.Core import CoreWebView2NavigationCompletedEventArgs
|
||
|
||
def handler(sender, args):
|
||
try:
|
||
src = self.core.Source or ""
|
||
except Exception:
|
||
src = "?"
|
||
print(f"[WV2] {_ts()} NavigationCompleted src={src}")
|
||
# 预热页（about:blank）的加载完成不触发 loadFinished（避免 JS 探针空转）
|
||
if src.startswith("about:blank"):
|
||
return
|
||
if self.on_load_finished:
|
||
self.on_load_finished()
|
||
return EventHandler[CoreWebView2NavigationCompletedEventArgs](handler)
|
||
|
||
# ---------- 子窗口发现（.NET 不暴露 HWND，轮询枚举） ----------
|
||
def find_child_once(self):
|
||
"""单次探测；找到返回 hwnd，否则 0（非阻塞，供 UI 线程定时器调）"""
|
||
import ctypes
|
||
import ctypes.wintypes as wt
|
||
u32 = ctypes.windll.user32
|
||
parent = wt.HWND(self.controller.ParentWindow.ToInt64())
|
||
found = []
|
||
|
||
@ctypes.WINFUNCTYPE(wt.BOOL, wt.HWND, wt.LPVOID)
|
||
def cb(h, _):
|
||
buf = ctypes.create_unicode_buffer(256)
|
||
u32.GetClassNameW(h, buf, 256)
|
||
if buf.value.startswith("Chrome_WidgetWin"):
|
||
found.append(h)
|
||
return True
|
||
|
||
u32.EnumChildWindows(parent, cb, None)
|
||
if found:
|
||
self.child_hwnd = int(found[0])
|
||
return self.child_hwnd
|
||
return 0
|
||
|
||
def find_child(self):
|
||
"""同步等待并返回 Chrome_WidgetWin_* 子 HWND（最多 10s）"""
|
||
import ctypes
|
||
import ctypes.wintypes as wt
|
||
u32 = ctypes.windll.user32
|
||
parent = wt.HWND(self.controller.ParentWindow.ToInt64())
|
||
for _ in range(200):
|
||
found = []
|
||
|
||
@ctypes.WINFUNCTYPE(wt.BOOL, wt.HWND, wt.LPVOID)
|
||
def cb(h, _):
|
||
buf = ctypes.create_unicode_buffer(256)
|
||
u32.GetClassNameW(h, buf, 256)
|
||
if buf.value.startswith("Chrome_WidgetWin"):
|
||
found.append(h)
|
||
return True
|
||
|
||
u32.EnumChildWindows(parent, cb, None)
|
||
if found:
|
||
self.child_hwnd = int(found[0])
|
||
return self.child_hwnd
|
||
self.app.processEvents()
|
||
time.sleep(0.05)
|
||
return 0
|
||
|
||
# ---------- 几何 ----------
|
||
def set_visible(self, visible: bool):
|
||
try:
|
||
self.controller.IsVisible = bool(visible)
|
||
except Exception:
|
||
pass
|
||
|
||
def set_bounds(self, left, top, width, height):
|
||
"""父窗客户区物理像素"""
|
||
try:
|
||
from System.Drawing import Rectangle as _Rect
|
||
self.controller.SetBoundsAndZoomFactor(_Rect(int(left), int(top),
|
||
int(width), int(height)), 1.0)
|
||
except Exception as ex:
|
||
print("[WV2] set_bounds error:", ex)
|
||
|
||
def child_size(self):
|
||
"""子窗口当前屏幕像素尺寸（观察用）"""
|
||
if not self.child_hwnd:
|
||
return (0, 0)
|
||
import ctypes
|
||
import ctypes.wintypes as wt
|
||
r = wt.RECT()
|
||
if ctypes.windll.user32.GetWindowRect(wt.HWND(self.child_hwnd), ctypes.byref(r)):
|
||
return (r.right - r.left, r.bottom - r.top)
|
||
return (0, 0)
|
||
|
||
# ---------- JS ----------
|
||
def navigate(self, url: str):
|
||
print(f"[WV2] {_ts()} navigate {url[:80]}")
|
||
try:
|
||
self.core.Navigate(url)
|
||
except Exception as ex:
|
||
print("[WV2] navigate error:", ex)
|
||
|
||
def execute_js(self, script: str):
|
||
"""fire-and-forget（ChatBridge.run_js 的替换，JS 文本完全同构）"""
|
||
self._js_run(script, None)
|
||
|
||
def execute_js_async(self, script: str, cb):
|
||
"""带回调执行：真异步，回调在主线程定时器 tick 中发出（绝不阻塞）"""
|
||
self._js_run(script, cb)
|
||
|
||
def _js_run(self, script: str, cb):
|
||
try:
|
||
task = self.core.ExecuteScriptWithResultAsync(script)
|
||
except Exception as ex:
|
||
print("[WV2] execute_js error:", ex)
|
||
if cb:
|
||
try:
|
||
cb(None)
|
||
except Exception:
|
||
pass
|
||
return
|
||
self._js_pending.append([task, cb, time.time()])
|
||
|
||
def _js_pump_tick(self):
|
||
if not self._js_pending:
|
||
return
|
||
import json
|
||
remaining = []
|
||
for task, cb, t0 in self._js_pending:
|
||
done = False
|
||
try:
|
||
done = task.IsCompleted
|
||
except Exception:
|
||
done = True
|
||
if not done and time.time() - t0 > 10:
|
||
done = True # 10s 安全超时（渲染器死亡时不永久卡队列）
|
||
if not done:
|
||
remaining.append([task, cb, t0])
|
||
continue
|
||
result = None
|
||
try:
|
||
if task.IsCompleted and not task.IsFaulted:
|
||
r = task.Result # CoreWebView2ExecuteScriptResult 包装结构
|
||
if getattr(r, "Succeeded", True):
|
||
s = r.ResultAsJson # JSON 编码字符串（或 null）
|
||
if s:
|
||
result = json.loads(s)
|
||
if isinstance(result, str):
|
||
try:
|
||
result = json.loads(result)
|
||
except Exception:
|
||
pass
|
||
except Exception as ex:
|
||
print("[WV2] js result decode error:", ex)
|
||
if cb:
|
||
try:
|
||
cb(result)
|
||
except Exception as ex:
|
||
print("[WV2] js callback error:", ex)
|
||
self._js_pending = remaining
|
||
|
||
def close(self):
|
||
try:
|
||
self._js_pump.stop()
|
||
except Exception:
|
||
pass
|
||
self._js_pending = []
|
||
try:
|
||
self.core.remove_WebMessageReceived(self._msg_handler)
|
||
except Exception:
|
||
pass
|
||
try:
|
||
self.core.remove_NavigationCompleted(self._nw_handler)
|
||
except Exception:
|
||
pass
|
||
try:
|
||
self.controller.Close()
|
||
except Exception:
|
||
pass
|