Coverage for agentos/system/browser.py: 24%

1""" 

2浏览器自动化模块 — 通过 CDP (Chrome DevTools Protocol) 控制浏览器。 

3 

4设计: 

5- 底层使用 Playwright 连接 Chromium 浏览器 

6- 操作统一为 BrowserAction 结构 

7- 支持导航、点击、填表、截图、提取文本、执行JS 

8- 权限级别: BROWSER (需用户授权) 

9""" 

10 

11from __future__ import annotations 

12 

13import asyncio 

14from dataclasses import dataclass 

15 

16from agentos.system.permissions import ( 

17 SystemPermissionManager, 

18 PermissionTier, 

19 PermissionDenied, 

20) 

21 

22 

23# ── 浏览器动作定义 ───────────────────────────────────────────── 

24 

25 

26@dataclass 

27class BrowserAction: 

28 """浏览器操作定义。""" 

29 action_type: str # navigate / click / type / screenshot / extract / js / wait / scroll 

30 url: str = "" # 导航目标 URL 

31 selector: str = "" # CSS/XPath 选择器 

32 value: str = "" # 输入值 / JS 代码 / 等待时间 

33 screenshot_path: str = "" # 截图保存路径 

34 wait_until: str = "load" # 等待条件: load / networkidle / domcontentloaded 

35 

36 

37@dataclass 

38class BrowserResult: 

39 """浏览器操作结果。""" 

40 success: bool 

41 action: str 

42 url: str = "" 

43 text: str = "" # 提取的文本 

44 html: str = "" # 页面 HTML 

45 screenshot_path: str = "" # 截图文件路径 

46 title: str = "" # 页面标题 

47 error: str = "" 

48 duration_ms: float = 0 

49 

50 

51# ── CDP 浏览器会话 ───────────────────────────────────────────── 

52 

53 

54class BrowserSession: 

55 """基于 Playwright 的浏览器会话，封装 CDP 底层协议。 

56 

57 使用方式: 

58 async with BrowserSession() as browser: 

59 await browser.navigate("https://example.com") 

60 text = await browser.extract_text("body") 

61 await browser.screenshot("page.png") 

62 """ 

63 

64 def __init__(self, headless: bool = True, slow_mo: int = 0, 

65 viewport_width: int = 1280, viewport_height: int = 720): 

66 self._headless = headless 

67 self._slow_mo = slow_mo 

68 self._viewport = {"width": viewport_width, "height": viewport_height} 

69 self._playwright = None 

70 self._browser = None 

71 self._page = None 

72 self._current_url = "" 

73 

74 async def __aenter__(self): 

75 await self.start() 

76 return self 

77 

78 async def __aexit__(self, *args): 

79 await self.close() 

80 

81 async def start(self) -> None: 

82 """启动浏览器实例。""" 

83 try: 

84 from playwright.async_api import async_playwright 

85 except ImportError: 

86 raise ImportError( 

87 "浏览器自动化需要 playwright。安装: pip install playwright && playwright install chromium" 

88 ) 

89 

90 self._playwright = await async_playwright().start() 

91 self._browser = await self._playwright.chromium.launch( 

92 headless=self._headless, 

93 slow_mo=self._slow_mo, 

94 args=[ 

95 "--no-sandbox", 

96 "--disable-setuid-sandbox", 

97 "--disable-dev-shm-usage", 

98 "--disable-gpu", 

99 ], 

100 ) 

101 self._page = await self._browser.new_page(viewport=self._viewport) 

102 

103 async def close(self) -> None: 

104 """关闭浏览器。""" 

105 if self._browser: 

106 await self._browser.close() 

107 if self._playwright: 

108 await self._playwright.stop() 

109 

110 # ── 核心操作 ── 

111 

112 async def navigate(self, url: str, wait_until: str = "load") -> BrowserResult: 

113 """导航到指定 URL。""" 

114 import time 

115 t0 = time.time() 

116 try: 

117 resp = await self._page.goto(url, wait_until=wait_until, timeout=30000) 

118 self._current_url = self._page.url 

119 title = await self._page.title() 

120 duration = (time.time() - t0) * 1000 

121 return BrowserResult( 

122 success=resp and resp.ok, 

123 action="navigate", 

124 url=self._current_url, 

125 title=title, 

126 duration_ms=duration, 

127 ) 

128 except Exception as e: 

129 return BrowserResult( 

130 success=False, action="navigate", url=url, 

131 error=str(e), duration_ms=(time.time() - t0) * 1000, 

132 ) 

133 

134 async def click(self, selector: str) -> BrowserResult: 

135 """点击元素。""" 

136 import time 

137 t0 = time.time() 

138 try: 

139 await self._page.click(selector, timeout=10000) 

140 return BrowserResult( 

141 success=True, action="click", 

142 url=self._page.url, selector=selector, 

143 duration_ms=(time.time() - t0) * 1000, 

144 ) 

145 except Exception as e: 

146 return BrowserResult( 

147 success=False, action="click", selector=selector, 

148 error=str(e), duration_ms=(time.time() - t0) * 1000, 

149 ) 

150 

151 async def type_text(self, selector: str, text: str) -> BrowserResult: 

152 """在输入框中输入文本。""" 

153 import time 

154 t0 = time.time() 

155 try: 

156 await self._page.fill(selector, text, timeout=10000) 

157 return BrowserResult( 

158 success=True, action="type", 

159 url=self._page.url, selector=selector, text=text, 

160 duration_ms=(time.time() - t0) * 1000, 

161 ) 

162 except Exception as e: 

163 return BrowserResult( 

164 success=False, action="type", selector=selector, 

165 error=str(e), duration_ms=(time.time() - t0) * 1000, 

166 ) 

167 

168 async def extract_text(self, selector: str = "body") -> BrowserResult: 

169 """提取页面文本。""" 

170 import time 

171 t0 = time.time() 

172 try: 

173 element = await self._page.query_selector(selector) 

174 if element: 

175 text = await element.inner_text() 

176 else: 

177 text = "" 

178 return BrowserResult( 

179 success=True, action="extract", 

180 url=self._page.url, text=text, 

181 duration_ms=(time.time() - t0) * 1000, 

182 ) 

183 except Exception as e: 

184 return BrowserResult( 

185 success=False, action="extract", 

186 error=str(e), duration_ms=(time.time() - t0) * 1000, 

187 ) 

188 

189 async def extract_html(self) -> BrowserResult: 

190 """获取完整 HTML。""" 

191 import time 

192 t0 = time.time() 

193 try: 

194 html = await self._page.content() 

195 return BrowserResult( 

196 success=True, action="extract", 

197 url=self._page.url, html=html, 

198 duration_ms=(time.time() - t0) * 1000, 

199 ) 

200 except Exception as e: 

201 return BrowserResult( 

202 success=False, action="extract", 

203 error=str(e), duration_ms=(time.time() - t0) * 1000, 

204 ) 

205 

206 async def screenshot(self, path: str = "", full_page: bool = True) -> BrowserResult: 

207 """截取页面截图。""" 

208 import time 

209 t0 = time.time() 

210 save_path = path or f"/tmp/agentos_screenshot_{int(t0)}.png" 

211 try: 

212 await self._page.screenshot(path=save_path, full_page=full_page) 

213 return BrowserResult( 

214 success=True, action="screenshot", 

215 url=self._page.url, screenshot_path=save_path, 

216 duration_ms=(time.time() - t0) * 1000, 

217 ) 

218 except Exception as e: 

219 return BrowserResult( 

220 success=False, action="screenshot", 

221 error=str(e), duration_ms=(time.time() - t0) * 1000, 

222 ) 

223 

224 async def execute_js(self, code: str) -> BrowserResult: 

225 """在页面中执行 JavaScript。""" 

226 import time 

227 t0 = time.time() 

228 try: 

229 result = await self._page.evaluate(code) 

230 return BrowserResult( 

231 success=True, action="js", 

232 url=self._page.url, text=str(result), 

233 duration_ms=(time.time() - t0) * 1000, 

234 ) 

235 except Exception as e: 

236 return BrowserResult( 

237 success=False, action="js", 

238 error=str(e), duration_ms=(time.time() - t0) * 1000, 

239 ) 

240 

241 async def wait(self, selector: str = "", milliseconds: int = 1000) -> BrowserResult: 

242 """等待元素出现或等待指定毫秒。""" 

243 import time 

244 t0 = time.time() 

245 try: 

246 if selector: 

247 await self._page.wait_for_selector(selector, timeout=10000) 

248 else: 

249 await asyncio.sleep(milliseconds / 1000) 

250 return BrowserResult( 

251 success=True, action="wait", 

252 url=self._page.url, selector=selector, 

253 duration_ms=(time.time() - t0) * 1000, 

254 ) 

255 except Exception as e: 

256 return BrowserResult( 

257 success=False, action="wait", selector=selector, 

258 error=str(e), duration_ms=(time.time() - t0) * 1000, 

259 ) 

260 

261 async def scroll(self, direction: str = "down", amount: int = 500) -> BrowserResult: 

262 """滚动页面。""" 

263 import time 

264 t0 = time.time() 

265 try: 

266 if direction == "down": 

267 await self._page.evaluate(f"window.scrollBy(0, {amount})") 

268 elif direction == "up": 

269 await self._page.evaluate(f"window.scrollBy(0, -{amount})") 

270 elif direction == "bottom": 

271 await self._page.evaluate("window.scrollTo(0, document.body.scrollHeight)") 

272 elif direction == "top": 

273 await self._page.evaluate("window.scrollTo(0, 0)") 

274 return BrowserResult( 

275 success=True, action="scroll", 

276 url=self._page.url, text=f"已滚动 {direction}", 

277 duration_ms=(time.time() - t0) * 1000, 

278 ) 

279 except Exception as e: 

280 return BrowserResult( 

281 success=False, action="scroll", 

282 error=str(e), duration_ms=(time.time() - t0) * 1000, 

283 ) 

284 

285 @property 

286 def current_url(self) -> str: 

287 return self._page.url if self._page else "" 

288 

289 

290# ── CDP 浏览器管理器 ─────────────────────────────────────────── 

291 

292 

293class CDPBrowser: 

294 """浏览器管理器 — 带权限控制的浏览器自动化入口。 

295 

296 使用: 

297 pm = SystemPermissionManager() 

298 browser = CDPBrowser(pm, "session-123") 

299 

300 async with browser.session() as sess: 

301 await sess.navigate("https://example.com") 

302 text = await sess.extract_text() 

303 """ 

304 

305 def __init__(self, perm_manager: SystemPermissionManager, session_id: str, 

306 headless: bool = True): 

307 self._pm = perm_manager 

308 self._sid = session_id 

309 self._headless = headless 

310 self._current_session: BrowserSession | None = None 

311 

312 def session(self, headless: bool | None = None) -> BrowserSession: 

313 """创建浏览器会话（上下文管理器）。""" 

314 # 权限检查 

315 try: 

316 self._pm.require(self._sid, PermissionTier.BROWSER, "browser:*") 

317 except PermissionDenied as e: 

318 raise PermissionDenied( 

319 PermissionTier.BROWSER, "browser:*", 

320 f"浏览器自动化需要 BROWSER 权限: {e}", 

321 ) 

322 

323 hl = headless if headless is not None else self._headless 

324 self._current_session = BrowserSession(headless=hl) 

325 return self._current_session 

326 

327 async def quick_fetch(self, url: str, extract_text: bool = True) -> BrowserResult: 

328 """快速抓取页面（自动打开关闭浏览器）。""" 

329 async with self.session() as sess: 

330 nav = await sess.navigate(url) 

331 if not nav.success: 

332 return nav 

333 if extract_text: 

334 return await sess.extract_text() 

335 return await sess.extract_html() 

336 

337 async def quick_screenshot(self, url: str, save_path: str) -> BrowserResult: 

338 """快速截图页面。""" 

339 async with self.session() as sess: 

340 nav = await sess.navigate(url) 

341 if not nav.success: 

342 return nav 

343 return await sess.screenshot(save_path) 

344 

345 async def execute_action(self, action: BrowserAction) -> BrowserResult: 

346 """执行单个浏览器动作。""" 

347 if not self._current_session: 

348 raise RuntimeError("没有活跃的浏览器会话，请使用 async with browser.session()") 

349 

350 sess = self._current_session 

351 

352 if action.action_type == "navigate": 

353 return await sess.navigate(action.url, action.wait_until) 

354 elif action.action_type == "click": 

355 return await sess.click(action.selector) 

356 elif action.action_type == "type": 

357 return await sess.type_text(action.selector, action.value) 

358 elif action.action_type == "screenshot": 

359 return await sess.screenshot(action.screenshot_path) 

360 elif action.action_type == "extract": 

361 return await sess.extract_text(action.selector or "body") 

362 elif action.action_type == "js": 

363 return await sess.execute_js(action.value) 

364 elif action.action_type == "wait": 

365 ms = int(action.value) if action.value.isdigit() else 1000 

366 return await sess.wait(action.selector, ms) 

367 elif action.action_type == "scroll": 

368 return await sess.scroll(action.value or "down") 

369 else: 

370 return BrowserResult(success=False, action=action.action_type, error=f"未知动作: {action.action_type}")