Spaces:
Runtime error
Runtime error
| 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) | |