"""
浏览器管理模块
使用 Playwright 管理浏览器实例和页面
"""
from typing import Optional, List, Dict, Any
from playwright.async_api import (
async_playwright,
Browser,
BrowserContext,
Page,
Playwright,
)
from loguru import logger
from cookies.cookies import Cookier, create_cookie_manager
from configs.settings import settings
from anti_detection import BrowserFingerprint, get_anti_detection_manager
class BrowserManager:
"""浏览器管理器"""
def __init__(
self,
headless: bool = True,
bin_path: Optional[str] = None,
cookie_manager: Optional[Cookier] = None,
enable_stealth: bool = True,
):
self.headless = headless
self.bin_path = bin_path
self.cookie_manager = cookie_manager or create_cookie_manager()
self.enable_stealth = enable_stealth
self._playwright: Optional[Playwright] = None
self._browser: Optional[Browser] = None
self._context: Optional[BrowserContext] = None
self._anti_detection = get_anti_detection_manager(enable_stealth=enable_stealth)
async def __aenter__(self):
"""异步上下文管理器入口"""
await self.start()
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
"""异步上下文管理器退出"""
await self.close()
async def start(self) -> None:
"""启动浏览器"""
logger.info("Starting browser...")
# 启动 Playwright
self._playwright = await async_playwright().start()
# 浏览器启动参数
launch_options = {
"headless": self.headless,
}
if self.bin_path:
launch_options["executable_path"] = self.bin_path
logger.info(f"Using custom browser binary: {self.bin_path}")
# 启动浏览器
self._browser = await self._playwright.chromium.launch(**launch_options)
# 获取反指纹配置
fingerprint_options = BrowserFingerprint.get_context_options()
# 尝试加载完整的 storage_state（包括 cookies 和 localStorage）
context_options = fingerprint_options.copy()
# 先尝试从 storage_state.json 加载完整状态
import os
storage_state_path = "storage_state.json"
if os.path.exists(storage_state_path):
try:
context_options["storage_state"] = storage_state_path
logger.info(f"Loaded storage state from {storage_state_path}")
except Exception as e:
logger.warning(f"Failed to load storage state: {e}")
# 降级到只加载 cookies
cookies = await self.cookie_manager.load_cookies()
if cookies:
context_options["storage_state"] = {"cookies": cookies, "origins": []}
logger.info(f"Loaded {len(cookies)} cookies (fallback)")
else:
# 如果没有 storage_state.json，只加载 cookies
cookies = await self.cookie_manager.load_cookies()
if cookies:
context_options["storage_state"] = {"cookies": cookies, "origins": []}
logger.info(f"Loaded {len(cookies)} cookies")
self._context = await self._browser.new_context(**context_options)
logger.info("Browser started successfully")
async def close(self) -> None:
"""关闭浏览器"""
logger.info("Closing browser...")
if self._context:
# 保存完整的 storage_state（包括 cookies 和 localStorage）
try:
storage_state = await self._context.storage_state()
# 保存完整的 storage_state 到文件
import json
with open("storage_state.json", "w") as f:
json.dump(storage_state, f, indent=2)
logger.info(f"Saved storage state to storage_state.json")
# 同时保存 cookies 到 cookies.json（向后兼容）
cookies = storage_state.get("cookies", [])
await self.cookie_manager.save_cookies(cookies)
logger.info(f"Saved {len(cookies)} cookies")
except Exception as e:
logger.error(f"Failed to save storage state: {e}")
await self._context.close()
self._context = None
if self._browser:
await self._browser.close()
self._browser = None
if self._playwright:
await self._playwright.stop()
self._playwright = None
logger.info("Browser closed")
async def new_page(self) -> Page:
"""创建新页面"""
if not self._context:
raise RuntimeError("Browser not started. Call start() first.")
page = await self._context.new_page()
# 应用反检测措施
if self.enable_stealth:
await self._anti_detection.prepare_page(page)
logger.debug("Created new page with anti-detection")
return page
async def get_cookies(self) -> List[Dict[str, Any]]:
"""获取当前 cookies"""
if not self._context:
raise RuntimeError("Browser not started. Call start() first.")
storage_state = await self._context.storage_state()
return storage_state.get("cookies", [])
async def save_cookies(self) -> None:
"""保存当前 cookies"""
if not self._context:
raise RuntimeError("Browser not started. Call start() first.")
cookies = await self.get_cookies()
await self.cookie_manager.save_cookies(cookies)
logger.info(f"Saved {len(cookies)} cookies")
@property
def context(self) -> BrowserContext:
"""获取浏览器上下文"""
if not self._context:
raise RuntimeError("Browser not started. Call start() first.")
return self._context
@property
def browser(self) -> Browser:
"""获取浏览器实例"""
if not self._browser:
raise RuntimeError("Browser not started. Call start() first.")
return self._browser
async def create_browser(
headless: Optional[bool] = None,
bin_path: Optional[str] = None,
enable_stealth: Optional[bool] = None,
) -> BrowserManager:
"""创建浏览器管理器"""
if headless is None:
headless = settings.headless
if bin_path is None:
bin_path = settings.browser_bin_path
if enable_stealth is None:
enable_stealth = settings.enable_stealth
manager = BrowserManager(
headless=headless,
bin_path=bin_path,
enable_stealth=enable_stealth
)
await manager.start()
return manager
