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
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
| import asyncio
import json
import websockets
from typing import Optional, List, Dict, Callable
class CDPTargetManager:
"""CDP 多标签页管理器"""
def __init__(self, ws_url: str):
self.ws_url = ws_url
self.ws = None
self._sessions: Dict[str, str] = {}
self._cmd_id = 0
async def connect(self):
"""建立 WebSocket 连接"""
self.ws = await websockets.connect(self.ws_url)
return self
async def disconnect(self):
"""关闭连接"""
if self.ws:
await self.ws.close()
self.ws = None
async def _cdp(self, method: str, params: dict = None,
session_id: str = None) -> dict:
"""发送 CDP 命令"""
self._cmd_id += 1
msg = {"id": self._cmd_id, "method": method,
"params": params or {}}
if session_id:
msg["sessionId"] = session_id
await self.ws.send(json.dumps(msg))
async for resp in self.ws:
data = json.loads(resp)
if data.get("id") == self._cmd_id:
return data.get("result", {})
async def list_targets(self, target_type: str = None) -> List[dict]:
"""列出所有目标，可选按类型筛选"""
result = await self._cdp("Target.getTargets")
targets = result.get("targetInfos", [])
if target_type:
targets = [t for t in targets if t.get("type") == target_type]
return targets
async def create_target(self, url: str = "about:blank",
width: int = None, height: int = None,
new_window: bool = False) -> str:
"""创建新标签页，返回 targetId"""
params = {"url": url}
if width and height:
params["width"] = width
params["height"] = height
if new_window:
params["newWindow"] = True
result = await self._cdp("Target.createTarget", params)
return result.get("targetId")
async def attach(self, target_id: str) -> str:
"""附加到指定目标，返回 sessionId"""
result = await self._cdp("Target.attachToTarget", {
"targetId": target_id,
"flatten": True
})
session_id = result.get("sessionId")
self._sessions[target_id] = session_id
return session_id
async def detach(self, target_id: str):
"""从目标分离"""
session_id = self._sessions.get(target_id)
if session_id:
await self._cdp("Target.detachFromTarget",
{"sessionId": session_id})
self._sessions.pop(target_id, None)
async def close_target(self, target_id: str) -> bool:
"""关闭目标"""
result = await self._cdp("Target.closeTarget",
{"targetId": target_id})
self._sessions.pop(target_id, None)
return result.get("success", False)
async def navigate(self, target_id: str, url: str) -> dict:
"""在指定标签页中导航"""
session_id = self._sessions.get(target_id)
if not session_id:
raise ValueError(f"未连接到目标 {target_id}")
return await self._cdp("Page.navigate",
{"url": url}, session_id)
async def evaluate(self, target_id: str, expression: str) -> dict:
"""在指定标签页中执行 JS"""
session_id = self._sessions.get(target_id)
if not session_id:
raise ValueError(f"未连接到目标 {target_id}")
return await self._cdp("Runtime.evaluate",
{"expression": expression}, session_id)
async def activate_target(self, target_id: str):
"""激活（前置）指定标签页"""
await self._cdp("Target.activateTarget",
{"targetId": target_id})
async def find_target(self, url_pattern: str = None,
title_pattern: str = None) -> Optional[dict]:
"""按 URL 或标题查找目标"""
targets = await self.list_targets()
for t in targets:
url = t.get("url", "")
title = t.get("title", "")
if url_pattern and url_pattern in url:
return t
if title_pattern and title_pattern.lower() in title.lower():
return t
return None
async def set_discover_targets(self, discover: bool = True):
"""启用/停用目标发现"""
await self._cdp("Target.setDiscoverTargets",
{"discover": discover})
async def get_session(self, target_id: str) -> Optional[str]:
"""获取指定目标的 sessionId"""
return self._sessions.get(target_id)
def get_attached_targets(self) -> List[str]:
"""获取所有已附加的目标 ID"""
return list(self._sessions.keys())
async def create_with_attach(self, url: str = "about:blank") -> tuple:
"""创建标签页并自动附加，返回 (targetId, sessionId)"""
target_id = await self.create_target(url)
session_id = await self.attach(target_id)
return target_id, session_id
async def close_all_pages(self, exclude_ids: List[str] = None):
"""关闭所有 page 类型目标"""
exclude = set(exclude_ids or [])
targets = await self.list_targets("page")
count = 0
for t in targets:
tid = t["targetId"]
if tid not in exclude:
if await self.close_target(tid):
count += 1
return count
async def snapshot_all_titles(self) -> Dict[str, str]:
"""获取所有已附加标签页的标题"""
results = {}
tasks = []
for tid in self._sessions:
tasks.append(self.evaluate(tid, "document.title"))
if tasks:
titles = await asyncio.gather(*tasks, return_exceptions=True)
for tid, title_result in zip(self._sessions.keys(), titles):
if isinstance(title_result, Exception):
results[tid] = f"Error: {title_result}"
else:
val = title_result.get("result", {}).get("value", "")
results[tid] = val
return results
|
