artemis/titles/idac/frontend.py

391 lines
12 KiB
Python

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