""" 浏览器管理器模块 负责管理全局唯一的 Playwright 浏览器实例,避免频繁启动/关闭浏览器的开销。 """ import asyncio from typing import Optional from playwright.async_api import async_playwright, Browser, Playwright, Page from ..utils.logger import logger class BrowserManager: """ 浏览器管理器(异步单例) """ _instance = None _playwright: Optional[Playwright] = None _browser: Optional[Browser] = None def __new__(cls): if cls._instance is None: cls._instance = super().__new__(cls) return cls._instance async def initialize(self): """ 初始化 Playwright 和 Browser """ if self._browser is None: try: logger.info("正在启动无头浏览器...") self._playwright = await async_playwright().start() # 启动 Chromium,headless=True 表示无头模式 self._browser = await self._playwright.chromium.launch(headless=True) logger.success("无头浏览器启动成功!") except Exception as e: logger.exception(f"无头浏览器启动失败: {e}") self._browser = None async def get_new_page(self) -> Optional[Page]: """ 获取一个新的页面 (Page) 使用完毕后,调用者应该负责关闭该页面 (await page.close()) """ if self._browser is None: logger.warning("浏览器尚未初始化,尝试重新初始化...") await self.initialize() if self._browser: try: return await self._browser.new_page() except Exception as e: logger.error(f"创建新页面失败: {e}") return None return None async def shutdown(self): """ 关闭浏览器和 Playwright """ if self._browser: await self._browser.close() self._browser = None logger.info("浏览器已关闭") if self._playwright: await self._playwright.stop() self._playwright = None logger.info("Playwright 已停止") # 全局浏览器管理器实例 browser_manager = BrowserManager()