import json
import websocket
import requests
import time
def get_websocket_url(port=9222):
"""获取 WebSocket 调试 URL"""
response = requests.get(f'http://localhost:{port}/json')
tabs = response.json()
return tabs[0]['webSocketDebuggerUrl']
class CDPDebugger:
"""CDP 调试器封装类"""
def __init__(self, ws_url):
self.ws = websocket.create_connection(ws_url)
self.command_id = 0
def send_command(self, method, params=None):
"""发送 CDP 命令"""
self.command_id += 1
command = {
"id": self.command_id,
"method": method,
"params": params or {}
}
print(f"\n>>> 发送命令: {method}")
print(f" 参数: {json.dumps(params or {}, indent=6)}")
self.ws.send(json.dumps(command))
response = json.loads(self.ws.recv())
print(f"<<< 响应: {json.dumps(response, indent=4, ensure_ascii=False)}")
return response
def wait_for_event(self, event_name, timeout=5):
"""等待特定事件"""
print(f"\n--- 等待事件: {event_name} ---")
start_time = time.time()
while time.time() - start_time < timeout:
try:
self.ws.settimeout(1)
message = self.ws.recv()
data = json.loads(message)
if 'method' in data:
print(f" 收到事件: {data['method']}")
if data['method'] == event_name:
print(f" 事件详情: {json.dumps(data['params'], indent=8, ensure_ascii=False)}")
return data
except:
continue
print(f" 超时: 未收到 {event_name} 事件")
return None
def close(self):
self.ws.close()
def example_basic_debugging():
"""
演示基础调试命令组合:
1. Debugger.enable - 启用调试器
2. Runtime.enable - 启用运行时
3. Page.enable - 启用页面事件
4. Page.navigate - 导航到页面
"""
print("\n" + "="*60)
print("示例 1: 基础调试流程")
print("="*60)
debugger = CDPDebugger(get_websocket_url())
debugger.send_command("Debugger.enable")
debugger.send_command("Runtime.enable")
debugger.send_command("Page.enable")
debugger.send_command("Page.navigate", {
"url": "https://example.com"
})
debugger.wait_for_event("Page.loadEventFired")
debugger.close()
def example_breakpoints():
"""
演示断点相关命令:
1. Debugger.setBreakpointByUrl - 按 URL 设置断点
2. Debugger.pause - 暂停执行
3. Debugger.resume - 恢复执行
4. Debugger.stepOver - 单步跳过
5. Debugger.stepInto - 单步进入
6. Debugger.stepOut - 单步跳出
"""
print("\n" + "="*60)
print("示例 2: 断点和单步调试")
print("="*60)
debugger = CDPDebugger(get_websocket_url())
debugger.send_command("Debugger.enable")
debugger.send_command("Runtime.enable")
debugger.send_command("Debugger.setBreakpointByUrl", {
"lineNumber": 10,
"url": "https://example.com/script.js",
"columnNumber": 0,
"condition": "x > 5"
})
debugger.send_command("Debugger.pause")
paused_event = debugger.wait_for_event("Debugger.paused")
if paused_event:
debugger.send_command("Debugger.stepOver")
debugger.send_command("Debugger.stepInto")
debugger.send_command("Debugger.stepOut")
debugger.send_command("Debugger.resume")
debugger.close()
def example_runtime_evaluation():
"""
演示运行时评估命令:
1. Runtime.evaluate - 执行 JavaScript
2. Debugger.evaluateOnCallFrame - 在调用帧上评估
3. Runtime.getProperties - 获取对象属性
"""
print("\n" + "="*60)
print("示例 3: JavaScript 执行和评估")
print("="*60)
debugger = CDPDebugger(get_websocket_url())
debugger.send_command("Debugger.enable")
debugger.send_command("Runtime.enable")
result = debugger.send_command("Runtime.evaluate", {
"expression": "document.title",
"returnByValue": True
})
debugger.send_command("Runtime.evaluate", {
"expression": """
(function() {
return {
url: window.location.href,
userAgent: navigator.userAgent,
screenWidth: screen.width
};
})()
""",
"returnByValue": True
})
result = debugger.send_command("Runtime.evaluate", {
"expression": "window",
"returnByValue": False
})
if 'result' in result and 'result' in result['result']:
object_id = result['result']['result'].get('objectId')
if object_id:
debugger.send_command("Runtime.getProperties", {
"objectId": object_id,
"ownProperties": True
})
debugger.close()
def example_script_management():
"""
演示脚本相关命令:
1. Debugger.getScriptSource - 获取脚本源代码
2. Debugger.setScriptSource - 热更新脚本(实验性)
3. Debugger.searchInContent - 在脚本中搜索
"""
print("\n" + "="*60)
print("示例 4: 脚本管理")
print("="*60)
debugger = CDPDebugger(get_websocket_url())
debugger.send_command("Debugger.enable")
script_event = debugger.wait_for_event("Debugger.scriptParsed", timeout=3)
if script_event and 'params' in script_event:
script_id = script_event['params'].get('scriptId')
if script_id:
debugger.send_command("Debugger.getScriptSource", {
"scriptId": script_id
})
debugger.send_command("Debugger.searchInContent", {
"scriptId": script_id,
"query": "function",
"caseSensitive": False,
"isRegex": False
})
debugger.close()
def example_exception_handling():
"""
演示异常处理命令:
1. Debugger.setPauseOnExceptions - 设置异常时暂停
2. Debugger.setAsyncCallStackDepth - 设置异步调用栈深度
"""
print("\n" + "="*60)
print("示例 5: 异常处理")
print("="*60)
debugger = CDPDebugger(get_websocket_url())
debugger.send_command("Debugger.enable")
debugger.send_command("Runtime.enable")
debugger.send_command("Debugger.setPauseOnExceptions", {
"state": "all"
})
debugger.send_command("Debugger.setAsyncCallStackDepth", {
"maxDepth": 32
})
debugger.send_command("Runtime.evaluate", {
"expression": "throw new Error('测试异常')",
"returnByValue": True
})
debugger.wait_for_event("Debugger.paused")
debugger.close()
def example_performance_profiling():
"""
演示性能分析组合:
1. Profiler.enable - 启用性能分析器
2. Profiler.start - 开始记录
3. Profiler.stop - 停止记录
4. Coverage.enable - 启用代码覆盖率
"""
print("\n" + "="*60)
print("示例 6: 性能分析")
print("="*60)
debugger = CDPDebugger(get_websocket_url())
debugger.send_command("Debugger.enable")
debugger.send_command("Profiler.enable")
debugger.send_command("Profiler.start")
debugger.send_command("Runtime.evaluate", {
"expression": """
for(let i = 0; i < 1000000; i++) {
Math.sqrt(i);
}
"""
})
profile = debugger.send_command("Profiler.stop")
debugger.send_command("Profiler.startPreciseCoverage", {
"callCount": True,
"detailed": True
})
time.sleep(2)
debugger.send_command("Profiler.takePreciseCoverage")
debugger.close()
if __name__ == "__main__":
print("\n" + "="*60)
print("CDP Debugger 常用命令示例集")
print("="*60)
print("\n请确保 Chrome 已使用以下命令启动:")
print("chrome --remote-debugging-port=9222 --user-data-dir=/tmp/chrome-debug")
print("\n")
try:
example_basic_debugging()
time.sleep(1)
example_breakpoints()
time.sleep(1)
example_runtime_evaluation()
time.sleep(1)
example_script_management()
time.sleep(1)
example_exception_handling()
time.sleep(1)
example_performance_profiling()
print("\n" + "="*60)
print("所有示例执行完成!")
print("="*60)
except requests.exceptions.ConnectionError:
print("\n❌ 错误: 无法连接到 Chrome")
print("请确保 Chrome 已使用 --remote-debugging-port=9222 启动")
except Exception as e:
print(f"\n❌ 发生错误: {e}")
import traceback
traceback.print_exc()