Upload 3 files

Dockerfile

@@ -0,0 +1,13 @@
+FROM python:3.9
+
+WORKDIR /code
+
+COPY ./requirements.txt /code/requirements.txt
+
+RUN pip install --no-cache-dir --upgrade -r /code/requirements.txt
+
+COPY . .
+
+EXPOSE 5000
+
+CMD ["uvicorn", "server:app", "--host", "0.0.0.0", "--port", "7860"]

requirements.txt

@@ -0,0 +1,2 @@
+fastapi
+uvicorn

server.py

@@ -0,0 +1,87 @@
+import random
+from fastapi import FastAPI, WebSocket, WebSocketDisconnect
+import uvicorn
+
+app = FastAPI()
+
+parties = {}
+users = {}
+
+class User:
+    def __init__(self, username, websocket, status):
+        self.username = username
+        self.websocket = websocket
+        self.status = status
+
+def generate_party_code():
+    return ''.join([str(random.randint(0, 9)) for _ in range(6)])
+
+async def broadcast_message(party_id, sender, message):
+    if party_id in parties:
+        if message.startswith("TMPN:") and sender.status == "owner":
+            formatted_message = message
+        else:
+            status = 'Owner' if sender.status == 'owner' else 'User'
+            formatted_message = f"{sender.username}:[{status}]:{message}"
+
+        for user_name in parties[party_id]['members']:
+            user = users[user_name]
+            if user != sender:
+                await user.websocket.send_text(formatted_message)
+
+@app.websocket("/ws")
+async def party_server(websocket: WebSocket):
+    await websocket.accept()
+    username = None
+    party_code = None
+    user_status = None
+
+    await websocket.send_text("[system]:Enter your name to start.")
+    data = await websocket.receive_text()
+    username = data
+
+    users[username] = User(username, websocket, "user")
+
+    await websocket.send_text("[system]:Would you like to create or join a party? Type 'create' or 'join'.")
+    action = await websocket.receive_text()
+
+    if action == "create":
+        party_code = generate_party_code()
+        parties[party_code] = {'owner': username, 'members': {username}}
+        users[username].status = "owner"
+        await websocket.send_text(f"[system]:Party created! Your party code is {party_code}.")
+        await broadcast_message(party_code, users[username], f'[system]:{username} has created the party.')
+
+    elif action == "join":
+        await websocket.send_text("[system]:Enter the party code to join.")
+        party_code = await websocket.receive_text()
+
+        if party_code in parties:
+            parties[party_code]['members'].add(username)
+            users[username].status = "user"
+            await websocket.send_text(f"[system]:You have joined the party with code {party_code}.")
+            await broadcast_message(party_code, users[username], f'[system]:{username} has joined the party.')
+        else:
+            await websocket.send_text("[system]:Invalid party code.")
+
+    try:
+        while True:
+            data = await websocket.receive_text()
+            if party_code:
+                await broadcast_message(party_code, users[username], data)
+    except WebSocketDisconnect:
+        if party_code and party_code in parties:
+            parties[party_code]['members'].remove(username)
+            if parties[party_code]['owner'] == username:
+                await broadcast_message(party_code, users[username], f'[system]:{username} (Owner) has left the party. All members are being kicked out.')
+                for member in list(parties[party_code]['members']):
+                    member_user = users[member]
+                    await member_user.websocket.send_text("[system]:The owner has left, and the party is deleted.")
+                    parties[party_code]['members'].remove(member)
+                del parties[party_code]
+            else:
+                await broadcast_message(party_code, users[username], f'[system]:{username} has left the party.')
+        del users[username]
+
+if __name__ == "__main__":
+    uvicorn.run(app, host="0.0.0.0", port=5000)