diff --git a/.cursor/worktrees.json b/.cursor/worktrees.json
new file mode 100644
index 0000000..77e9744
--- /dev/null
+++ b/.cursor/worktrees.json
@@ -0,0 +1,5 @@
+{
+ "setup-worktree": [
+ "npm install"
+ ]
+}
diff --git a/Dockerfile b/Dockerfile
new file mode 100644
index 0000000..36cb293
--- /dev/null
+++ b/Dockerfile
@@ -0,0 +1,37 @@
+FROM python:3.11-slim
+
+# 设置工作目录
+WORKDIR /app
+
+# 安装系统依赖和 Playwright 依赖
+RUN apt-get update && apt-get install -y \
+ wget \
+ gnupg \
+ && rm -rf /var/lib/apt/lists/*
+
+# 复制项目文件
+COPY requirements.txt .
+
+# 安装 Python 依赖
+RUN pip install --no-cache-dir -r requirements.txt
+
+# 安装 Playwright 浏览器和系统依赖
+# 注意：chromium 是必须的
+RUN playwright install chromium
+RUN playwright install-deps chromium
+
+# 复制源代码
+COPY . .
+
+# 编译 proto 文件
+RUN python -m grpc_tools.protoc -I./proto --python_out=. --grpc_python_out=. ./proto/spider.proto
+
+# 创建日志目录
+RUN mkdir -p log
+
+# 暴露端口
+EXPOSE 8000 50051
+
+# 启动命令 (默认启动 HTTP Server)
+CMD ["python", "http_server.py"]
+
diff --git a/cdp_client.py b/cdp_client.py
index fd4cd8b..5dd1a20 100644
--- a/cdp_client.py
+++ b/cdp_client.py
@@ -83,7 +83,13 @@ async def on_response(self, response):
'body': body.decode('utf-8', errors='ignore')
}
except Exception as e:
- logger.warning(f"获取响应体失败: {e}")
+ error_msg = str(e)
+ if "No data found" in error_msg or "Target closed" in error_msg:
+ # 忽略预期内的错误，使用 debug 级别
+ logger.debug(f"获取响应体失败 (忽略): {error_msg}")
+ else:
+ logger.warning(f"获取响应体失败: {error_msg}")
+
req['response'] = {
'status_code': response.status,
'headers': response.headers,
@@ -107,6 +113,8 @@ def __init__(self, session_id: str):
self.context: Optional[BrowserContext] = None
self.page: Optional[Page] = None
self.playwright = None
+ self.cdp_session = None
+ self.screencast_listener = None
self.network_interceptor = NetworkInterceptor()
self.custom_headers: Dict[str, str] = {}
self.created_at = time.time()
@@ -236,6 +244,338 @@ async def execute_script(self, script: str) -> Any:
except Exception as e:
logger.error(f"脚本执行失败: {e}")
raise
+
+ async def start_screencast(self, callback):
+ """开启屏幕投射"""
+ if not self.page:
+ raise RuntimeError("浏览器会话未初始化")
+
+ try:
+ # 建立 CDP 会话
+ if not self.cdp_session:
+ self.cdp_session = await self.context.new_cdp_session(self.page)
+
+ # 清理旧的监听器
+ if self.screencast_listener:
+ self.cdp_session.remove_listener("Page.screencastFrame", self.screencast_listener)
+ self.screencast_listener = None
+
+ # 监听 screencast 帧
+ async def on_screencast_frame(frame):
+ try:
+ # frame 包含 data (base64), sessionId, metadata
+ # 调用回调函数发送给 WebSocket
+ if callback:
+ await callback(frame['data'], frame['metadata'])
+
+ # 必须确认帧，否则流会停止
+ # 注意：必须使用 frame 中的 sessionId 确认，因为这可能是来自非主页面的帧（如果支持）
+ # 但 Page.screencastFrameAck 需要在开启 screencast 的 session 上发送
+ if self.cdp_session:
+ await self.cdp_session.send("Page.screencastFrameAck", {"sessionId": frame['sessionId']})
+ except Exception as e:
+ logger.error(f"Screencast frame error: {e}")
+
+ # 保存并添加新监听器
+ self.screencast_listener = on_screencast_frame
+ self.cdp_session.on("Page.screencastFrame", self.screencast_listener)
+
+ # 开启投射
+ await self.cdp_session.send("Page.startScreencast", {
+ "format": "jpeg",
+ "quality": 80,
+ "maxWidth": 1280, # 限制最大宽度，减少传输量
+ "maxHeight": 720, # 限制最大高度
+ "everyNthFrame": 1
+ })
+ logger.info(f"Screencast started for session {self.session_id}")
+
+ # 监听新页面
+ self.context.on("page", self._on_new_page)
+
+ except Exception as e:
+ logger.error(f"Start screencast failed: {e}")
+ raise
+
+ async def _on_new_page(self, page: Page):
+ """处理新页面创建"""
+ try:
+ logger.info(f"New page detected: {page.url}")
+ await page.wait_for_load_state()
+
+ # 更新当前页面引用
+ old_page = self.page
+ self.page = page
+
+ # 重新应用 stealth
+ await stealth_async(self.page)
+
+ # 更新网络拦截
+ self.page.on('request', self.network_interceptor.on_request)
+ self.page.on('response', self.network_interceptor.on_response)
+
+ # 切换 CDP 会话到新页面
+ # 1. 停止旧页面的 screencast
+ if self.cdp_session:
+ try:
+ await self.cdp_session.send("Page.stopScreencast")
+ if self.screencast_listener:
+ self.cdp_session.remove_listener("Page.screencastFrame", self.screencast_listener)
+ except Exception as e:
+ logger.warning(f"Failed to stop old screencast: {e}")
+
+ # 2. 创建新 CDP 会话
+ self.cdp_session = await self.context.new_cdp_session(self.page)
+
+ # 3. 重新绑定监听器并开启 screencast
+ if self.screencast_listener:
+ self.cdp_session.on("Page.screencastFrame", self.screencast_listener)
+ await self.cdp_session.send("Page.startScreencast", {
+ "format": "jpeg",
+ "quality": 80,
+ "maxWidth": 1280,
+ "maxHeight": 720,
+ "everyNthFrame": 1
+ })
+ logger.info(f"Screencast switched to new page: {page.url}")
+
+ # 通知前端更新标签页
+ # 这里需要一种机制通知 WebSocket，目前通过简单的日志或者回调扩展实现
+ # 由于没有直接的 ws 引用，这里先依靠前端轮询或后端主动推送（如果设计支持）
+ # 暂时只能自动切换，前端看到的会变，但列表没变
+
+ except Exception as e:
+ logger.error(f"Failed to switch to new page: {e}")
+
+ async def get_pages(self) -> List[Dict[str, str]]:
+ """获取所有页面信息"""
+ if not self.context:
+ return []
+
+ pages = []
+ for p in self.context.pages:
+ try:
+ title = await p.title()
+ except:
+ title = "Loading..."
+
+ # 使用 id 作为唯一标识，playwright page 没有公开 id，可以用 guid
+ # 这里简单用索引或对象哈希，或者给 page 附加上 id
+ if not hasattr(p, '_guid'):
+ p._guid = str(uuid.uuid4())
+
+ pages.append({
+ 'id': p._guid,
+ 'url': p.url,
+ 'title': title
+ })
+ return pages
+
+ async def switch_to_page(self, page_id: str):
+ """切换到指定页面"""
+ if not self.context:
+ return
+
+ target_page = None
+ for p in self.context.pages:
+ if hasattr(p, '_guid') and p._guid == page_id:
+ target_page = p
+ break
+
+ if target_page and target_page != self.page:
+ logger.info(f"Switching to page {page_id}")
+ await self._on_new_page(target_page)
+
+ async def stop_screencast(self):
+ """停止屏幕投射"""
+ try:
+ # 移除页面监听
+ if self.context:
+ self.context.remove_listener("page", self._on_new_page)
+
+ if self.cdp_session:
+ await self.cdp_session.send("Page.stopScreencast")
+
+ # 移除监听器
+ if self.screencast_listener:
+ self.cdp_session.remove_listener("Page.screencastFrame", self.screencast_listener)
+ self.screencast_listener = None
+
+ logger.info(f"Screencast stopped for session {self.session_id}")
+ except Exception as e:
+ logger.error(f"Stop screencast failed: {e}")
+
+ async def handle_input_event(self, event: Dict[str, Any]):
+ """处理输入事件 - 使用 CDP 原生方法"""
+ if not self.page:
+ return
+
+ try:
+ event_type = event.get('type')
+ x = event.get('x', 0)
+ y = event.get('y', 0)
+
+ # 确保 CDP session 存在
+ if not self.cdp_session:
+ self.cdp_session = await self.context.new_cdp_session(self.page)
+
+ if event_type == 'mousemove':
+ await self.cdp_session.send("Input.dispatchMouseEvent", {
+ "type": "mouseMoved",
+ "x": x,
+ "y": y,
+ "modifiers": event.get('modifiers', 0)
+ })
+
+ elif event_type == 'mousedown':
+ button = event.get('button', 'left')
+ cdp_button = {'left': 'left', 'middle': 'middle', 'right': 'right'}.get(button, 'left')
+ buttons = {'left': 1, 'middle': 4, 'right': 2}.get(button, 1)
+
+ await self.cdp_session.send("Input.dispatchMouseEvent", {
+ "type": "mousePressed",
+ "x": x,
+ "y": y,
+ "button": cdp_button,
+ "buttons": buttons,
+ "clickCount": 1,
+ "modifiers": event.get('modifiers', 0)
+ })
+
+ elif event_type == 'mouseup':
+ button = event.get('button', 'left')
+ cdp_button = {'left': 'left', 'middle': 'middle', 'right': 'right'}.get(button, 'left')
+
+ await self.cdp_session.send("Input.dispatchMouseEvent", {
+ "type": "mouseReleased",
+ "x": x,
+ "y": y,
+ "button": cdp_button,
+ "buttons": 0,
+ "clickCount": 1,
+ "modifiers": event.get('modifiers', 0)
+ })
+
+ elif event_type == 'click':
+ # Click is a sequence, we should respect modifiers for all
+ button = event.get('button', 'left')
+ cdp_button = {'left': 'left', 'middle': 'middle', 'right': 'right'}.get(button, 'left')
+ buttons = {'left': 1, 'middle': 4, 'right': 2}.get(button, 1)
+ modifiers = event.get('modifiers', 0)
+
+ await self.cdp_session.send("Input.dispatchMouseEvent", {
+ "type": "mouseMoved",
+ "x": x,
+ "y": y,
+ "modifiers": modifiers
+ })
+ await self.cdp_session.send("Input.dispatchMouseEvent", {
+ "type": "mousePressed",
+ "x": x,
+ "y": y,
+ "button": cdp_button,
+ "buttons": buttons,
+ "clickCount": 1,
+ "modifiers": modifiers
+ })
+ await self.cdp_session.send("Input.dispatchMouseEvent", {
+ "type": "mouseReleased",
+ "x": x,
+ "y": y,
+ "button": cdp_button,
+ "buttons": 0,
+ "clickCount": 1,
+ "modifiers": modifiers
+ })
+
+ elif event_type == 'keydown':
+ params = {
+ "type": "keyDown",
+ "key": event.get('key', ''),
+ "code": event.get('code', ''),
+ "modifiers": event.get('modifiers', 0),
+ "windowsVirtualKeyCode": event.get('windowsVirtualKeyCode', 0),
+ "nativeVirtualKeyCode": event.get('nativeVirtualKeyCode', 0),
+ "location": event.get('location', 0),
+ "autoRepeat": False, # We filter repeats in frontend
+ "isKeypad": event.get('location', 0) == 3
+ }
+
+ key = event.get('key', '')
+ if len(key) == 1:
+ params["text"] = key
+ params["unmodifiedText"] = key
+
+ if event.get('commands'): # Optional if we support it later
+ params['commands'] = event.get('commands')
+
+ await self.cdp_session.send("Input.dispatchKeyEvent", params)
+
+ elif event_type == 'keyup':
+ await self.cdp_session.send("Input.dispatchKeyEvent", {
+ "type": "keyUp",
+ "key": event.get('key', ''),
+ "code": event.get('code', ''),
+ "modifiers": event.get('modifiers', 0),
+ "location": event.get('location', 0)
+ })
+
+ elif event_type == 'keypress':
+ key = event.get('key', '')
+ if key:
+ await self.cdp_session.send("Input.dispatchKeyEvent", {
+ "type": "keyDown",
+ "key": key,
+ "text": key if len(key) == 1 else ""
+ })
+ await self.cdp_session.send("Input.dispatchKeyEvent", {
+ "type": "keyUp",
+ "key": key
+ })
+
+ elif event_type == 'scroll':
+ deltaX = event.get('deltaX', 0)
+ deltaY = event.get('deltaY', 0)
+ await self.cdp_session.send("Input.dispatchMouseEvent", {
+ "type": "mouseWheel",
+ "x": x,
+ "y": y,
+ "deltaX": deltaX,
+ "deltaY": deltaY,
+ "modifiers": event.get('modifiers', 0)
+ })
+
+ self.last_activity = time.time()
+
+ except Exception as e:
+ logger.error(f"Handle input event failed: {e}, Event: {event}")
+
+ async def replay_events(self, events: List[dict]):
+ """重放事件序列"""
+ if not events: return
+
+ logger.info(f"开始重放任务，共 {len(events)} 个事件")
+ start_time = events[0].get('timestamp', 0)
+
+ # 为了保证重放的准确性，使用相对时间
+ for i, event in enumerate(events):
+ try:
+ # 计算需要等待的时间
+ if i > 0:
+ prev_time = events[i-1].get('timestamp', 0)
+ curr_time = event.get('timestamp', 0)
+ delay = curr_time - prev_time
+ # 如果间隔太大（比如超过5秒），可能是在思考，可以适当缩短或者按原样等待
+ if delay > 0:
+ await asyncio.sleep(delay)
+
+ # 执行操作
+ await self.handle_input_event(event)
+ except Exception as e:
+ logger.error(f"重放事件失败: {e}, index: {i}")
+
+ logger.info("任务重放完成")
+
async def get_content(self) -> str:
"""获取页面内容"""
diff --git a/config.py b/config.py
index 532690e..1751141 100644
--- a/config.py
+++ b/config.py
@@ -31,40 +31,37 @@ class BrowserRPCConfig:
LOG_LEVEL: str = 'INFO'
LOG_DIR: str = 'browser_rpc/log'
- # 代理配置（可选）
- PROXY_SERVER: Optional[str] = None
+ # Redis 配置
+ REDIS_URL: str = 'redis://localhost:6379/0'
+ # 安全配置
+ CLUSTER_SECRET: str = "browser-rpc-secret-key-change-me" # 内部通信密钥
+ API_KEY: Optional[str] = None # 客户端访问密钥 (可选)
+
def __init__(self):
"""从环境变量加载配置"""
- # 尝试从 .env 文件加载
- env_file = os.path.join(os.path.dirname(__file__), '.env')
- if os.path.exists(env_file):
- with open(env_file) as f:
- for line in f:
- line = line.strip()
- if line and not line.startswith('#') and '=' in line:
- key, value = line.split('=', 1)
- key = key.strip()
- value = value.strip()
- # 转换类型
- if hasattr(self, key):
- current_value = getattr(self, key)
- if isinstance(current_value, bool):
- value = value.lower() in ('true', '1', 'yes')
- elif isinstance(current_value, int):
- value = int(value)
- setattr(self, key, value)
-
+ # 尝试从 .env 文件加载 (如果 python-dotenv 已安装)
+ try:
+ from dotenv import load_dotenv
+ load_dotenv()
+ except ImportError:
+ pass
+
# 环境变量优先
for key in dir(self):
if key.isupper():
env_value = os.getenv(key)
if env_value is not None:
current_value = getattr(self, key)
+ # 类型转换
if isinstance(current_value, bool):
env_value = env_value.lower() in ('true', '1', 'yes')
elif isinstance(current_value, int):
- env_value = int(env_value)
+ try:
+ env_value = int(env_value)
+ except ValueError:
+ # 如果无法转换为 int，保持原值或忽略
+ continue
setattr(self, key, env_value)
diff --git a/core/registry.py b/core/registry.py
new file mode 100644
index 0000000..a407220
--- /dev/null
+++ b/core/registry.py
@@ -0,0 +1,140 @@
+import asyncio
+import json
+import logging
+import socket
+import time
+import uuid
+import os
+from typing import Dict, List, Optional
+
+import redis.asyncio as redis
+
+logger = logging.getLogger(__name__)
+
+class NodeRegistry:
+ """节点注册与发现"""
+
+ def __init__(self, redis_url: str):
+ self.redis_url = redis_url
+ self.redis: Optional[redis.Redis] = None
+ self.node_id = str(uuid.uuid4())
+ self.host = self._get_local_ip()
+ self.port = int(os.getenv('HTTP_PORT', 8000))
+ self.max_sessions = int(os.getenv('MAX_SESSIONS', 10))
+ self.is_running = False
+
+ def _get_local_ip(self):
+ try:
+ # 获取本机IP，在Docker中这通常是容器IP
+ # 在某些网络配置下可能需要手动指定
+ hostname = socket.gethostname()
+ return socket.gethostbyname(hostname)
+ except:
+ return '127.0.0.1'
+
+ async def connect(self):
+ """连接 Redis"""
+ if not self.redis:
+ self.redis = redis.from_url(self.redis_url, decode_responses=True)
+ logger.info(f"Connected to Redis: {self.redis_url}")
+
+ async def register_node(self):
+ """注册节点"""
+ await self.connect()
+ self.is_running = True
+ asyncio.create_task(self._heartbeat_loop())
+ logger.info(f"Node registered: {self.node_id} ({self.host}:{self.port})")
+
+ async def _heartbeat_loop(self):
+ """心跳循环"""
+ while self.is_running:
+ try:
+ # 更新节点信息
+ node_info = {
+ 'id': self.node_id,
+ 'host': self.host,
+ 'port': self.port,
+ 'max_sessions': self.max_sessions,
+ 'last_seen': time.time()
+ }
+
+ # 使用 Hash 存储节点信息
+ await self.redis.hset('nodes', self.node_id, json.dumps(node_info))
+
+ # 设置过期时间（例如 10 秒后过期，心跳每 5 秒一次）
+ # 注意：Redis Hash 的字段不能单独设置过期，所以我们可以在应用层过滤
+ # 或者使用单独的 key: node:{id} 并设置 expire
+ await self.redis.set(f"node_heartbeat:{self.node_id}", "1", ex=15)
+
+ logger.debug(f"Heartbeat sent for {self.node_id}")
+ except Exception as e:
+ logger.error(f"Heartbeat failed: {e}")
+
+ await asyncio.sleep(5)
+
+ async def update_load(self, active_sessions: int):
+ """更新负载信息"""
+ if not self.redis: return
+ try:
+ await self.redis.hset(f"node_load:{self.node_id}", "active", active_sessions)
+ except Exception as e:
+ logger.error(f"Failed to update load: {e}")
+
+ async def get_best_node(self, exclude_nodes: List[str] = None) -> Optional[Dict]:
+ """获取最佳可用节点 (简单的最少连接数策略)"""
+ await self.connect()
+ exclude_nodes = exclude_nodes or []
+ try:
+ nodes = await self.redis.hgetall('nodes')
+ best_node = None
+ min_load = float('inf')
+
+ for node_id, info_str in nodes.items():
+ if node_id in exclude_nodes:
+ continue
+
+ # 检查心跳
+ if not await self.redis.exists(f"node_heartbeat:{node_id}"):
+ # 清理过期节点
+ await self.redis.hdel('nodes', node_id)
+ continue
+
+ info = json.loads(info_str)
+
+ # 获取负载
+ load = await self.redis.hget(f"node_load:{node_id}", "active")
+ current_load = int(load) if load else 0
+
+ if current_load < info['max_sessions']:
+ if current_load < min_load:
+ min_load = current_load
+ best_node = info
+
+ return best_node
+ except Exception as e:
+ logger.error(f"Failed to get best node: {e}")
+ return None
+
+ async def register_session(self, session_id: str, node_id: str):
+ """注册会话位置"""
+ await self.connect()
+ # session -> node_id
+ await self.redis.set(f"session_route:{session_id}", node_id, ex=3600)
+
+ async def get_session_node(self, session_id: str) -> Optional[Dict]:
+ """获取会话所在的节点信息"""
+ await self.connect()
+ node_id = await self.redis.get(f"session_route:{session_id}")
+ if not node_id:
+ return None
+
+ info_str = await self.redis.hget('nodes', node_id)
+ if info_str:
+ return json.loads(info_str)
+ return None
+
+ async def close(self):
+ self.is_running = False
+ if self.redis:
+ await self.redis.close()
+
diff --git a/docker-compose.yml b/docker-compose.yml
new file mode 100644
index 0000000..4033bdf
--- /dev/null
+++ b/docker-compose.yml
@@ -0,0 +1,52 @@
+version: '3.8'
+
+services:
+ redis:
+ image: redis:7-alpine
+ ports:
+ - "6379:6379"
+
+ # 网关 (Gateway) - 统一入口
+ gateway:
+ build: .
+ ports:
+ - "8000:8000"
+ environment:
+ - REDIS_URL=redis://redis:6379/0
+ - API_KEY=dev-test-key
+ - CLUSTER_SECRET=dev-cluster-secret
+ depends_on:
+ - redis
+ command: python gateway.py
+
+ # 浏览器节点 (Worker)
+ browser-node-1:
+ build: .
+ environment:
+ - HTTP_PORT=8001
+ - MAX_SESSIONS=5
+ - REDIS_URL=redis://redis:6379/0
+ - CLUSTER_SECRET=dev-cluster-secret
+ volumes:
+ - ./log:/app/log
+ depends_on:
+ - redis
+ command: python http_server.py
+
+ # 浏览器节点 2 (Worker) - 模拟扩展
+ browser-node-2:
+ build: .
+ environment:
+ - HTTP_PORT=8002
+ - MAX_SESSIONS=5
+ - REDIS_URL=redis://redis:6379/0
+ - CLUSTER_SECRET=dev-cluster-secret
+ volumes:
+ - ./log:/app/log
+ depends_on:
+ - redis
+ command: python http_server.py
+
+networks:
+ default:
+ driver: bridge
\ No newline at end of file
diff --git a/docs/daily_summary_20251217.md b/docs/daily_summary_20251217.md
new file mode 100644
index 0000000..802a207
--- /dev/null
+++ b/docs/daily_summary_20251217.md
@@ -0,0 +1,59 @@
+# 2025-12-17 工作总结：分布式架构演进
+
+## 1. 工作概览
+今日主要完成了项目从**单体应用**向**分布式微服务架构**的演进。通过引入 Redis 作为服务注册中心，并实现统一网关（Gateway），系统现在具备了水平扩展能力，能够支持大规模调度场景。
+
+## 2. 完成事项
+
+### 2.1 容器化与编排
+- **Dockerfile**: 创建了统一的构建文件，支持 Worker 和 Gateway 的镜像构建。
+- **Docker Compose**: 定义了本地集群拓扑，包含 Redis、Gateway 和多个 Worker 节点，实现了“一键启动”的开发环境。
+- **配置管理**: 改造 `config.py`，支持通过环境变量注入配置（如 `REDIS_URL`, `HTTP_PORT`），适应容器化部署。
+
+### 2.2 服务注册与发现 (Service Discovery)
+- **NodeRegistry (`core/registry.py`)**:
+ - 实现了基于 Redis 的节点注册机制。
+ - 实现了**心跳机制 (Heartbeat)**，自动剔除失效节点。
+ - 实现了**负载上报**，Worker 节点实时更新当前活跃会话数。
+ - 实现了简单的**负载均衡策略**（Least Connections），网关优先选择负载最低的节点。
+
+### 2.3 统一网关 (Gateway)
+- **API 网关 (`gateway.py`)**:
+ - 作为集群的单一入口（Entrypoint），监听 8000 端口。
+ - 实现了 HTTP 请求的透明代理，根据 `session_id` 自动路由到对应 Worker。
+ - **智能路由**: 在创建会话（`POST /api/sessions`）时，自动计算并分发到最佳节点。
+- **WebSocket 代理**:
+ - 实现了 WebSocket 流量的双向转发，确保前端可以无感知地连接到后端任意 Worker 的 CDP 流。
+
+### 2.4 Worker 节点改造
+- 修改 `http_server.py`，在启动时自动连接 Redis 并注册自身。
+- 在会话创建/销毁时同步更新 Redis 中的负载数据。
+
+## 3. 架构现状
+
+```mermaid
+graph TD
+ Client[客户端/前端] --> Gateway[统一网关 :8000]
+ Gateway -->|服务发现 & 路由| Redis[(Redis 注册中心)]
+ Gateway -->|HTTP/WS 代理| Node1[Worker 节点 1 :8001]
+ Gateway -->|HTTP/WS 代理| Node2[Worker 节点 2 :8002]
+ Node1 -.->|心跳 & 负载| Redis
+ Node2 -.->|心跳 & 负载| Redis
+```
+
+## 4. 待办事项 (To-Do)
+
+### 优先事项
+1. **环境验证**: 在真实的 Docker 环境（非沙箱）中运行 `docker-compose up`，验证完整的集群通信链路。
+2. **健壮性增强**:
+ - 网关层增加重试机制和断路器。
+ - 优化 WebSocket 代理的异常处理（如后端节点重启时的重连）。
+3. **安全性**:
+ - 实现网关与 Worker 之间的内部认证（防止绕过网关直接访问 Worker）。
+ - 增加 API Key 认证。
+
+### 长期规划
+- **跨地域调度**: 引入地域标签（Region/Zone），实现就近调度。
+- **K8s 部署**: 编写 Kubernetes 部署清单（Deployment, Service, Ingress）。
+- **监控告警**: 集成 Prometheus + Grafana 监控节点负载和健康状态。
+
diff --git a/docs/daily_summary_20251218.md b/docs/daily_summary_20251218.md
new file mode 100644
index 0000000..788f64d
--- /dev/null
+++ b/docs/daily_summary_20251218.md
@@ -0,0 +1,46 @@
+# 2025-12-18 工作总结：健壮性与安全性增强
+
+## 1. 工作概览
+今日主要针对分布式架构进行了**健壮性 (Robustness)** 和 **安全性 (Security)** 的增强。实现了网关层的重试机制、WebSocket 异常处理，以及服务间和客户端的认证机制。
+
+## 2. 完成事项
+
+### 2.1 健壮性增强 (Gateway Robustness)
+- **HTTP 重试机制 (`gateway.py`)**:
+ - `create_session`: 实现了**3次重试**逻辑。如果首选节点连接失败，会自动将该节点加入排除列表，并尝试获取下一个最佳节点。
+ - `proxy_request`: 针对网络超时（Timeout）和连接错误（ConnectError）实现了重试，提高了瞬时网络抖动下的稳定性。
+- **WebSocket 异常处理**:
+ - 优化了 `websocket_proxy` 的错误捕获。
+ - 实现了更优雅的双向关闭逻辑，当一端断开时，正确关闭另一端并记录日志。
+
+### 2.2 安全性增强 (Security)
+- **内部集群认证 (Cluster Auth)**:
+ - 引入 `X-Cluster-Secret` 头。
+ - **Gateway**: 在转发请求给 Worker 时，自动注入配置的 `CLUSTER_SECRET`。
+ - **Worker (`http_server.py`)**: 在所有敏感接口（创建/关闭会话、控制指令、WebSocket）增加依赖注入 `verify_cluster_secret`，拒绝无密钥的非法访问。
+- **客户端认证 (Client Auth)**:
+ - 引入 `X-API-Key` 头。
+ - **Gateway**: 增加了 `verify_client_auth` 依赖，如果配置了 `API_KEY`，则验证客户端请求。
+
+### 2.3 配置更新
+- `config.py`: 新增 `CLUSTER_SECRET` 和 `API_KEY` 配置项。
+
+## 3. 架构现状更新
+
+```mermaid
+graph TD
+ Client[客户端] -->|X-API-Key| Gateway[统一网关]
+ Gateway -->|X-Cluster-Secret| Node1[Worker 节点 1]
+ Gateway -->|X-Cluster-Secret| Node2[Worker 节点 2]
+
+ subgraph Security
+ Gateway --验证--> Client
+ Node1 --验证--> Gateway
+ Node2 --验证--> Gateway
+ end
+```
+
+## 4. 后续规划
+- **集成测试**: 需要在真实环境中验证 Auth 失败的情况和 Retry 成功的情况。
+- **监控**: 接入 Prometheus 监控重试次数和错误率。
+
diff --git a/docs/remote_control_issues.md b/docs/remote_control_issues.md
new file mode 100644
index 0000000..5929f7d
--- /dev/null
+++ b/docs/remote_control_issues.md
@@ -0,0 +1,98 @@
+# Remote Control 功能问题记录
+
+## 功能概述
+实现浏览器远程 GUI 操作能力和任务重放功能。
+
+## 已完成的工作
+
+### 1. 架构设计
+- WebSocket 接口 `/ws/sessions/{session_id}` 用于实时通信
+- CDP `Page.startScreencast` 实现视频流推送
+- CDP `Input.dispatchMouseEvent` / `Input.dispatchKeyEvent` 处理输入事件
+
+### 2. 前端实现 (`static/remote.html`)
+- 视频流接收和显示
+- 鼠标事件捕获和坐标转换
+- 键盘事件捕获
+- 录制/重放控制界面
+- 调试模式（显示坐标转换信息）
+
+### 3. 后端实现 (`cdp_client.py`, `http_server.py`)
+- `start_screencast` / `stop_screencast` 方法
+- `handle_input_event` 方法（使用 CDP 原生 Input 事件）
+- `replay_events` 方法
+- WebSocket 路由处理
+
+## 当前问题
+
+### 问题1: 点击操作不稳定
+**现象**: 有的区域可以点击，有的区域无法点击
+
+**可能原因**:
+1. 坐标计算仍有偏差
+2. CDP screencast 图片尺寸与 viewport 尺寸的关系不确定
+3. 高 DPI 屏幕的缩放问题
+
+**已尝试的方案**:
+- 使用相对位置计算 `(imgX / rect.width) * deviceWidth`
+- 移除图片的 `max-width: 100%` 限制
+- 使用 CDP 原生 `Input.dispatchMouseEvent` 替代 Playwright 方法
+
+**待验证**:
+- [ ] 检查 CDP screencast metadata 中是否有缩放因子信息
+- [ ] 对比 `screen.naturalWidth` 和 `metadata.deviceWidth` 的关系
+- [ ] 测试固定 viewport 尺寸是否能解决问题
+
+### 问题2: 键盘输入可能不完整
+**现象**: 特殊按键可能无法正确发送
+
+**可能原因**:
+- CDP `Input.dispatchKeyEvent` 需要更多参数（如 `code`, `windowsVirtualKeyCode` 等）
+
+**待修复**:
+- [ ] 补充键盘事件的完整参数
+- [ ] 测试中文输入
+
+## 调试方法
+
+### 前端调试
+1. 勾选页面上的"调试"复选框
+2. 查看右下角的调试信息：
+ - 鼠标位置
+ - 图片内位置
+ - 图片显示尺寸 vs 原始尺寸
+ - viewport 尺寸
+ - 发送坐标
+
+### 后端调试
+查看服务器日志：
+```bash
+tail -f log/http_server.log
+```
+
+## 下一步计划
+
+1. **深入分析坐标问题**
+ - 在调试模式下，对比点击位置和实际响应位置
+ - 检查 CDP screencast 的具体行为
+
+2. **完善键盘输入**
+ - 参考 Playwright 的键盘事件实现
+ - 支持组合键（Ctrl+C 等）
+
+3. **优化用户体验**
+ - 降低视频延迟（调整 screencast 参数）
+ - 添加加载状态指示
+
+## 相关文件
+
+- `static/remote.html` - 前端控制页面
+- `static/test_page.html` - 测试页面
+- `cdp_client.py` - CDP 客户端封装
+- `http_server.py` - HTTP/WebSocket 服务器
+
+## 参考资料
+
+- [CDP Input Domain](https://chromedevtools.github.io/devtools-protocol/tot/Input/)
+- [CDP Page.startScreencast](https://chromedevtools.github.io/devtools-protocol/tot/Page/#method-startScreencast)
+
diff --git a/gateway.py b/gateway.py
new file mode 100644
index 0000000..0802a28
--- /dev/null
+++ b/gateway.py
@@ -0,0 +1,243 @@
+import logging
+import os
+import sys
+from typing import Optional
+
+from fastapi import FastAPI, Request, HTTPException, WebSocket, WebSocketDisconnect, Security, Depends
+from fastapi.responses import JSONResponse, RedirectResponse
+from fastapi.security.api_key import APIKeyHeader
+import httpx
+import websockets
+from websockets.client import connect as ws_connect
+
+from core.registry import NodeRegistry
+from config import get_config
+
+# 配置日志
+logging.basicConfig(
+ level=logging.INFO,
+ format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
+ handlers=[logging.StreamHandler(sys.stdout)]
+)
+logger = logging.getLogger(__name__)
+
+app = FastAPI(
+ title="Browser RPC Gateway",
+ description="Unified Gateway for Browser RPC Cluster",
+ version="1.0.0"
+)
+
+config = get_config()
+registry: Optional[NodeRegistry] = None
+# Disable SSL verification for sandbox compatibility
+http_client = httpx.AsyncClient(verify=False)
+api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)
+
+async def verify_client_auth(api_key: str = Security(api_key_header)):
+ """验证客户端 API Key"""
+ if config.API_KEY and api_key != config.API_KEY:
+ raise HTTPException(status_code=403, detail="Invalid API Key")
+
+def get_upstream_headers(original_headers: dict = None) -> dict:
+ """构造上游请求头，注入内部通信密钥"""
+ headers = dict(original_headers or {})
+ headers["X-Cluster-Secret"] = config.CLUSTER_SECRET
+ # 移除可能引起冲突的 Header
+ headers.pop("host", None)
+ headers.pop("content-length", None)
+ return headers
+
+@app.on_event("startup")
+async def startup_event():
+ global registry
+ if config.REDIS_URL:
+ registry = NodeRegistry(config.REDIS_URL)
+ await registry.connect()
+ logger.info("Gateway connected to Redis")
+
+@app.on_event("shutdown")
+async def shutdown_event():
+ if registry:
+ await registry.close()
+ await http_client.aclose()
+
+@app.post("/api/sessions", dependencies=[Depends(verify_client_auth)])
+async def create_session(request: Request):
+ """创建会话 - 负载均衡路由 (带重试机制)"""
+ exclude_nodes = []
+ last_exception = None
+
+ # 读取一次 body，因为 request.json() 是 async 的且 stream 可能会被消耗
+ try:
+ body = await request.json()
+ except Exception:
+ body = {}
+
+ # 最多尝试 3 个不同节点
+ for attempt in range(3):
+ try:
+ # 获取最佳节点
+ node = await registry.get_best_node(exclude_nodes=exclude_nodes)
+ if not node:
+ break
+
+ # 转发请求
+ target_url = f"http://{node['host']}:{node['port']}/api/sessions"
+
+ logger.info(f"Forwarding create_session to {target_url} (Attempt {attempt+1})")
+
+ # 注入内部通信密钥
+ headers = get_upstream_headers()
+
+ response = await http_client.post(target_url, json=body, headers=headers, timeout=10.0)
+
+ if response.status_code >= 500:
+ # 如果是服务端错误，可能是该节点有问题，尝试其他节点
+ logger.warning(f"Node {node['id']} returned {response.status_code}")
+ exclude_nodes.append(node['id'])
+ continue
+
+ return JSONResponse(content=response.json(), status_code=response.status_code)
+
+ except (httpx.ConnectError, httpx.TimeoutException, httpx.ReadError) as e:
+ logger.warning(f"Failed to connect to node {node.get('id', 'unknown')}: {e}")
+ if node:
+ exclude_nodes.append(node['id'])
+ last_exception = e
+ except Exception as e:
+ logger.error(f"Gateway create session failed: {e}")
+ # 非网络错误直接抛出
+ raise HTTPException(status_code=500, detail=str(e))
+
+ error_msg = f"Failed to create session after 3 attempts. Last error: {last_exception}"
+ logger.error(error_msg)
+ raise HTTPException(status_code=503, detail=error_msg)
+
+async def get_node_for_session(session_id: str):
+ """查找会话所在的节点"""
+ node_info = await registry.get_session_node(session_id)
+ if not node_info:
+ raise HTTPException(status_code=404, detail="Session not found or expired")
+ return node_info
+
+@app.api_route("/api/sessions/{session_id}/{path:path}", methods=["GET", "POST", "DELETE", "PUT"], dependencies=[Depends(verify_client_auth)])
+async def proxy_request(session_id: str, path: str, request: Request):
+ """通用请求代理 - 基于 Session ID 路由"""
+ try:
+ node = await get_node_for_session(session_id)
+ target_url = f"http://{node['host']}:{node['port']}/api/sessions/{session_id}/{path}"
+
+ # 构造请求参数
+ headers = get_upstream_headers(request.headers)
+
+ params = {
+ "method": request.method,
+ "url": target_url,
+ "headers": headers,
+ "params": dict(request.query_params),
+ "timeout": 30.0
+ }
+
+ # 读取 Body
+ if request.method not in ["GET", "HEAD"]:
+ body = await request.body()
+ params["content"] = body
+
+ # 重试逻辑 (针对网络波动)
+ last_error = None
+ for attempt in range(3):
+ try:
+ logger.info(f"Proxying {request.method} {path} to {target_url} (Attempt {attempt+1})")
+ response = await http_client.request(**params)
+
+ # 透传响应状态码和内容
+ content = response.content
+ try:
+ # 尝试解析 JSON，如果不是 JSON 则直接返回内容
+ json_content = response.json()
+ return JSONResponse(content=json_content, status_code=response.status_code)
+ except:
+ from fastapi.responses import Response
+ return Response(content=content, status_code=response.status_code)
+
+ except (httpx.ConnectError, httpx.TimeoutException, httpx.ReadError) as e:
+ logger.warning(f"Proxy request failed (Attempt {attempt+1}): {e}")
+ last_error = e
+ await asyncio.sleep(0.5)
+
+ raise HTTPException(status_code=504, detail=f"Upstream request failed after 3 attempts: {last_error}")
+
+ except HTTPException:
+ raise
+ except Exception as e:
+ logger.error(f"Proxy failed: {e}")
+ raise HTTPException(status_code=500, detail=str(e))
+
+@app.delete("/api/sessions/{session_id}", dependencies=[Depends(verify_client_auth)])
+async def close_session(session_id: str):
+ """关闭会话代理"""
+ # 单独处理 DELETE，因为它是根路径
+ return await proxy_request(session_id, "", Request(scope={"method": "DELETE", "type": "http"}))
+
+# WebSocket 代理
+# 注意：FastAPI 实现 WebSocket 代理比较复杂，这里使用简化的管道模式
+@app.websocket("/ws/sessions/{session_id}")
+async def websocket_proxy(client_ws: WebSocket, session_id: str):
+ await client_ws.accept()
+
+ try:
+ node = await get_node_for_session(session_id)
+ target_ws_url = f"ws://{node['host']}:{node['port']}/ws/sessions/{session_id}"
+
+ logger.info(f"Connecting proxy to backend: {target_ws_url}")
+
+ # 注入内部密钥
+ extra_headers = {"X-Cluster-Secret": config.CLUSTER_SECRET}
+
+ async with ws_connect(target_ws_url, extra_headers=extra_headers) as node_ws:
+ # 双向转发
+ async def forward_to_node():
+ try:
+ while True:
+ data = await client_ws.receive_text()
+ await node_ws.send(data)
+ except WebSocketDisconnect:
+ logger.info(f"Client disconnected for session {session_id}")
+ except Exception as e:
+ logger.error(f"Error forwarding to node: {e}")
+
+ async def forward_to_client():
+ try:
+ while True:
+ data = await node_ws.recv()
+ await client_ws.send_text(data)
+ except websockets.exceptions.ConnectionClosed as e:
+ logger.info(f"Node disconnected for session {session_id}: {e.code}")
+ await client_ws.close(code=1011, reason="Upstream closed")
+ except Exception as e:
+ logger.error(f"Error forwarding to client: {e}")
+
+ # 使用 asyncio.wait 等待任一任务结束
+ done, pending = await asyncio.wait(
+ [asyncio.create_task(forward_to_node()), asyncio.create_task(forward_to_client())],
+ return_when=asyncio.FIRST_COMPLETED
+ )
+
+ # 取消剩余任务
+ for task in pending:
+ task.cancel()
+
+ except HTTPException as e:
+ logger.error(f"WebSocket handshake failed: {e.detail}")
+ await client_ws.close(code=1008, reason=e.detail)
+ except Exception as e:
+ logger.error(f"WebSocket proxy error: {e}")
+ try:
+ await client_ws.close(code=1011, reason="Proxy internal error")
+ except:
+ pass
+
+if __name__ == "__main__":
+ import uvicorn
+ uvicorn.run(app, host="0.0.0.0", port=8000)
+
diff --git a/http_client.py b/http_client.py
index 4b9ba13..2bf6c07 100644
--- a/http_client.py
+++ b/http_client.py
@@ -26,7 +26,8 @@ def __init__(self, base_url: str = 'http://localhost:8000'):
base_url: HTTP 服务器地址，例如 'http://localhost:8000'
"""
self.base_url = base_url.rstrip('/')
- self.client = httpx.AsyncClient(timeout=60.0)
+ # Disable SSL verification for local development/sandbox compatibility
+ self.client = httpx.AsyncClient(timeout=60.0, verify=False)
self.session_id: Optional[str] = None
async def close(self):
diff --git a/http_server.py b/http_server.py
index fd3fe12..dfaaf68 100644
--- a/http_server.py
+++ b/http_server.py
@@ -10,11 +10,15 @@
import sys
from typing import List, Optional, Dict, Any
-from fastapi import FastAPI, HTTPException
+from fastapi import FastAPI, HTTPException, WebSocket, WebSocketDisconnect, Header, Depends
from fastapi.responses import JSONResponse, Response
+from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel, Field
from cdp_client import BrowserPool
+from core.registry import NodeRegistry
+import time
+import os
# 配置日志
import os
@@ -37,8 +41,23 @@
version="1.0.0"
)
+# 挂载静态文件
+static_dir = os.path.join(os.path.dirname(__file__), 'static')
+os.makedirs(static_dir, exist_ok=True)
+app.mount("/static", StaticFiles(directory=static_dir), name="static")
+
# 全局浏览器池
browser_pool: Optional[BrowserPool] = None
+node_registry: Optional[NodeRegistry] = None
+
+
+async def verify_cluster_secret(x_cluster_secret: Optional[str] = Header(None, alias="X-Cluster-Secret")):
+ """验证内部集群通信密钥"""
+ from config import get_config
+ config = get_config()
+ if config.CLUSTER_SECRET:
+ if x_cluster_secret != config.CLUSTER_SECRET:
+ raise HTTPException(status_code=403, detail="Invalid Cluster Secret")
# Pydantic 模型定义
@@ -199,21 +218,33 @@ def get_session(session_id: str):
@app.on_event("startup")
async def startup_event():
- """应用启动时初始化浏览器池"""
- global browser_pool
+ """应用启动时初始化浏览器池和注册节点"""
+ global browser_pool, node_registry
from config import get_config
config = get_config()
browser_pool = BrowserPool(config.MAX_SESSIONS, config.SESSION_TIMEOUT)
logger.info(f"HTTP 服务器启动完成 (最大会话数: {config.MAX_SESSIONS})")
+
+ # 初始化节点注册
+ if config.REDIS_URL:
+ try:
+ node_registry = NodeRegistry(config.REDIS_URL)
+ await node_registry.register_node()
+ logger.info("节点注册成功")
+ except Exception as e:
+ logger.error(f"节点注册失败: {e}")
@app.on_event("shutdown")
async def shutdown_event():
- """应用关闭时清理浏览器池"""
- global browser_pool
+ """应用关闭时清理"""
+ global browser_pool, node_registry
if browser_pool:
- await browser_pool.cleanup()
+ await browser_pool.close_all()
logger.info("浏览器池已清理")
+
+ if node_registry:
+ await node_registry.close()
@app.get("/")
@@ -226,7 +257,7 @@ async def root():
}
-@app.post("/api/sessions", response_model=CreateSessionResponse)
+@app.post("/api/sessions", response_model=CreateSessionResponse, dependencies=[Depends(verify_cluster_secret)])
async def create_session(request: CreateSessionRequest):
"""创建浏览器会话"""
try:
@@ -242,6 +273,12 @@ async def create_session(request: CreateSessionRequest):
height=request.height
)
+ # 更新负载信息
+ if node_registry:
+ await node_registry.update_load(len(browser_pool.sessions))
+ # 注册会话路由
+ await node_registry.register_session(session_id, node_registry.node_id)
+
logger.info(f"创建会话成功: {session_id}")
return CreateSessionResponse(
session_id=session_id,
@@ -253,11 +290,16 @@ async def create_session(request: CreateSessionRequest):
raise HTTPException(status_code=500, detail=f"创建会话失败: {str(e)}")
-@app.delete("/api/sessions/{session_id}", response_model=CloseSessionResponse)
+@app.delete("/api/sessions/{session_id}", response_model=CloseSessionResponse, dependencies=[Depends(verify_cluster_secret)])
async def close_session(session_id: str):
"""关闭浏览器会话"""
try:
success = await browser_pool.close_session(session_id)
+
+ # 更新负载信息
+ if node_registry:
+ await node_registry.update_load(len(browser_pool.sessions))
+
if success:
logger.info(f"关闭会话成功: {session_id}")
return CloseSessionResponse(
@@ -273,7 +315,7 @@ async def close_session(session_id: str):
raise HTTPException(status_code=500, detail=f"关闭会话失败: {str(e)}")
-@app.post("/api/sessions/{session_id}/navigate", response_model=NavigateResponse)
+@app.post("/api/sessions/{session_id}/navigate", response_model=NavigateResponse, dependencies=[Depends(verify_cluster_secret)])
async def navigate(session_id: str, request: NavigateRequest):
"""导航到指定 URL"""
try:
@@ -293,7 +335,7 @@ async def navigate(session_id: str, request: NavigateRequest):
raise HTTPException(status_code=500, detail=f"导航失败: {str(e)}")
-@app.post("/api/sessions/{session_id}/execute", response_model=ExecuteScriptResponse)
+@app.post("/api/sessions/{session_id}/execute", response_model=ExecuteScriptResponse, dependencies=[Depends(verify_cluster_secret)])
async def execute_script(session_id: str, request: ExecuteScriptRequest):
"""执行 JavaScript"""
try:
@@ -315,7 +357,7 @@ async def execute_script(session_id: str, request: ExecuteScriptRequest):
)
-@app.get("/api/sessions/{session_id}/content", response_model=GetPageContentResponse)
+@app.get("/api/sessions/{session_id}/content", response_model=GetPageContentResponse, dependencies=[Depends(verify_cluster_secret)])
async def get_page_content(session_id: str):
"""获取页面内容"""
try:
@@ -335,7 +377,7 @@ async def get_page_content(session_id: str):
raise HTTPException(status_code=500, detail=f"获取失败: {str(e)}")
-@app.post("/api/sessions/{session_id}/network", response_model=GetNetworkRequestsResponse)
+@app.post("/api/sessions/{session_id}/network", response_model=GetNetworkRequestsResponse, dependencies=[Depends(verify_cluster_secret)])
async def get_network_requests(session_id: str, request: GetNetworkRequestsRequest):
"""获取拦截的网络请求"""
try:
@@ -371,7 +413,7 @@ async def get_network_requests(session_id: str, request: GetNetworkRequestsReque
raise HTTPException(status_code=500, detail=f"获取失败: {str(e)}")
-@app.post("/api/sessions/{session_id}/wait", response_model=WaitForElementResponse)
+@app.post("/api/sessions/{session_id}/wait", response_model=WaitForElementResponse, dependencies=[Depends(verify_cluster_secret)])
async def wait_for_element(session_id: str, request: WaitForElementRequest):
"""等待元素出现"""
try:
@@ -390,7 +432,7 @@ async def wait_for_element(session_id: str, request: WaitForElementRequest):
raise HTTPException(status_code=500, detail=f"等待失败: {str(e)}")
-@app.post("/api/sessions/{session_id}/click", response_model=ClickElementResponse)
+@app.post("/api/sessions/{session_id}/click", response_model=ClickElementResponse, dependencies=[Depends(verify_cluster_secret)])
async def click_element(session_id: str, request: ClickElementRequest):
"""点击元素"""
try:
@@ -409,7 +451,7 @@ async def click_element(session_id: str, request: ClickElementRequest):
raise HTTPException(status_code=500, detail=f"点击失败: {str(e)}")
-@app.post("/api/sessions/{session_id}/type", response_model=TypeTextResponse)
+@app.post("/api/sessions/{session_id}/type", response_model=TypeTextResponse, dependencies=[Depends(verify_cluster_secret)])
async def type_text(session_id: str, request: TypeTextRequest):
"""输入文本"""
try:
@@ -428,7 +470,7 @@ async def type_text(session_id: str, request: TypeTextRequest):
raise HTTPException(status_code=500, detail=f"输入失败: {str(e)}")
-@app.post("/api/sessions/{session_id}/screenshot", response_model=TakeScreenshotResponse)
+@app.post("/api/sessions/{session_id}/screenshot", response_model=TakeScreenshotResponse, dependencies=[Depends(verify_cluster_secret)])
async def take_screenshot(session_id: str, request: TakeScreenshotRequest):
"""页面截图"""
try:
@@ -454,7 +496,7 @@ async def take_screenshot(session_id: str, request: TakeScreenshotRequest):
raise HTTPException(status_code=500, detail=f"截图失败: {str(e)}")
-@app.post("/api/sessions/{session_id}/headers", response_model=SetHeadersResponse)
+@app.post("/api/sessions/{session_id}/headers", response_model=SetHeadersResponse, dependencies=[Depends(verify_cluster_secret)])
async def set_headers(session_id: str, request: SetHeadersRequest):
"""设置请求头"""
try:
@@ -473,7 +515,7 @@ async def set_headers(session_id: str, request: SetHeadersRequest):
raise HTTPException(status_code=500, detail=f"设置失败: {str(e)}")
-@app.post("/api/sessions/{session_id}/cookies", response_model=SetCookiesResponse)
+@app.post("/api/sessions/{session_id}/cookies", response_model=SetCookiesResponse, dependencies=[Depends(verify_cluster_secret)])
async def set_cookies(session_id: str, request: SetCookiesRequest):
"""设置 Cookie"""
try:
@@ -534,7 +576,7 @@ async def set_cookies(session_id: str, request: SetCookiesRequest):
raise HTTPException(status_code=500, detail=f"设置失败: {str(e)}")
-@app.get("/api/sessions/{session_id}/cookies", response_model=GetCookiesResponse)
+@app.get("/api/sessions/{session_id}/cookies", response_model=GetCookiesResponse, dependencies=[Depends(verify_cluster_secret)])
async def get_cookies(session_id: str, url: Optional[str] = None):
"""获取 Cookie"""
try:
@@ -568,6 +610,153 @@ async def get_cookies(session_id: str, url: Optional[str] = None):
raise HTTPException(status_code=500, detail=f"获取失败: {str(e)}")
+# 简单的任务存储 (内存中)
+# 实际生产环境应该使用数据库
+recorded_tasks: Dict[str, List[Dict]] = {}
+
+@app.websocket("/ws/sessions/{session_id}", dependencies=[Depends(verify_cluster_secret)])
+async def websocket_endpoint(websocket: WebSocket, session_id: str):
+ """WebSocket 远程控制接口"""
+ await websocket.accept()
+
+ try:
+ # 获取会话 (不使用 get_session 包装器，避免 HTTPException)
+ session = browser_pool.get_session(session_id)
+ if not session:
+ await websocket.close(code=4004, reason="Session not found")
+ return
+
+ # 录制状态
+ is_recording = False
+ current_recording = []
+
+ # 定义发送帧的回调
+ async def send_frame(data, metadata):
+ try:
+ # 顺便检查是否有新标签页，如果有，推送列表
+ # 这是一个简化的做法，理想情况是基于事件
+ # 为了性能，每 60 帧 (约1秒) 检查一次
+ nonlocal frame_counter
+ frame_counter += 1
+ if frame_counter % 60 == 0:
+ pages = await session.get_pages()
+ if pages:
+ active_id = getattr(session.page, '_guid', None)
+ await websocket.send_json({
+ "type": "tabs",
+ "tabs": pages,
+ "activeId": active_id
+ })
+
+ await websocket.send_json({
+ "type": "frame",
+ "data": data,
+ "metadata": metadata
+ })
+ except Exception as e:
+ logger.error(f"WebSocket send frame error: {e}")
+ # 连接断开时可能会报错，这里暂时忽略
+
+ try:
+ frame_counter = 0
+ # 开启投射
+ await session.start_screencast(send_frame)
+ logger.info(f"WebSocket connected for session {session_id}")
+
+ # 初始发送一次标签页列表
+ pages = await session.get_pages()
+ active_id = getattr(session.page, '_guid', None)
+ await websocket.send_json({
+ "type": "tabs",
+ "tabs": pages,
+ "activeId": active_id
+ })
+
+ while True:
+ # 接收前端指令
+ data = await websocket.receive_json()
+ msg_type = data.get('type')
+
+ if msg_type == 'input':
+ # 执行操作
+ event = data.get('event', {})
+ await session.handle_input_event(event)
+
+ # 录制
+ if is_recording:
+ event_record = event.copy()
+ event_record['timestamp'] = time.time()
+ current_recording.append(event_record)
+
+ elif msg_type == 'control':
+ cmd = data.get('command')
+ if cmd == 'start_record':
+ is_recording = True
+ current_recording = []
+ logger.info(f"Session {session_id} started recording")
+ await websocket.send_json({"type": "status", "message": "Recording started"})
+
+ elif cmd == 'stop_record':
+ is_recording = False
+ task_name = data.get('taskName', f"task_{int(time.time())}")
+ recorded_tasks[task_name] = current_recording
+ logger.info(f"Session {session_id} stopped recording, saved as {task_name}")
+ await websocket.send_json({
+ "type": "status",
+ "message": f"Recording saved as {task_name}",
+ "taskName": task_name,
+ "eventCount": len(current_recording)
+ })
+
+ elif cmd == 'replay':
+ task_name = data.get('taskName')
+ if task_name in recorded_tasks:
+ logger.info(f"Session {session_id} replaying task {task_name}")
+ await websocket.send_json({"type": "status", "message": f"Replaying {task_name}..."})
+ # 异步执行重放，避免阻塞 WebSocket 接收循环
+ asyncio.create_task(session.replay_events(recorded_tasks[task_name]))
+ else:
+ await websocket.send_json({"type": "error", "message": f"Task {task_name} not found"})
+
+ elif cmd == 'switch_tab':
+ tab_id = data.get('tabId')
+ if tab_id:
+ await session.switch_to_page(tab_id)
+ # 切换后发送新的标签页列表
+ pages = await session.get_pages()
+ active_id = getattr(session.page, '_guid', None)
+ await websocket.send_json({
+ "type": "tabs",
+ "tabs": pages,
+ "activeId": active_id
+ })
+
+ elif msg_type == 'ping':
+ await websocket.send_json({"type": "pong"})
+
+ # 定时推送标签页信息 (简单的轮询机制，或者在 screencast 帧中附带)
+ # 在这里我们无法轻松插入循环，只能依赖前端交互或事件触发
+ # 改进：修改 _on_new_page 回调机制，允许传入 callback
+
+ except WebSocketDisconnect:
+ logger.info(f"WebSocket disconnected for session {session_id}")
+ except Exception as e:
+ logger.error(f"WebSocket loop error: {e}")
+ finally:
+ # 停止投射
+ try:
+ await session.stop_screencast()
+ except:
+ pass
+
+ except Exception as e:
+ logger.error(f"WebSocket connection error: {e}")
+ try:
+ await websocket.close(code=1011)
+ except:
+ pass
+
+
if __name__ == "__main__":
import uvicorn
from config import get_config
diff --git a/requirements.txt b/requirements.txt
index bbea987..785cc92 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,21 +1 @@
-# 浏览器自动化
-playwright==1.40.0
-playwright-stealth==1.0.6
-
-# gRPC 框架
-grpcio==1.60.0
-grpcio-tools==1.60.0
-protobuf==4.25.1
-
-# 日志
-loguru==0.7.2
-
-# 工具库
-python-dotenv==1.0.0
-
-# HTTP API 框架
-fastapi==0.109.0
-uvicorn[standard]==0.27.0
-httpx==0.26.0
-pydantic==2.5.3
-
+redis==5.0.1
diff --git a/scripts/test_local_cluster.py b/scripts/test_local_cluster.py
new file mode 100644
index 0000000..c508292
--- /dev/null
+++ b/scripts/test_local_cluster.py
@@ -0,0 +1,202 @@
+import asyncio
+import os
+import signal
+import sys
+import time
+import subprocess
+import requests
+import json
+
+# 配置
+REDIS_PORT = 6379
+GATEWAY_PORT = 8000
+NODE1_PORT = 8001
+NODE2_PORT = 8002
+
+processes = []
+
+def start_process(cmd, name, env=None):
+ """启动子进程"""
+ print(f"Starting {name}...")
+ full_env = os.environ.copy()
+ if env:
+ full_env.update(env)
+
+ # 重定向输出到文件
+ log_file = open(f"{name}.log", "w")
+ p = subprocess.Popen(
+ cmd,
+ shell=True,
+ env=full_env,
+ stdout=log_file,
+ stderr=subprocess.STDOUT
+ )
+ processes.append(p)
+ return p
+
+def cleanup():
+ """清理进程"""
+ print("\nCleaning up...")
+ for p in processes:
+ p.terminate()
+ # p.kill() # 如果 terminate 不起作用
+
+ # 尝试杀死可能残留的 Python 进程 (通过端口查找更准确，这里简化处理)
+ subprocess.run("pkill -f 'python gateway.py'", shell=True)
+ subprocess.run("pkill -f 'python http_server.py'", shell=True)
+ print("Cleanup complete.")
+
+async def wait_for_port(port, timeout=30):
+ """等待端口就绪"""
+ start_time = time.time()
+ while time.time() - start_time < timeout:
+ try:
+ # 尝试连接端口
+ reader, writer = await asyncio.open_connection('127.0.0.1', port)
+ writer.close()
+ await writer.wait_closed()
+ return True
+ except:
+ await asyncio.sleep(0.5)
+ return False
+
+async def run_tests():
+ """执行测试"""
+ print("\n=== Starting Integration Tests ===\n")
+
+ # 1. 测试 Gateway 健康检查 (根路径)
+ try:
+ resp = requests.get(f"http://127.0.0.1:{GATEWAY_PORT}/")
+ print(f"Gateway Root Check: {resp.status_code}")
+ if resp.status_code == 200:
+ print(" -> SUCCESS")
+ else:
+ print(" -> FAILED")
+ except Exception as e:
+ print(f"Gateway Root Check FAILED: {e}")
+
+ # 2. 测试创建会话 (无 Auth - 应该失败)
+ try:
+ resp = requests.post(f"http://127.0.0.1:{GATEWAY_PORT}/api/sessions", json={"headless": True})
+ print(f"\nCreate Session (No Auth): {resp.status_code}")
+ # 注意：如果 config.API_KEY 未设置，这可能成功；如果设置了，应该 403
+ # 我们默认 config 中 API_KEY 是 None，除非在 test_env 中设置
+ # 在本次测试脚本中，我们将设置 API_KEY
+ if resp.status_code == 403:
+ print(" -> SUCCESS (Auth working)")
+ else:
+ print(f" -> UNEXPECTED (Expected 403, got {resp.status_code})")
+ except Exception as e:
+ print(f"Create Session (No Auth) FAILED: {e}")
+
+ # 3. 测试创建会话 (有 Auth)
+ session_id = None
+ headers = {"X-API-Key": "test-api-key"}
+ try:
+ resp = requests.post(
+ f"http://127.0.0.1:{GATEWAY_PORT}/api/sessions",
+ json={"headless": True},
+ headers=headers
+ )
+ print(f"\nCreate Session (With Auth): {resp.status_code}")
+ if resp.status_code == 200:
+ data = resp.json()
+ session_id = data.get("session_id")
+ print(f" -> SUCCESS: Session ID {session_id}")
+ else:
+ print(f" -> FAILED: {resp.text}")
+ except Exception as e:
+ print(f"Create Session (With Auth) FAILED: {e}")
+
+ # 4. 如果创建成功，测试导航
+ if session_id:
+ try:
+ resp = requests.post(
+ f"http://127.0.0.1:{GATEWAY_PORT}/api/sessions/{session_id}/navigate",
+ json={"url": "https://example.com"},
+ headers=headers
+ )
+ print(f"\nNavigate: {resp.status_code}")
+ if resp.status_code == 200:
+ print(" -> SUCCESS")
+ else:
+ print(f" -> FAILED: {resp.text}")
+ except Exception as e:
+ print(f"Navigate FAILED: {e}")
+
+ # 5. 关闭会话
+ try:
+ resp = requests.delete(
+ f"http://127.0.0.1:{GATEWAY_PORT}/api/sessions/{session_id}",
+ headers=headers
+ )
+ print(f"\nClose Session: {resp.status_code}")
+ if resp.status_code == 200:
+ print(" -> SUCCESS")
+ else:
+ print(f" -> FAILED")
+ except Exception as e:
+ print(f"Close Session FAILED: {e}")
+
+ print("\n=== Tests Completed ===")
+
+async def main():
+ # 0. 检查 Redis 是否运行 (本地模拟前提)
+ # 如果没有 Redis，这个测试脚本无法完全运行
+ # 这里我们假设用户本地可能有 Redis，或者我们可以尝试跳过 Redis 依赖的某些部分？
+ # 不，核心逻辑依赖 Redis。
+ # 既然 Docker 失败，我们只能寄希望于本地有 Redis 或 python redis 库能连接到某个地方
+ # 如果完全没有 Redis，我们只能测试 Gateway 的启动和 Auth 拒绝，无法测试转发
+
+ # 设置环境变量
+ common_env = {
+ "REDIS_URL": "redis://localhost:6379/0", # 假设本地有 Redis
+ "CLUSTER_SECRET": "test-secret",
+ "API_KEY": "test-api-key"
+ }
+
+ # 启动 Gateway
+ start_process(
+ "python gateway.py",
+ "gateway",
+ env=common_env
+ )
+
+ # 启动 Node 1
+ start_process(
+ "python http_server.py",
+ "node1",
+ env={**common_env, "HTTP_PORT": str(NODE1_PORT)}
+ )
+
+ # 启动 Node 2
+ start_process(
+ "python http_server.py",
+ "node2",
+ env={**common_env, "HTTP_PORT": str(NODE2_PORT)}
+ )
+
+ print("Waiting for services to start...")
+
+ # 等待端口
+ if not await wait_for_port(GATEWAY_PORT):
+ print("Gateway failed to start (timeout)")
+ cleanup()
+ return
+
+ # 等待至少一个 Node 启动 (依赖 Redis 成功连接)
+ # 如果本地没有 Redis，Node 启动会报错，Gateway 也会报错
+ # 我们可以通过检查日志来确认
+
+ await asyncio.sleep(5) # 给一点额外时间
+
+ await run_tests()
+
+ cleanup()
+
+if __name__ == "__main__":
+ try:
+ asyncio.run(main())
+ except KeyboardInterrupt:
+ cleanup()
+
diff --git a/static/diagnose_coordinates.html b/static/diagnose_coordinates.html
new file mode 100644
index 0000000..997a26d
--- /dev/null
+++ b/static/diagnose_coordinates.html
@@ -0,0 +1,103 @@
+
+
+
+
+ 坐标诊断工具
+
+
+
+ 远程控制坐标诊断
+
+
+
+
+
+
+
+ ![Remote Screen]()
+
+
+
+
diff --git a/static/remote.html b/static/remote.html
new file mode 100644
index 0000000..8b13789
--- /dev/null
+++ b/static/remote.html
@@ -0,0 +1 @@
+
diff --git a/static/test_page.html b/static/test_page.html
new file mode 100644
index 0000000..56a4058
--- /dev/null
+++ b/static/test_page.html
@@ -0,0 +1,68 @@
+
+
+
+ Click Test Page
+
+
+
+ 点击测试页面
+
+
+ 点我！
+
+
+
+
+ 事件日志：\n
+
+
+
+