修改浏览器方案为同步方案

This commit is contained in:
zk
2026-07-24 16:56:57 +08:00
parent fc74c9f2b0
commit 39e162ae0f
+68 -48
View File
@@ -4,18 +4,21 @@
- open_page(url, wait_ms): 打开页面,返回 page_id - open_page(url, wait_ms): 打开页面,返回 page_id
- query(page_id, selector): 选中节点并提取文字、图片 URL 与属性 - query(page_id, selector): 选中节点并提取文字、图片 URL 与属性
实现上内部常驻一个 Browser;每次 open_page 创建独立 BrowserContext 和 Page 实现上内部常驻一个 Browser;每次 open_page 创建独立 BrowserContext 和 Page
query 完成后自动释放对应资源 Playwright 只在一个后台线程里运行,但外层 API 保持同步
""" """
from __future__ import annotations from __future__ import annotations
import asyncio import atexit
import queue
import threading import threading
import uuid import uuid
from concurrent.futures import Future
from dataclasses import dataclass from dataclasses import dataclass
from typing import Any, Callable
from playwright.async_api import Browser, BrowserContext, Page, async_playwright from playwright.sync_api import Browser, BrowserContext, Page, sync_playwright
@dataclass @dataclass
@@ -35,41 +38,59 @@ class _PageSession:
class _BrowserRuntime: class _BrowserRuntime:
"""后台事件循环中管理 Playwright 生命周期。""" """单独线程中管理 Playwright 生命周期。"""
def __init__(self) -> None: def __init__(self) -> None:
self._thread: threading.Thread | None = None self._thread: threading.Thread | None = None
self._loop: asyncio.AbstractEventLoop | None = None
self._ready = threading.Event() self._ready = threading.Event()
self._closed = threading.Event()
self._job_queue: queue.Queue[
tuple[Callable[..., Any], tuple[Any, ...], dict[str, Any], Future[Any]]
] = queue.Queue()
self._lock = threading.RLock() self._lock = threading.RLock()
self._playwright = None self._playwright = None
self._browser: Browser | None = None self._browser: Browser | None = None
self._pages: dict[str, _PageSession] = {} self._pages: dict[str, _PageSession] = {}
atexit.register(self.close)
def _thread_main(self) -> None: def _thread_main(self) -> None:
loop = asyncio.new_event_loop() with sync_playwright() as playwright:
asyncio.set_event_loop(loop) self._playwright = playwright
self._loop = loop self._browser = playwright.chromium.launch(headless=True)
self._ready.set() self._ready.set()
loop.run_forever()
pending = asyncio.all_tasks(loop) while True:
for task in pending: job = self._job_queue.get()
task.cancel() if job is None:
break
if pending: fn, args, kwargs, fut = job
loop.run_until_complete(asyncio.gather(*pending, return_exceptions=True)) try:
result = fn(*args, **kwargs)
except BaseException as exc: # pragma: no cover - propagate failures
fut.set_exception(exc)
else:
fut.set_result(result)
for session in self._pages.values():
try:
session.page.close()
except Exception:
pass
try:
session.context.close()
except Exception:
pass
self._pages.clear()
if self._browser is not None: if self._browser is not None:
loop.run_until_complete(self._browser.close()) try:
self._browser.close()
finally:
self._browser = None self._browser = None
if self._playwright is not None:
loop.run_until_complete(self._playwright.stop())
self._playwright = None self._playwright = None
self._closed.set()
loop.run_until_complete(loop.shutdown_asyncgens())
loop.close()
def _ensure_thread(self) -> None: def _ensure_thread(self) -> None:
with self._lock: with self._lock:
@@ -84,42 +105,45 @@ class _BrowserRuntime:
self._thread.start() self._thread.start()
self._ready.wait() self._ready.wait()
def _run(self, coro): def _run(self, fn: Callable[..., Any], *args: Any, **kwargs: Any) -> Any:
self._ensure_thread() self._ensure_thread()
assert self._loop is not None fut: Future[Any] = Future()
future = asyncio.run_coroutine_threadsafe(coro, self._loop) self._job_queue.put((fn, args, kwargs, fut))
return future.result() return fut.result()
async def _ensure_browser(self) -> Browser: def close(self) -> None:
if self._browser is not None: with self._lock:
if self._closed.is_set():
return
if self._thread is None:
return
self._job_queue.put(None)
self._closed.wait(timeout=10)
def _ensure_browser(self) -> Browser:
if self._browser is None:
raise RuntimeError("Browser runtime is not ready.")
return self._browser return self._browser
if self._playwright is None: def _open_page_impl(self, url: str, wait_ms: int) -> str:
self._playwright = await async_playwright().start() browser = self._ensure_browser()
context = browser.new_context()
self._browser = await self._playwright.chromium.launch(headless=True) page = context.new_page()
return self._browser page.goto(url, wait_until="domcontentloaded", timeout=30000)
async def _open_page(self, url: str, wait_ms: int) -> str:
browser = await self._ensure_browser()
context = await browser.new_context()
page = await context.new_page()
await page.goto(url, wait_until="domcontentloaded", timeout=30000)
if wait_ms > 0: if wait_ms > 0:
await page.wait_for_timeout(wait_ms) page.wait_for_timeout(wait_ms)
page_id = uuid.uuid4().hex page_id = uuid.uuid4().hex
self._pages[page_id] = _PageSession(context=context, page=page) self._pages[page_id] = _PageSession(context=context, page=page)
return page_id return page_id
async def _query(self, page_id: str, selector: str) -> list[NodeSnapshot]: def _query_impl(self, page_id: str, selector: str) -> list[NodeSnapshot]:
session = self._pages.get(page_id) session = self._pages.get(page_id)
if session is None: if session is None:
raise KeyError(f"Unknown page_id: {page_id}") raise KeyError(f"Unknown page_id: {page_id}")
try:
locator = session.page.locator(selector) locator = session.page.locator(selector)
payload = await locator.evaluate_all( payload = locator.evaluate_all(
r"""(elements) => elements.map((el) => { r"""(elements) => elements.map((el) => {
const attrs = {}; const attrs = {};
for (const attr of Array.from(el.attributes || [])) { for (const attr of Array.from(el.attributes || [])) {
@@ -223,10 +247,6 @@ class _BrowserRuntime:
) )
for item in payload for item in payload
] ]
finally:
await session.page.close()
await session.context.close()
self._pages.pop(page_id, None)
_runtime = _BrowserRuntime() _runtime = _BrowserRuntime()
@@ -234,12 +254,12 @@ _runtime = _BrowserRuntime()
def open_page(url: str, wait_ms: int = 3000) -> str: def open_page(url: str, wait_ms: int = 3000) -> str:
"""打开页面并返回 page_id。""" """打开页面并返回 page_id。"""
return _runtime._run(_runtime._open_page(url, wait_ms)) return _runtime._run(_runtime._open_page_impl, url, wait_ms)
def query(page_id: str, selector: str) -> list[NodeSnapshot]: def query(page_id: str, selector: str) -> list[NodeSnapshot]:
"""按选择器抓取节点,提取文本、图片 URL 与属性。""" """按选择器抓取节点,提取文本、图片 URL 与属性。"""
items = _runtime._run(_runtime._query(page_id, selector)) items = _runtime._run(_runtime._query_impl, page_id, selector)
seen_urls: set[str] = set() seen_urls: set[str] = set()
for item in items: for item in items: