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
| import asyncio
import websockets
import json
CDP_URL = "ws://127.0.0.1:9222/devtools/browser/..."
CMD_ID = [0]
async def cdp(ws, session_id, method, params=None):
CMD_ID[0] += 1
cmd_id = CMD_ID[0]
await ws.send(json.dumps({
"sessionId": session_id,
"id": cmd_id,
"method": method,
"params": params or {}
}))
async for msg in ws:
resp = json.loads(msg)
if resp.get("id") == cmd_id:
return resp.get("result", {})
async def connect_page(ws):
targets = await cdp(ws, None, "Target.getTargets")
target_id = targets["targetInfos"][0]["targetId"]
session = await cdp(ws, None, "Target.attachToTarget", {
"targetId": target_id, "flatten": True
})
return session["sessionId"]
async def get_local_storage(ws, session_id):
"""获取当前页面的所有 LocalStorage 数据"""
result = await cdp(ws, session_id, "Runtime.evaluate", {
"expression": "JSON.stringify(window.localStorage)"
})
return json.loads(result["result"]["value"])
async def set_local_storage_item(ws, session_id, key, value):
"""设置 LocalStorage 项"""
await cdp(ws, session_id, "Runtime.evaluate", {
"expression": f"window.localStorage.setItem('{key}', '{value}')"
})
async def remove_local_storage_item(ws, session_id, key):
"""删除 LocalStorage 项"""
await cdp(ws, session_id, "Runtime.evaluate", {
"expression": f"window.localStorage.removeItem('{key}')"
})
async def clear_local_storage(ws, session_id):
"""清空 LocalStorage"""
await cdp(ws, session_id, "Runtime.evaluate", {
"expression": "window.localStorage.clear()"
})
async def get_session_storage(ws, session_id):
"""获取 SessionStorage"""
result = await cdp(ws, session_id, "Runtime.evaluate", {
"expression": "JSON.stringify(window.sessionStorage)"
})
return json.loads(result["result"]["value"])
|
