artemis/titles/idac/frontend.py
2024-03-12 17:48:12 +01:00

399 lines
13 KiB
Python

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")