# 声明：本代码仅供学习和研究目的使用。使用者应遵守以下原则： # 1. 不得用于任何商业用途。 # 2. 使用时应遵守目标平台的使用条款和robots.txt规则。 # 3. 不得进行大规模爬取或对平台造成运营干扰。 # 4. 应合理控制请求频率，避免给目标平台带来不必要的负担。 # 5. 不得用于任何非法或不当的用途。 # # 详细许可条款请参阅项目根目录下的LICENSE文件。 # 使用本代码即表示您同意遵守上述原则和LICENSE中的所有条款。 import os import asyncio from typing import Optional, Dict, Any from playwright.async_api import Browser, BrowserContext, Playwright import config from tools.browser_launcher import BrowserLauncher from tools import utils class CDPBrowserManager: """ CDP浏览器管理器，负责启动和管理通过CDP连接的浏览器 """ def __init__(self): self.launcher = BrowserLauncher() self.browser: Optional[Browser] = None self.browser_context: Optional[BrowserContext] = None self.debug_port: Optional[int] = None async def launch_and_connect(self, playwright: Playwright, playwright_proxy: Optional[Dict] = None, user_agent: Optional[str] = None, headless: bool = False) -> BrowserContext: """ 启动浏览器并通过CDP连接 """ try: # 1. 检测浏览器路径 browser_path = await self._get_browser_path() # 2. 获取可用端口 self.debug_port = self.launcher.find_available_port(config.CDP_DEBUG_PORT) # 3. 启动浏览器 await self._launch_browser(browser_path, headless) # 4. 通过CDP连接 await self._connect_via_cdp(playwright) # 5. 创建浏览器上下文 browser_context = await self._create_browser_context( playwright_proxy, user_agent ) self.browser_context = browser_context return browser_context except Exception as e: utils.logger.error(f"[CDPBrowserManager] CDP浏览器启动失败: {e}") await self.cleanup() raise async def _get_browser_path(self) -> str: """ 获取浏览器路径 """ # 优先使用用户自定义路径 if config.CUSTOM_BROWSER_PATH and os.path.isfile(config.CUSTOM_BROWSER_PATH): utils.logger.info(f"[CDPBrowserManager] 使用自定义浏览器路径: {config.CUSTOM_BROWSER_PATH}") return config.CUSTOM_BROWSER_PATH # 自动检测浏览器路径 browser_paths = self.launcher.detect_browser_paths() if not browser_paths: raise RuntimeError( "未找到可用的浏览器。请确保已安装Chrome或Edge浏览器，" "或在配置文件中设置CUSTOM_BROWSER_PATH指定浏览器路径。" ) browser_path = browser_paths[0] # 使用第一个找到的浏览器 browser_name, browser_version = self.launcher.get_browser_info(browser_path) utils.logger.info(f"[CDPBrowserManager] 检测到浏览器: {browser_name} ({browser_version})") utils.logger.info(f"[CDPBrowserManager] 浏览器路径: {browser_path}") return browser_path async def _launch_browser(self, browser_path: str, headless: bool): """ 启动浏览器进程 """ # 设置用户数据目录（如果启用了保存登录状态） user_data_dir = None if config.SAVE_LOGIN_STATE: user_data_dir = os.path.join( os.getcwd(), "browser_data", f"cdp_{config.USER_DATA_DIR % config.PLATFORM}" ) os.makedirs(user_data_dir, exist_ok=True) utils.logger.info(f"[CDPBrowserManager] 用户数据目录: {user_data_dir}") # 启动浏览器 self.launcher.browser_process = self.launcher.launch_browser( browser_path=browser_path, debug_port=self.debug_port, headless=headless, user_data_dir=user_data_dir ) # 等待浏览器准备就绪 if not self.launcher.wait_for_browser_ready( self.debug_port, config.BROWSER_LAUNCH_TIMEOUT ): raise RuntimeError(f"浏览器在 {config.BROWSER_LAUNCH_TIMEOUT} 秒内未能启动") async def _connect_via_cdp(self, playwright: Playwright): """ 通过CDP连接到浏览器 """ cdp_url = f"http://localhost:{self.debug_port}" utils.logger.info(f"[CDPBrowserManager] 正在通过CDP连接到浏览器: {cdp_url}") try: # 使用Playwright的connectOverCDP方法连接 self.browser = await playwright.chromium.connect_over_cdp(cdp_url) if self.browser.is_connected(): utils.logger.info("[CDPBrowserManager] 成功连接到浏览器") utils.logger.info(f"[CDPBrowserManager] 浏览器上下文数量: {len(self.browser.contexts)}") else: raise RuntimeError("CDP连接失败") except Exception as e: utils.logger.error(f"[CDPBrowserManager] CDP连接失败: {e}") raise async def _create_browser_context(self, playwright_proxy: Optional[Dict] = None, user_agent: Optional[str] = None) -> BrowserContext: """ 创建或获取浏览器上下文 """ if not self.browser: raise RuntimeError("浏览器未连接") # 获取现有上下文或创建新的上下文 contexts = self.browser.contexts if contexts: # 使用现有的第一个上下文 browser_context = contexts[0] utils.logger.info("[CDPBrowserManager] 使用现有的浏览器上下文") else: # 创建新的上下文 context_options = { "viewport": {"width": 1920, "height": 1080}, "accept_downloads": True, } # 设置用户代理 if user_agent: context_options["user_agent"] = user_agent utils.logger.info(f"[CDPBrowserManager] 设置用户代理: {user_agent}") # 注意：CDP模式下代理设置可能不生效，因为浏览器已经启动 if playwright_proxy: utils.logger.warning( "[CDPBrowserManager] 警告: CDP模式下代理设置可能不生效，" "建议在浏览器启动前配置系统代理或浏览器代理扩展" ) browser_context = await self.browser.new_context(**context_options) utils.logger.info("[CDPBrowserManager] 创建新的浏览器上下文") return browser_context async def add_stealth_script(self, script_path: str = "libs/stealth.min.js"): """ 添加反检测脚本 """ if self.browser_context and os.path.exists(script_path): try: await self.browser_context.add_init_script(path=script_path) utils.logger.info(f"[CDPBrowserManager] 已添加反检测脚本: {script_path}") except Exception as e: utils.logger.warning(f"[CDPBrowserManager] 添加反检测脚本失败: {e}") async def add_cookies(self, cookies: list): """ 添加Cookie """ if self.browser_context: try: await self.browser_context.add_cookies(cookies) utils.logger.info(f"[CDPBrowserManager] 已添加 {len(cookies)} 个Cookie") except Exception as e: utils.logger.warning(f"[CDPBrowserManager] 添加Cookie失败: {e}") async def get_cookies(self) -> list: """ 获取当前Cookie """ if self.browser_context: try: cookies = await self.browser_context.cookies() return cookies except Exception as e: utils.logger.warning(f"[CDPBrowserManager] 获取Cookie失败: {e}") return [] return [] async def cleanup(self): """ 清理资源 """ try: # 关闭浏览器上下文 if self.browser_context: await self.browser_context.close() self.browser_context = None utils.logger.info("[CDPBrowserManager] 浏览器上下文已关闭") # 断开浏览器连接 if self.browser: await self.browser.close() self.browser = None utils.logger.info("[CDPBrowserManager] 浏览器连接已断开") # 关闭浏览器进程（如果配置为自动关闭） if config.AUTO_CLOSE_BROWSER: self.launcher.cleanup() else: utils.logger.info("[CDPBrowserManager] 浏览器进程保持运行（AUTO_CLOSE_BROWSER=False）") except Exception as e: utils.logger.error(f"[CDPBrowserManager] 清理资源时出错: {e}") def is_connected(self) -> bool: """ 检查是否已连接到浏览器 """ return self.browser is not None and self.browser.is_connected() async def get_browser_info(self) -> Dict[str, Any]: """ 获取浏览器信息 """ if not self.browser: return {} try: version = self.browser.version contexts_count = len(self.browser.contexts) return { "version": version, "contexts_count": contexts_count, "debug_port": self.debug_port, "is_connected": self.is_connected() } except Exception as e: utils.logger.warning(f"[CDPBrowserManager] 获取浏览器信息失败: {e}") return {}