artemis/titles/idac/frontend.py

import json
import yaml
import jinja2

from os import path
from typing import List, Any, Type
from starlette.routing import Route, Mount
from starlette.responses import Response, RedirectResponse, JSONResponse
from starlette.requests import Request

from core.frontend import FE_Base, UserSession
from core.config import CoreConfig
from titles.idac.database import IDACData
from titles.idac.schema.profile import *
from titles.idac.schema.item import *
from titles.idac.config import IDACConfig
from titles.idac.const import IDACConstants


class RankingData:
    def __init__(
        self,
        rank: int,
        name: str,
        record: int,
        eval_id: int,
        store: str,
        style_car_id: int,
        update_date: str,
    ) -> None:
        self.rank: int = rank
        self.name: str = name
        self.record: str = record
        self.store: str = store
        self.eval_id: int = eval_id
        self.style_car_id: int = style_car_id
        self.update_date: str = update_date

    def make(self):
        return vars(self)


class RequestValidator:
    def __init__(self) -> None:
        self.success: bool = True
        self.error: str = ""

    def validate_param(
        self,
        request_args: Dict[bytes, bytes],
        param_name: str,
        param_type: Type[None],
        default=None,
        required: bool = True,
    ) -> None:
        # Check if the parameter is missing
        if param_name not in request_args:
            if required:
                self.success = False
                self.error += f"Missing parameter: '{param_name}'. "
            else:
                # If the parameter is not required,
                # return the default value if it exists
                return default
            return None

        param_value = request_args[param_name]

        # Check if the parameter type is not empty
        if param_type:
            try:
                # Attempt to convert the parameter value to the specified type
                param_value = param_type(param_value)
            except ValueError:
                # If the conversion fails, return an error
                self.success = False
                self.error += f"Invalid parameter type for '{param_name}'. "
                return None

        return param_value


class RankingRequest(RequestValidator):
    def __init__(self, request_args: Dict[bytes, bytes]) -> None:
        super().__init__()

        self.course_id: int = self.validate_param(request_args, "courseId", int)
        self.page_number: int = self.validate_param(
            request_args, "pageNumber", int, default=1, required=False
        )


class RankingResponse:
    def __init__(self) -> None:
        self.success: bool = False
        self.error: str = ""
        self.total_pages: int = 0
        self.total_records: int = 0
        self.updated_at: str = ""
        self.ranking: list[RankingData] = []

    def make(self):
        ret = vars(self)
        self.error = (
            "Unknown error." if not self.success and self.error == "" else self.error
        )
        ret["ranking"] = [rank.make() for rank in self.ranking]

        return ret

class IDACFrontend(FE_Base):
    isLeaf = False
    children: Dict[str, Any] = {}

    def __init__(
        self, cfg: CoreConfig, environment: jinja2.Environment, cfg_dir: str
    ) -> None:
        super().__init__(cfg, environment)
        self.data = IDACData(cfg)
        self.core_cfg = cfg
        self.game_cfg = IDACConfig()
        if path.exists(f"{cfg_dir}/{IDACConstants.CONFIG_NAME}"):
            self.game_cfg.update(
                yaml.safe_load(open(f"{cfg_dir}/{IDACConstants.CONFIG_NAME}"))
            )
        self.nav_name = "頭文字D THE ARCADE"
        # self.nav_name = "IDAC"
        # TODO: Add version list
        self.version = IDACConstants.VER_IDAC_SEASON_2

        self.profile = IDACProfileFrontend(cfg, self.environment)
        self.ranking = IDACRankingFrontend(cfg, self.environment)
    
    def get_routes(self) -> List[Route]:
        return [
            Route("/", self.render_GET),
            Mount("/profile", routes=[
                Route("/", self.profile.render_GET),
                # dirty hack
                Route("/export.get", self.profile.render_GET),
            ]),
            Mount("/ranking", routes=[
                Route("/", self.ranking.render_GET),
                # dirty hack
                Route("/const.get", self.ranking.render_GET),
                Route("/ranking.get", self.ranking.render_GET),
            ]),
        ]

    async def render_GET(self, request: Request) -> bytes:
        uri: str = request.url.path

        template = self.environment.get_template(
            "titles/idac/templates/idac_index.jinja"
        )
        usr_sesh = self.validate_session(request)
        if not usr_sesh:
            usr_sesh = UserSession()

        # redirect to the ranking page
        if uri.startswith("/game/idac"):
            return RedirectResponse("/game/idac/ranking", 303)

        return Response(template.render(
            title=f"{self.core_config.server.name} | {self.nav_name}",
            game_list=self.environment.globals["game_list"],
            sesh=vars(usr_sesh),
            active_page="idac",
        ), media_type="text/html; charset=utf-8")

    async def render_POST(self, request: Request) -> bytes:
        pass


class IDACRankingFrontend(FE_Base):
    def __init__(self, cfg: CoreConfig, environment: jinja2.Environment) -> None:
        super().__init__(cfg, environment)
        self.data = IDACData(cfg)
        self.core_cfg = cfg

        self.nav_name = "頭文字D THE ARCADE"
        # TODO: Add version list
        self.version = IDACConstants.VER_IDAC_SEASON_2
    
    async def render_GET(self, request: Request) -> bytes:
        uri: str = request.url.path

        template = self.environment.get_template(
            "titles/idac/templates/ranking/index.jinja"
        )
        usr_sesh = self.validate_session(request)
        if not usr_sesh:
            usr_sesh = UserSession()
        user_id = usr_sesh.user_id

        # IDAC constants
        if uri.startswith("/game/idac/ranking/const.get"):
            # get the constants
            with open("titles/idac/templates/const.json", "r", encoding="utf-8") as f:
                constants = json.load(f)

            return JSONResponse(constants)

        # leaderboard ranking
        elif uri.startswith("/game/idac/ranking/ranking.get"):
            req = RankingRequest(request.query_params._dict)
            resp = RankingResponse()

            if not req.success:
                resp.error = req.error
                return JSONResponse(resp.make())

            # get the total number of records
            total_records = await self.data.item.get_time_trial_ranking_by_course_total(
                self.version, req.course_id
            )
            # return an error if there are no records
            if total_records is None or total_records == 0:
                resp.error = "No records found."
                return JSONResponse(resp.make())

            # get the total number of records
            total = total_records["count"]

            limit = 50
            offset = (req.page_number - 1) * limit

            ranking = await self.data.item.get_time_trial_ranking_by_course(
                self.version,
                req.course_id,
                limit=limit,
                offset=offset,
            )

            for i, rank in enumerate(ranking):
                user_id = rank["user"]

                # get the username, country and store from the profile
                profile = await self.data.profile.get_profile(user_id, self.version)
                arcade = await self.data.arcade.get_arcade(profile["store"])

                if arcade is None:
                    arcade = {}
                    arcade["name"] = self.core_config.server.name

                # should never happen
                if profile is None:
                    continue

                resp.ranking.append(
                    RankingData(
                        rank=offset + i + 1,
                        name=profile["username"],
                        record=rank["goal_time"],
                        store=arcade["name"],
                        eval_id=rank["eval_id"],
                        style_car_id=rank["style_car_id"],
                        update_date=str(rank["play_dt"]),
                    )
                )

            # now return the json data, with the total number of pages and records
            # round up the total pages
            resp.success = True
            resp.total_pages = (total // limit) + 1
            resp.total_records = total
            return JSONResponse(resp.make())

        return Response(template.render(
            title=f"{self.core_config.server.name} | {self.nav_name}",
            game_list=self.environment.globals["game_list"],
            sesh=vars(usr_sesh),
            active_page="idac",
            active_tab="ranking",
        ), media_type="text/html; charset=utf-8")


class IDACProfileFrontend(FE_Base):
    def __init__(self, cfg: CoreConfig, environment: jinja2.Environment) -> None:
        super().__init__(cfg, environment)
        self.data = IDACData(cfg)
        self.core_cfg = cfg

        self.nav_name = "頭文字D THE ARCADE"
        # TODO: Add version list
        self.version = IDACConstants.VER_IDAC_SEASON_2

        self.ticket_names = {
            3: "car_dressup_points",
            5: "avatar_points",
            25: "full_tune_tickets",
            34: "full_tune_fragments",
        }

    async def generate_all_tables_json(self, user_id: int):
        json_export = {}

        idac_tables = {
            profile,
            config,
            avatar,
            rank,
            stock,
            theory,
            car,
            ticket,
            story,
            episode,
            difficulty,
            course,
            trial,
            challenge,
            theory_course,
            theory_partner,
            theory_running,
            vs_info,
            stamp,
            timetrial_event,
        }

        for table in idac_tables:
            sql = select(table).where(
                table.c.user == user_id,
            )

            # check if the table has a version column
            if "version" in table.c:
                sql = sql.where(table.c.version == self.version)

            # lol use the profile connection for items, dirty hack
            result = await self.data.profile.execute(sql)
            data_list = result.fetchall()

            # add the list to the json export with the correct table name
            json_export[table.name] = []
            for data in data_list:
                tmp = data._asdict()
                tmp.pop("id")
                tmp.pop("user")
                json_export[table.name].append(tmp)

        return json.dumps(json_export, indent=4, default=str, ensure_ascii=False)

    async def render_GET(self, request: Request) -> bytes:
        uri: str = request.url.path

        template = self.environment.get_template(
            "titles/idac/templates/profile/index.jinja"
        )
        usr_sesh = self.validate_session(request)
        if not usr_sesh:
            usr_sesh = UserSession()
        user_id = usr_sesh.user_id

        user = await self.data.user.get_user(user_id)
        if user is None:
            self.logger.debug(f"User {user_id} not found")
            return RedirectResponse("/user/", 303)

        # profile export
        if uri.startswith("/game/idac/profile/export.get"):
            if user_id == 0:
                return RedirectResponse("/game/idac", 303)

            # set the file name, content type and size to download the json
            content = await self.generate_all_tables_json(user_id)

            self.logger.info(f"User {user_id} exported their IDAC data")
            return Response(
                content.encode("utf-8"), 
                200, 
                {'content-disposition': 'attachment; filename=idac_profile.json'},
                "application/octet-stream"
            )

        profile_data, tickets, rank = None, None, None
        if user_id > 0:
            profile_data = await self.data.profile.get_profile(user_id, self.version)
            ticket_data = await self.data.item.get_tickets(user_id)
            rank = await self.data.profile.get_profile_rank(user_id, self.version)

            if ticket_data:
                tickets = {
                    self.ticket_names[ticket["ticket_id"]]: ticket["ticket_cnt"]
                    for ticket in ticket_data
                }

        return Response(template.render(
            title=f"{self.core_config.server.name} | {self.nav_name}",
            game_list=self.environment.globals["game_list"],
            profile=profile_data,
            tickets=tickets,
            rank=rank,
            sesh=vars(usr_sesh),
            username=user["username"],
            active_page="idac",
            active_tab="profile",
        ), media_type="text/html; charset=utf-8")