226 lines
7.6 KiB
Python
226 lines
7.6 KiB
Python
"""
|
||
imgui_webview.py
|
||
后台 playwright 无头浏览器 + 截图 → Panda3D 纹理，供 ImGui 面板显示。
|
||
"""
|
||
from __future__ import annotations
|
||
import threading
|
||
import io
|
||
import time
|
||
|
||
|
||
class ImGuiWebView:
|
||
"""
|
||
后台线程运行 playwright Chromium，定期截图并转换为 Panda3D 纹理。
|
||
ImGui 直接用 imgui.image() 显示纹理，鼠标/滚轮事件转发给浏览器。
|
||
"""
|
||
|
||
def __init__(self, width: int = 1280, height: int = 720):
|
||
self.view_width = width
|
||
self.view_height = height
|
||
|
||
# 截图数据（bytes）
|
||
self._screenshot: bytes | None = None
|
||
self._screenshot_lock = threading.Lock()
|
||
self.tex_dirty = False # 有新截图待上传 GPU
|
||
|
||
# 状态
|
||
self.current_url = ""
|
||
self.title = ""
|
||
self.is_loading = False
|
||
self.error: str | None = None
|
||
|
||
# 待处理指令（由 ImGui 线程写，浏览器线程读）
|
||
self._cmd_navigate: str | None = None
|
||
self._cmd_click: tuple | None = None # (x_ratio, y_ratio)
|
||
self._cmd_scroll: float | None = None # pixels
|
||
self._cmd_back = False
|
||
self._cmd_forward = False
|
||
self._cmd_reload = False
|
||
self._lock = threading.Lock()
|
||
|
||
self._running = False
|
||
self._thread: threading.Thread | None = None
|
||
|
||
# ------------------------------------------------------------------ #
|
||
# 公开控制 API（由 ImGui 线程调用，线程安全）
|
||
# ------------------------------------------------------------------ #
|
||
|
||
def start(self, url: str):
|
||
if self._running:
|
||
return
|
||
self._running = True
|
||
self._cmd_navigate = url
|
||
self._thread = threading.Thread(target=self._run, daemon=True,
|
||
name="imgui-webview")
|
||
self._thread.start()
|
||
|
||
def stop(self):
|
||
self._running = False
|
||
|
||
def navigate(self, url: str):
|
||
if not url.startswith(('http://', 'https://', 'file://')):
|
||
url = 'https://' + url
|
||
with self._lock:
|
||
self._cmd_navigate = url
|
||
self.is_loading = True
|
||
|
||
def click(self, x_ratio: float, y_ratio: float):
|
||
with self._lock:
|
||
self._cmd_click = (x_ratio, y_ratio)
|
||
|
||
def scroll(self, delta_px: float):
|
||
with self._lock:
|
||
self._cmd_scroll = delta_px
|
||
|
||
def go_back(self):
|
||
with self._lock:
|
||
self._cmd_back = True
|
||
|
||
def go_forward(self):
|
||
with self._lock:
|
||
self._cmd_forward = True
|
||
|
||
def reload(self):
|
||
with self._lock:
|
||
self._cmd_reload = True
|
||
|
||
def get_screenshot_bytes(self) -> bytes | None:
|
||
with self._screenshot_lock:
|
||
return self._screenshot
|
||
|
||
# ------------------------------------------------------------------ #
|
||
# 内部线程
|
||
# ------------------------------------------------------------------ #
|
||
|
||
def _run(self):
|
||
import asyncio
|
||
loop = asyncio.new_event_loop()
|
||
asyncio.set_event_loop(loop)
|
||
try:
|
||
loop.run_until_complete(self._async_run())
|
||
except Exception as exc:
|
||
self.error = f"WebView 线程异常: {exc}"
|
||
import traceback; traceback.print_exc()
|
||
finally:
|
||
loop.close()
|
||
|
||
async def _async_run(self):
|
||
try:
|
||
from playwright.async_api import async_playwright
|
||
except ImportError:
|
||
self.error = (
|
||
"playwright 未安装。\n"
|
||
"请运行: pip install playwright\n"
|
||
"然后运行: playwright install chromium"
|
||
)
|
||
self._running = False
|
||
return
|
||
|
||
try:
|
||
async with async_playwright() as pw:
|
||
browser = await pw.chromium.launch(headless=True)
|
||
ctx = await browser.new_context(
|
||
viewport={"width": self.view_width,
|
||
"height": self.view_height},
|
||
user_agent=(
|
||
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
|
||
"AppleWebKit/537.36 (KHTML, like Gecko) "
|
||
"Chrome/121.0.0.0 Safari/537.36"
|
||
),
|
||
)
|
||
page = await ctx.new_page()
|
||
|
||
# 初次导航
|
||
start_url = self._cmd_navigate or "about:blank"
|
||
self._cmd_navigate = None
|
||
await self._goto(page, start_url)
|
||
await self._snap(page)
|
||
|
||
# 主循环
|
||
import asyncio
|
||
while self._running:
|
||
await asyncio.sleep(0.05)
|
||
|
||
with self._lock:
|
||
nav_url = self._cmd_navigate; self._cmd_navigate = None
|
||
clk = self._cmd_click; self._cmd_click = None
|
||
scr = self._cmd_scroll; self._cmd_scroll = None
|
||
do_back = self._cmd_back; self._cmd_back = False
|
||
do_fwd = self._cmd_forward; self._cmd_forward = False
|
||
do_reload = self._cmd_reload; self._cmd_reload = False
|
||
|
||
changed = False
|
||
|
||
if nav_url:
|
||
self.is_loading = True
|
||
await self._goto(page, nav_url)
|
||
changed = True
|
||
self.is_loading = False
|
||
|
||
if do_back:
|
||
await page.go_back()
|
||
await asyncio.sleep(0.5)
|
||
changed = True
|
||
|
||
if do_fwd:
|
||
await page.go_forward()
|
||
await asyncio.sleep(0.5)
|
||
changed = True
|
||
|
||
if do_reload:
|
||
await page.reload(wait_until="domcontentloaded")
|
||
changed = True
|
||
|
||
if clk is not None:
|
||
xr, yr = clk
|
||
x = int(xr * self.view_width)
|
||
y = int(yr * self.view_height)
|
||
await page.mouse.click(x, y)
|
||
await asyncio.sleep(0.4)
|
||
changed = True
|
||
|
||
if scr is not None:
|
||
await page.evaluate(
|
||
f"window.scrollBy(0, {int(scr)})"
|
||
)
|
||
await asyncio.sleep(0.15)
|
||
changed = True
|
||
|
||
if changed:
|
||
self.current_url = page.url
|
||
try:
|
||
self.title = await page.title()
|
||
except Exception:
|
||
pass
|
||
await self._snap(page)
|
||
|
||
await browser.close()
|
||
|
||
except Exception as exc:
|
||
self.error = str(exc)
|
||
import traceback; traceback.print_exc()
|
||
finally:
|
||
self._running = False
|
||
|
||
async def _goto(self, page, url: str):
|
||
import asyncio
|
||
try:
|
||
await page.goto(url, wait_until="domcontentloaded", timeout=20_000)
|
||
self.current_url = page.url
|
||
try:
|
||
self.title = await page.title()
|
||
except Exception:
|
||
pass
|
||
except Exception as exc:
|
||
print(f"[WebView] 导航失败 {url}: {exc}")
|
||
|
||
async def _snap(self, page):
|
||
"""截图并更新 self._screenshot"""
|
||
try:
|
||
data = await page.screenshot(type="png", full_page=False)
|
||
with self._screenshot_lock:
|
||
self._screenshot = data
|
||
self.tex_dirty = True
|
||
except Exception as exc:
|
||
print(f"[WebView] 截图失败: {exc}")
|