artemis/titles/idac/frontend.py

import json
import yaml
import jinja2

from os import path
from typing import Any, Type
from twisted.web import resource
from twisted.web.util import redirectTo
from twisted.web.http import Request
from twisted.web.server import Session

from core.frontend import FE_Base, IUserSession
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.encode() 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.encode()][0].decode()

        # 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

    def to_json(self):
        return json.dumps(self.make(), default=str, ensure_ascii=False).encode("utf-8")


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"
        # TODO: Add version list
        self.version = IDACConstants.VER_IDAC_SEASON_2

        self.putChild(b"profile", IDACProfileFrontend(cfg, self.environment))
        self.putChild(b"ranking", IDACRankingFrontend(cfg, self.environment))


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

        template = self.environment.get_template(
            "titles/idac/frontend/idac_index.jinja"
        )
        sesh: Session = request.getSession()
        usr_sesh = IUserSession(sesh)

        # redirect to the ranking page
        if uri.startswith("/game/idac"):
            return redirectTo(b"/game/idac/ranking", request)

        return 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",
        ).encode("utf-16")

    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
    
    def render_GET(self, request: Request) -> bytes:
        uri: str = request.uri.decode()

        template = self.environment.get_template(
            "titles/idac/frontend/ranking/index.jinja"
        )
        sesh: Session = request.getSession()
        usr_sesh = IUserSession(sesh)
        user_id = usr_sesh.userId
        # user_id = usr_sesh.user_id

        # IDAC constants
        if uri.startswith("/game/idac/ranking/const.get"):
            # set the content type to json
            request.responseHeaders.addRawHeader(b"content-type", b"application/json")

            # get the constants
            with open("titles/idac/frontend/const.json", "r", encoding="utf-8") as f:
                constants = json.load(f)

            return json.dumps(constants, ensure_ascii=False).encode("utf-8")

        # leaderboard ranking
        elif uri.startswith("/game/idac/ranking/ranking.get"):
            # set the content type to json
            request.responseHeaders.addRawHeader(b"content-type", b"application/json")

            req = RankingRequest(request.args)
            resp = RankingResponse()

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

            # get the total number of records
            total_records = 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 resp.to_json()

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

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

            ranking = 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 = self.data.profile.get_profile(user_id, self.version)
                arcade = 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 resp.to_json()

        return 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",
        ).encode("utf-16")


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",
        }
    
    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 = 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)


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

        template = self.environment.get_template(
            "titles/idac/frontend/profile/index.jinja"
        )
        sesh: Session = request.getSession()
        usr_sesh = IUserSession(sesh)
        user_id = usr_sesh.userId
        # user_id = usr_sesh.user_id

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

            # set the file name, content type and size to download the json
            content = self.generate_all_tables_json(user_id).encode("utf-8")
            request.responseHeaders.addRawHeader(
                b"content-type", b"application/octet-stream"
            )
            request.responseHeaders.addRawHeader(
                b"content-disposition", b"attachment; filename=idac_profile.json"
            )
            request.responseHeaders.addRawHeader(
                b"content-length", str(len(content)).encode("utf-8")
            )

            self.logger.info(f"User {user_id} exported their IDAC data")
            return content

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

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

        return 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),
            active_page="idac",
            active_tab="profile",
        ).encode("utf-16")