cutt/main.py

from typing import Annotated
from fastapi import APIRouter, Depends, FastAPI, HTTPException, Security, status
from fastapi.responses import JSONResponse
from fastapi.staticfiles import StaticFiles
from db import Player, Team, Chemistry, MVPRanking, engine
from sqlmodel import (
    Session,
    func,
    select,
)
from fastapi.middleware.cors import CORSMiddleware
from analysis import analysis_router
from security import (
    change_password,
    get_current_active_user,
    login_for_access_token,
    logout,
    read_player_me,
    read_own_items,
    set_first_password,
)

C = Chemistry
R = MVPRanking
P = Player

app = FastAPI(title="cutt")
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 add_player(player: Player):
    with Session(engine) as session:
        session.add(player)
        session.commit()


def add_players(players: list[Player]):
    with Session(engine) as session:
        for player in players:
            session.add(player)
        session.commit()


async def list_players(team_id: int):
    with Session(engine) as session:
        statement = select(Team).where(Team.id == team_id)
        players = [t.players for t in session.exec(statement)][0]
        if players:
            return [
                player.model_dump(include={"id", "display_name", "number"})
                for player in players
            ]


async def read_teams_me(user: Annotated[Player, Depends(get_current_active_user)]):
    with Session(engine) as session:
        return [p.teams for p in session.exec(select(P).where(P.id == user.id))][0]


def list_teams():
    with Session(engine) as session:
        statement = select(Team)
        return session.exec(statement).fetchall()


player_router = APIRouter(prefix="/player")
player_router.add_api_route("/list", endpoint=list_players, methods=["GET"])
player_router.add_api_route("/add", endpoint=add_player, methods=["POST"])
player_router.add_api_route("/me", endpoint=read_player_me, methods=["GET"])
player_router.add_api_route("/me/teams", endpoint=read_teams_me, methods=["GET"])
player_router.add_api_route("/me/items", endpoint=read_own_items, methods=["GET"])
player_router.add_api_route(
    "/change_password", endpoint=change_password, methods=["POST"]
)

team_router = APIRouter(
    prefix="/team",
    dependencies=[Security(get_current_active_user, scopes=["admin"])],
)
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...",
)


@api_router.put("/mvps")
def submit_mvps(
    mvps: MVPRanking,
    user: Annotated[Player, Depends(get_current_active_user)],
):
    if user.id == mvps.user:
        with Session(engine) as session:
            session.add(mvps)
            session.commit()
            return JSONResponse("success!")
    else:
        raise wrong_user_id_exception


@api_router.get("/mvps")
def get_mvps(
    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)
            .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_404_NOT_FOUND,
                detail="no previous state was found",
            )


@api_router.put("/chemistry")
def submit_chemistry(
    chemistry: Chemistry, user: Annotated[Player, Depends(get_current_active_user)]
):
    if user.id == chemistry.user:
        with Session(engine) as session:
            session.add(chemistry)
            session.commit()
            return JSONResponse("success!")
    else:
        raise wrong_user_id_exception


@api_router.get("/chemistry")
def get_chemistry(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)
            .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:
            return chemistry
        else:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                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,
    dependencies=[Security(get_current_active_user, scopes=["analysis"])],
)
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("/logout", endpoint=logout, methods=["POST"])
app.include_router(api_router)
app.mount("/", SPAStaticFiles(directory="dist", html=True), name="site")