You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
143 lines
3.7 KiB
143 lines
3.7 KiB
import json
|
|
from dataclasses import dataclass
|
|
from typing import Dict, List, Optional, Union
|
|
|
|
from fastapi import APIRouter, WebSocket, WebSocketDisconnect
|
|
from pynput.keyboard import Controller as KeyboardController
|
|
from pynput.keyboard import Key
|
|
from pynput.mouse import Button, Controller
|
|
|
|
router = APIRouter()
|
|
|
|
# 存储活跃的WebSocket连接
|
|
active_connection: Optional[WebSocket] = None
|
|
|
|
|
|
@dataclass
|
|
class WebSocketMessage:
|
|
"""WebSocket消息数据类"""
|
|
|
|
type: str
|
|
msg: str = ""
|
|
title: str = "提示"
|
|
duration: int = 1
|
|
data: Dict = None
|
|
|
|
def __post_init__(self):
|
|
if self.data is None:
|
|
self.data = {"x": 0, "y": 0, "key_str": ""}
|
|
|
|
def to_dict(self) -> Dict:
|
|
"""将消息转换为字典格式"""
|
|
return {
|
|
"type": self.type,
|
|
"msg": self.msg,
|
|
"title": self.title,
|
|
"duration": self.duration,
|
|
"data": self.data,
|
|
}
|
|
|
|
|
|
class WebSocketMessageType:
|
|
"""WebSocket消息类型常量"""
|
|
|
|
# 系统消息类型
|
|
INFO = "info"
|
|
SUCCESS = "success"
|
|
WARNING = "warning"
|
|
ERROR = "error"
|
|
|
|
# 鼠标操作类型
|
|
MOUSE_MOVE = "mouse_move"
|
|
MOUSE_CLICK = "mouse_click"
|
|
|
|
|
|
class MessageSender:
|
|
"""消息发送器,发送给前端"""
|
|
|
|
@staticmethod
|
|
async def send_message(
|
|
ws_data_type: str, msg: str, title: str = "提示", duration: int = 1
|
|
) -> None:
|
|
"""
|
|
发送消息到WebSocket客户端
|
|
|
|
Args:
|
|
ws_data_type: 消息类型
|
|
msg: 消息内容
|
|
title: 消息标题
|
|
duration: 显示持续时间
|
|
"""
|
|
if not active_connection:
|
|
return
|
|
|
|
try:
|
|
message = WebSocketMessage(
|
|
type=ws_data_type, msg=msg, title=title, duration=duration
|
|
)
|
|
await active_connection.send_json(message.to_dict())
|
|
except Exception as e:
|
|
print(f"发送消息失败: {e}")
|
|
|
|
|
|
class GestureHandler:
|
|
"""手势控制器"""
|
|
|
|
def __init__(self):
|
|
self.keyboard = KeyboardController()
|
|
self.mouse = Controller()
|
|
|
|
def move_mouse(self, x: int, y: int) -> None:
|
|
"""移动鼠标到指定位置"""
|
|
print(f"👉 正在移动鼠标到位置: ({x}, {y})")
|
|
self.mouse.position = (x, y)
|
|
|
|
def click_mouse(self) -> None:
|
|
"""点击鼠标左键"""
|
|
print("👉 正在点击鼠标")
|
|
self.mouse.click(Button.left)
|
|
|
|
|
|
|
|
@router.websocket("/ws_wavecontrol")
|
|
async def websocket_endpoint(websocket: WebSocket):
|
|
"""WebSocket端点处理函数"""
|
|
global active_connection
|
|
|
|
await websocket.accept()
|
|
active_connection = websocket
|
|
|
|
gesture_handler = GestureHandler()
|
|
try:
|
|
while True:
|
|
data_str = await websocket.receive_text()
|
|
await _handle_message(data_str, websocket, gesture_handler)
|
|
except WebSocketDisconnect:
|
|
active_connection = None
|
|
except Exception as e:
|
|
print(f"WebSocket处理错误: {e}")
|
|
active_connection = None
|
|
|
|
|
|
async def _handle_message(
|
|
data_str: str,
|
|
websocket: WebSocket,
|
|
gesture_handler: GestureHandler,
|
|
) -> None:
|
|
"""处理WebSocket消息"""
|
|
try:
|
|
message = WebSocketMessage(**json.loads(data_str))
|
|
data = message.data
|
|
print(f"[收到指令] type={message.type}, data={data}") # 调试输出
|
|
|
|
|
|
# 处理鼠标操作
|
|
if message.type == WebSocketMessageType.MOUSE_MOVE:
|
|
gesture_handler.move_mouse(data["x"], data["y"])
|
|
elif message.type == WebSocketMessageType.MOUSE_CLICK:
|
|
gesture_handler.click_mouse()
|
|
|
|
except json.JSONDecodeError:
|
|
print("无效的JSON数据")
|
|
except Exception as e:
|
|
print(f"处理消息失败: {e}")
|