73 lines
2.3 KiB
Python
73 lines
2.3 KiB
Python
"""
|
||
浏览器管理器模块
|
||
|
||
负责管理全局唯一的 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()
|