mirror of
https://github.com/qwerfd2/OverRide-Server.git
synced 2026-09-25 00:08:02 +03:00
413 lines
14 KiB
Python
413 lines
14 KiB
Python
from starlette.applications import Starlette
|
|
from starlette.responses import FileResponse, Response, JSONResponse
|
|
from starlette.routing import Route
|
|
from starlette.requests import Request
|
|
import contextlib
|
|
import os
|
|
import databases
|
|
import sqlalchemy
|
|
import json
|
|
import secrets
|
|
from sqlalchemy import Table, Column, Integer, String, JSON, Boolean, Float, ForeignKey, update, insert
|
|
from sqlalchemy.ext.asyncio import create_async_engine
|
|
from sqlalchemy.future import select
|
|
import time
|
|
|
|
from config import HOST, PORT, DEBUG, SSL_CERT, SSL_KEY, REGISTRATION
|
|
|
|
#------------- DB -------------
|
|
|
|
DB_NAME = "player.db"
|
|
DB_PATH = os.path.join(os.getcwd(), DB_NAME)
|
|
DATABASE_URL = f"sqlite+aiosqlite:///{DB_PATH}"
|
|
|
|
database = databases.Database(DATABASE_URL)
|
|
metadata = sqlalchemy.MetaData()
|
|
|
|
player = Table(
|
|
"player",
|
|
metadata,
|
|
Column("id", String(64), primary_key=True),
|
|
Column("name", String(64), unique=True, nullable=False),
|
|
Column("banned", Boolean, nullable=False),
|
|
Column("rating", Float, nullable=False),
|
|
Column("data", JSON, nullable=False),
|
|
)
|
|
|
|
rank = Table(
|
|
"rank",
|
|
metadata,
|
|
Column("id", Integer, primary_key=True, autoincrement=True),
|
|
Column("player_id", String(64), ForeignKey("player.id"), nullable=False),
|
|
Column("key", String(64), nullable=False),
|
|
Column("score", Integer, nullable=False),
|
|
Column("isFC", Boolean, nullable=False),
|
|
)
|
|
|
|
redeem = Table(
|
|
"redeem",
|
|
metadata,
|
|
Column("id", Integer, primary_key=True, autoincrement=True),
|
|
Column("code", String(64), nullable=False),
|
|
Column("item", JSON, nullable=False),
|
|
Column("expire", Integer, nullable=False),
|
|
Column("limit", Integer, nullable=False),
|
|
Column("roster", JSON, nullable=False),
|
|
)
|
|
|
|
async def init_db():
|
|
|
|
if not os.path.exists(DB_PATH):
|
|
print("[DB] Creating new database:", DB_PATH)
|
|
|
|
engine = create_async_engine(DATABASE_URL, echo=False)
|
|
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(metadata.create_all)
|
|
|
|
await engine.dispose()
|
|
print("[DB] Database initialized successfully.")
|
|
|
|
async def get_player_data(user_id):
|
|
query = player.select().where(player.c.id == user_id)
|
|
player_info = await database.fetch_one(query=query)
|
|
if player_info:
|
|
return dict(player_info)
|
|
else:
|
|
return None
|
|
|
|
async def set_player_data(user_id, data_field, new_data):
|
|
if data_field not in player.c:
|
|
raise ValueError(f"Invalid column name: {data_field}")
|
|
|
|
query = player.update().where(player.c.id == user_id).values({data_field: new_data})
|
|
await database.execute(query)
|
|
|
|
# ------------ CDN -------------
|
|
|
|
root_folder = os.path.dirname(os.path.abspath(__file__))
|
|
allowed_folders = ["OverRapid", "Resources", "OverRide"]
|
|
|
|
async def serve_file(request):
|
|
path = request.path_params['path']
|
|
first_level_folder = path.split('/')[0]
|
|
if first_level_folder in allowed_folders:
|
|
file_path = os.path.realpath(os.path.join(os.getcwd(), path))
|
|
if os.path.isfile(file_path):
|
|
return FileResponse(file_path)
|
|
|
|
return Response("File not found", status_code=404)
|
|
|
|
#------------ API -------------
|
|
|
|
async def api_function(request: Request):
|
|
data = await request.json()
|
|
content = data.get("Content", "")
|
|
name = data.get("Name", "")
|
|
user_id = data.get("UserId", "")
|
|
player_info = await get_player_data(user_id)
|
|
content_dict = json.loads(content)
|
|
|
|
if player_info is None:
|
|
print("WARNING: User not found in database.")
|
|
|
|
if name == "dataDownload":
|
|
user_id = content_dict.get("key", "")
|
|
if player_info:
|
|
return JSONResponse(player_info['data'])
|
|
else:
|
|
return JSONResponse({})
|
|
|
|
elif name == "UIDCheck":
|
|
result = await get_player_data(content_dict.get("key", ""))
|
|
return JSONResponse({"exists": result is not None})
|
|
|
|
elif name == "dataUpload":
|
|
content_json = content_dict["jsonString"]
|
|
if player_info:
|
|
await set_player_data(user_id, "data", content_json)
|
|
elif REGISTRATION:
|
|
query = insert(player).values(id=user_id, name=content_json["name"], banned=False, rating=0.0, data=content_json)
|
|
await database.execute(query)
|
|
|
|
return JSONResponse({"dummy": True})
|
|
|
|
elif name == "nameCheck":
|
|
player_name = content_dict.get("name", "")
|
|
query = player.select().where(player.c.name == player_name)
|
|
result = await database.fetch_one(query)
|
|
if not REGISTRATION:
|
|
return JSONResponse({"regClosed": True})
|
|
|
|
return JSONResponse({"exists": result is not None})
|
|
|
|
elif name == "getName":
|
|
if player_info:
|
|
if player_info['name']:
|
|
if isinstance(player_info['banned'], str) or player_info['banned'] is None:
|
|
return JSONResponse({"banned": player_info['banned']})
|
|
else:
|
|
return JSONResponse({"name": player_info['name']})
|
|
|
|
return JSONResponse({})
|
|
|
|
elif name == "ratingUpload":
|
|
try:
|
|
score = float(content_dict.get("score", ""))
|
|
except ValueError:
|
|
return JSONResponse({"dummy": False}, status_code=400)
|
|
|
|
if player_info['rating'] is not None:
|
|
await set_player_data(user_id, "rating", score)
|
|
return JSONResponse({"dummy": True})
|
|
else:
|
|
return JSONResponse({"dummy": False}, status_code=400)
|
|
|
|
elif name == "rename":
|
|
new_name = content_dict.get("name", "")
|
|
|
|
if player_info['banned'] > 0 or player_info['banned'] is None:
|
|
return JSONResponse({"banned": True})
|
|
else:
|
|
await set_player_data(user_id, "name", new_name)
|
|
return JSONResponse({"banned": False})
|
|
|
|
elif name == "ratingPosition":
|
|
if player_info:
|
|
query = select(sqlalchemy.func.count()).where(player.c.rating > player_info['rating'])
|
|
higher_count = (await database.fetch_one(query))[0]
|
|
|
|
query = select(sqlalchemy.func.count())
|
|
total_count = (await database.fetch_one(query))[0]
|
|
|
|
response_data = {
|
|
"userExists": True,
|
|
"total": total_count,
|
|
"count": higher_count
|
|
}
|
|
else:
|
|
response_data = {"userExists": False}
|
|
|
|
return JSONResponse(response_data)
|
|
|
|
elif name == "rankUpload":
|
|
key = content_dict.get("key", "")
|
|
score = content_dict.get("score", 0)
|
|
isFC = content_dict.get("isFC", False)
|
|
query = select(rank.c.score).where(rank.c.player_id == user_id, rank.c.key == key)
|
|
result = await database.fetch_one(query)
|
|
|
|
if result is not None:
|
|
if result[0] < score:
|
|
query = update(rank).where(rank.c.player_id == user_id, rank.c.key == key).values(score=score, isFC=isFC)
|
|
await database.execute(query)
|
|
else:
|
|
query = insert(rank).values(player_id=user_id, key=key, score=score, isFC=isFC)
|
|
await database.execute(query)
|
|
|
|
return JSONResponse({"dummy": True})
|
|
|
|
elif name == "rankLeaderboard":
|
|
key_to_match = content_dict.get("key", "")
|
|
max_key = content_dict.get("maxKey", 99)
|
|
|
|
query = select(rank.c.player_id, rank.c.score, rank.c.isFC).where(rank.c.key == key_to_match)
|
|
qualified_ranks = await database.fetch_all(query)
|
|
|
|
qualified_ranks = sorted(qualified_ranks, key=lambda x: x["score"], reverse=True)
|
|
|
|
top_scores = qualified_ranks[:max_key]
|
|
|
|
result_list = []
|
|
for rank_entry in top_scores:
|
|
player_id = rank_entry["player_id"]
|
|
score = rank_entry["score"]
|
|
isFC = rank_entry["isFC"]
|
|
|
|
query = select(player.c.name).where(player.c.id == player_id)
|
|
player_name = (await database.fetch_one(query))["name"]
|
|
|
|
result_list.append({"name": player_name, "score": score, "isFC": bool(isFC)})
|
|
|
|
return JSONResponse(result_list)
|
|
|
|
elif name == "ratingLeaderboard":
|
|
query = select(player.c.name, player.c.rating).order_by(player.c.rating.desc())
|
|
result = await database.fetch_all(query)
|
|
|
|
leaderboard = [{"name": row["name"], "score": row["rating"]} for row in result]
|
|
return JSONResponse(leaderboard)
|
|
|
|
elif name == "codeCheck":
|
|
code = content_dict.get("code", "")
|
|
|
|
query = select(redeem).where(redeem.c.code == code)
|
|
redeem_info = await database.fetch_one(query)
|
|
|
|
if redeem_info:
|
|
item = redeem_info["item"]
|
|
expire = redeem_info["expire"]
|
|
limit = redeem_info["limit"]
|
|
roster = redeem_info["roster"]
|
|
current_time = int(time.time())
|
|
if current_time > expire or len(roster) >= limit or user_id in roster:
|
|
return JSONResponse({"success": False})
|
|
|
|
roster.append(user_id)
|
|
query = update(redeem).where(redeem.c.code == code).values(roster=roster)
|
|
await database.execute(query)
|
|
return JSONResponse({"success": True, "item": item})
|
|
else:
|
|
return JSONResponse({"success": False})
|
|
|
|
elif name == "withdraw":
|
|
return JSONResponse({})
|
|
|
|
#------------ PVP -------------
|
|
|
|
queue = []
|
|
|
|
def find_lobby_by_id(lobby_id):
|
|
for lobby in queue:
|
|
if lobby.get("id") == lobby_id:
|
|
return lobby
|
|
return None
|
|
|
|
async def pvp_function(request: Request):
|
|
data = await request.json()
|
|
type = data.get("type", "")
|
|
request_code = data.get("request", "")
|
|
lobby_id = data.get("id", "")
|
|
|
|
if type == "match":
|
|
if request_code == "list":
|
|
filtered_lobbies = [lobby for lobby in queue if not lobby.get("selected", False)]
|
|
return JSONResponse(filtered_lobbies)
|
|
|
|
elif request_code == "create":
|
|
player_id = data.get("pid", "")
|
|
session_id = secrets.token_urlsafe(24)
|
|
lobby_data = {
|
|
"data": {
|
|
"mp3": data.get("mp3", ""),
|
|
"diff": data.get("diff", ""),
|
|
"fm": data.get("fm", False),
|
|
},
|
|
"timestamp": int(time.time()),
|
|
"players": {
|
|
player_id: {
|
|
"icon": data.get("icon", ""),
|
|
"level": data.get("level", 1),
|
|
"is4Key": data.get("is4Key", False),
|
|
"diff": data.get("diff", ""),
|
|
"options": data.get("options", []),
|
|
"username": data.get("username", ""),
|
|
"rating": data.get("rating", 0.0),
|
|
"achivement": data.get("achivement", "")
|
|
}
|
|
},
|
|
"waiting": True,
|
|
"id": session_id,
|
|
"match": {}
|
|
}
|
|
queue.append(lobby_data)
|
|
return JSONResponse({"id": session_id})
|
|
|
|
elif request_code == "purge":
|
|
for index, lobby in enumerate(queue):
|
|
if lobby.get("id") == lobby_id:
|
|
del queue[index]
|
|
return JSONResponse(queue)
|
|
return JSONResponse(queue)
|
|
|
|
elif request_code == "players":
|
|
lobby = find_lobby_by_id(lobby_id)
|
|
if lobby:
|
|
return JSONResponse(lobby.get("players"))
|
|
return JSONResponse({})
|
|
|
|
elif request_code == "join":
|
|
lobby = find_lobby_by_id(lobby_id)
|
|
if lobby:
|
|
player_id = data.get("pid", "")
|
|
lobby["players"][player_id] = {
|
|
"icon": data.get("icon", ""),
|
|
"level": data.get("level", 1),
|
|
"is4Key": data.get("is4Key", False),
|
|
"diff": data.get("diff", ""),
|
|
"options": data.get("options", []),
|
|
"username": data.get("username", ""),
|
|
"rating": data.get("rating", 0.0),
|
|
"achivement": data.get("achivement", "")
|
|
}
|
|
lobby["waiting"] = False
|
|
lobby["timestamp"] = int(time.time())
|
|
return JSONResponse(lobby)
|
|
else:
|
|
return JSONResponse({})
|
|
|
|
elif request_code == "select":
|
|
lobby = find_lobby_by_id(lobby_id)
|
|
if lobby:
|
|
is_already_selected = lobby.get("selected", False)
|
|
if not is_already_selected:
|
|
lobby["selected"] = True
|
|
return JSONResponse({"isAlreadySelected": False})
|
|
else:
|
|
return JSONResponse({"isAlreadySelected": True})
|
|
else:
|
|
return JSONResponse({"isAlreadySelected": True})
|
|
|
|
elif request_code == "unselect":
|
|
lobby = find_lobby_by_id(lobby_id)
|
|
if lobby:
|
|
lobby["selected"] = False
|
|
return JSONResponse({"message": ""})
|
|
else:
|
|
return JSONResponse({"message": ""})
|
|
|
|
elif request_code == "match":
|
|
lobby = find_lobby_by_id(lobby_id)
|
|
if lobby:
|
|
player_id = data.get("pid", "")
|
|
match_value = data.get("match", "")
|
|
current_timestamp = int(time.time())
|
|
if player_id in lobby["match"]:
|
|
lobby["match"][player_id]["timestamp"] = current_timestamp
|
|
lobby["match"][player_id]["match"] = match_value
|
|
else:
|
|
lobby["match"][player_id] = {
|
|
"timestamp": current_timestamp,
|
|
"match": match_value
|
|
}
|
|
return JSONResponse(lobby["match"])
|
|
else:
|
|
return JSONResponse({})
|
|
|
|
else:
|
|
return JSONResponse({})
|
|
|
|
routes = [
|
|
Route("/Api", api_function, methods=["POST"]),
|
|
Route("/Pvp", pvp_function, methods=["POST"])
|
|
]
|
|
|
|
routes.append(Route("/{path:path}", serve_file))
|
|
|
|
@contextlib.asynccontextmanager
|
|
async def lifespan(app: Starlette):
|
|
await database.connect()
|
|
await init_db()
|
|
try:
|
|
yield
|
|
finally:
|
|
await database.disconnect()
|
|
|
|
app = Starlette(debug=DEBUG, routes=routes, lifespan=lifespan)
|
|
|
|
if __name__ == "__main__":
|
|
import uvicorn
|
|
ssl_context = (SSL_CERT, SSL_KEY) if SSL_CERT and SSL_KEY else None
|
|
uvicorn.run(app, host=HOST, port=PORT, ssl_certfile=SSL_CERT, ssl_keyfile=SSL_KEY)
|
|
|
|
# Made By Tony 2025.5.4 |