From 6e70258208dde40ec25120bc7f38deb17b6d6809 Mon Sep 17 00:00:00 2001 From: zk Date: Mon, 27 Jul 2026 15:53:47 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BC=98=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/tool/browser.py | 418 ++++++++++++++++++++++----------------- app/tool/page_extract.py | 7 +- 2 files changed, 243 insertions(+), 182 deletions(-) diff --git a/app/tool/browser.py b/app/tool/browser.py index 3dae968..853e2d5 100644 --- a/app/tool/browser.py +++ b/app/tool/browser.py @@ -1,24 +1,29 @@ """Playwright 页面抓取工具。 -对外仅暴露两个动作： +对外暴露三个动作： - open_page(url, wait_ms): 打开页面，返回 page_id - query(page_id, selector): 选中节点并提取文字、图片 URL 与属性 +- close_page(page_id): 关闭页面，释放浏览器上下文 实现上内部常驻一个 Browser；每次 open_page 创建独立 BrowserContext 和 Page。 -Playwright 只在一个后台线程里运行，但外层 API 保持同步。 +Playwright 用 async API 跑在一个后台事件循环线程里，多个调用方线程可以真正并发 +（页面加载彼此重叠，不会互相排队）；外层 API 仍保持同步。 """ from __future__ import annotations +import asyncio import atexit -import queue import threading import uuid -from concurrent.futures import Future +from collections.abc import Coroutine from dataclasses import dataclass -from typing import Any, Callable +from typing import Any -from playwright.sync_api import Browser, BrowserContext, Page, sync_playwright +from playwright.async_api import Browser, BrowserContext, Page, async_playwright + +# 页面打开超时（毫秒） +_GOTO_TIMEOUT = 30000 @dataclass @@ -37,207 +42,194 @@ class _PageSession: page: Page +# 节点提取脚本：取标签名、可见文字、图片地址（含懒加载）与全部属性 +_EXTRACT_JS = r"""(elements) => elements.map((el) => { + const attrs = {}; + for (const attr of Array.from(el.attributes || [])) { + attrs[attr.name] = attr.value; + } + + const imageUrls = []; + const addImageUrl = (url) => { + if (!url) return; + try { + const absUrl = new URL(url, location.href).href; + if ( + absUrl.startsWith('data:') || + absUrl.startsWith('blob:') || + absUrl === 'about:blank' + ) { + return; + } + imageUrls.push(absUrl); + } catch { + if ( + url.startsWith('data:') || + url.startsWith('blob:') || + url === 'about:blank' + ) { + return; + } + imageUrls.push(url); + } + }; + + const resolveImageUrl = (img) => { + const candidates = [ + img.getAttribute('data-src'), + img.getAttribute('data-original'), + img.getAttribute('data-croporisrc'), + img.currentSrc, + img.src, + img.getAttribute('src'), + ]; + + const srcset = img.getAttribute('srcset'); + if (srcset) { + for (const part of srcset.split(',')) { + const candidate = part.trim().split(/\s+/)[0]; + candidates.push(candidate); + } + } + + for (const candidate of candidates) { + if (!candidate) continue; + if ( + candidate.startsWith('data:') || + candidate.startsWith('blob:') || + candidate === 'about:blank' + ) { + continue; + } + try { + const absUrl = new URL(candidate, location.href).href; + if ( + absUrl.startsWith('data:') || + absUrl.startsWith('blob:') || + absUrl === 'about:blank' + ) { + continue; + } + return absUrl; + } catch { + return candidate; + } + } + + return null; + }; + + if (el.tagName === 'IMG') { + addImageUrl(resolveImageUrl(el)); + } + + for (const img of Array.from(el.querySelectorAll('img'))) { + addImageUrl(resolveImageUrl(img)); + } + + const text = (el.innerText || el.textContent || '').trim(); + + return { + tag: (el.tagName || '').toLowerCase(), + text, + image_urls: Array.from(new Set(imageUrls)), + attrs, + }; +})""" + + class _BrowserRuntime: - """在单独线程中管理 Playwright 生命周期。""" + """在后台事件循环线程中管理 Playwright 生命周期。 + + 所有 Playwright 调用都以协程形式提交到同一个事件循环， + 因此多个业务线程的页面操作可以并发交叠执行。 + """ def __init__(self) -> None: self._thread: threading.Thread | None = None + self._loop: asyncio.AbstractEventLoop | None = None + self._launch_lock: asyncio.Lock | None = None 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._playwright = None + self._playwright: Any = None self._browser: Browser | None = None self._pages: dict[str, _PageSession] = {} atexit.register(self.close) + # ──────────── 事件循环线程 ──────────── + def _thread_main(self) -> None: - with sync_playwright() as playwright: - self._playwright = playwright - self._browser = playwright.chromium.launch(headless=True) - self._ready.set() + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + self._loop = loop + # 在循环所属线程内创建，避免多协程各自新建锁 + self._launch_lock = asyncio.Lock() + self._ready.set() + try: + loop.run_forever() + finally: + loop.close() - while True: - job = self._job_queue.get() - if job is None: - break - - fn, args, kwargs, fut = job - 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: - try: - self._browser.close() - finally: - self._browser = None - - self._playwright = None - self._closed.set() - - def _ensure_thread(self) -> None: + def _ensure_loop(self) -> asyncio.AbstractEventLoop: with self._lock: - if self._thread and self._thread.is_alive(): - return + if self._thread is None or not self._thread.is_alive(): + self._ready.clear() + self._thread = threading.Thread( + target=self._thread_main, + name="offerpie-playwright", + daemon=True, + ) + self._thread.start() + self._ready.wait() - self._thread = threading.Thread( - target=self._thread_main, - name="offerpie-playwright", - daemon=True, - ) - self._thread.start() - self._ready.wait() + loop = self._loop + if loop is None: + raise RuntimeError("Browser runtime loop is not ready.") + return loop - def _run(self, fn: Callable[..., Any], *args: Any, **kwargs: Any) -> Any: - self._ensure_thread() - fut: Future[Any] = Future() - self._job_queue.put((fn, args, kwargs, fut)) - return fut.result() + def _submit(self, coro: Coroutine[Any, Any, Any]) -> Any: + """把协程提交到事件循环并同步等待结果。""" + loop = self._ensure_loop() + return asyncio.run_coroutine_threadsafe(coro, loop).result() - def close(self) -> 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) + # ──────────── Playwright 操作 ──────────── - def _ensure_browser(self) -> Browser: - if self._browser is None: - raise RuntimeError("Browser runtime is not ready.") + async def _ensure_browser(self) -> Browser: + if self._browser is not None: + return self._browser + + assert self._launch_lock is not None + async with self._launch_lock: + if self._browser is None: + self._playwright = await async_playwright().start() + self._browser = await self._playwright.chromium.launch(headless=True) return self._browser - def _open_page_impl(self, url: str, wait_ms: int) -> str: - browser = self._ensure_browser() - context = browser.new_context() - page = context.new_page() - page.goto(url, wait_until="domcontentloaded", timeout=30000) - if wait_ms > 0: - page.wait_for_timeout(wait_ms) + async def _open_page_impl(self, url: str, wait_ms: int) -> str: + browser = await self._ensure_browser() + context = await browser.new_context() + page = await context.new_page() + + try: + await page.goto(url, wait_until="domcontentloaded", timeout=_GOTO_TIMEOUT) + if wait_ms > 0: + await page.wait_for_timeout(wait_ms) + except BaseException: + # 打开失败也要释放上下文，避免泄漏 + await self._dispose(context, page) + raise page_id = uuid.uuid4().hex self._pages[page_id] = _PageSession(context=context, page=page) return page_id - def _query_impl(self, page_id: str, selector: str) -> list[NodeSnapshot]: + async def _query_impl(self, page_id: str, selector: str) -> list[NodeSnapshot]: session = self._pages.get(page_id) if session is None: raise KeyError(f"Unknown page_id: {page_id}") locator = session.page.locator(selector) - payload = locator.evaluate_all( - r"""(elements) => elements.map((el) => { - const attrs = {}; - for (const attr of Array.from(el.attributes || [])) { - attrs[attr.name] = attr.value; - } - - const imageUrls = []; - const addImageUrl = (url) => { - if (!url) return; - try { - const absUrl = new URL(url, location.href).href; - if ( - absUrl.startsWith('data:') || - absUrl.startsWith('blob:') || - absUrl === 'about:blank' - ) { - return; - } - imageUrls.push(absUrl); - } catch { - if ( - url.startsWith('data:') || - url.startsWith('blob:') || - url === 'about:blank' - ) { - return; - } - imageUrls.push(url); - } - }; - - const resolveImageUrl = (img) => { - const candidates = [ - img.getAttribute('data-src'), - img.getAttribute('data-original'), - img.getAttribute('data-croporisrc'), - img.currentSrc, - img.src, - img.getAttribute('src'), - ]; - - const srcset = img.getAttribute('srcset'); - if (srcset) { - for (const part of srcset.split(',')) { - const candidate = part.trim().split(/\s+/)[0]; - candidates.push(candidate); - } - } - - for (const candidate of candidates) { - if (!candidate) continue; - if ( - candidate.startsWith('data:') || - candidate.startsWith('blob:') || - candidate === 'about:blank' - ) { - continue; - } - try { - const absUrl = new URL(candidate, location.href).href; - if ( - absUrl.startsWith('data:') || - absUrl.startsWith('blob:') || - absUrl === 'about:blank' - ) { - continue; - } - return absUrl; - } catch { - return candidate; - } - } - - return null; - }; - - if (el.tagName === 'IMG') { - addImageUrl(resolveImageUrl(el)); - } - - for (const img of Array.from(el.querySelectorAll('img'))) { - addImageUrl(resolveImageUrl(img)); - } - - const text = (el.innerText || el.textContent || '').trim(); - - return { - tag: (el.tagName || '').toLowerCase(), - text, - image_urls: Array.from(new Set(imageUrls)), - attrs, - }; - })""" - ) + payload = await locator.evaluate_all(_EXTRACT_JS) return [ NodeSnapshot( tag=item.get("tag", ""), @@ -248,18 +240,77 @@ class _BrowserRuntime: for item in payload ] + @staticmethod + async def _dispose(context: BrowserContext, page: Page | None) -> None: + if page is not None: + try: + await page.close() + except Exception: + pass + try: + await context.close() + except Exception: + pass + + async def _close_page_impl(self, page_id: str) -> None: + session = self._pages.pop(page_id, None) + if session is None: + return + await self._dispose(session.context, session.page) + + async def _shutdown_impl(self) -> None: + for page_id in list(self._pages): + await self._close_page_impl(page_id) + + if self._browser is not None: + try: + await self._browser.close() + except Exception: + pass + self._browser = None + + if self._playwright is not None: + try: + await self._playwright.stop() + except Exception: + pass + self._playwright = None + + # ──────────── 生命周期 ──────────── + + def close(self) -> None: + """关闭全部页面与浏览器，停止事件循环。""" + with self._lock: + thread = self._thread + loop = self._loop + if thread is None or not thread.is_alive() or loop is None: + return + + try: + asyncio.run_coroutine_threadsafe( + self._shutdown_impl(), loop + ).result(timeout=15) + except Exception: + pass + + loop.call_soon_threadsafe(loop.stop) + thread.join(timeout=10) + self._thread = None + self._loop = None + self._launch_lock = None + _runtime = _BrowserRuntime() -def open_page(url: str, wait_ms: int = 3000) -> str: +def open_page(url: str, wait_ms: int = 8000) -> str: """打开页面并返回 page_id。""" - return _runtime._run(_runtime._open_page_impl, url, wait_ms) + return _runtime._submit(_runtime._open_page_impl(url, wait_ms)) def query(page_id: str, selector: str) -> list[NodeSnapshot]: """按选择器抓取节点，提取文本、图片 URL 与属性。""" - items = _runtime._run(_runtime._query_impl, page_id, selector) + items = _runtime._submit(_runtime._query_impl(page_id, selector)) seen_urls: set[str] = set() for item in items: @@ -272,3 +323,8 @@ def query(page_id: str, selector: str) -> list[NodeSnapshot]: item.image_urls = unique_urls return items + + +def close_page(page_id: str) -> None: + """关闭页面，释放对应的浏览器上下文。""" + _runtime._submit(_runtime._close_page_impl(page_id)) diff --git a/app/tool/page_extract.py b/app/tool/page_extract.py index a20132d..6a21512 100644 --- a/app/tool/page_extract.py +++ b/app/tool/page_extract.py @@ -13,7 +13,7 @@ from __future__ import annotations from app.core.logger import log -from app.tool.browser import open_page, query +from app.tool.browser import close_page, open_page, query from app.tool.cv import has_qr from app.tool.image_download import download_image from app.tool.ocr import ocr @@ -39,12 +39,17 @@ def extract_page(url: str) -> str: if not url or not url.startswith("http"): return "" + page_id: str | None = None try: page_id = open_page(url, _WAIT_MS) nodes = query(page_id, _SELECTOR) except Exception as exc: log.error(f"公告页打开失败: {url} | {exc}") return "" + finally: + # 页面数据已取出，尽早释放浏览器上下文，不占着资源等后续 OCR + if page_id is not None: + close_page(page_id) texts: list[str] = [] image_urls: list[str] = []