| 12 | |
| 13 | |
| 14 | class Browser: |
| 15 | def __init__(self): |
| 16 | self.playwright = None |
| 17 | self.browser = None |
| 18 | self.page = None |
| 19 | self.agent = AgentState() |
| 20 | |
| 21 | async def start(self): |
| 22 | self.playwright = await async_playwright().start() |
| 23 | self.browser = await self.playwright.chromium.launch(headless=True) |
| 24 | self.page = await self.browser.new_page() |
| 25 | return self |
| 26 | |
| 27 | # def new_page(self): |
| 28 | # return self.browser.new_page() |
| 29 | |
| 30 | async def go_to(self, url): |
| 31 | try: |
| 32 | await self.page.goto(url, timeout=20000) |
| 33 | |
| 34 | except TimeoutError as e: |
| 35 | print(f"TimeoutError: {e} when trying to navigate to {url}") |
| 36 | return False |
| 37 | return True |
| 38 | |
| 39 | async def screenshot(self, project_name): |
| 40 | screenshots_save_path = Config().get_screenshots_dir() |
| 41 | |
| 42 | page_metadata = await self.page.evaluate("() => { return { url: document.location.href, title: document.title } }") |
| 43 | page_url = page_metadata['url'] |
| 44 | random_filename = os.urandom(20).hex() |
| 45 | filename_to_save = f"{random_filename}.png" |
| 46 | path_to_save = os.path.join(screenshots_save_path, filename_to_save) |
| 47 | |
| 48 | await self.page.emulate_media(media="screen") |
| 49 | await self.page.screenshot(path=path_to_save, full_page=True) |
| 50 | screenshot = await self.page.screenshot() |
| 51 | screenshot_bytes = base64.b64encode(screenshot).decode() |
| 52 | new_state = self.agent.new_state() |
| 53 | new_state["internal_monologue"] = "Browsing the web right now..." |
| 54 | new_state["browser_session"]["url"] = page_url |
| 55 | new_state["browser_session"]["screenshot"] = path_to_save |
| 56 | self.agent.add_to_current_state(project_name, new_state) |
| 57 | # self.close() |
| 58 | return path_to_save, screenshot_bytes |
| 59 | |
| 60 | def get_html(self): |
| 61 | return self.page.content() |
| 62 | |
| 63 | def get_markdown(self): |
| 64 | return md(self.page.content()) |
| 65 | |
| 66 | def get_pdf(self): |
| 67 | pdfs_save_path = Config().get_pdfs_dir() |
| 68 | |
| 69 | page_metadata = self.page.evaluate("() => { return { url: document.location.href, title: document.title } }") |
| 70 | filename_to_save = f"{page_metadata['title']}.pdf" |
| 71 | save_path = os.path.join(pdfs_save_path, filename_to_save) |