"""浏览器池：管理 Playwright 浏览器实例的生命周期和并发访问 核心能力： - 启动/关闭 Chromium 浏览器实例（进程级单例） - 通过信号量控制并发页面数量，避免 API 并发把浏览器资源打爆 - 支持本地调试模式下保留浏览器窗口（出错或成功时），方便人工排查 """ from __future__ import annotations import asyncio import logging from collections.abc import AsyncIterator from contextlib import asynccontextmanager from typing import Any from app.core.config import Settings from app.core.errors import BrowserUnavailableError, ResourceBusyError logger = logging.getLogger(__name__) class BrowserPool: """浏览器实例池，提供并发安全的页面会话获取""" def __init__(self, settings: Settings): self._settings = settings self._playwright: Any | None = None self._playwright_cm: Any | None = None # stealth 钩子包装的 playwright 上下文管理器 self._browser: Any | None = None self._semaphore = asyncio.Semaphore(settings.max_site_concurrency) self._restart_lock = asyncio.Lock() self._started = False self._startup_error: str | None = None self._retained_contexts: list[Any] = [] self._retained_pages: list[Any] = [] # 主动回收相关计数：累计创建的上下文数、本次浏览器启动时刻、当前在用上下文数 self._contexts_since_start = 0 self._browser_started_at: float | None = None self._active_contexts = 0 @property def enabled(self) -> bool: """浏览器功能是否启用""" return self._settings.browser_enabled @property def ready(self) -> bool: """浏览器是否已就绪（未启用或已启动成功均视为就绪）""" return not self.enabled or self._started @property def startup_error(self) -> str | None: """浏览器启动失败时的错误信息""" return self._startup_error async def start(self) -> None: """启动 Playwright 浏览器实例 仅在 browser_enabled=True 且尚未启动时执行。 启动失败会记录错误信息到 startup_error，但不会抛异常（服务仍可启动）。 """ if not self.enabled or self._started: return try: from playwright.async_api import async_playwright headless = self._settings.browser_headless_effective logger.info( "启动浏览器：enabled=%s headless=%s keep_open_on_error=%s channel=%s proxy=%s stealth=%s env=%s", self._settings.browser_enabled, headless, self._settings.browser_keep_open_on_error_effective, self._settings.browser_channel, bool(self._settings.proxy_server), self._settings.browser_stealth_enabled, self._settings.app_env, ) if self._settings.browser_stealth_enabled: # 用 playwright-stealth 钩子包装 playwright：后续 new_context/new_page # 会自动应用 stealth（含 HTTP sec-ch-ua 客户端提示），并令 navigator.languages # 与日语环境保持一致，避免指纹自相矛盾。 from playwright_stealth import Stealth stealth = Stealth( navigator_languages_override=("ja-JP", "ja", "en-US", "en"), ) self._playwright_cm = stealth.use_async(async_playwright()) self._playwright = await self._playwright_cm.__aenter__() else: self._playwright_cm = None self._playwright = await async_playwright().start() chromium = self._playwright.chromium launch_args = [ "--disable-blink-features=AutomationControlled", "--disable-dev-shm-usage", "--disable-web-security", "--no-first-run", "--disable-infobars", "--no-sandbox", "--disable-setuid-sandbox", "--disable-gpu", "--use-mock-keychain", "--password-store=basic", ] # 本地可视化调试时强制窗口化，避免浏览器启动后不可见 if not headless: launch_args.extend(["--start-maximized", "--window-position=120,120"]) self._browser = await chromium.launch( headless=headless, channel=self._settings.browser_channel, proxy=self._settings.proxy, args=launch_args, ) self._started = True self._startup_error = None self._contexts_since_start = 0 self._browser_started_at = asyncio.get_running_loop().time() logger.info("浏览器启动完成") except Exception as exc: self._playwright = None self._browser = None self._started = False self._startup_error = str(exc) logger.exception("浏览器启动失败：%s", self._startup_error) def _is_browser_connected(self) -> bool: """检查当前浏览器对象是否仍与 Playwright driver 保持连接。""" browser = self._browser if not self._started or browser is None: return False is_connected = getattr(browser, "is_connected", None) if callable(is_connected): try: return bool(is_connected()) except Exception: return False return True async def close(self) -> None: """关闭浏览器与 Playwright 运行时，通常发生在服务退出时""" self._retained_pages.clear() for context in self._retained_contexts: try: await context.close() except Exception: logger.exception("关闭保留的调试上下文失败") self._retained_contexts.clear() if self._browser is not None: try: await self._browser.close() except Exception: logger.exception("关闭浏览器实例失败") if self._playwright_cm is not None: # stealth 钩子包装：通过上下文管理器退出来停止 playwright try: await self._playwright_cm.__aexit__(None, None, None) except Exception: logger.exception("停止 Playwright（stealth 包装）运行时失败") elif self._playwright is not None: try: await self._playwright.stop() except Exception: logger.exception("停止 Playwright 运行时失败") self._browser = None self._playwright = None self._playwright_cm = None self._started = False async def _restart_browser(self, reason: str) -> None: """在浏览器掉线后串行重建 Playwright 运行时。""" async with self._restart_lock: if self._is_browser_connected(): return logger.warning("检测到浏览器连接不可用，准备重建浏览器实例：reason=%s", reason) await self.close() await self.start() if not self._is_browser_connected(): raise BrowserUnavailableError(self._startup_error or f"Browser restart failed: {reason}") async def _ensure_browser_ready(self) -> None: """确保浏览器池处于可用状态；若连接已断开则自动重建。""" if not self.enabled: raise BrowserUnavailableError("Browser support is disabled") if self._is_browser_connected(): return await self._restart_browser("browser is disconnected before acquiring page session") def _recycle_due(self) -> bool: """判断浏览器是否达到主动回收阈值（累计上下文数或运行时长任一超限）。""" max_contexts = self._settings.browser_recycle_after_contexts if max_contexts > 0 and self._contexts_since_start >= max_contexts: return True max_age = self._settings.browser_recycle_after_seconds if max_age > 0 and self._browser_started_at is not None: elapsed = asyncio.get_running_loop().time() - self._browser_started_at if elapsed >= max_age: return True return False async def _maybe_recycle(self) -> None: """达到回收阈值且当前无在用上下文时，主动重建浏览器以释放累积内存。 采用「空闲时回收」策略：若仍有其他会话在使用上下文，则跳过本次， 交由后续空闲时机处理，避免关闭浏览器时打断进行中的请求。 """ if not self._recycle_due(): return async with self._restart_lock: # 持锁后二次确认，避免与并发回收/重建重复执行 if not self._recycle_due(): return if self._active_contexts > 0: logger.debug( "浏览器达到回收阈值但仍有 %s 个上下文在用，延迟回收", self._active_contexts, ) return logger.info( "浏览器达到回收阈值，主动重建：contexts=%s elapsed=%.0fs", self._contexts_since_start, 0.0 if self._browser_started_at is None else asyncio.get_running_loop().time() - self._browser_started_at, ) await self.close() await self.start() if not self._is_browser_connected(): raise BrowserUnavailableError(self._startup_error or "Browser recycle failed") @staticmethod def _should_retry_context_creation(exc: Exception) -> bool: """识别 Playwright driver 断链类错误，允许执行一次重建后重试。""" message = str(exc).lower() retry_markers = ( "connection closed while reading from the driver", "browser has been closed", "target page, context or browser has been closed", "connection closed", ) return any(marker in message for marker in retry_markers) async def _retain_context(self, context: Any, page: Any | None, reason: str) -> None: """保留浏览器上下文用于本地调试排查 当本地调试模式开启时，出错或成功后不关闭浏览器窗口， 方便人工查看页面状态。保留数量超过池大小时自动关闭最早的。 """ if page is not None: try: await page.bring_to_front() logger.debug("%s：已将页面置顶：url=%s", reason, page.url) except Exception: logger.exception("%s：页面置顶失败", reason) self._retained_pages.append(page) self._retained_contexts.append(context) # 超出保留上限时关闭最早的上下文 limit = max(1, int(self._settings.browser_pool_size)) while len(self._retained_contexts) > limit: old = self._retained_contexts.pop(0) try: await old.close() except Exception: logger.exception("关闭旧的保留调试上下文失败") async def _build_context_obj(self, storage_state: dict[str, Any] | None) -> Any: """创建浏览器上下文（含断链重试），仅负责创建本身。 计入回收阈值统计；不负责并发信号量与关闭，由调用方管理生命周期。 启用 stealth 时，playwright 已被 use_async 钩子包装，new_context 会自动 应用 stealth（含 HTTP sec-ch-ua），此处无需再手动 apply。 """ context = None for attempt in range(2): try: context = await self._browser.new_context( locale=self._settings.browser_locale, timezone_id=self._settings.browser_timezone_id, user_agent=self._settings.browser_user_agent, viewport=None if not self._settings.browser_headless_effective else {"width": 1440, "height": 900}, storage_state=storage_state, ) break except Exception as exc: if attempt == 0 and self._should_retry_context_creation(exc): logger.warning("创建浏览器上下文失败，尝试重建浏览器后重试一次：err=%s", exc) await self._restart_browser(f"new_context failed: {exc}") continue raise # 上下文创建成功，计入回收阈值统计 self._contexts_since_start += 1 return context @staticmethod async def _configure_page(context: Any) -> Any: """在上下文中创建页面并设置反检测请求头。""" page = await context.new_page() await page.set_extra_http_headers( { "Accept-Language": "ja-JP,ja;q=0.9,en-US;q=0.8,en;q=0.7", "Cache-Control": "no-cache", "Pragma": "no-cache", } ) return page async def create_persistent_context(self, storage_state: dict[str, Any] | None = None) -> Any: """创建一个由调用方负责生命周期的长存上下文。 用于 cf_clearance 保活复用：通过挑战后保留该上下文，后续请求在其内开新 标签页（page_in_context）访问目标页——同上下文 cookie 共享，clearance 有效。 调用方需在不再使用时调用 close_context 释放。 """ await self._ensure_browser_ready() return await self._build_context_obj(storage_state) @asynccontextmanager async def page_in_context(self, context: Any) -> AsyncIterator[Any]: """在既有（长存）上下文中开一个新标签页，复用并发信号量。 结束时仅关闭该页面，不关闭上下文（上下文生命周期由调用方维护）。 同一上下文可并发开多个页面，各自独立导航。 Raises: ResourceBusyError: 等待可用槽位超时 """ await self._ensure_browser_ready() try: await asyncio.wait_for( self._semaphore.acquire(), timeout=self._settings.page_acquire_timeout_seconds, ) except TimeoutError as exc: raise ResourceBusyError() from exc page = None counted = False try: self._active_contexts += 1 counted = True page = await self._configure_page(context) yield page finally: if page is not None: try: await page.close() except Exception: logger.debug("关闭复用上下文页面失败", exc_info=True) if counted: self._active_contexts -= 1 self._semaphore.release() async def close_context(self, context: Any) -> None: """关闭一个长存上下文（忽略已关闭等异常）。""" try: await context.close() except Exception: logger.debug("关闭长存上下文失败", exc_info=True) @asynccontextmanager async def page_session( self, storage_state: dict[str, Any] | None = None, *, keep_open_on_success: bool | None = None, ) -> AsyncIterator[tuple[Any, Any]]: """获取一个浏览器页面会话（上下文管理器） 内部通过信号量做并发控制，超时未获取到槽位会抛出 ResourceBusyError。 页面创建时会设置日语环境、User-Agent 等反检测参数。 Args: storage_state: 浏览器存储状态（含 Cookie），用于会话复用 keep_open_on_success: 请求成功后是否保留浏览器窗口（用于本地调试） Yields: (context, page) 元组，context 为浏览器上下文，page 为页面实例 Raises: BrowserUnavailableError: 浏览器未启用或未启动 ResourceBusyError: 等待可用槽位超时 """ await self._ensure_browser_ready() should_keep_open_on_success = ( self._settings.browser_keep_open_effective if keep_open_on_success is None else keep_open_on_success ) try: await asyncio.wait_for( self._semaphore.acquire(), timeout=self._settings.page_acquire_timeout_seconds, ) except TimeoutError as exc: raise ResourceBusyError() from exc context = None page = None context_counted = False try: # 创建上下文前检查是否需要主动回收（仅在空闲时执行，不打断进行中的会话） await self._maybe_recycle() context = await self._build_context_obj(storage_state) # 上下文创建成功，标记为在用（回收阈值计数已在 _build_context_obj 内完成） self._active_contexts += 1 context_counted = True page = await self._configure_page(context) yield context, page except Exception: if self._settings.browser_keep_open_on_error_effective and context is not None: await self._retain_context(context, page, reason="抓取失败，已保留页面用于排查") logger.warning( "本地调试模式已保留浏览器页面用于排查，窗口不会自动关闭；请手动查看页面。当前保留上下文数=%s", len(self._retained_contexts), ) page = None context = None raise finally: # 本地调试模式下保留窗口 if context is not None and should_keep_open_on_success: await self._retain_context(context, page, reason="本地调试：已保留页面（请求成功）") page = None context = None if page is not None: await page.close() if context is not None: await context.close() if context_counted: self._active_contexts -= 1 self._semaphore.release()