import dotenv from flask import Flask, Response, jsonify, request from flask_cors import CORS from flask_socketio import SocketIO, emit from .room import Room, test_algo from .song import Song, init_db # from .song_fetch import * dotenv.load_dotenv() app = Flask(__name__) app.config["SECRET_KEY"] = "your_secret_key" socketio = SocketIO(app) CORS(app) ROOMS: dict[int, Room] = {} # { room_id: room, ... } def error(msg: str, status: int = 400) -> Response: res = jsonify({"success": False, "error": msg}) res.status_code = status return res @app.get("/api/join") def join(): room_id = request.args.get("room") code = request.args.get("code") if room_id is None: return error("Missing room id") if (room := ROOMS.get(int(room_id))) is None: return error("Invalid room") if room.pin is not None and room.pin != code: return error("Invalid code") return {"success": True, "ws": f"/ws/{room_id}"} @app.get("/api/queue") def queue(): if (room_id := request.args.get("room")) is None: return error("Missing room id") if (room := ROOMS.get(int(room_id))) is None: return error("Invalid room") return {"success": True, "queue": room.playing} @app.post("/api/queue/next") def queue_next(): if (room_id := request.args.get("room")) is None: return error("Missing room id") if (room := ROOMS.get(int(room_id))) is None: return error("Invalid room") room.playing_idx += 1 if room.playing_idx >= len(room.playing): ## queue ended # room.renew_queue() emit("update_songs", {"songs": [1, 2, 3]}, broadcast=True, namespace="/") return {"success": True, "ended": True, "index": room.playing_idx, "queue": room.playing} return {"success": True, "ended": False, "index": room.playing_idx} @app.post("/api/room/new") def room_new(): if (room_name := request.args.get("name")) is None: return error("Missing room name") if (room_cords := request.args.get("coords")) is None: return error("Missing room coords") if room_pin := request.args.get("pin"): room_pin = int(room_pin) else: room_pin = None lat, lon = room_cords.split(",") room = Room( id=max(ROOMS or [0]) + 1, # coord=(float(lat), float(lon)), name=room_name, pin=room_pin, tags=set([tag for tag in request.args.get("tags", "").split(",") if tag]), songs={}, history=[], playing=[], playing_idx=-1, ) ROOMS[room.id] = room return {"success": True, "room_id": room.id} @app.get("/api/room") def room(): return [ { "id": room.id, "name": room.name, "private": room.pin is not None, "coords": room.coord, } for room in ROOMS.values() ] init_db() test_algo() exit() if __name__ == "__main__": socketio.run(app, debug=True)