|
from fastapi import FastAPI, WebSocket, HTTPException |
|
from fastapi.staticfiles import StaticFiles |
|
from pydantic import BaseModel |
|
from fastapi.responses import FileResponse |
|
import uuid, random, asyncio |
|
|
|
app = FastAPI() |
|
|
|
|
|
@app.get("/") |
|
def read_index(): |
|
return FileResponse("index.html") |
|
|
|
|
|
class CreateLobby(BaseModel): |
|
max_players: int |
|
starting_amount: int |
|
passkey: str |
|
|
|
class JoinLobby(BaseModel): |
|
lobby_id: str |
|
player_name: str |
|
passkey: str |
|
|
|
|
|
lobbies = {} |
|
|
|
class Player: |
|
def __init__(self, name, amount): |
|
self.name = name |
|
self.amount = amount |
|
self.cards = [] |
|
self.active = True |
|
|
|
class Lobby: |
|
def __init__(self, max_players, start_amt, passkey): |
|
self.id = str(uuid.uuid4())[:8] |
|
self.passkey = passkey |
|
self.max_players = max_players |
|
self.start_amt = start_amt |
|
self.players = [] |
|
self.state = "waiting" |
|
self.deck = [] |
|
self.pot = 0 |
|
self.community = [] |
|
|
|
def build_deck(self): |
|
suits = ['♠','♥','♦','♣'] |
|
ranks = [str(r) for r in range(2,11)] + list("JQKA") |
|
self.deck = [r + s for s in suits for r in ranks] |
|
random.shuffle(self.deck) |
|
print(f"[Lobby {self.id}] deck shuffled") |
|
|
|
def deal(self): |
|
self.build_deck() |
|
|
|
for p in self.players: |
|
p.cards = [self.deck.pop(), self.deck.pop()] |
|
print(f"[Lobby {self.id}] Dealt to {p.name}: {p.cards}") |
|
|
|
|
|
@app.post("/create_lobby") |
|
async def create_lobby(data: CreateLobby): |
|
lobby = Lobby(data.max_players, data.starting_amount, data.passkey) |
|
lobbies[lobby.id] = lobby |
|
print(f"Lobby {lobby.id} created, max_players={lobby.max_players}") |
|
return {"lobby_id": lobby.id} |
|
|
|
@app.post("/join_lobby") |
|
async def join_lobby(data: JoinLobby): |
|
lobby = lobbies.get(data.lobby_id) |
|
if not lobby or lobby.passkey != data.passkey: |
|
raise HTTPException(404, "Lobby not found or wrong passkey") |
|
if len(lobby.players) >= lobby.max_players: |
|
raise HTTPException(400, "Lobby is full") |
|
p = Player(data.player_name, lobby.start_amt) |
|
lobby.players.append(p) |
|
print(f"{p.name} joined Lobby {lobby.id}") |
|
|
|
if len(lobby.players) == lobby.max_players: |
|
lobby.state = "dealing" |
|
lobby.deal() |
|
return {"status": "joined"} |
|
|
|
@app.get("/lobby/{lobby_id}/state") |
|
async def lobby_state(lobby_id: str): |
|
lobby = lobbies.get(lobby_id) |
|
if not lobby: |
|
raise HTTPException(404, "Lobby not found") |
|
|
|
return { |
|
"state": lobby.state, |
|
"players": [ |
|
{"name": p.name, "amount": p.amount, "cards": p.cards if lobby.state!="waiting" else []} |
|
for p in lobby.players |
|
], |
|
"community": lobby.community, |
|
"pot": lobby.pot |
|
} |
|
|
|
|
|
|
|
if __name__ == "__main__": |
|
import uvicorn |
|
uvicorn.run("app:app", host="0.0.0.0", port=7860, log_level="info") |