cdp-socket

import asyncio import os import shutil from cdp_socket.utils.utils import launch_chrome, random_port from cdp_socket.utils.conn import get_websock_url from cdp_socket.socket import SingleCDPSocket async def main(): data_dir = os.path.join(os.getcwd(), "cdp_data_dir") port = random_port() process = None try: process = launch_chrome(data_dir, port) websock_url = await get_websock_url(port, timeout=5) async with SingleCDPSocket(websock_url, timeout=5) as sock: targets = await sock.exec("Target.getTargets") print(f"Active targets: {targets}") # Example: Navigate to a page # This assumes you have a page target. The 'Target.getTargets' output helps identify it. # For browser-level commands, you directly use the 'sock' object. # For page-specific commands, you might need to attach to a page target first. # For simplicity, this example just fetches targets. except Exception as e: print(f"An error occurred: {e}") finally: if process and process.poll() is None: # Check if process is still running os.kill(process.pid, 15) # Terminate Chrome gracefully if os.path.exists(data_dir): shutil.rmtree(data_dir) if __name__ == "__main__": asyncio.run(main())