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
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
| """
headless_chrome_manager.py
完整的 Headless Chrome 管理器，覆盖启动、停止、重启、健康检查、多实例管理。
依赖：pip install websockets psutil
"""
import asyncio
import json
import os
import signal
import subprocess
import time
import urllib.request
from dataclasses import dataclass, field
from typing import Callable, Optional
@dataclass
class ChromeInstance:
"""单个 Chrome 实例的状态"""
port: int
pid: int
process: subprocess.Popen
start_time: float
headless_mode: str
xvfb_process: Optional[subprocess.Popen] = None
session_count: int = 0
last_health_check: float = 0.0
healthy: bool = True
class HeadlessChromeManager:
"""
Headless Chrome 管理器
功能：
- 启动/停止 Chrome 实例（支持多种无头模式）
- 自动健康检查与重启
- 多实例管理与端口分配
- 资源使用追踪
- 支持 Xvfb 有头模式
"""
CMD_ID = [0]
@staticmethod
async def cdp(ws, method, params=None, session_id=None):
HeadlessChromeManager.CMD_ID[0] += 1
msg = {"id": HeadlessChromeManager.CMD_ID[0], "method": method, "params": params or {}}
if session_id:
msg["sessionId"] = session_id
await ws.send(json.dumps(msg))
async for resp in ws:
data = json.loads(resp)
if data.get("id") == HeadlessChromeManager.CMD_ID[0]:
return data.get("result", {})
def __init__(
self,
headless_mode: str = "new",
base_port: int = 9222,
max_instances: int = 3,
check_interval: int = 30,
max_retries: int = 3,
chrome_path: Optional[str] = None,
auth_token: Optional[str] = None,
user_data_dir: str = "/tmp/chrome-cdp-data",
):
self.headless_mode = headless_mode
self.base_port = base_port
self.max_instances = max_instances
self.check_interval = check_interval
self.max_retries = max_retries
self.chrome_path = chrome_path or find_chrome_path()
self.auth_token = auth_token or ""
self.user_data_dir = user_data_dir
self.instances: dict[int, ChromeInstance] = {}
self._running = False
self._monitor_task: Optional[asyncio.Task] = None
os.makedirs(user_data_dir, exist_ok=True)
def _build_args(self, port: int, instance_dir: str) -> list:
"""构建单实例启动参数"""
args = [
self.chrome_path,
f"--remote-debugging-port={port}",
"--remote-allow-origins=*",
"--no-first-run",
"--no-default-browser-check",
"--disable-dev-shm-usage",
"--disable-gpu",
f"--window-size=1920,1080",
f"--user-data-dir={instance_dir}",
"--disable-background-networking",
"--disable-background-timer-throttling",
"--disable-breakpad",
"--disable-component-update",
"--disable-sync",
"--mute-audio",
"--force-color-profile=srgb",
]
if self.headless_mode and self.headless_mode.lower() != "none":
args.append(f"--headless={self.headless_mode}")
if os.environ.get("CHROME_NO_SANDBOX", "true").lower() in ("1", "true"):
args.append("--no-sandbox")
return args
async def start_instance(self, port: Optional[int] = None, xvfb: bool = False) -> int:
"""启动一个新的 Chrome 实例"""
if len(self.instances) >= self.max_instances:
raise RuntimeError(f"已达到最大实例数 ({self.max_instances})")
port = port or self._find_free_port()
instance_dir = os.path.join(self.user_data_dir, f"instance-{port}")
xvfb_proc = None
if xvfb and self.headless_mode.lower() == "none":
display = f":{port - self.base_port + 99}"
xvfb_proc = subprocess.Popen(
["Xvfb", display, "-screen", "0", "1920x1080x24", "-ac"],
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
)
os.environ["DISPLAY"] = display
await asyncio.sleep(1)
args = self._build_args(port, instance_dir)
proc = subprocess.Popen(
args,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
preexec_fn=os.setsid if os.name != "nt" else None,
)
await self._wait_for_port(port)
instance = ChromeInstance(
port=port,
pid=proc.pid,
process=proc,
start_time=time.time(),
headless_mode=self.headless_mode,
xvfb_process=xvfb_proc,
)
self.instances[port] = instance
print(f"[{port}] 实例已启动 (PID: {proc.pid})")
return port
async def stop_instance(self, port: int):
"""停止指定实例"""
instance = self.instances.pop(port, None)
if not instance:
print(f"[{port}] 实例不存在")
return
self._graceful_kill(instance.process)
if instance.xvfb_process:
self._graceful_kill(instance.xvfb_process)
print(f"[{port}] 实例已停止 (运行时长: {time.time() - instance.start_time:.1f}s)")
async def stop_all(self):
"""停止所有实例"""
for port in list(self.instances.keys()):
await self.stop_instance(port)
print("所有实例已停止")
async def restart_instance(self, port: int, xvfb: bool = False):
"""重启指定实例"""
print(f"[{port}] 正在重启...")
await self.stop_instance(port)
await asyncio.sleep(2)
await self.start_instance(port=port, xvfb=xvfb)
print(f"[{port}] 重启完成")
async def health_check(self, port: int) -> bool:
"""检查单个实例健康状态"""
instance = self.instances.get(port)
if not instance:
return False
try:
resp = urllib.request.urlopen(
f"http://127.0.0.1:{port}/json/version", timeout=5
)
if resp.status == 200:
data = json.loads(resp.read())
instance.healthy = True
instance.last_health_check = time.time()
instance.session_count = self._count_sessions(port)
return True
except Exception:
pass
instance.healthy = False
return False
async def start_monitoring(self):
"""启动后台健康监控"""
self._running = True
retries = {}
while self._running:
await asyncio.sleep(self.check_interval)
for port in list(self.instances.keys()):
if not await self.health_check(port):
retries[port] = retries.get(port, 0) + 1
print(f"[{port}] 健康检查失败 ({retries[port]}/{self.max_retries})")
if retries[port] >= self.max_retries:
print(f"[{port}] 达到最大重试次数，自动重启")
await self.restart_instance(port)
retries[port] = 0
else:
retries[port] = 0
def stop_monitoring(self):
"""停止健康监控"""
self._running = False
def get_stats(self) -> dict:
"""获取所有实例的统计信息"""
return {
"total_instances": len(self.instances),
"max_instances": self.max_instances,
"headless_mode": self.headless_mode,
"chrome_path": self.chrome_path,
"auth_enabled": bool(self.auth_token),
"instances": [
{
"port": inst.port,
"pid": inst.pid,
"uptime": time.time() - inst.start_time,
"healthy": inst.healthy,
"sessions": inst.session_count,
"xvfb": inst.xvfb_process is not None,
}
for inst in sorted(self.instances.values(), key=lambda i: i.port)
],
}
def _find_free_port(self) -> int:
"""查找可用端口"""
import socket
for offset in range(100):
port = self.base_port + offset
if port not in self.instances:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
if s.connect_ex(("127.0.0.1", port)) != 0:
return port
raise RuntimeError("无法找到可用端口")
@staticmethod
async def _wait_for_port(port: int, timeout: int = 30):
"""等待端口就绪"""
start = time.time()
while time.time() - start < timeout:
try:
resp = urllib.request.urlopen(
f"http://127.0.0.1:{port}/json/version", timeout=5
)
if resp.status == 200:
return
except Exception:
pass
await asyncio.sleep(1)
raise TimeoutError(f"端口 {port} 未在 {timeout} 秒内就绪")
@staticmethod
def _count_sessions(port: int) -> int:
"""统计当前会话数"""
try:
resp = urllib.request.urlopen(f"http://127.0.0.1:{port}/json", timeout=3)
return len(json.loads(resp.read()))
except Exception:
return 0
@staticmethod
def _graceful_kill(proc: subprocess.Popen):
"""优雅杀死进程"""
if proc is None or proc.poll() is not None:
return
proc.terminate()
try:
proc.wait(timeout=10)
except subprocess.TimeoutExpired:
proc.kill()
proc.wait(timeout=5)
async def __aenter__(self):
"""上下文管理器入口"""
return self
async def __aexit__(self, *args):
"""上下文管理器退出"""
self.stop_monitoring()
await self.stop_all()
|
