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.
This commit is contained in:
prokopis3
@@ -180,42 +180,83 @@ class BrowserProfiler:
|
|||||||
|
|
||||||
# Run keyboard input loop in a separate task
|
# Run keyboard input loop in a separate task
|
||||||
async def listen_for_quit_command():
|
async def listen_for_quit_command():
|
||||||
import termios
|
import sys
|
||||||
import tty
|
|
||||||
import select
|
|
||||||
|
|
||||||
# First output the prompt
|
# First output the prompt
|
||||||
self.logger.info("Press 'q' when you've finished using the browser...", tag="PROFILE")
|
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
|
||||||
|
)
|
||||||
|
|
||||||
# Save original terminal settings
|
async def check_browser_process():
|
||||||
fd = sys.stdin.fileno()
|
if (
|
||||||
old_settings = termios.tcgetattr(fd)
|
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
|
||||||
|
|
||||||
try:
|
# Platform-specific handling
|
||||||
# Switch to non-canonical mode (no line buffering)
|
if sys.platform == "win32":
|
||||||
tty.setcbreak(fd)
|
import msvcrt
|
||||||
|
|
||||||
while True:
|
while True:
|
||||||
# Check if input is available (non-blocking)
|
if msvcrt.kbhit():
|
||||||
readable, _, _ = select.select([sys.stdin], [], [], 0.5)
|
key = msvcrt.getch().decode("utf-8")
|
||||||
if readable:
|
if key.lower() == "q":
|
||||||
key = sys.stdin.read(1)
|
self.logger.info(
|
||||||
if key.lower() == 'q':
|
"Closing browser and saving profile...",
|
||||||
self.logger.info("Closing browser and saving profile...", tag="PROFILE", base_color=LogColor.GREEN)
|
tag="PROFILE",
|
||||||
|
base_color=LogColor.GREEN
|
||||||
|
)
|
||||||
user_done_event.set()
|
user_done_event.set()
|
||||||
return
|
return
|
||||||
|
|
||||||
# Check if the browser process has already exited
|
if await 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
|
return
|
||||||
|
|
||||||
await asyncio.sleep(0.1)
|
await asyncio.sleep(0.1)
|
||||||
|
|
||||||
finally:
|
else: # Unix-like
|
||||||
# Restore terminal settings
|
import termios
|
||||||
termios.tcsetattr(fd, termios.TCSADRAIN, old_settings)
|
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:
|
try:
|
||||||
from playwright.async_api import async_playwright
|
from playwright.async_api import async_playwright
|
||||||
@@ -682,42 +723,62 @@ class BrowserProfiler:
|
|||||||
|
|
||||||
# Run keyboard input loop in a separate task
|
# Run keyboard input loop in a separate task
|
||||||
async def listen_for_quit_command():
|
async def listen_for_quit_command():
|
||||||
import termios
|
import sys
|
||||||
import tty
|
|
||||||
import select
|
|
||||||
|
|
||||||
# First output the prompt
|
# First output the prompt
|
||||||
self.logger.info("Press 'q' to stop the browser and exit...", tag="CDP")
|
self.logger.info(
|
||||||
|
"Press {segment} to stop the browser and exit...",
|
||||||
|
tag="CDP",
|
||||||
|
params={"segment": "'q'"}, colors={"segment": LogColor.YELLOW},
|
||||||
|
base_color=LogColor.CYAN
|
||||||
|
)
|
||||||
|
|
||||||
# Save original terminal settings
|
async def check_browser_process():
|
||||||
fd = sys.stdin.fileno()
|
if managed_browser.browser_process and managed_browser.browser_process.poll() is not None:
|
||||||
old_settings = termios.tcgetattr(fd)
|
self.logger.info("Browser already closed. Ending input listener.", tag="CDP")
|
||||||
|
user_done_event.set()
|
||||||
|
return True
|
||||||
|
return False
|
||||||
|
|
||||||
try:
|
if sys.platform == "win32":
|
||||||
# Switch to non-canonical mode (no line buffering)
|
import msvcrt
|
||||||
tty.setcbreak(fd)
|
|
||||||
|
|
||||||
while True:
|
while True:
|
||||||
# Check if input is available (non-blocking)
|
if msvcrt.kbhit():
|
||||||
readable, _, _ = select.select([sys.stdin], [], [], 0.5)
|
key = msvcrt.getch().decode("utf-8")
|
||||||
if readable:
|
if key.lower() == "q":
|
||||||
key = sys.stdin.read(1)
|
|
||||||
if key.lower() == 'q':
|
|
||||||
self.logger.info("Closing browser...", tag="CDP")
|
self.logger.info("Closing browser...", tag="CDP")
|
||||||
user_done_event.set()
|
user_done_event.set()
|
||||||
return
|
return
|
||||||
|
|
||||||
# Check if the browser process has already exited
|
if await 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
|
return
|
||||||
|
|
||||||
await asyncio.sleep(0.1)
|
await asyncio.sleep(0.1)
|
||||||
|
else:
|
||||||
|
import termios
|
||||||
|
import tty
|
||||||
|
import select
|
||||||
|
|
||||||
finally:
|
fd = sys.stdin.fileno()
|
||||||
# Restore terminal settings
|
old_settings = termios.tcgetattr(fd)
|
||||||
termios.tcsetattr(fd, termios.TCSADRAIN, old_settings)
|
|
||||||
|
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
|
# Function to retrieve and display CDP JSON config
|
||||||
async def get_cdp_json(port):
|
async def get_cdp_json(port):
|
||||||
|
|||||||
Reference in New Issue
Block a user