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
| import asyncio, json
from datetime import datetime
class CDPWorkerDebugger:
"""CDP Worker 调试器"""
def __init__(self, ws, session_id):
self.ws, self._psid = ws, session_id
self._cid = 0
self.workers = {}
async def _cmd(self, method, params=None, sid=None):
self._cid += 1
msg = {"id": self._cid, "method": method, "params": params or {}}
if sid or self._psid:
msg["sessionId"] = sid or self._psid
await self.ws.send(json.dumps(msg))
async for r in self.ws:
d = json.loads(r)
if d.get("id") == self._cid:
return d.get("result", {})
async def start(self):
await self._cmd("Target.setAutoAttach", {"autoAttach": True, "waitForDebuggerOnStart": False, "flatten": True})
async def listen(self, duration=30):
start = asyncio.get_event_loop().time()
while (asyncio.get_event_loop().time() - start) < duration:
try:
msg = await asyncio.wait_for(self.ws.__anext__(), timeout=1)
d, m, p = json.loads(msg), d.get("method",""), d.get("params",{})
if m == "Target.attachedToTarget":
info = p["targetInfo"]
if info["type"] in ("worker", "shared_worker", "service_worker"):
self.workers[info["targetId"]] = {
"sid": p["sessionId"], "url": info.get("url",""),
"type": info["type"], "attached": datetime.now().isoformat()
}
await self._cmd("Runtime.enable", sid=p["sessionId"])
elif m == "Target.detachedFromTarget":
if p.get("targetId") in self.workers:
self.workers[p["targetId"]]["detached"] = datetime.now().isoformat()
except asyncio.TimeoutError:
continue
async def evaluate(self, worker_id, expr):
if worker_id not in self.workers: raise ValueError(f"Worker {worker_id} not found")
r = await self._cmd("Runtime.evaluate", {"expression": expr, "returnByValue": True}, sid=self.workers[worker_id]["sid"])
return r.get("result",{}).get("value")
async def send_message(self, worker_id, msg):
return await self.evaluate(worker_id, f"self.dispatchEvent(new MessageEvent('message', {{data: {json.dumps(msg)}}}));")
def summary(self):
return {"total": len(self.workers), "active": sum(1 for w in self.workers.values() if "detached" not in w)}
|
