369 lines
15 KiB
Python
369 lines
15 KiB
Python
"""
|
||
刮刮乐远程控制模块
|
||
通过 WebSocket 实时传输页面截图到前端，并接收用户操作
|
||
"""
|
||
|
||
import asyncio
|
||
import base64
|
||
import json
|
||
from typing import Optional, Dict, Any
|
||
from loguru import logger
|
||
from playwright.async_api import Page
|
||
|
||
|
||
class CaptchaRemoteController:
|
||
"""刮刮乐远程控制器"""
|
||
|
||
def __init__(self):
|
||
self.active_sessions: Dict[str, Dict[str, Any]] = {}
|
||
self.websocket_connections: Dict[str, Any] = {}
|
||
|
||
async def create_session(self, session_id: str, page: Page) -> Dict[str, str]:
|
||
"""
|
||
创建远程控制会话
|
||
|
||
Args:
|
||
session_id: 会话ID（通常是用户ID）
|
||
page: Playwright Page 对象
|
||
|
||
Returns:
|
||
包含会话信息的字典
|
||
"""
|
||
# 获取滑块元素位置
|
||
captcha_info = await self._get_captcha_info(page)
|
||
|
||
# 只截取滑块区域，不截取整个页面（性能优化）
|
||
screenshot_bytes = await self._screenshot_captcha_area(page, captcha_info)
|
||
screenshot_base64 = base64.b64encode(screenshot_bytes).decode('utf-8')
|
||
|
||
# 获取视口大小
|
||
try:
|
||
viewport = page.viewport_size
|
||
if viewport is None:
|
||
# 如果没有设置viewport，使用默认值或通过JS获取
|
||
viewport = await page.evaluate("() => ({width: window.innerWidth, height: window.innerHeight})")
|
||
except:
|
||
viewport = {'width': 1280, 'height': 720} # 默认值
|
||
|
||
# 存储会话
|
||
self.active_sessions[session_id] = {
|
||
'page': page,
|
||
'screenshot': screenshot_base64,
|
||
'captcha_info': captcha_info,
|
||
'completed': False,
|
||
'viewport': viewport
|
||
}
|
||
|
||
logger.info(f"✅ 创建远程控制会话: {session_id}")
|
||
|
||
return {
|
||
'session_id': session_id,
|
||
'screenshot': screenshot_base64,
|
||
'captcha_info': captcha_info,
|
||
'viewport': self.active_sessions[session_id]['viewport']
|
||
}
|
||
|
||
async def _screenshot_captcha_area(self, page: Page, captcha_info: Dict[str, Any]) -> bytes:
|
||
"""截取整个验证码容器区域"""
|
||
try:
|
||
if captcha_info and 'x' in captcha_info:
|
||
# 直接截取整个容器，稍微留一点边距
|
||
x = max(0, captcha_info['x'] - 10)
|
||
y = max(0, captcha_info['y'] - 10)
|
||
width = captcha_info['width'] + 20
|
||
height = captcha_info['height'] + 20
|
||
|
||
# 截取整个验证码容器
|
||
screenshot_bytes = await page.screenshot(
|
||
type='jpeg',
|
||
quality=80, # 验证码区域用高质量
|
||
clip={
|
||
'x': x,
|
||
'y': y,
|
||
'width': width,
|
||
'height': height
|
||
}
|
||
)
|
||
logger.info(f"✅ 截取验证码容器: {width}x{height} (包含完整验证码)")
|
||
return screenshot_bytes
|
||
else:
|
||
# 如果没有找到滑块，截取整个页面
|
||
logger.warning("未找到滑块位置，截取整个页面")
|
||
return await page.screenshot(type='jpeg', quality=75, full_page=False)
|
||
|
||
except Exception as e:
|
||
logger.warning(f"截取滑块区域失败，使用全页面: {e}")
|
||
return await page.screenshot(type='jpeg', quality=75, full_page=False)
|
||
|
||
async def _get_captcha_info(self, page: Page) -> Dict[str, Any]:
|
||
"""获取滑块验证码信息（查找整个容器）"""
|
||
try:
|
||
# 优先查找整个验证码容器（不是按钮）
|
||
container_selectors = [
|
||
'#nocaptcha', # 完整的验证码容器
|
||
'.scratch-captcha-container',
|
||
'[id*="captcha"]',
|
||
'.nc-container'
|
||
]
|
||
|
||
# 先在主页面查找
|
||
for selector in container_selectors:
|
||
try:
|
||
element = await page.query_selector(selector)
|
||
if element:
|
||
box = await element.bounding_box()
|
||
if box and box['width'] > 100 and box['height'] > 100: # 确保找到的是容器
|
||
logger.info(f"✅ 在主页面找到验证码容器: {selector}, 大小: {box['width']}x{box['height']}")
|
||
return {
|
||
'selector': selector,
|
||
'x': box['x'],
|
||
'y': box['y'],
|
||
'width': box['width'],
|
||
'height': box['height'],
|
||
'in_iframe': False
|
||
}
|
||
except Exception as e:
|
||
logger.debug(f"检查选择器 {selector} 失败: {e}")
|
||
continue
|
||
|
||
# 在 iframe 中查找
|
||
frames = page.frames
|
||
for frame in frames:
|
||
if frame != page.main_frame:
|
||
for selector in container_selectors:
|
||
try:
|
||
element = await frame.query_selector(selector)
|
||
if element:
|
||
box = await element.bounding_box()
|
||
if box and box['width'] > 100 and box['height'] > 100:
|
||
logger.info(f"✅ 在iframe找到验证码容器: {selector}, 大小: {box['width']}x{box['height']}")
|
||
return {
|
||
'selector': selector,
|
||
'x': box['x'],
|
||
'y': box['y'],
|
||
'width': box['width'],
|
||
'height': box['height'],
|
||
'in_iframe': True
|
||
# 注意：不保存 frame 对象，因为不能被 JSON 序列化
|
||
}
|
||
except Exception as e:
|
||
logger.debug(f"iframe检查选择器 {selector} 失败: {e}")
|
||
continue
|
||
|
||
logger.warning("⚠️ 未找到验证码容器")
|
||
return None
|
||
|
||
except Exception as e:
|
||
logger.error(f"获取滑块信息失败: {e}")
|
||
return None
|
||
|
||
async def update_screenshot(self, session_id: str, quality: int = 75) -> Optional[str]:
|
||
"""更新会话的截图（截取整个验证码容器）"""
|
||
if session_id not in self.active_sessions:
|
||
return None
|
||
|
||
try:
|
||
page = self.active_sessions[session_id]['page']
|
||
captcha_info = self.active_sessions[session_id].get('captcha_info')
|
||
|
||
# 截取整个验证码容器
|
||
if captcha_info and 'x' in captcha_info:
|
||
x = max(0, captcha_info['x'] - 10)
|
||
y = max(0, captcha_info['y'] - 10)
|
||
width = captcha_info['width'] + 20
|
||
height = captcha_info['height'] + 20
|
||
|
||
screenshot_bytes = await page.screenshot(
|
||
type='jpeg',
|
||
quality=quality,
|
||
clip={'x': x, 'y': y, 'width': width, 'height': height}
|
||
)
|
||
else:
|
||
# 降级方案：截取整个页面
|
||
screenshot_bytes = await page.screenshot(
|
||
type='jpeg',
|
||
quality=quality,
|
||
full_page=False
|
||
)
|
||
|
||
screenshot_base64 = base64.b64encode(screenshot_bytes).decode('utf-8')
|
||
self.active_sessions[session_id]['screenshot'] = screenshot_base64
|
||
return screenshot_base64
|
||
|
||
except Exception as e:
|
||
logger.error(f"更新截图失败: {e}")
|
||
return None
|
||
|
||
async def handle_mouse_event(self, session_id: str, event_type: str, x: int, y: int) -> bool:
|
||
"""
|
||
处理鼠标事件
|
||
|
||
Args:
|
||
session_id: 会话ID
|
||
event_type: 事件类型 (down/move/up)
|
||
x: X坐标
|
||
y: Y坐标
|
||
|
||
Returns:
|
||
是否成功
|
||
"""
|
||
if session_id not in self.active_sessions:
|
||
logger.warning(f"会话不存在: {session_id}")
|
||
return False
|
||
|
||
try:
|
||
page = self.active_sessions[session_id]['page']
|
||
|
||
if event_type == 'down':
|
||
await page.mouse.move(x, y)
|
||
await page.mouse.down()
|
||
logger.debug(f"鼠标按下: ({x}, {y})")
|
||
|
||
elif event_type == 'move':
|
||
await page.mouse.move(x, y)
|
||
logger.debug(f"鼠标移动: ({x}, {y})")
|
||
|
||
elif event_type == 'up':
|
||
await page.mouse.up()
|
||
logger.debug(f"鼠标释放: ({x}, {y})")
|
||
|
||
else:
|
||
logger.warning(f"未知事件类型: {event_type}")
|
||
return False
|
||
|
||
return True
|
||
|
||
except Exception as e:
|
||
logger.error(f"处理鼠标事件失败: {e}")
|
||
return False
|
||
|
||
async def check_completion(self, session_id: str) -> bool:
|
||
"""检查验证是否完成（更严格的判断）"""
|
||
if session_id not in self.active_sessions:
|
||
return False
|
||
|
||
try:
|
||
page = self.active_sessions[session_id]['page']
|
||
|
||
# 多个选择器检查，确保更准确
|
||
captcha_selectors = [
|
||
'#nocaptcha',
|
||
'#scratch-captcha-btn',
|
||
'.scratch-captcha-container',
|
||
'.scratch-captcha-slider'
|
||
]
|
||
|
||
found_visible_captcha = False
|
||
|
||
# 检查主页面
|
||
for selector in captcha_selectors:
|
||
try:
|
||
element = await page.query_selector(selector)
|
||
if element:
|
||
is_visible = await element.is_visible()
|
||
if is_visible:
|
||
logger.debug(f"主页面发现可见滑块: {selector}")
|
||
found_visible_captcha = True
|
||
break
|
||
except:
|
||
continue
|
||
|
||
if found_visible_captcha:
|
||
return False
|
||
|
||
# 检查所有 iframe
|
||
frames = page.frames
|
||
for frame in frames:
|
||
if frame != page.main_frame:
|
||
for selector in captcha_selectors:
|
||
try:
|
||
element = await frame.query_selector(selector)
|
||
if element:
|
||
is_visible = await element.is_visible()
|
||
if is_visible:
|
||
logger.debug(f"iframe中发现可见滑块: {selector}")
|
||
found_visible_captcha = True
|
||
break
|
||
except:
|
||
continue
|
||
if found_visible_captcha:
|
||
break
|
||
|
||
if found_visible_captcha:
|
||
return False
|
||
|
||
# 额外检查：看页面内容是否还包含滑块相关文字
|
||
try:
|
||
page_content = await page.content()
|
||
captcha_keywords = ['scratch-captcha', 'nocaptcha', 'slider-btn']
|
||
|
||
# 如果页面中仍然有大量滑块相关内容，可能还未完成
|
||
keyword_count = sum(1 for kw in captcha_keywords if kw in page_content)
|
||
if keyword_count >= 2:
|
||
logger.debug(f"页面中仍有 {keyword_count} 个滑块关键词")
|
||
return False
|
||
except:
|
||
pass
|
||
|
||
# 所有检查都通过，认为验证完成
|
||
logger.success(f"✅ 验证完成（所有滑块元素已消失）: {session_id}")
|
||
self.active_sessions[session_id]['completed'] = True
|
||
return True
|
||
|
||
except Exception as e:
|
||
logger.error(f"检查完成状态失败: {e}")
|
||
# 出错时返回 False，不要误判为成功
|
||
return False
|
||
|
||
def is_completed(self, session_id: str) -> bool:
|
||
"""检查会话是否已完成"""
|
||
if session_id not in self.active_sessions:
|
||
return False
|
||
return self.active_sessions[session_id].get('completed', False)
|
||
|
||
def session_exists(self, session_id: str) -> bool:
|
||
"""检查会话是否存在"""
|
||
return session_id in self.active_sessions
|
||
|
||
async def close_session(self, session_id: str):
|
||
"""关闭会话"""
|
||
if session_id in self.active_sessions:
|
||
del self.active_sessions[session_id]
|
||
logger.info(f"🔒 关闭远程控制会话: {session_id}")
|
||
|
||
async def auto_refresh_screenshot(self, session_id: str, interval: float = 1.0):
|
||
"""自动刷新截图（优化版：按需更新）"""
|
||
last_update_time = asyncio.get_event_loop().time()
|
||
|
||
while session_id in self.active_sessions and not self.is_completed(session_id):
|
||
try:
|
||
current_time = asyncio.get_event_loop().time()
|
||
|
||
# 使用自适应刷新：空闲时降低频率
|
||
if current_time - last_update_time >= interval:
|
||
screenshot = await self.update_screenshot(session_id, quality=55) # 降低质量提升性能
|
||
|
||
if screenshot and session_id in self.websocket_connections:
|
||
try:
|
||
ws = self.websocket_connections[session_id]
|
||
await ws.send_json({
|
||
'type': 'screenshot_update',
|
||
'screenshot': screenshot
|
||
})
|
||
last_update_time = current_time
|
||
except:
|
||
# WebSocket 可能已断开
|
||
break
|
||
|
||
# 降低检查频率，减少 CPU 使用
|
||
await asyncio.sleep(0.5)
|
||
|
||
except Exception as e:
|
||
logger.error(f"自动刷新截图失败: {e}")
|
||
await asyncio.sleep(1) # 出错时等待更长时间
|
||
|
||
|
||
# 全局实例
|
||
captcha_controller = CaptchaRemoteController()
|
||
|