Spaces:
Sleeping
Sleeping
from flask import Flask, render_template, request, jsonify | |
from flask_socketio import SocketIO, emit | |
from datetime import datetime | |
import os | |
app = Flask(__name__) | |
app.config['SECRET_KEY'] = 'marauders_map_secret!' | |
socketio = SocketIO(app, cors_allowed_origins="*") | |
# 存储用户位置信息的字典 | |
# 结构: { | |
# 'username': { | |
# 'location': {'lat': float, 'lng': float}, | |
# 'sid': string, | |
# 'last_update': datetime string, | |
# 'footprints': list # 存储最近的脚印位置 | |
# } | |
# } | |
users = {} | |
def index(): | |
return render_template('index.html') | |
def handle_connect(): | |
print(f'Client connected: {request.sid}') | |
def handle_disconnect(): | |
user_sid = request.sid | |
disconnected_user = None | |
# 查找断开连接的用户 | |
for username, data in users.items(): | |
if data['sid'] == user_sid: | |
disconnected_user = username | |
break | |
# 如果找到断开连接的用户,从用户字典中移除 | |
if disconnected_user: | |
del users[disconnected_user] | |
print(f'User disconnected: {disconnected_user}') | |
# 广播用户断开连接的消息 | |
emit('user_disconnected', {'username': disconnected_user}, broadcast=True) | |
def handle_location_update(data): | |
username = data['username'] | |
location = data['location'] | |
# 更新用户位置信息 | |
if username not in users: | |
users[username] = { | |
'location': location, | |
'sid': request.sid, | |
'last_update': datetime.now().isoformat(), | |
'footprints': [] | |
} | |
else: | |
users[username].update({ | |
'location': location, | |
'last_update': datetime.now().isoformat() | |
}) | |
# 添加新的脚印到用户的脚印列表 | |
users[username]['footprints'].append(location) | |
# 只保留最近的20个脚印 | |
if len(users[username]['footprints']) > 20: | |
users[username]['footprints'] = users[username]['footprints'][-20:] | |
# 广播更新后的用户信息 | |
emit('users_update', users, broadcast=True) | |
def handle_username_update(data): | |
old_username = data['old_username'] | |
new_username = data['new_username'] | |
# 如果旧用户名存在,更新为新用户名 | |
if old_username in users: | |
users[new_username] = users.pop(old_username) | |
print(f'Username updated: {old_username} -> {new_username}') | |
# 广播用户名更新信息 | |
emit('users_update', users, broadcast=True) | |
def handle_error(error): | |
print(f'Error occurred: {error}') | |
if __name__ == '__main__': | |
# 确保static和templates目录存在 | |
for dir_name in ['static', 'templates']: | |
if not os.path.exists(dir_name): | |
os.makedirs(dir_name) | |
print(f'Created directory: {dir_name}') | |
# 启动服务器 | |
print('Starting Marauder\'s Map server...') | |
# socketio.run(app, debug=True, host='0.0.0.0', port=5000) | |
socketio.run(app, debug=True, host='0.0.0.0', port=7860, allow_unsafe_werkzeug=True) | |