artemis/titles/idac/frontend.py

145 lines
4.6 KiB
Python

import json
from typing import List
from starlette.routing import Route
from starlette.responses import Response, RedirectResponse
import yaml
import jinja2
from os import path
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 IDACFrontend(FE_Base):
def __init__(
self, cfg: CoreConfig, environment: jinja2.Environment, cfg_dir: str
) -> None:
super().__init__(cfg, environment)
self.data = IDACData(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.ticket_names = {
3: "car_dressup_points",
5: "avatar_points",
25: "full_tune_tickets",
34: "full_tune_fragments",
}
def get_routes(self) -> List[Route]:
return [
Route("/", self.render_GET)
]
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/idac_index.jinja"
)
usr_sesh = self.validate_session(request)
if not usr_sesh:
usr_sesh = UserSession()
user_id = usr_sesh.user_id
# user_id = usr_sesh.user_id
# profile export
if uri.startswith("/game/idac/export"):
if user_id == 0:
return RedirectResponse(b"/game/idac", request)
# set the file name, content type and size to download the json
content = await self.generate_all_tables_json(user_id).encode("utf-8")
self.logger.info(f"User {user_id} exported their IDAC data")
return Response(
content,
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),
active_page="idac",
), media_type="text/html; charset=utf-8")