Coverage for agentos/system/browser.py: 24%
191 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-06 19:15 +0800
1"""
2浏览器自动化模块 — 通过 CDP (Chrome DevTools Protocol) 控制浏览器。
4设计:
5- 底层使用 Playwright 连接 Chromium 浏览器
6- 操作统一为 BrowserAction 结构
7- 支持导航、点击、填表、截图、提取文本、执行JS
8- 权限级别: BROWSER (需用户授权)
9"""
11from __future__ import annotations
13import asyncio
14from dataclasses import dataclass
16from agentos.system.permissions import (
17 PermissionDenied,
18 PermissionTier,
19 SystemPermissionManager,
20)
22# ── 浏览器动作定义 ─────────────────────────────────────────────
25@dataclass
26class BrowserAction:
27 """浏览器操作定义。"""
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
37@dataclass
38class BrowserResult:
39 """浏览器操作结果。"""
41 success: bool
42 action: str
43 url: str = ""
44 text: str = "" # 提取的文本
45 html: str = "" # 页面 HTML
46 screenshot_path: str = "" # 截图文件路径
47 title: str = "" # 页面标题
48 error: str = ""
49 duration_ms: float = 0
52# ── CDP 浏览器会话 ─────────────────────────────────────────────
55class BrowserSession:
56 """基于 Playwright 的浏览器会话，封装 CDP 底层协议。
58 使用方式:
59 async with BrowserSession() as browser:
60 await browser.navigate("https://example.com")
61 text = await browser.extract_text("body")
62 await browser.screenshot("page.png")
63 """
65 def __init__(
66 self,
67 headless: bool = True,
68 slow_mo: int = 0,
69 viewport_width: int = 1280,
70 viewport_height: int = 720,
71 ):
72 self._headless = headless
73 self._slow_mo = slow_mo
74 self._viewport = {"width": viewport_width, "height": viewport_height}
75 self._playwright = None
76 self._browser = None
77 self._page = None
78 self._current_url = ""
80 async def __aenter__(self):
81 await self.start()
82 return self
84 async def __aexit__(self, *args):
85 await self.close()
87 async def start(self) -> None:
88 """启动浏览器实例。"""
89 try:
90 from playwright.async_api import async_playwright
91 except ImportError:
92 raise ImportError(
93 "浏览器自动化需要 playwright。安装: pip install playwright && playwright install chromium"
94 )
96 self._playwright = await async_playwright().start()
97 self._browser = await self._playwright.chromium.launch(
98 headless=self._headless,
99 slow_mo=self._slow_mo,
100 args=[
101 "--no-sandbox",
102 "--disable-setuid-sandbox",
103 "--disable-dev-shm-usage",
104 "--disable-gpu",
105 ],
106 )
107 self._page = await self._browser.new_page(viewport=self._viewport)
109 async def close(self) -> None:
110 """关闭浏览器。"""
111 if self._browser:
112 await self._browser.close()
113 if self._playwright:
114 await self._playwright.stop()
116 # ── 核心操作 ──
118 async def navigate(self, url: str, wait_until: str = "load") -> BrowserResult:
119 """导航到指定 URL。"""
120 import time
122 t0 = time.time()
123 try:
124 resp = await self._page.goto(url, wait_until=wait_until, timeout=30000)
125 self._current_url = self._page.url
126 title = await self._page.title()
127 duration = (time.time() - t0) * 1000
128 return BrowserResult(
129 success=resp and resp.ok,
130 action="navigate",
131 url=self._current_url,
132 title=title,
133 duration_ms=duration,
134 )
135 except Exception as e:
136 return BrowserResult(
137 success=False,
138 action="navigate",
139 url=url,
140 error=str(e),
141 duration_ms=(time.time() - t0) * 1000,
142 )
144 async def click(self, selector: str) -> BrowserResult:
145 """点击元素。"""
146 import time
148 t0 = time.time()
149 try:
150 await self._page.click(selector, timeout=10000)
151 return BrowserResult(
152 success=True,
153 action="click",
154 url=self._page.url,
155 selector=selector,
156 duration_ms=(time.time() - t0) * 1000,
157 )
158 except Exception as e:
159 return BrowserResult(
160 success=False,
161 action="click",
162 selector=selector,
163 error=str(e),
164 duration_ms=(time.time() - t0) * 1000,
165 )
167 async def type_text(self, selector: str, text: str) -> BrowserResult:
168 """在输入框中输入文本。"""
169 import time
171 t0 = time.time()
172 try:
173 await self._page.fill(selector, text, timeout=10000)
174 return BrowserResult(
175 success=True,
176 action="type",
177 url=self._page.url,
178 selector=selector,
179 text=text,
180 duration_ms=(time.time() - t0) * 1000,
181 )
182 except Exception as e:
183 return BrowserResult(
184 success=False,
185 action="type",
186 selector=selector,
187 error=str(e),
188 duration_ms=(time.time() - t0) * 1000,
189 )
191 async def extract_text(self, selector: str = "body") -> BrowserResult:
192 """提取页面文本。"""
193 import time
195 t0 = time.time()
196 try:
197 element = await self._page.query_selector(selector)
198 if element:
199 text = await element.inner_text()
200 else:
201 text = ""
202 return BrowserResult(
203 success=True,
204 action="extract",
205 url=self._page.url,
206 text=text,
207 duration_ms=(time.time() - t0) * 1000,
208 )
209 except Exception as e:
210 return BrowserResult(
211 success=False,
212 action="extract",
213 error=str(e),
214 duration_ms=(time.time() - t0) * 1000,
215 )
217 async def extract_html(self) -> BrowserResult:
218 """获取完整 HTML。"""
219 import time
221 t0 = time.time()
222 try:
223 html = await self._page.content()
224 return BrowserResult(
225 success=True,
226 action="extract",
227 url=self._page.url,
228 html=html,
229 duration_ms=(time.time() - t0) * 1000,
230 )
231 except Exception as e:
232 return BrowserResult(
233 success=False,
234 action="extract",
235 error=str(e),
236 duration_ms=(time.time() - t0) * 1000,
237 )
239 async def screenshot(self, path: str = "", full_page: bool = True) -> BrowserResult:
240 """截取页面截图。"""
241 import time
243 t0 = time.time()
244 save_path = path or f"/tmp/agentos_screenshot_{int(t0)}.png"
245 try:
246 await self._page.screenshot(path=save_path, full_page=full_page)
247 return BrowserResult(
248 success=True,
249 action="screenshot",
250 url=self._page.url,
251 screenshot_path=save_path,
252 duration_ms=(time.time() - t0) * 1000,
253 )
254 except Exception as e:
255 return BrowserResult(
256 success=False,
257 action="screenshot",
258 error=str(e),
259 duration_ms=(time.time() - t0) * 1000,
260 )
262 async def execute_js(self, code: str) -> BrowserResult:
263 """在页面中执行 JavaScript。"""
264 import time
266 t0 = time.time()
267 try:
268 result = await self._page.evaluate(code)
269 return BrowserResult(
270 success=True,
271 action="js",
272 url=self._page.url,
273 text=str(result),
274 duration_ms=(time.time() - t0) * 1000,
275 )
276 except Exception as e:
277 return BrowserResult(
278 success=False,
279 action="js",
280 error=str(e),
281 duration_ms=(time.time() - t0) * 1000,
282 )
284 async def wait(self, selector: str = "", milliseconds: int = 1000) -> BrowserResult:
285 """等待元素出现或等待指定毫秒。"""
286 import time
288 t0 = time.time()
289 try:
290 if selector:
291 await self._page.wait_for_selector(selector, timeout=10000)
292 else:
293 await asyncio.sleep(milliseconds / 1000)
294 return BrowserResult(
295 success=True,
296 action="wait",
297 url=self._page.url,
298 selector=selector,
299 duration_ms=(time.time() - t0) * 1000,
300 )
301 except Exception as e:
302 return BrowserResult(
303 success=False,
304 action="wait",
305 selector=selector,
306 error=str(e),
307 duration_ms=(time.time() - t0) * 1000,
308 )
310 async def scroll(self, direction: str = "down", amount: int = 500) -> BrowserResult:
311 """滚动页面。"""
312 import time
314 t0 = time.time()
315 try:
316 if direction == "down":
317 await self._page.evaluate(f"window.scrollBy(0, {amount})")
318 elif direction == "up":
319 await self._page.evaluate(f"window.scrollBy(0, -{amount})")
320 elif direction == "bottom":
321 await self._page.evaluate("window.scrollTo(0, document.body.scrollHeight)")
322 elif direction == "top":
323 await self._page.evaluate("window.scrollTo(0, 0)")
324 return BrowserResult(
325 success=True,
326 action="scroll",
327 url=self._page.url,
328 text=f"已滚动 {direction}",
329 duration_ms=(time.time() - t0) * 1000,
330 )
331 except Exception as e:
332 return BrowserResult(
333 success=False,
334 action="scroll",
335 error=str(e),
336 duration_ms=(time.time() - t0) * 1000,
337 )
339 @property
340 def current_url(self) -> str:
341 return self._page.url if self._page else ""
344# ── CDP 浏览器管理器 ───────────────────────────────────────────
347class CDPBrowser:
348 """浏览器管理器 — 带权限控制的浏览器自动化入口。
350 使用:
351 pm = SystemPermissionManager()
352 browser = CDPBrowser(pm, "session-123")
354 async with browser.session() as sess:
355 await sess.navigate("https://example.com")
356 text = await sess.extract_text()
357 """
359 def __init__(
360 self, perm_manager: SystemPermissionManager, session_id: str, headless: bool = True
361 ):
362 self._pm = perm_manager
363 self._sid = session_id
364 self._headless = headless
365 self._current_session: BrowserSession | None = None
367 def session(self, headless: bool | None = None) -> BrowserSession:
368 """创建浏览器会话（上下文管理器）。"""
369 # 权限检查
370 try:
371 self._pm.require(self._sid, PermissionTier.BROWSER, "browser:*")
372 except PermissionDenied as e:
373 raise PermissionDenied(
374 PermissionTier.BROWSER,
375 "browser:*",
376 f"浏览器自动化需要 BROWSER 权限: {e}",
377 )
379 hl = headless if headless is not None else self._headless
380 self._current_session = BrowserSession(headless=hl)
381 return self._current_session
383 async def quick_fetch(self, url: str, extract_text: bool = True) -> BrowserResult:
384 """快速抓取页面（自动打开关闭浏览器）。"""
385 async with self.session() as sess:
386 nav = await sess.navigate(url)
387 if not nav.success:
388 return nav
389 if extract_text:
390 return await sess.extract_text()
391 return await sess.extract_html()
393 async def quick_screenshot(self, url: str, save_path: str) -> BrowserResult:
394 """快速截图页面。"""
395 async with self.session() as sess:
396 nav = await sess.navigate(url)
397 if not nav.success:
398 return nav
399 return await sess.screenshot(save_path)
401 async def execute_action(self, action: BrowserAction) -> BrowserResult:
402 """执行单个浏览器动作。"""
403 if not self._current_session:
404 raise RuntimeError("没有活跃的浏览器会话，请使用 async with browser.session()")
406 sess = self._current_session
408 if action.action_type == "navigate":
409 return await sess.navigate(action.url, action.wait_until)
410 elif action.action_type == "click":
411 return await sess.click(action.selector)
412 elif action.action_type == "type":
413 return await sess.type_text(action.selector, action.value)
414 elif action.action_type == "screenshot":
415 return await sess.screenshot(action.screenshot_path)
416 elif action.action_type == "extract":
417 return await sess.extract_text(action.selector or "body")
418 elif action.action_type == "js":
419 return await sess.execute_js(action.value)
420 elif action.action_type == "wait":
421 ms = int(action.value) if action.value.isdigit() else 1000
422 return await sess.wait(action.selector, ms)
423 elif action.action_type == "scroll":
424 return await sess.scroll(action.value or "down")
425 else:
426 return BrowserResult(
427 success=False, action=action.action_type, error=f"未知动作: {action.action_type}"
428 )