forked from Hay1tsme/artemis
399 lines
13 KiB
Python
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")
|