CDP基本认识

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68

from DrissionPage import ChromiumPage, ChromiumOptions
from test1 import set_log_breakpoint
import time
import threading

co = ChromiumOptions()
co.set_browser_path('/Applications/Google Chrome.app/Contents/MacOS/Google Chrome')
co.set_argument('--remote-debugging-port=9222')
co.set_argument('--user-data-dir=/Users/scan/Library/Application Support/Google/Chrome/Default')
co.set_argument('--auto-open-devtools-for-tabs')
co.set_argument('--remote-allow-origins=*')
page = ChromiumPage(addr_or_opts=co)

def console_monitor():
    page.console.start()
    try:
        while True:
            for msg in page.console.steps():
                # 分级显示不同日志类型
                if msg.level == 'error':
                    print(f"🔴 [ERROR] {msg.text}")
                elif msg.level == 'warning':
                    print(f"🟠 [WARNING] {msg.text}")
                elif '日志断点触发' in msg.text:
                    print(f"🔵 [BREAKPOINT] {msg.text}")
                else:
                    print(f"⚪ [LOG] {msg.text}")
            time.sleep(0.1)
    except Exception as e:
        print(f"❌ 监控线程异常: {str(e)}")
    finally:
        print("🛑 控制台监听器已停止")

def main():
    # 启动监控线程（带状态提示）
    print("🧵 正在启动监控线程...")
    monitor_thread = threading.Thread(target=console_monitor, daemon=True)
    monitor_thread.start()
    print("✅ 监控线程已启动")
    # 打开目标页面（带加载进度）
    print("🌐 正在加载页面 https://www.baidu.com ...")
    page.get("https://www.baidu.com/")
    page.wait.doc_loaded()
    print("✅ 页面加载完成")
    # 设置日志断点（带详细状态报告）
    print("⏳ 正在设置日志断点...")
    try:
        set_log_breakpoint(
            js_url="https://pss.bdstatic.com/r/www/cache/static/protocol/https/amd_modules/@baidu/search-sug_ce0f74a.js",
            line=1795,  ##行数
            column=35,  ##列数
            condition="console.log('日志断点触发:', dataArray.map(x=>x.value)) || false",
            target_page_url="https://www.baidu.com/"
        )
        print("✅ 日志断点设置成功")
    except Exception as e:
        print(f"❌ 断点设置失败: {str(e)}")
    # 主线程保持运行（带退出提示）
    print("🚀 程序已进入监控状态，按Ctrl+C退出...")
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        print("\n🛑 用户请求终止程序")
    finally:
        page.close()
if __name__ == "__main__":
    main()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66

#LogPoint.py
import json
import requests
from websocket import create_connection


def get_target_ws_url(target_page_url: str) -> str:
    """获取目标页面的WebSocket调试地址"""
    targets = requests.get('http://localhost:9222/json').json()
    for t in targets:
        if t['url'].startswith(target_page_url):
            return t['webSocketDebuggerUrl']
    raise Exception(f"未找到匹配页面: {target_page_url}")


def set_log_breakpoint(js_url: str, line: int, column: int, condition: str, target_page_url: str):
    """同步即时设置断点（无需等待脚本加载）"""
    ws_url = get_target_ws_url(target_page_url)
    ws = create_connection(ws_url)
    msg_id = 1

    # 1. 启用Debugger
    ws.send(json.dumps({
        "id": msg_id,
        "method": "Debugger.enable"
    }))
    _wait_for_id(ws, msg_id)
    msg_id += 1

    # 2. 直接设置断点（关键修改点）
    print(f"⚡ 直接设置断点: {js_url}:{line}")
    ws.send(json.dumps({
        "id": msg_id,
        "method": "Debugger.setBreakpointByUrl",
        "params": {
            "url": js_url,
            "lineNumber": line,
            "columnNumber": column,
            "condition": condition
        }
    }))

    # 3. 获取设置结果
    resp = _wait_for_id(ws, msg_id)
    if "error" in resp:
        raise Exception(f"断点设置失败: {resp['error']}")
    print(f"✅ 断点已激活 (ID: {resp['result']['breakpointId']})")

    # 4. 保持连接监听断点触发
    while True:
        msg = json.loads(ws.recv())
        if msg.get("method") == "Debugger.paused":
            print("🔥 断点命中!")
            # 可在此处添加断点命中后的处理逻辑
            ws.send(json.dumps({
                "id": msg_id + 1,
                "method": "Debugger.resume"
            }))


def _wait_for_id(ws, expect_id):
    """等待指定ID的响应"""
    while True:
        msg = json.loads(ws.recv())
        if msg.get("id") == expect_id:
            return msg

1

{"command":"Fetch.enable","parameters":{"patterns":[{"urlPattern":"*.mi.com/*","resourceType":"XHR","requestStage":"Request"}],"handleAuthRequests":false}}

1

Fetch.requestPaused