From 263d362daabb6de65fdcc8b57f46fd080fba0202 Mon Sep 17 00:00:00 2001 From: prokopis3 Date: Fri, 30 May 2025 14:43:18 +0300 Subject: [PATCH] fix(browser_profiler): cross-platform 'q' to quit This commit introduces platform-specific handling for the 'q' key press to quit the browser profiler, ensuring compatibility with both Windows and Unix-like systems. It also adds a check to see if the browser process has already exited, terminating the input listener if so. - Implemented `msvcrt` for Windows to capture keyboard input without requiring a newline. - Retained `termios`, `tty`, and `select` for Unix-like systems. - Added a check for browser process termination to gracefully exit the input listener. - Updated logger messages to use colored output for better user experience. --- crawl4ai/browser_profiler.py | 179 +++++++++++++++++++++++------------ 1 file changed, 120 insertions(+), 59 deletions(-) diff --git a/crawl4ai/browser_profiler.py b/crawl4ai/browser_profiler.py index bc902f61..a00eecab 100644 --- a/crawl4ai/browser_profiler.py +++ b/crawl4ai/browser_profiler.py @@ -180,42 +180,83 @@ class BrowserProfiler: # Run keyboard input loop in a separate task async def listen_for_quit_command(): - import termios - import tty - import select - + import sys + # First output the prompt - self.logger.info("Press 'q' when you've finished using the browser...", tag="PROFILE") - - # Save original terminal settings - fd = sys.stdin.fileno() - old_settings = termios.tcgetattr(fd) - - try: - # Switch to non-canonical mode (no line buffering) - tty.setcbreak(fd) - + self.logger.info( + "Press {segment} when you've finished using the browser...", + tag="PROFILE", + params={"segment": "'q'"}, colors={"segment": LogColor.YELLOW}, + base_color=LogColor.CYAN + ) + + async def check_browser_process(): + if ( + managed_browser.browser_process + and managed_browser.browser_process.poll() is not None + ): + self.logger.info( + "Browser already closed. Ending input listener.", tag="PROFILE" + ) + user_done_event.set() + return True + return False + + # Platform-specific handling + if sys.platform == "win32": + import msvcrt + while True: - # Check if input is available (non-blocking) - readable, _, _ = select.select([sys.stdin], [], [], 0.5) - if readable: - key = sys.stdin.read(1) - if key.lower() == 'q': - self.logger.info("Closing browser and saving profile...", tag="PROFILE", base_color=LogColor.GREEN) + if msvcrt.kbhit(): + key = msvcrt.getch().decode("utf-8") + if key.lower() == "q": + self.logger.info( + "Closing browser and saving profile...", + tag="PROFILE", + base_color=LogColor.GREEN + ) user_done_event.set() return - - # Check if the browser process has already exited - if managed_browser.browser_process and managed_browser.browser_process.poll() is not None: - self.logger.info("Browser already closed. Ending input listener.", tag="PROFILE") - user_done_event.set() + + if await check_browser_process(): return - + await asyncio.sleep(0.1) - - finally: - # Restore terminal settings - termios.tcsetattr(fd, termios.TCSADRAIN, old_settings) + + else: # Unix-like + import termios + import tty + import select + + # Save original terminal settings + fd = sys.stdin.fileno() + old_settings = termios.tcgetattr(fd) + + try: + # Switch to non-canonical mode (no line buffering) + tty.setcbreak(fd) + + while True: + # Check if input is available (non-blocking) + readable, _, _ = select.select([sys.stdin], [], [], 0.5) + if readable: + key = sys.stdin.read(1) + if key.lower() == "q": + self.logger.info( + "Closing browser and saving profile...", + tag="PROFILE", + base_color=LogColor.GREEN + ) + user_done_event.set() + return + + if await check_browser_process(): + return + + await asyncio.sleep(0.1) + finally: + # Restore terminal settings + termios.tcsetattr(fd, termios.TCSADRAIN, old_settings) try: from playwright.async_api import async_playwright @@ -682,42 +723,62 @@ class BrowserProfiler: # Run keyboard input loop in a separate task async def listen_for_quit_command(): - import termios - import tty - import select - + import sys + # First output the prompt - self.logger.info("Press 'q' to stop the browser and exit...", tag="CDP") - - # Save original terminal settings - fd = sys.stdin.fileno() - old_settings = termios.tcgetattr(fd) - - try: - # Switch to non-canonical mode (no line buffering) - tty.setcbreak(fd) - + self.logger.info( + "Press {segment} to stop the browser and exit...", + tag="CDP", + params={"segment": "'q'"}, colors={"segment": LogColor.YELLOW}, + base_color=LogColor.CYAN + ) + + async def check_browser_process(): + if managed_browser.browser_process and managed_browser.browser_process.poll() is not None: + self.logger.info("Browser already closed. Ending input listener.", tag="CDP") + user_done_event.set() + return True + return False + + if sys.platform == "win32": + import msvcrt + while True: - # Check if input is available (non-blocking) - readable, _, _ = select.select([sys.stdin], [], [], 0.5) - if readable: - key = sys.stdin.read(1) - if key.lower() == 'q': + if msvcrt.kbhit(): + key = msvcrt.getch().decode("utf-8") + if key.lower() == "q": self.logger.info("Closing browser...", tag="CDP") user_done_event.set() return - - # Check if the browser process has already exited - if managed_browser.browser_process and managed_browser.browser_process.poll() is not None: - self.logger.info("Browser already closed. Ending input listener.", tag="CDP") - user_done_event.set() + + if await check_browser_process(): return - + await asyncio.sleep(0.1) - - finally: - # Restore terminal settings - termios.tcsetattr(fd, termios.TCSADRAIN, old_settings) + else: + import termios + import tty + import select + + fd = sys.stdin.fileno() + old_settings = termios.tcgetattr(fd) + + try: + tty.setcbreak(fd) + while True: + readable, _, _ = select.select([sys.stdin], [], [], 0.5) + if readable: + key = sys.stdin.read(1) + if key.lower() == "q": + self.logger.info("Closing browser...", tag="CDP") + user_done_event.set() + return + + if await check_browser_process(): + return + await asyncio.sleep(0.1) + finally: + termios.tcsetattr(fd, termios.TCSADRAIN, old_settings) # Function to retrieve and display CDP JSON config async def get_cdp_json(port):