Init
This commit is contained in:
parent
56701189ca
commit
36dff3d99a
12 changed files with 1296 additions and 0 deletions
1
customiwmserver/__init__.py
Normal file
1
customiwmserver/__init__.py
Normal file
|
@ -0,0 +1 @@
|
|||
__version__ = "0.1.0"
|
3
customiwmserver/__main__.py
Normal file
3
customiwmserver/__main__.py
Normal file
|
@ -0,0 +1,3 @@
|
|||
import uvicorn
|
||||
if __name__ == "__main__":
|
||||
uvicorn.run("customiwmserver.main:app", host="0.0.0.0", port=8001, reload=True)
|
281
customiwmserver/data_types.py
Normal file
281
customiwmserver/data_types.py
Normal file
|
@ -0,0 +1,281 @@
|
|||
from pydantic import BaseModel
|
||||
from typing import List, Optional, Union
|
||||
|
||||
|
||||
class User(BaseModel):
|
||||
"""Pydantic model for user data.
|
||||
Created from the user response (`/api/v1/user/x`),
|
||||
some values might be unused."""
|
||||
|
||||
Admin: bool = True
|
||||
Banned: bool = False
|
||||
BulletSpr: int = 0
|
||||
CapeColor: int = 0
|
||||
Costume: int = 0
|
||||
Country: int = 0
|
||||
CreatedAt: str = "2020-02-14T10:03:40Z"
|
||||
DeathEffect: int = 0
|
||||
DeletedAt: Optional[str] = None
|
||||
FacialExpression: int = 0
|
||||
Followed: bool = False
|
||||
FollowerColor: int = 0
|
||||
FollowerSpr: int = 0
|
||||
GunSpr: int = 0
|
||||
HairColor: int = 0
|
||||
HairSpr: int = 0
|
||||
HatColor: int = 0
|
||||
HatColorInv: int = 0
|
||||
HatSpr: int = 0
|
||||
ID: int = 0
|
||||
NumMapsClear: int = 0
|
||||
NumMapsCreated: int = 0
|
||||
NumSpeedrunRecords: int = 0
|
||||
PantsColor: int = 0
|
||||
RecordEndurance0: int = 0
|
||||
RecordEndurance1: int = 0
|
||||
RecordEndurance2: int = 0
|
||||
RecordEndurance3: int = 0
|
||||
RecordExplorer: int = 0
|
||||
RecordHardcore: int = 0
|
||||
RecordRoulette: int = 0
|
||||
RecordScribble0: int = 0
|
||||
RecordScribble1: int = 0
|
||||
RecordScribble2: int = 0
|
||||
RecordScribble3: int = 0
|
||||
SaveEffect: int = 0
|
||||
ShirtColor: int = 0
|
||||
ShoesColor: int = 0
|
||||
SkinColor: int = 0
|
||||
SwordSpr: int = 0
|
||||
TextSnd: int = 0
|
||||
Unlocks: str = ""
|
||||
UpdatedAt: str = "2020-03-26T05:22:32Z"
|
||||
Username: str = "magmaus3"
|
||||
Email: str = "user@example.com"
|
||||
|
||||
|
||||
class Notification(BaseModel):
|
||||
"""Pydantic model for Notifications
|
||||
|
||||
Available notification types:
|
||||
- 0: <user> took your record on <map> with time <time>
|
||||
- NotifData seems to need following variables:
|
||||
- `MapID`: `int`
|
||||
- `MapName`: `str`, displayed map name
|
||||
- `ByUserID`: `int`, who took the record
|
||||
- `ByUserName`: `str`, who took the record (displayed name)
|
||||
- `NewTime`: `int`, might be the either the amount of seconds or the time id (TODO: correct the info)
|
||||
|
||||
- 1: Your time on <map> was removed
|
||||
- Reason IDs:
|
||||
- 0: no reason
|
||||
- 1: empty reason (?)
|
||||
- 2: Autofire/macros/scripting
|
||||
- 3: Replay was broken
|
||||
- Other reason IDs return the following message: `Unrecognized reason: <reason>`
|
||||
- Other than `Reason`, NotifData requires following variables as well:
|
||||
- MapID
|
||||
- MapName
|
||||
|
||||
- 2: Your level <level> was removed
|
||||
- Reason IDs:
|
||||
- 0: no reason
|
||||
- 1: empty reason (?)
|
||||
- 2: Inappropriate name or description
|
||||
- 3: Submitted using autofire/macros/scripting
|
||||
- 4: Inappropriate content
|
||||
- 5: Unreasonably long forced waiting
|
||||
- 6: Intentionally lagging the game
|
||||
- Like with type 1, other reason IDs show the "unrecognized reason" message
|
||||
- Other than `Reason`, NotifData requires following variables as well:
|
||||
- MapID
|
||||
- MapName
|
||||
|
||||
- 3: You were temporarily banned.
|
||||
- Reason IDs:
|
||||
- 0: no reason
|
||||
- 1: empty reason (?)
|
||||
- 2: Inappropriate level name or description
|
||||
- 3: Inappropriate user name
|
||||
- 4: Inappropriate playlist name
|
||||
- 5: Autofire/macros/scripting
|
||||
- 6: Inappropriate level content
|
||||
- 7: Level having unreasonably long forced waiting
|
||||
- 8: Level intentionally lagging the game
|
||||
- 9: Exploiting the game
|
||||
- Like with types 1 and 2, other reason IDs return the "unrecognized reason" message
|
||||
- 4: Message from an administrator:
|
||||
- NotifData requires the following variables:
|
||||
- Message: str
|
||||
|
||||
All other types are displayed as an empty notification, without any special handling.
|
||||
|
||||
Note: NotifData must be returned to the game as a string,
|
||||
otherwise the game crashes (as of `Early Access Ver 0.786`)
|
||||
"""
|
||||
|
||||
CreatedAt: str = "2008-01-18T00:00:00Z"
|
||||
UpdatedAt: str = "2008-01-18T00:00:00Z"
|
||||
DeletedAt: Optional[str] = None
|
||||
ForUserID: int = 0
|
||||
ID: int = 0
|
||||
IsRead: bool = False
|
||||
MapID: int = 0 # Might be optional for some notification types
|
||||
|
||||
# NotifData also accepts str for compatibility with the game (see docstring)
|
||||
NotifData: Union[str, dict] = {
|
||||
"MapID": 215884,
|
||||
"MapName": "speedrun_tower.map",
|
||||
"ByUserID": 1,
|
||||
"ByUserName": "user",
|
||||
"NewTime": 1172,
|
||||
"Reason": 2,
|
||||
}
|
||||
Type: int = 0
|
||||
|
||||
class Map(BaseModel):
|
||||
"""Pydantic model for maps.
|
||||
It looks like a lot of variables are also included in the user
|
||||
data.
|
||||
"""
|
||||
ID: int = 0
|
||||
CreatorId: int = 0
|
||||
|
||||
# CreatorName might be also retrieved from the user data
|
||||
# using the CreatorId
|
||||
CreatorName: str = ""
|
||||
|
||||
Name: str = "Interesting level"
|
||||
Description: str = ""
|
||||
Version: int = 90
|
||||
CreatedAt: str = "2022-06-18T15:10:02Z"
|
||||
DeletedAt: Union[str, None] = None
|
||||
MapCode: str = "AAABBAAA"
|
||||
Listed: bool = True
|
||||
HiddenInChallenges: bool = False
|
||||
DragonCoins: bool = False
|
||||
|
||||
# Might be duplicates from the User data
|
||||
|
||||
ShoesColor: int = 0
|
||||
PantsColor: int = 0
|
||||
ShirtColor: int = 10897172
|
||||
CapeColor: int = 0
|
||||
SkinColor: int = 0
|
||||
HairColor: int = 0
|
||||
HatSpr: int = 1
|
||||
Country: int = 143
|
||||
HairSpr: int = 0
|
||||
HatColor: int = 0
|
||||
HatColorInv: int = 0
|
||||
FacialExpression: int = 0
|
||||
DeathEffect: int = 0
|
||||
GunSpr: int = 0
|
||||
BulletSpr: int = 0
|
||||
SwordSpr: int = 0
|
||||
Costume: int = 0
|
||||
FollowerSpr: int = 0
|
||||
FollowerColor: int = 0
|
||||
SaveEffect: int = 0
|
||||
TextSnd: int = 0
|
||||
|
||||
AverageUserDifficulty: float = 0
|
||||
ComputedDifficulty: float = 0
|
||||
|
||||
# Variables below might be used when sorted
|
||||
# by rating
|
||||
AverageRating: float = 0
|
||||
NumRatings: int = 0
|
||||
NumThumbsUp: int = 0
|
||||
NumThumbsDown: int = 0
|
||||
|
||||
|
||||
TagIDs: str = ""
|
||||
TagNames: str = ""
|
||||
TagFreqs: str = ""
|
||||
|
||||
|
||||
PlayCount: int = 0
|
||||
ClearCount: int = 0
|
||||
|
||||
# I'm not sure how ClearRate is handled
|
||||
# in the official server, but I assume it's
|
||||
# calculated as ClearCount / PlayCount.
|
||||
ClearRate: float = 0.0
|
||||
FavoriteCount: int = 0
|
||||
DeathCount: int = 0
|
||||
PlayerCount: int = 0
|
||||
|
||||
BestTimeUserID: int = 0
|
||||
BestTimeUsername: str = ""
|
||||
BestTimePlaytime: int = 0
|
||||
MyBestPlaytime: int = 0
|
||||
|
||||
FirstClearUserID: Union[int, None] = 0
|
||||
|
||||
# Propably not needed, because you can simply read the
|
||||
# FirstClearUserID, and get the name from there
|
||||
# It's still useful for requests though.
|
||||
FirstClearUsername: Union[str, None] = "uwu"
|
||||
MapData: str = ""
|
||||
MapReplay: str = "" # Might be also stored in DB as a dict of user IDs and the replays
|
||||
Played: bool = False
|
||||
Clear: bool = False
|
||||
FullClear: bool = False
|
||||
|
||||
# Leaderboard scores
|
||||
Leaderboard: list = []
|
||||
|
||||
|
||||
class MapLeaderboard(BaseModel):
|
||||
"""Pydantic model for Map record leaderboards."""
|
||||
BestPlaytime: int
|
||||
BestPlaytimeTime: str
|
||||
BestReplay: str
|
||||
CreatorName: str
|
||||
UserID: int
|
||||
|
||||
ShoesColor: int = 0
|
||||
PantsColor: int = 0
|
||||
ShirtColor: int = 10897172
|
||||
CapeColor: int = 0
|
||||
SkinColor: int = 0
|
||||
HairColor: int = 0
|
||||
HatSpr: int = 1
|
||||
Country: int = 143
|
||||
HairSpr: int = 0
|
||||
HatColor: int = 0
|
||||
HatColorInv: int = 0
|
||||
FacialExpression: int = 0
|
||||
DeathEffect: int = 0
|
||||
GunSpr: int = 0
|
||||
BulletSpr: int = 0
|
||||
SwordSpr: int = 0
|
||||
Costume: int = 0
|
||||
FollowerSpr: int = 0
|
||||
FollowerColor: int = 0
|
||||
SaveEffect: int = 0
|
||||
TextSnd: int = 0
|
||||
|
||||
|
||||
ids_to_names = {
|
||||
0: "Adventure/Variety",
|
||||
6: "Gimmick",
|
||||
7: "Trap/Troll",
|
||||
3: "Joke/Meme",
|
||||
2: "Needle",
|
||||
5: "Puzzle",
|
||||
1: "Boss/Avoidance",
|
||||
9: "Art",
|
||||
8: "Music",
|
||||
4: "Auto",
|
||||
}
|
||||
|
||||
def convertTagsToNames(tags):
|
||||
"""Converts tag IDs to names"""
|
||||
global ids_to_names
|
||||
tagNames = []
|
||||
for i in tags.split(','):
|
||||
tagNames.append(ids_to_names[int(i)])
|
||||
return tagNames
|
||||
|
43
customiwmserver/database.py
Normal file
43
customiwmserver/database.py
Normal file
|
@ -0,0 +1,43 @@
|
|||
from datetime import datetime
|
||||
from pymongo import MongoClient
|
||||
|
||||
|
||||
client = MongoClient("mongodb://root:catboys@mongo:27017")
|
||||
|
||||
|
||||
db = client["IWM_CustomServer_DB"]
|
||||
|
||||
user_collection = db.users
|
||||
maps_collection = db.maps
|
||||
reports_collection = db.reports
|
||||
|
||||
general_collection = db.general
|
||||
admin_log_collection = db.admin_log
|
||||
|
||||
def LogAdminAction(action_type: str, action_data: dict, UserID: int = None, success: bool = True):
|
||||
"""Log administrator action."""
|
||||
admin_log_collection.insert_one({
|
||||
"date": datetime.utcnow(),
|
||||
"action_type": action_type,
|
||||
"action_data": action_data
|
||||
})
|
||||
|
||||
def auth_check(Authorization):
|
||||
"""Checks credentials.
|
||||
Returns a tuple with result (for example False, "nouser").
|
||||
|
||||
Results:
|
||||
- nouser = user not found
|
||||
- wrongpass = wrong password
|
||||
- [dictionary] = query
|
||||
"""
|
||||
username, password = Authorization.split(":")
|
||||
|
||||
query = user_collection.find_one({"Username": username})
|
||||
if not query:
|
||||
return False, "nouser"
|
||||
|
||||
if query['Password'] != password:
|
||||
return False, "wrongpass"
|
||||
|
||||
return True, query
|
411
customiwmserver/main.py
Normal file
411
customiwmserver/main.py
Normal file
|
@ -0,0 +1,411 @@
|
|||
from fastapi import FastAPI, Form, File, UploadFile, Header, HTTPException, Body
|
||||
from fastapi.responses import PlainTextResponse
|
||||
from fastapi.exceptions import RequestValidationError
|
||||
from starlette.exceptions import HTTPException as StarletteHTTPException
|
||||
|
||||
from typing import Union, Optional, Any
|
||||
import uvicorn
|
||||
from . import data_types as types
|
||||
from . import database as db
|
||||
import pymongo
|
||||
|
||||
app = FastAPI()
|
||||
|
||||
@app.exception_handler(RequestValidationError)
|
||||
async def http_exception_handler(request, exc):
|
||||
return PlainTextResponse(f"{str(exc)}", status_code=422)
|
||||
|
||||
@app.exception_handler(StarletteHTTPException)
|
||||
async def http_exception_handler(request, exc):
|
||||
return PlainTextResponse(f"[{exc.status_code}] {str(exc.detail)}", status_code=exc.status_code)
|
||||
|
||||
## Users
|
||||
|
||||
@app.post("/api/v1/login")
|
||||
async def login(username: str = Form(), password: str = Form(), version: str = Form()):
|
||||
"""User login"""
|
||||
#FIXME Add login
|
||||
|
||||
return {"token": username + ":" + password, "userId": 1}
|
||||
|
||||
|
||||
@app.put("/api/v1/user")
|
||||
async def create_user(username: str = Form(), password: str = Form(), email: str = Form(), version: str = Form()):
|
||||
"""Create a user with specified credentials"""
|
||||
token = username + ":" + password
|
||||
|
||||
UserID = len(list(db.user_collection.find({})))
|
||||
if UserID == 0: UserID = 1
|
||||
|
||||
insert_data = {
|
||||
**types.User(
|
||||
Username=username,
|
||||
Email=email,
|
||||
ID=UserID
|
||||
).dict(),
|
||||
"Password": password,
|
||||
}
|
||||
|
||||
db.user_collection.insert_one(insert_data)
|
||||
return {"token": token, "userId": UserID}
|
||||
|
||||
@app.get("/api/v1/notifunread")
|
||||
async def notifunread():
|
||||
"""Returns the number of unread notifications."""
|
||||
#FIXME Add notifications
|
||||
return 0
|
||||
|
||||
@app.get("/api/v1/refresh")
|
||||
async def refresh_login(Authorization: Union[str, None] = Header(default=None)):
|
||||
"""Intended for refreshing user token."""
|
||||
return {"token": Authorization}
|
||||
|
||||
@app.get("/api/v1/user/{user_id}")
|
||||
async def get_user(user_id: int):
|
||||
"""Returns specified user's profile."""
|
||||
#FIXME use database
|
||||
query = db.user_collection.find_one({"ID": user_id})
|
||||
del query["Password"]
|
||||
return types.User(**query)
|
||||
|
||||
## Maps
|
||||
|
||||
@app.get("/api/v1/mapcount")
|
||||
async def mapcount():
|
||||
return len(list(db.maps_collection.find({})))
|
||||
@app.get("/api/v1/useruploadcooldown")
|
||||
async def useruploadcooldown():
|
||||
"""Limits the amount of levels that user can upload."""
|
||||
return {"success": True}
|
||||
|
||||
@app.get("/api/v1/map")
|
||||
async def search_for_maps(
|
||||
start: int = 0,
|
||||
limit: int = 5,
|
||||
min_diff: float = 0.0,
|
||||
max_diff: float = 5.0,
|
||||
order: str = '[{ "Dir": "desc", "Name": "created_at" }]',
|
||||
name: str = "",
|
||||
author: str = "",
|
||||
author_id: Optional[int] = None,
|
||||
last_x_hours: Optional[int] = None
|
||||
):
|
||||
"""Search for maps."""
|
||||
|
||||
# query = db.maps_collection.find({ "CreatorId": author_id }).limit(limit)
|
||||
query = db.maps_collection.find({}).limit(limit)
|
||||
entries = []
|
||||
for i in query:
|
||||
del i['_id']
|
||||
del i['MapData']
|
||||
entries.append(i)
|
||||
return entries
|
||||
|
||||
@app.get("/api/v1/map/{mapID}")
|
||||
async def getMap(mapID: int):
|
||||
query = db.maps_collection.find_one({ "ID": mapID })
|
||||
del query['_id']
|
||||
return query
|
||||
|
||||
@app.post("/api/v1/map/{mapID}/start")
|
||||
async def getMap(mapID: int):
|
||||
query = db.maps_collection.find_one({ "ID": mapID })
|
||||
del query['_id']
|
||||
|
||||
returned_resp = {
|
||||
"BestDeaths": 0,
|
||||
"BestPlaytime": 0,
|
||||
"Clear": False,
|
||||
"CurMap": query,
|
||||
"Difficulty": 0,
|
||||
"Followed": False,
|
||||
"Played": True,
|
||||
"Rating": 5,
|
||||
"TagIDs": "1,8,9",
|
||||
"TagNames": "Boss/Avoidance,Music,Art"
|
||||
}
|
||||
return returned_resp
|
||||
|
||||
@app.post("/api/v1/map/{mapID}/stop")
|
||||
async def stopMapPlay(
|
||||
mapID: int,
|
||||
clear: int = Form(),
|
||||
deaths: int = Form(),
|
||||
playtime: int = Form(),
|
||||
totalDeaths: int = Form(),
|
||||
totalTime: int = Form(),
|
||||
replayData: str = Form(),
|
||||
Authorization: Union[str, None] = Header(default=None),
|
||||
):
|
||||
"""Saves the map replay, and informs the user if their play is a record"""
|
||||
authcheck = db.auth_check(Authorization)
|
||||
if not authcheck[0] and authcheck[1] == "nouser":
|
||||
raise HTTPException(404, detail="User not found")
|
||||
elif not authcheck[0] and authcheck[1] == "wrongpass":
|
||||
raise HTTPException(403, detail="Wrong password")
|
||||
elif authcheck[0]:
|
||||
NewMapRecord = False
|
||||
FirstClear = False
|
||||
|
||||
userData = types.User(**authcheck[1])
|
||||
|
||||
query = db.maps_collection.find_one({"ID": mapID, "Leaderboard.UserID": userData.ID})
|
||||
if query is not None: del query['_id']
|
||||
print(__import__("json").dumps(query,indent=4))
|
||||
|
||||
BestUserTime = None
|
||||
BestTime = None
|
||||
if query is not None and 'Leaderboard' in query:
|
||||
print("-"*3)
|
||||
for i in query['Leaderboard']:
|
||||
if i['UserID'] == userData.ID:
|
||||
if BestUserTime is None or BestUserTime > i['BestPlaytime'] :
|
||||
BestUserTime = i['BestPlaytime']
|
||||
if BestTime is None or BestTime > i['BestPlaytime']:
|
||||
BestTime = i["BestPlaytime"]
|
||||
if len(query['Leaderboard']) <= 1 and i['UserID'] != userData.ID:
|
||||
FirstClear = True
|
||||
print(BestUserTime, BestTime)
|
||||
if BestUserTime is None or playtime < BestUserTime:
|
||||
print(BestUserTime)
|
||||
|
||||
updateQuery = db.maps_collection.update_one({"ID": mapID},
|
||||
{
|
||||
|
||||
"$push": {
|
||||
"Leaderboard": types.MapLeaderboard(
|
||||
ShoesColor=userData.ShoesColor,
|
||||
PantsColor=userData.PantsColor,
|
||||
ShirtColor=userData.ShirtColor,
|
||||
CapeColor=userData.CapeColor,
|
||||
SkinColor=userData.SkinColor,
|
||||
HairColor=userData.HairColor,
|
||||
HatSpr=userData.HatSpr,
|
||||
Country=userData.Country,
|
||||
HairSpr=userData.HairSpr,
|
||||
HatColor=userData.HatColor,
|
||||
HatColorInv=userData.HatColorInv,
|
||||
FacialExpression=userData.FacialExpression,
|
||||
DeathEffect=userData.DeathEffect,
|
||||
GunSpr=userData.GunSpr,
|
||||
BulletSpr=userData.BulletSpr,
|
||||
SwordSpr=userData.SwordSpr,
|
||||
Costume=userData.Costume,
|
||||
FollowerSpr=userData.FollowerSpr,
|
||||
FollowerColor=userData.FollowerColor,
|
||||
SaveEffect=userData.SaveEffect,
|
||||
TextSnd=userData.TextSnd,
|
||||
|
||||
BestPlaytime=playtime,
|
||||
BestPlaytimeTime="2020-02-13T15:19:33Z",
|
||||
BestReplay=replayData,
|
||||
CreatorName=userData.Username,
|
||||
UserID=userData.ID
|
||||
).dict()
|
||||
|
||||
},
|
||||
})
|
||||
if BestTime is None or playtime < BestTime:
|
||||
print(BestTime, playtime)
|
||||
NewMapRecord = True
|
||||
|
||||
return {"FirstClear": FirstClear, "NewMapRecord": NewMapRecord}
|
||||
@app.put("/api/v1/map")
|
||||
async def upload_map(
|
||||
Authorization: Union[str, None] = Header(default=None),
|
||||
mapName: str = Form(),
|
||||
mapDescription: str = Form(default=''),
|
||||
mapVersion: int = Form(),
|
||||
mapData: str = Form(),
|
||||
file: Optional[UploadFile] = None,
|
||||
mapReplay: str = Form(),
|
||||
deaths: int = Form(),
|
||||
playtime: int = Form(),
|
||||
totalDeaths: int = Form(),
|
||||
totalTime: int = Form(),
|
||||
listed: int = Form(),
|
||||
requiresCancels: int = Form(),
|
||||
hideInChallenges: int = Form(),
|
||||
tags: str = Form(),
|
||||
rng: int = Form(),
|
||||
clientVersion: float = Form(),
|
||||
):
|
||||
authcheck = db.auth_check(Authorization)
|
||||
if not authcheck[0] and authcheck[1] == "nouser":
|
||||
raise HTTPException(404, detail="User not found")
|
||||
elif not authcheck[0] and authcheck[1] == "wrongpass":
|
||||
raise HTTPException(403, detail="Wrong password")
|
||||
elif authcheck[0]:
|
||||
print(authcheck)
|
||||
userData = types.User(**authcheck[1])
|
||||
MapCode = "SUSYBAKA" #FIXME Add mapcodes correctly
|
||||
db.maps_collection.insert_one({
|
||||
**types.Map(
|
||||
CreatorName=userData.Username,
|
||||
CreatorId=userData.ID,
|
||||
|
||||
ID=len(list(db.maps_collection.find({}))) + 1,
|
||||
Name=mapName,
|
||||
Description=mapDescription,
|
||||
Version=mapVersion,
|
||||
MapCode=MapCode,
|
||||
MapData=mapData,
|
||||
Listed=bool(listed),
|
||||
HiddenInChallenges=hideInChallenges,
|
||||
TagIDs=tags,
|
||||
TagNames=",".join(types.convertTagsToNames(tags)),
|
||||
|
||||
ShoesColor=userData.ShoesColor,
|
||||
PantsColor=userData.PantsColor,
|
||||
ShirtColor=userData.ShirtColor,
|
||||
CapeColor=userData.CapeColor,
|
||||
SkinColor=userData.SkinColor,
|
||||
HairColor=userData.HairColor,
|
||||
HatSpr=userData.HatSpr,
|
||||
Country=userData.Country,
|
||||
HairSpr=userData.HairSpr,
|
||||
HatColor=userData.HatColor,
|
||||
HatColorInv=userData.HatColorInv,
|
||||
FacialExpression=userData.FacialExpression,
|
||||
DeathEffect=userData.DeathEffect,
|
||||
GunSpr=userData.GunSpr,
|
||||
BulletSpr=userData.BulletSpr,
|
||||
SwordSpr=userData.SwordSpr,
|
||||
Costume=userData.Costume,
|
||||
FollowerSpr=userData.FollowerSpr,
|
||||
FollowerColor=userData.FollowerColor,
|
||||
SaveEffect=userData.SaveEffect,
|
||||
TextSnd=userData.TextSnd,
|
||||
|
||||
Leaderboard=[
|
||||
types.MapLeaderboard(
|
||||
BestPlaytime=playtime,
|
||||
BestPlaytimeTime="2020-02-13T15:19:33Z",
|
||||
BestReplay=mapReplay,
|
||||
CreatorName=userData.Username,
|
||||
UserID=userData.ID
|
||||
|
||||
).dict()
|
||||
]
|
||||
).dict()
|
||||
})
|
||||
return {"MapCode": MapCode}
|
||||
#raise HTTPException(501)
|
||||
|
||||
@app.get("/api/v1/map/{mapID}/besttimes/{maxEntries}")
|
||||
async def getMapLeaderboard(mapID, maxEntries):
|
||||
"""Returns maxEntries records for the specified level"""
|
||||
query = db.maps_collection.find_one({ "ID": int(mapID) })
|
||||
if not query:
|
||||
raise HTTPException(404, detail="Map not found")
|
||||
del query['_id']
|
||||
leaderboard = query['Leaderboard']
|
||||
|
||||
return leaderboard
|
||||
@app.get("/api/v1/map/{mapID}/userbesttime/{userID}")
|
||||
async def getPlayerRecord(mapID, userID):
|
||||
"""Returns specific replay"""
|
||||
query = db.maps_collection.find_one({ "ID": int(mapID) })
|
||||
if not query:
|
||||
raise HTTPException(404, detail="Map not found")
|
||||
del query['_id']
|
||||
leaderboard = query['Leaderboard']
|
||||
|
||||
# Find user
|
||||
replayIndex = None
|
||||
for i in range(len(leaderboard)):
|
||||
print(f"DBG: {i}, {leaderboard[i]}\t{leaderboard[i]['UserID'] == int(userID):}")
|
||||
if leaderboard[i]["UserID"] == int(userID):
|
||||
replayIndex = i
|
||||
break
|
||||
print(leaderboard[replayIndex])
|
||||
|
||||
if replayIndex is not None: return {"BestMapTime": leaderboard[replayIndex], "Exists": True}
|
||||
else: return {"BestMapTime": None, "Exists": False}
|
||||
|
||||
@app.get("/api/v1/map/{mapID}/besttime")
|
||||
async def getBestRecord(mapID: int):
|
||||
query = db.maps_collection.find_one({ "ID": int(mapID) })
|
||||
BestTime = None
|
||||
BestPlay = None
|
||||
for i in query['Leaderboard']:
|
||||
if BestTime is None or BestTime > i['BestPlaytime']:
|
||||
BestTime = i["BestPlaytime"]
|
||||
BestPlay = i
|
||||
|
||||
return {"BestMapTime": BestPlay, "Exists": True}
|
||||
|
||||
@app.post("/api/v1/map/{mapID}/invalidatealltimes")
|
||||
async def invalidateAllTimes(
|
||||
mapID: int,
|
||||
Reason: int = Body(),
|
||||
CustomReason: str = Body(default=""),
|
||||
Authorization: Union[str, None] = Header(default=None),
|
||||
):
|
||||
"""Removes ALL records from a specific map"""
|
||||
|
||||
authcheck = db.auth_check(Authorization)
|
||||
if not authcheck[0] and authcheck[1] == "nouser":
|
||||
raise HTTPException(404, detail="User not found")
|
||||
elif not authcheck[0] and authcheck[1] == "wrongpass":
|
||||
raise HTTPException(403, detail="Wrong password")
|
||||
elif authcheck[0]:
|
||||
userData = types.User(**authcheck[1])
|
||||
if userData.Admin:
|
||||
query = db.maps_collection.update_one({"ID": mapID}, {"$push": {
|
||||
"Leaderboard": {
|
||||
"$each": [],
|
||||
"$slice": 0
|
||||
}
|
||||
}
|
||||
})
|
||||
db.LogAdminAction(action_type="invalidateAllTimes",
|
||||
action_data={
|
||||
"Reason": Reason,
|
||||
"CustomReason": CustomReason,
|
||||
"mapID": mapID
|
||||
},
|
||||
UserID=userData.ID
|
||||
)
|
||||
else:
|
||||
db.LogAdminAction(action_type="invalidateAllTimes", action_data={
|
||||
"Reason": Reason,
|
||||
"CustomReason": CustomReason,
|
||||
"mapID": mapID,
|
||||
"unauthorized": True
|
||||
}, success=False)
|
||||
raise HTTPException(403, detail="Attempted to perform an administrator action without permission. This will be reported.")
|
||||
## General
|
||||
|
||||
@app.post("/api/v1/reports")
|
||||
async def reportContent(
|
||||
Authorization: Union[str, None] = Header(default=None),
|
||||
user_id: int = Form(),
|
||||
map_id: Optional[int] = Form(default=None),
|
||||
report_type: int = Form(),
|
||||
content: str = Form()
|
||||
):
|
||||
|
||||
authcheck = db.auth_check(Authorization)
|
||||
if not authcheck[0] and authcheck[1] == "nouser":
|
||||
raise HTTPException(404, detail="User not found")
|
||||
elif not authcheck[0] and authcheck[1] == "wrongpass":
|
||||
raise HTTPException(403, detail="Wrong password")
|
||||
elif authcheck[0]:
|
||||
db.reports_collection.insert_one({"user_id": user_id, "map_id": map_id, "report_type": report_type, "content": content})
|
||||
return True
|
||||
|
||||
@app.get("/api/v1/featuredlist")
|
||||
async def featuredlist():
|
||||
"""Returns the list id of the weekly levels list."""
|
||||
#FIXME Add playlists
|
||||
return
|
||||
|
||||
@app.get("/api/v1/followcheck")
|
||||
async def followcheck():
|
||||
#FIXME Find the purpouse of this endpoint
|
||||
return 1
|
||||
|
||||
def start():
|
||||
"""Launched with `poetry run start` at root level"""
|
||||
uvicorn.run("customiwmserver.main:app", host="0.0.0.0", port=8001, reload=True)
|
Loading…
Add table
Add a link
Reference in a new issue