| 1 | # import asyncio |
| 2 | # import re |
| 3 | # from bs4 import BeautifulSoup |
| 4 | # from playwright.async_api import ( |
| 5 | # async_playwright, |
| 6 | # Browser as PlaywrightBrowser, |
| 7 | # Page, |
| 8 | # Frame, |
| 9 | # BrowserContext, |
| 10 | # ) |
| 11 | |
| 12 | # from helpers import files |
| 13 | |
| 14 | |
| 15 | # class NoPageError(Exception): |
| 16 | # pass |
| 17 | |
| 18 | |
| 19 | # class Browser: |
| 20 | |
| 21 | # load_timeout = 10000 |
| 22 | # interact_timeout = 3000 |
| 23 | # selector_name = "data-a0sel3ct0r" |
| 24 | |
| 25 | # def __init__(self, headless=True): |
| 26 | # self.browser: PlaywrightBrowser = None # type: ignore |
| 27 | # self.context: BrowserContext = None # type: ignore |
| 28 | # self.page: Page = None # type: ignore |
| 29 | # self._playwright = None |
| 30 | # self.headless = headless |
| 31 | # self.contexts = {} |
| 32 | # self.last_selector = "" |
| 33 | # self.page_loaded = False |
| 34 | # self.navigation_count = 0 |
| 35 | |
| 36 | # async def __aenter__(self): |
| 37 | # await self.start() |
| 38 | # return self |
| 39 | |
| 40 | # async def __aexit__(self, exc_type, exc_val, exc_tb): |
| 41 | # await self.close() |
| 42 | |
| 43 | # async def start(self): |
| 44 | # """Start browser session""" |
| 45 | # self._playwright = await async_playwright().start() |
| 46 | # if not self.browser: |
| 47 | # self.browser = await self._playwright.chromium.launch( |
| 48 | # headless=self.headless, args=["--disable-http2"] |
| 49 | # ) |
| 50 | # if not self.context: |
| 51 | # self.context = await self.browser.new_context( |
| 52 | # user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.6422.141 Safari/537.36" |
| 53 | # ) |
| 54 | |
| 55 | # self.page = await self.context.new_page() |
| 56 | # await self.page.set_viewport_size({"width": 1200, "height": 1200}) |
| 57 | |
| 58 | # # Inject the JavaScript to modify the attachShadow method |
| 59 | # js_override = files.read_file("lib/browser/init_override.js") |
| 60 | # await self.page.add_init_script(js_override) |
| 61 | |
| 62 | # # Setup frame handling |
| 63 | # async def inject_script_into_frames(frame): |
| 64 | # try: |
| 65 | # await self.wait_tick() |
| 66 | # if not frame.is_detached(): |
| 67 | # async with asyncio.timeout(0.25): |
| 68 | # await frame.evaluate(js_override) |
| 69 | # print(f"Injected script into frame: {frame.url[:100]}") |
| 70 | # except Exception as e: |
| 71 | # # Frame might have been detached during injection, which is normal |
| 72 | # print( |
| 73 | # f"Could not inject into frame (possibly detached): {str(e)[:100]}" |
| 74 | # ) |
| 75 | |
| 76 | # self.page.on( |
| 77 | # "frameattached", |
| 78 | # lambda frame: asyncio.ensure_future(inject_script_into_frames(frame)), |
| 79 | # ) |
| 80 | |
| 81 | # # Handle page navigation events |
| 82 | # async def handle_navigation(frame): |
| 83 | # if frame == self.page.main_frame: |
| 84 | # print(f"Page navigated to: {frame.url[:100]}") |
| 85 | # self.page_loaded = False |
| 86 | # self.navigation_count += 1 |
| 87 | |
| 88 | # async def handle_load(dummy): |
| 89 | # print("Page load completed") |
| 90 | # self.page_loaded = True |
| 91 | |
| 92 | # async def handle_request(request): |
| 93 | # if ( |
| 94 | # request.is_navigation_request() |
| 95 | # and request.frame == self.page.main_frame |
| 96 | # ): |
| 97 | # print(f"Navigation started to: {request.url[:100]}") |
| 98 | # self.page_loaded = False |
| 99 | # self.navigation_count += 1 |
| 100 | |
| 101 | # self.page.on("request", handle_request) |
| 102 | # self.page.on("framenavigated", handle_navigation) |
| 103 | # self.page.on("load", handle_load) |
| 104 | |
| 105 | # async def close(self): |
| 106 | # """Close browser session""" |
| 107 | # if self.browser: |
| 108 | # await self.browser.close() |
| 109 | # if self._playwright: |
| 110 | # await self._playwright.stop() |
| 111 | |
| 112 | # async def open(self, url: str): |
| 113 | # """Open a URL in the browser""" |
| 114 | # self.last_selector = "" |
| 115 | # self.contexts = {} |
| 116 | # if self.page: |
| 117 | # await self.page.close() |
| 118 | # await self.start() |
| 119 | # try: |
| 120 | # await self.page.goto( |
| 121 | # url, wait_until="networkidle", timeout=Browser.load_timeout |
| 122 | # ) |
| 123 | # except TimeoutError as e: |
| 124 | # pass |
| 125 | # except Exception as e: |
| 126 | # print(f"Error opening page: {e}") |
| 127 | # raise e |
| 128 | # await self.wait_tick() |
| 129 | |
| 130 | # async def get_full_dom(self) -> str: |
| 131 | # """Get full DOM with unique selectors""" |
| 132 | # await self._check_page() |
| 133 | # js_code = files.read_file("lib/browser/extract_dom.js") |
| 134 | |
| 135 | # # Get all frames |
| 136 | # self.contexts = {} |
| 137 | # frame_contents = {} |
| 138 | |
| 139 | # # Extract content from each frame |
| 140 | # i = -1 |
| 141 | # for frame in self.page.frames: |
| 142 | # try: |
| 143 | # if frame.url: # and frame != self.page.main_frame: |
| 144 | # i += 1 |
| 145 | # frame_mark = self._num_to_alpha(i) |
| 146 | |
| 147 | # # Check if frame is still valid |
| 148 | # await self.wait_tick() |
| 149 | # if not frame.is_detached(): |
| 150 | # try: |
| 151 | # # short timeout to identify and skip unresponsive frames |
| 152 | # async with asyncio.timeout(0.25): |
| 153 | # await frame.evaluate("window.location.href") |
| 154 | # except TimeoutError as e: |
| 155 | # print(f"Skipping unresponsive frame: {frame.url}") |
| 156 | # continue |
| 157 | |
| 158 | # await frame.wait_for_load_state( |
| 159 | # "domcontentloaded", timeout=1000 |
| 160 | # ) |
| 161 | |
| 162 | # async with asyncio.timeout(1): |
| 163 | # content = await frame.evaluate( |
| 164 | # js_code, [frame_mark, self.selector_name] |
| 165 | # ) |
| 166 | # self.contexts[frame_mark] = frame |
| 167 | # frame_contents[frame.url] = content |
| 168 | # else: |
| 169 | # print(f"Warning: Frame was detached: {frame.url}") |
| 170 | # except Exception as e: |
| 171 | # print(f"Error extracting from frame {frame.url}: {e}") |
| 172 | |
| 173 | # # # Get main frame content |
| 174 | # # main_mark = self._num_to_alpha(0) |
| 175 | # # main_content = "" |
| 176 | # # try: |
| 177 | # # async with asyncio.timeout(1): |
| 178 | # # main_content = await self.page.evaluate(js_code, [main_mark, self.selector_name]) |
| 179 | # # self.contexts[main_mark] = self.page |
| 180 | # # except Exception as e: |
| 181 | # # print(f"Error when extracting from main frame: {e}") |
| 182 | |
| 183 | # # Replace iframe placeholders with actual content |
| 184 | # # for url, content in frame_contents.items(): |
| 185 | # # placeholder = f'<iframe src="{url}"' |
| 186 | # # main_content = main_content.replace(placeholder, f'{placeholder}>\n<!-- IFrame Content Start -->\n{content}\n<!-- IFrame Content End -->\n</iframe') |
| 187 | |
| 188 | # # return main_content + "".join(frame_contents.values()) |
| 189 | # return "".join(frame_contents.values()) |
| 190 | |
| 191 | # def strip_html_dom(self, html_content: str) -> str: |
| 192 | # """Clean and strip HTML content""" |
| 193 | # if not html_content: |
| 194 | # return "" |
| 195 | |
| 196 | # soup = BeautifulSoup(html_content, "html.parser") |
| 197 | |
| 198 | # for tag in soup.find_all( |
| 199 | # ["br", "hr", "style", "script", "noscript", "meta", "link", "svg"] |
| 200 | # ): |
| 201 | # tag.decompose() |
| 202 | |
| 203 | # for tag in soup.find_all(True): |
| 204 | # if tag.attrs and "invisible" in tag.attrs: |
| 205 | # tag.decompose() |
| 206 | |
| 207 | # for tag in soup.find_all(True): |
| 208 | # allowed_attrs = [ |
| 209 | # self.selector_name, |
| 210 | # "aria-label", |
| 211 | # "placeholder", |
| 212 | # "name", |
| 213 | # "value", |
| 214 | # "type", |
| 215 | # ] |
| 216 | # attrs = { |
| 217 | # "selector" if key == self.selector_name else key: tag.attrs[key] |
| 218 | # for key in allowed_attrs |
| 219 | # if key in tag.attrs and tag.attrs[key] |
| 220 | # } |
| 221 | # tag.attrs = attrs |
| 222 | |
| 223 | # def remove_empty(tag_name: str) -> None: |
| 224 | # for tag in soup.find_all(tag_name): |
| 225 | # if not tag.attrs: |
| 226 | # tag.unwrap() |
| 227 | |
| 228 | # remove_empty("span") |
| 229 | # remove_empty("p") |
| 230 | # remove_empty("strong") |
| 231 | |
| 232 | # return soup.prettify(formatter="minimal") |
| 233 | |
| 234 | # def process_html_with_selectors(self, html_content: str) -> str: |
| 235 | # """Process HTML content and add selectors to interactive elements""" |
| 236 | # if not html_content: |
| 237 | # return "" |
| 238 | |
| 239 | # html_content = re.sub(r"\s+", " ", html_content) |
| 240 | # soup = BeautifulSoup(html_content, "html.parser") |
| 241 | |
| 242 | # structural_tags = [ |
| 243 | # "html", |
| 244 | # "head", |
| 245 | # "body", |
| 246 | # "div", |
| 247 | # "span", |
| 248 | # "section", |
| 249 | # "main", |
| 250 | # "article", |
| 251 | # "header", |
| 252 | # "footer", |
| 253 | # "nav", |
| 254 | # "ul", |
| 255 | # "ol", |
| 256 | # "li", |
| 257 | # "tr", |
| 258 | # "td", |
| 259 | # "th", |
| 260 | # ] |
| 261 | # for tag in structural_tags: |
| 262 | # for element in soup.find_all(tag): |
| 263 | # element.unwrap() |
| 264 | |
| 265 | # out = str(soup).strip() |
| 266 | # out = re.sub(r">\s*<", "><", out) |
| 267 | # out = re.sub(r'aria-label="', 'label="', out) |
| 268 | |
| 269 | # # out = re.sub(r'selector="(\d+[a-zA-Z]+)"', r'selector=\1', out) |
| 270 | # return out |
| 271 | |
| 272 | # async def get_clean_dom(self) -> str: |
| 273 | # """Get clean DOM with selectors""" |
| 274 | # full_dom = await self.get_full_dom() |
| 275 | # clean_dom = self.strip_html_dom(full_dom) |
| 276 | # return self.process_html_with_selectors(clean_dom) |
| 277 | |
| 278 | # async def click(self, selector: str): |
| 279 | # await self._check_page() |
| 280 | # ctx, selector = self._parse_selector(selector) |
| 281 | # self.last_selector = selector |
| 282 | # # js_code = files.read_file("lib/browser/click.js") |
| 283 | # # result = await self.page.evaluate(js_code, [selector]) |
| 284 | # # if not result: |
| 285 | # result = await ctx.hover(selector, force=True, timeout=Browser.interact_timeout) |
| 286 | # await self.wait_tick() |
| 287 | # result = await ctx.click(selector, force=True, timeout=Browser.interact_timeout) |
| 288 | # await self.wait_tick() |
| 289 | |
| 290 | # # await self.page.wait_for_load_state("networkidle") |
| 291 | # return result |
| 292 | |
| 293 | # async def press(self, key: str): |
| 294 | # await self._check_page() |
| 295 | # if self.last_selector: |
| 296 | # await self.page.press( |
| 297 | # self.last_selector, key, timeout=Browser.interact_timeout |
| 298 | # ) |
| 299 | # else: |
| 300 | # await self.page.keyboard.press(key) |
| 301 | |
| 302 | # async def fill(self, selector: str, text: str): |
| 303 | # await self._check_page() |
| 304 | # ctx, selector = self._parse_selector(selector) |
| 305 | # self.last_selector = selector |
| 306 | # try: |
| 307 | # await self.click(selector) |
| 308 | # except Exception as e: |
| 309 | # pass |
| 310 | # await ctx.fill(selector, text, force=True, timeout=Browser.interact_timeout) |
| 311 | # await self.wait_tick() |
| 312 | |
| 313 | # async def execute(self, js_code: str): |
| 314 | # await self._check_page() |
| 315 | # result = await self.page.evaluate(js_code) |
| 316 | # return result |
| 317 | |
| 318 | # async def screenshot(self, path: str, full_page=False): |
| 319 | # await self._check_page() |
| 320 | # await self.page.screenshot(path=path, full_page=full_page) |
| 321 | |
| 322 | # def _parse_selector(self, selector: str) -> tuple[Page | Frame, str]: |
| 323 | # try: |
| 324 | # ctx = self.page |
| 325 | # # Check if selector is our UID, return |
| 326 | # if re.match(r"^\d+[a-zA-Z]+$", selector): |
| 327 | # alpha_part = "".join(filter(str.isalpha, selector)) |
| 328 | # ctx = self.contexts[alpha_part] |
| 329 | # selector = f"[{self.selector_name}='{selector}']" |
| 330 | # return (ctx, selector) |
| 331 | # except Exception as e: |
| 332 | # raise Exception(f"Error evaluating selector: {selector}") |
| 333 | |
| 334 | # async def _check_page(self): |
| 335 | # for _ in range(2): |
| 336 | # try: |
| 337 | # await self.wait_tick() |
| 338 | # self.page = self.context.pages[0] |
| 339 | # if not self.page: |
| 340 | # raise NoPageError( |
| 341 | # "No page is open in the browser. Please open a URL first." |
| 342 | # ) |
| 343 | # # await self.page.wait_for_load_state("networkidle",) |
| 344 | # async with asyncio.timeout(self.load_timeout / 1000): |
| 345 | # if not self.page_loaded: |
| 346 | # while not self.page_loaded: |
| 347 | # await asyncio.sleep(0.1) |
| 348 | # await self.wait_tick() |
| 349 | # return |
| 350 | # except TimeoutError as e: |
| 351 | # self.page_loaded = True |
| 352 | # return |
| 353 | # except NoPageError as e: |
| 354 | # raise e |
| 355 | # except Exception as e: |
| 356 | # print(f"Error checking page: {e}") |
| 357 | |
| 358 | # def _num_to_alpha(self, num: int) -> str: |
| 359 | # if num < 0: |
| 360 | # return "" |
| 361 | |
| 362 | # result = "" |
| 363 | # while num >= 0: |
| 364 | # result = chr(num % 26 + 97) + result |
| 365 | # num = num // 26 - 1 |
| 366 | |
| 367 | # return result |
| 368 | |
| 369 | # async def wait_tick(self): |
| 370 | # if self.page: |
| 371 | # await self.page.evaluate("window.location.href") |
| 372 | |
| 373 | # async def wait(self, seconds: float = 1.0): |
| 374 | # await asyncio.sleep(seconds) |
| 375 | # await self.wait_tick() |
| 376 | |
| 377 | # async def wait_for_action(self): |
| 378 | # nav_count = self.navigation_count |
| 379 | # for _ in range(5): |
| 380 | # await self._check_page() |
| 381 | # if nav_count != self.navigation_count: |
| 382 | # print("Navigation detected") |
| 383 | # await asyncio.sleep(1) |
| 384 | # return |
| 385 | # await asyncio.sleep(0.1) |