xingluo/computer-craft-web-file
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 3 Wiki Activity

更改ws为http

Browse Source
This commit is contained in:
nnwang
2025-12-05 19:02:19 +08:00
parent d3faa4b74b
commit 258bc8915a
5 changed files with 554 additions and 1050 deletions
Download Patch File Download Diff File Expand all files Collapse all files

545
PyServer/main.py
View File

@@ -1,14 +1,11 @@
import asyncio
import json
import logging
import uuid
import time
import os
from datetime import datetime, timedelta
from typing import Dict, Any, Set
import threading
from datetime import datetime
from typing import Dict, Any, List
from http.server import HTTPServer, BaseHTTPRequestHandler
import websockets
from urllib.parse import parse_qs, urlparse
import mimetypes
import re
@@ -19,9 +16,10 @@ logger = logging.getLogger(__name__)
# 存储房间信息
rooms = {}
connected_clients = {}
ws_port = 81 # ws服务外部端口
# 前端到客户端的消息队列
frontend_to_client_queues = {}
# 客户端到前端的消息队列
client_to_frontend_queues = {}
# 静态文件目录
STATIC_DIR = os.path.join(os.path.dirname(__file__), 'static')
@@ -33,67 +31,33 @@ class Room:
def __init__(self, room_id: str, server_host: str):
self.room_id = room_id
self.created_at = datetime.now()
self.last_activity = datetime.now()
self.clients: Set[str] = set()
# 从host中移除端口号
host_without_port = re.sub(r':\d+$', '', server_host)
# 使用80端口
self.ws_url = f"ws://{host_without_port}:{ws_port}"
self.frontend_url = f"http://{server_host}/?id={room_id}&ws={self.ws_url}"
def add_client(self, client_id: str):
self.clients.add(client_id)
self.last_activity = datetime.now()
logger.info(f"客户端 {client_id} 加入房间 {self.room_id}, 当前客户端数: {len(self.clients)}")
def remove_client(self, client_id: str):
if client_id in self.clients:
self.clients.remove(client_id)
self.last_activity = datetime.now()
logger.info(f"客户端 {client_id} 离开房间 {self.room_id}, 剩余客户端数: {len(self.clients)}")
def is_empty(self) -> bool:
return len(self.clients) == 0
self.frontend_url = f"http://{server_host}/?id={room_id}"
def to_dict(self) -> Dict[str, Any]:
return {
'room_id': self.room_id,
'frontend_url': self.frontend_url,
'ws_url': self.ws_url,
'client_count': len(self.clients),
'created_at': self.created_at.isoformat(),
'last_activity': self.last_activity.isoformat()
'created_at': self.created_at.isoformat()
}
def cleanup_empty_rooms():
"""定期清理空房间"""
while True:
time.sleep(300) # 每5分钟检查一次
current_time = datetime.now()
empty_rooms = []
for room_id, room in list(rooms.items()):
if room.is_empty() and current_time - room.last_activity > timedelta(minutes=10):
empty_rooms.append(room_id)
for room_id in empty_rooms:
if room_id in rooms:
del rooms[room_id]
logger.info(f"清理空房间: {room_id}")
def get_frontend_to_client_queue(room_id: str) -> List[Dict[str, Any]]:
if room_id not in frontend_to_client_queues:
frontend_to_client_queues[room_id] = []
return frontend_to_client_queues[room_id]
# 启动清理线程
cleanup_thread = threading.Thread(target=cleanup_empty_rooms, daemon=True)
cleanup_thread.start()
def get_client_to_frontend_queue(room_id: str) -> List[Dict[str, Any]]:
if room_id not in client_to_frontend_queues:
client_to_frontend_queues[room_id] = []
return client_to_frontend_queues[room_id]
class HTTPHandler(BaseHTTPRequestHandler):
def do_GET(self):
"""处理HTTP GET请求"""
try:
logger.info(f"收到GET请求: {self.path} from {self.client_address[0]}")
logger.info(f"请求头: {dict(self.headers)}")
parsed_path = urlparse(self.path)
path = parsed_path.path
query_params = parse_qs(parsed_path.query)
@@ -119,13 +83,22 @@ class HTTPHandler(BaseHTTPRequestHandler):
def do_POST(self):
"""处理HTTP POST请求"""
try:
logger.info(f"收到POST请求: {self.path} from {self.client_address[0]}")
content_length = int(self.headers.get('Content-Length', 0))
post_data = self.rfile.read(content_length) if content_length > 0 else b'{}'
parsed_path = urlparse(self.path)
path = parsed_path.path
if path == '/api/room':
self.handle_create_room()
self.handle_create_room(post_data)
elif path == '/api/frontend/send':
self.handle_frontend_send_message(post_data)
elif path == '/api/frontend/receive':
self.handle_frontend_receive_message(post_data)
elif path == '/api/client/send':
self.handle_client_send_message(post_data)
elif path == '/api/client/receive':
self.handle_client_receive_message(post_data)
else:
self.send_error(404, "Not Found")
except Exception as e:
@@ -135,7 +108,6 @@ class HTTPHandler(BaseHTTPRequestHandler):
def handle_root_path(self, query_params: Dict[str, Any]):
"""处理根路径请求"""
room_id = query_params.get('id', [None])[0]
ws_url = query_params.get('ws', [None])[0]
if room_id:
# 有房间ID参数，直接返回前端页面
@@ -157,8 +129,8 @@ class HTTPHandler(BaseHTTPRequestHandler):
logger.info(f"通过根路径创建新房间: {room_id}")
# 重定向到带房间ID和WebSocket URL的URL
redirect_url = f'/?id={room_id}&ws=ws://{host_without_port}:{ws_port}'
# 重定向到带房间ID的URL
redirect_url = f'/?id={room_id}'
self.send_response(302)
self.send_header('Location', redirect_url)
self.end_headers()
@@ -169,14 +141,11 @@ class HTTPHandler(BaseHTTPRequestHandler):
def handle_static_file(self, path: str):
"""处理静态文件请求"""
logger.info(f"处理静态文件请求: {path}")
# 安全检查：防止路径遍历攻击
if '..' in path:
logger.warning(f"检测到可疑路径: {path}")
self.send_error(403, "Forbidden: Path traversal not allowed")
return
# 规范化路径
if path == '/':
path = '/index.html'
@@ -188,29 +157,20 @@ class HTTPHandler(BaseHTTPRequestHandler):
# 构建完整文件路径
full_path = os.path.join(STATIC_DIR, file_path)
logger.info(f"尝试访问文件: {full_path}")
# 如果是目录，尝试查找index.html
if os.path.isdir(full_path):
index_path = os.path.join(full_path, 'index.html')
if os.path.exists(index_path):
full_path = index_path
logger.info(f"重定向到目录索引文件: {index_path}")
else:
logger.warning(f"目录不存在索引文件: {full_path}")
self.send_error(404, "Directory index not found")
return
# 检查文件是否存在且是普通文件
if not os.path.exists(full_path):
logger.warning(f"文件不存在: {full_path}")
if not os.path.exists(full_path) or not os.path.isfile(full_path):
self.send_error(404, f"File not found: {path}")
return
if not os.path.isfile(full_path):
logger.warning(f"路径不是文件: {full_path}")
self.send_error(403, "Not a file")
return
try:
# 读取文件内容
@@ -222,21 +182,15 @@ class HTTPHandler(BaseHTTPRequestHandler):
if mime_type is None:
mime_type = 'application/octet-stream'
logger.info(f"成功读取文件: {full_path}, 大小: {len(content)} bytes, MIME类型: {mime_type}")
# 发送响应头
self.send_response(200)
self.send_header('Content-Type', mime_type)
self.send_header('Content-Length', str(len(content)))
# 添加缓存控制头（可选）
self.send_header('Cache-Control', 'public, max-age=3600') # 缓存1小时
self.send_header('Cache-Control', 'public, max-age=3600')
self.end_headers()
# 发送文件内容
self.wfile.write(content)
logger.info(f"文件发送完成: {full_path}")
except Exception as e:
logger.error(f"读取或发送文件失败: {e}")
@@ -246,7 +200,7 @@ class HTTPHandler(BaseHTTPRequestHandler):
"""服务静态文件（内部方法）"""
self.handle_static_file(path)
def handle_create_room(self):
def handle_create_room(self, post_data=None):
"""创建新房间"""
try:
# 生成唯一房间ID
@@ -265,8 +219,7 @@ class HTTPHandler(BaseHTTPRequestHandler):
response = {
'success': True,
'room_id': room_id,
'frontend_url': room.frontend_url,
'ws_url': room.ws_url
'frontend_url': room.frontend_url
}
self.send_response(200)
@@ -279,37 +232,23 @@ class HTTPHandler(BaseHTTPRequestHandler):
logger.error(f"创建房间失败: {e}")
self.send_error(500, str(e))
def handle_get_room(self, room_id: str):
"""获取房间信息"""
def handle_frontend_send_message(self, post_data):
"""前端发送消息到客户端"""
try:
if room_id not in rooms:
self.send_error(404, '房间不存在')
data = json.loads(post_data.decode('utf-8'))
room_id = data.get('room_id')
message = data.get('message')
if not room_id or not message:
self.send_error(400, "需要room_id和message参数")
return
room = rooms[room_id]
response = {'success': True, 'data': room.to_dict()}
self.send_response(200)
self.send_header('Content-Type', 'application/json')
self.send_header('Access-Control-Allow-Origin', '*')
self.end_headers()
self.wfile.write(json.dumps(response).encode())
except Exception as e:
logger.error(f"获取房间信息失败: {e}")
self.send_error(500, str(e))
def handle_list_rooms(self):
"""列出所有活跃房间"""
try:
active_rooms = [room.to_dict() for room in rooms.values() if not room.is_empty()]
queue = get_frontend_to_client_queue(room_id)
queue.append(message)
response = {
'success': True,
'data': {
'total_rooms': len(active_rooms),
'rooms': active_rooms
}
'message': '消息已发送到客户端队列'
}
self.send_response(200)
@@ -319,7 +258,108 @@ class HTTPHandler(BaseHTTPRequestHandler):
self.wfile.write(json.dumps(response).encode())
except Exception as e:
logger.error(f"获取房间列表失败: {e}")
logger.error(f"前端发送消息失败: {e}")
self.send_error(500, str(e))
def handle_frontend_receive_message(self, post_data):
"""前端接收来自客户端的消息"""
try:
data = json.loads(post_data.decode('utf-8'))
room_id = data.get('room_id')
if not room_id:
self.send_error(400, "需要room_id参数")
return
queue = get_client_to_frontend_queue(room_id)
if queue:
# 返回队列中的第一个消息
message = queue.pop(0)
response = {
'success': True,
'message': message
}
else:
# 没有消息
response = {
'success': True,
'message': None
}
self.send_response(200)
self.send_header('Content-Type', 'application/json')
self.send_header('Access-Control-Allow-Origin', '*')
self.end_headers()
self.wfile.write(json.dumps(response).encode())
except Exception as e:
logger.error(f"前端接收消息失败: {e}")
self.send_error(500, str(e))
def handle_client_send_message(self, post_data):
"""客户端发送消息到前端"""
try:
data = json.loads(post_data.decode('utf-8'))
room_id = data.get('room_id')
message = data.get('message')
if not room_id or not message:
self.send_error(400, "需要room_id和message参数")
return
queue = get_client_to_frontend_queue(room_id)
queue.append(message)
response = {
'success': True,
'message': '消息已发送到前端队列'
}
self.send_response(200)
self.send_header('Content-Type', 'application/json')
self.send_header('Access-Control-Allow-Origin', '*')
self.end_headers()
self.wfile.write(json.dumps(response).encode())
except Exception as e:
logger.error(f"客户端发送消息失败: {e}")
self.send_error(500, str(e))
def handle_client_receive_message(self, post_data):
"""客户端接收来自前端的消息"""
try:
data = json.loads(post_data.decode('utf-8'))
room_id = data.get('room_id')
if not room_id:
self.send_error(400, "需要room_id参数")
return
queue = get_frontend_to_client_queue(room_id)
if queue:
# 返回队列中的第一个消息
message = queue.pop(0)
response = {
'success': True,
'message': message
}
else:
# 没有消息
response = {
'success': True,
'message': None
}
self.send_response(200)
self.send_header('Content-Type', 'application/json')
self.send_header('Access-Control-Allow-Origin', '*')
self.end_headers()
self.wfile.write(json.dumps(response).encode())
except Exception as e:
logger.error(f"客户端接收消息失败: {e}")
self.send_error(500, str(e))
def log_message(self, format, *args):
@@ -328,300 +368,15 @@ class HTTPHandler(BaseHTTPRequestHandler):
self.log_date_time_string(),
format % args))
class WebSocketHTTPRequestHandler(HTTPHandler):
"""支持WebSocket升级的HTTP请求处理器"""
def do_GET(self):
"""处理GET请求，支持WebSocket升级"""
if self.headers.get('Upgrade') == 'websocket':
self.handle_websocket_upgrade()
else:
super().do_GET()
def handle_websocket_upgrade(self):
"""处理WebSocket升级请求"""
# 这里我们只是记录日志，实际的WebSocket处理在websockets库中完成
logger.info(f"WebSocket连接请求: {self.path}")
self.send_error(426, "WebSocket upgrade required") # 这个错误不会被触发，因为websockets库会拦截请求
async def handle_websocket(websocket):
"""处理WebSocket连接 - 修复版本，移除了path参数"""
client_id = str(uuid.uuid4())
room_id = None
client_type = 'unknown'
try:
# 发送连接成功消息
await send_message(websocket, {
'type': 'connected',
'message': '连接成功',
'client_id': client_id,
'timestamp': datetime.now().isoformat()
})
logger.info(f"客户端连接: {client_id}")
# 处理消息
async for message in websocket:
try:
data = json.loads(message)
# 处理消息并获取房间和客户端类型信息
result = await handle_websocket_message(websocket, client_id, data)
if result:
room_id, client_type = result
except json.JSONDecodeError:
logger.error(f"消息格式错误: {message}")
except Exception as e:
logger.error(f"处理消息错误: {e}")
except websockets.exceptions.ConnectionClosed:
logger.info(f"客户端断开连接: {client_id}")
finally:
# 清理连接
if client_id in connected_clients:
info = connected_clients.pop(client_id)
room_id = info.get('room_id')
logger.info(f"清理客户端: {client_id}, 房间: {room_id}")
if room_id and room_id in rooms:
room = rooms[room_id]
room.remove_client(client_id)
# 通知房间内其他客户端
await broadcast_to_room(room_id, {
'type': 'user_left',
'client_id': client_id,
'message': '用户离开房间',
'room_size': len(room.clients),
'timestamp': datetime.now().isoformat()
}, exclude_client=client_id)
# 额外检查：即使不在 connected_clients 中，如果还在房间里也要移除
elif room_id and room_id in rooms:
room = rooms[room_id]
room.remove_client(client_id)
# 通知房间内其他客户端
await broadcast_to_room(room_id, {
'type': 'user_left',
'client_id': client_id,
'message': '用户离开房间',
'room_size': len(room.clients),
'timestamp': datetime.now().isoformat()
}, exclude_client=client_id)
async def handle_websocket_message(websocket, client_id: str, data: Dict[str, Any]):
"""处理WebSocket消息"""
message_type = data.get('type')
if message_type == 'join_room':
return await handle_join_room(websocket, client_id, data)
elif message_type == 'file_operation':
await handle_file_operation(websocket, client_id, data)
elif message_type == 'file_operation_response':
await handle_file_operation_response(websocket, client_id, data)
elif message_type == 'leave_room':
await handle_leave_room(websocket, client_id, data)
elif message_type == 'ping':
await send_message(websocket, {
'type': 'pong',
'message': 'pong',
'timestamp': datetime.now().isoformat()
})
else:
logger.warning(f"未知消息类型: {message_type}")
async def handle_join_room(websocket, client_id: str, data: Dict[str, Any]):
"""处理加入房间请求"""
room_id = data.get('room_id')
client_type = data.get('client_type', 'unknown')
if not room_id:
await send_message(websocket, {
'type': 'error',
'message': '房间ID不能为空'
})
return
if room_id not in rooms:
await send_message(websocket, {
'type': 'error',
'message': '房间不存在'
})
return
room = rooms[room_id]
room.add_client(client_id)
# 存储客户端信息
connected_clients[client_id] = {
'websocket': websocket,
'room_id': room_id,
'client_type': client_type
}
logger.info(f"客户端 {client_id} ({client_type}) 加入房间 {room_id}")
# 发送加入成功消息
await send_message(websocket, {
'type': 'joined_room',
'room_id': room_id,
'message': f'成功加入房间 {room_id}',
'client_count': len(room.clients),
'timestamp': datetime.now().isoformat()
})
# 通知房间内其他客户端
await broadcast_to_room(room_id, {
'type': 'user_joined',
'client_id': client_id,
'client_type': client_type,
'message': '新用户加入房间',
'room_size': len(room.clients),
'timestamp': datetime.now().isoformat()
}, exclude_client=client_id)
# 返回房间ID和客户端类型，用于finally块清理
return room_id, client_type
async def handle_leave_room(websocket, client_id: str, data: Dict[str, Any]):
"""处理离开房间请求"""
room_id = data.get('room_id')
if not room_id:
room_id = connected_clients.get(client_id, {}).get('room_id')
if room_id and room_id in rooms:
room = rooms[room_id]
room.remove_client(client_id)
# 从连接客户端中移除
if client_id in connected_clients:
del connected_clients[client_id]
# 通知房间内其他客户端
await broadcast_to_room(room_id, {
'type': 'user_left',
'client_id': client_id,
'message': '用户离开房间',
'room_size': len(room.clients),
'timestamp': datetime.now().isoformat()
}, exclude_client=client_id)
logger.info(f"客户端 {client_id} 主动离开房间 {room_id}")
async def handle_file_operation(websocket, client_id: str, data: Dict[str, Any]):
"""处理文件操作请求"""
room_id = data.get('room_id')
request_id = data.get('requestId')
operation_type = data.get('operation_type')
if not room_id or room_id not in rooms:
await send_message(websocket, {
'type': 'error',
'message': '无效的房间ID',
'requestId': request_id
})
return
# 记录操作
logger.info(f"收到文件操作请求: {operation_type} from {client_id} in room {room_id}")
# 查找文件客户端
file_clients = []
for cid, client_info in connected_clients.items():
if client_info['room_id'] == room_id and cid != client_id:
if client_info.get('client_type') == 'file_client':
file_clients.append(client_info['websocket'])
if file_clients:
# 转发给文件客户端
data['sender_id'] = client_id
await send_message(file_clients[0], {
'type': 'file_operation_request',
**data
})
else:
# 没有文件客户端，返回错误
await send_message(websocket, {
'type': 'file_operation_response',
'requestId': request_id,
'success': False,
'error': '房间内没有文件客户端',
'timestamp': datetime.now().isoformat()
})
async def handle_file_operation_response(websocket, client_id: str, data: Dict[str, Any]):
"""处理文件操作响应"""
request_id = data.get('requestId')
target_client_id = data.get('target_client_id')
logger.info(f"收到文件操作响应: {request_id} -> {target_client_id}")
if target_client_id and target_client_id in connected_clients:
# 发送给指定客户端
target_websocket = connected_clients[target_client_id]['websocket']
await send_message(target_websocket, {
'type': 'file_operation_response',
**data
})
else:
logger.error("文件操作响应没有指定目标客户端或目标客户端不存在")
async def send_message(websocket, message: Dict[str, Any]):
"""发送消息到WebSocket"""
try:
await websocket.send(json.dumps(message))
except websockets.exceptions.ConnectionClosed:
logger.warning("连接已关闭，无法发送消息")
async def broadcast_to_room(room_id: str, message: Dict[str, Any], exclude_client: str = None):
"""向房间内所有客户端广播消息"""
if room_id not in rooms:
return
room = rooms[room_id]
for client_id in room.clients:
if client_id != exclude_client and client_id in connected_clients:
try:
await send_message(connected_clients[client_id]['websocket'], message)
except Exception as e:
logger.error(f"广播消息失败: {e}")
def run_http_server():
"""运行HTTP服务器"""
try:
server = HTTPServer(('0.0.0.0', 80), WebSocketHTTPRequestHandler)
server = HTTPServer(('0.0.0.0', 80), HTTPHandler)
logger.info("HTTP服务器启动在端口 80")
server.serve_forever()
except Exception as e:
logger.error(f"HTTP服务器启动失败: {e}")
async def run_websocket_server():
"""运行WebSocket服务器"""
try:
# 使用新的函数签名，不传递path参数
server = await websockets.serve(handle_websocket, '0.0.0.0', 81)
logger.info("WebSocket服务器启动在端口 81")
await server.wait_closed()
except Exception as e:
logger.error(f"WebSocket服务器启动失败: {e}")
async def main():
"""主函数"""
# 启动HTTP服务器（在单独线程中）
http_thread = threading.Thread(target=run_http_server, daemon=True)
http_thread.start()
# 启动WebSocket服务器
await run_websocket_server()
if __name__ == '__main__':
logger.info("启动服务器...")
# 检查端口权限
try:
asyncio.run(main())
except PermissionError:
logger.error("需要管理员权限才能绑定到80端口")
logger.info("尝试使用8080端口...")
# 可以在这里添加回退到其他端口的逻辑
run_http_server()