253 lines
7.9 KiB
Python
253 lines
7.9 KiB
Python
from typing import Annotated
|
|
from fastapi import APIRouter, Depends, FastAPI, HTTPException, Security, status
|
|
from fastapi.responses import FileResponse, JSONResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
from cutt.db import Player, PlayerType, Team, Chemistry, MVPRanking, engine
|
|
from sqlmodel import (
|
|
Session,
|
|
func,
|
|
select,
|
|
)
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from cutt.analysis import analysis_router
|
|
from cutt.security import (
|
|
get_current_active_user,
|
|
login_for_access_token,
|
|
logout,
|
|
register,
|
|
set_first_password,
|
|
)
|
|
from cutt.player import player_router
|
|
|
|
C = Chemistry
|
|
PT = PlayerType
|
|
R = MVPRanking
|
|
P = Player
|
|
|
|
app = FastAPI(
|
|
title="cutt", swagger_ui_parameters={"syntaxHighlight": {"theme": "monokai"}}
|
|
)
|
|
api_router = APIRouter(prefix="/api")
|
|
origins = [
|
|
"https://cutt.0124816.xyz",
|
|
"http://localhost:5173",
|
|
]
|
|
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=origins,
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
|
|
def add_team(team: Team):
|
|
with Session(engine) as session:
|
|
session.add(team)
|
|
session.commit()
|
|
|
|
|
|
def list_teams():
|
|
with Session(engine) as session:
|
|
statement = select(Team)
|
|
return session.exec(statement).fetchall()
|
|
|
|
|
|
team_router = APIRouter(
|
|
prefix="/teams",
|
|
dependencies=[Security(get_current_active_user, scopes=["admin"])],
|
|
tags=["team"],
|
|
)
|
|
team_router.add_api_route("/list", endpoint=list_teams, methods=["GET"])
|
|
team_router.add_api_route("/add", endpoint=add_team, methods=["POST"])
|
|
|
|
|
|
wrong_user_id_exception = HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="you're not who you think you are...",
|
|
)
|
|
somethings_fishy = HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST, detail="something up..."
|
|
)
|
|
|
|
|
|
@api_router.put("/mvps", tags=["analysis"])
|
|
def submit_mvps(
|
|
mvps: MVPRanking,
|
|
user: Annotated[Player, Depends(get_current_active_user)],
|
|
):
|
|
if mvps.team == 42:
|
|
return JSONResponse("DEMO team, nothing happens")
|
|
if user.id == mvps.user:
|
|
with Session(engine) as session:
|
|
statement = select(Team).where(Team.id == mvps.team)
|
|
players = [t.players for t in session.exec(statement)][0]
|
|
if players:
|
|
player_ids = {p.id for p in players}
|
|
if player_ids >= set(mvps.mvps):
|
|
session.add(mvps)
|
|
session.commit()
|
|
return JSONResponse("success!")
|
|
raise somethings_fishy
|
|
else:
|
|
raise wrong_user_id_exception
|
|
|
|
|
|
@api_router.get("/mvps/{team_id}", tags=["analysis"])
|
|
def get_mvps(
|
|
team_id: int,
|
|
user: Annotated[Player, Depends(get_current_active_user)],
|
|
):
|
|
with Session(engine) as session:
|
|
subquery = (
|
|
select(R.user, func.max(R.time).label("latest"))
|
|
.where(R.user == user.id)
|
|
.where(R.team == team_id)
|
|
.group_by(R.user)
|
|
.subquery()
|
|
)
|
|
statement2 = select(R).join(
|
|
subquery, (R.user == subquery.c.user) & (R.time == subquery.c.latest)
|
|
)
|
|
mvps = session.exec(statement2).one_or_none()
|
|
if mvps:
|
|
return mvps
|
|
else:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="no previous state was found",
|
|
)
|
|
|
|
|
|
@api_router.put("/chemistry", tags=["analysis"])
|
|
def submit_chemistry(
|
|
chemistry: Chemistry, user: Annotated[Player, Depends(get_current_active_user)]
|
|
):
|
|
if chemistry.team == 42:
|
|
return JSONResponse("DEMO team, nothing happens")
|
|
if user.id == chemistry.user:
|
|
with Session(engine) as session:
|
|
statement = select(Team).where(Team.id == chemistry.team)
|
|
players = [t.players for t in session.exec(statement)][0]
|
|
if players:
|
|
player_ids = {p.id for p in players}
|
|
if player_ids >= (
|
|
set(chemistry.love) | set(chemistry.hate) | set(chemistry.undecided)
|
|
):
|
|
session.add(chemistry)
|
|
session.commit()
|
|
return JSONResponse("success!")
|
|
raise somethings_fishy
|
|
else:
|
|
raise wrong_user_id_exception
|
|
|
|
|
|
@api_router.get("/chemistry/{team_id}", tags=["analysis"])
|
|
def get_chemistry(
|
|
team_id: int, user: Annotated[Player, Depends(get_current_active_user)]
|
|
):
|
|
with Session(engine) as session:
|
|
subquery = (
|
|
select(C.user, func.max(C.time).label("latest"))
|
|
.where(C.user == user.id)
|
|
.where(C.team == team_id)
|
|
.group_by(C.user)
|
|
.subquery()
|
|
)
|
|
statement2 = select(C).join(
|
|
subquery, (C.user == subquery.c.user) & (C.time == subquery.c.latest)
|
|
)
|
|
chemistry = session.exec(statement2).one_or_none()
|
|
if chemistry is not None:
|
|
return chemistry
|
|
else:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="no previous state was found",
|
|
)
|
|
|
|
|
|
@api_router.put("/playertype", tags=["analysis"])
|
|
def submit_playertype(
|
|
playertype: PlayerType, user: Annotated[Player, Depends(get_current_active_user)]
|
|
):
|
|
if playertype.team == 42:
|
|
return JSONResponse("DEMO team, nothing happens")
|
|
if user.id == playertype.user:
|
|
with Session(engine) as session:
|
|
statement = select(Team).where(Team.id == playertype.team)
|
|
players = [t.players for t in session.exec(statement)][0]
|
|
if players:
|
|
player_ids = {p.id for p in players}
|
|
if player_ids >= (
|
|
set(playertype.handlers)
|
|
| set(playertype.combis)
|
|
| set(playertype.cutters)
|
|
):
|
|
session.add(playertype)
|
|
session.commit()
|
|
return JSONResponse("success!")
|
|
raise somethings_fishy
|
|
else:
|
|
raise wrong_user_id_exception
|
|
|
|
|
|
@api_router.get("/playertype/{team_id}", tags=["analysis"])
|
|
def get_playertype(
|
|
team_id: int, user: Annotated[Player, Depends(get_current_active_user)]
|
|
):
|
|
with Session(engine) as session:
|
|
subquery = (
|
|
select(PT.user, func.max(PT.time).label("latest"))
|
|
.where(PT.user == user.id)
|
|
.where(PT.team == team_id)
|
|
.group_by(PT.user)
|
|
.subquery()
|
|
)
|
|
statement2 = select(PT).join(
|
|
subquery, (PT.user == subquery.c.user) & (PT.time == subquery.c.latest)
|
|
)
|
|
playertype = session.exec(statement2).one_or_none()
|
|
if playertype is not None:
|
|
return playertype
|
|
else:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="no previous state was found",
|
|
)
|
|
|
|
|
|
class SPAStaticFiles(StaticFiles):
|
|
async def get_response(self, path: str, scope):
|
|
response = await super().get_response(path, scope)
|
|
if response.status_code == 404:
|
|
response = await super().get_response(".", scope)
|
|
return response
|
|
|
|
|
|
api_router.include_router(
|
|
player_router, dependencies=[Depends(get_current_active_user)]
|
|
)
|
|
api_router.include_router(team_router, dependencies=[Depends(get_current_active_user)])
|
|
api_router.include_router(analysis_router)
|
|
api_router.add_api_route("/token", endpoint=login_for_access_token, methods=["POST"])
|
|
api_router.add_api_route("/set_password", endpoint=set_first_password, methods=["POST"])
|
|
api_router.add_api_route("/register", endpoint=register, methods=["POST"])
|
|
api_router.add_api_route("/logout", endpoint=logout, methods=["POST"])
|
|
app.include_router(api_router)
|
|
|
|
|
|
# app.mount("/", SPAStaticFiles(directory="dist", html=True), name="site")
|
|
@app.get("/")
|
|
async def root():
|
|
return FileResponse("dist/index.html")
|
|
|
|
|
|
@app.exception_handler(404)
|
|
async def exception_404_handler(request, exc):
|
|
return FileResponse("dist/index.html")
|
|
|
|
|
|
app.mount("/", StaticFiles(directory="dist"), name="ui")
|