artemis/titles/mai2/frontend.py

302 lines
12 KiB
Python
Raw Normal View History

2024-06-09 07:05:57 +00:00
from typing import List
from starlette.routing import Route, Mount
from starlette.requests import Request
from starlette.responses import Response, RedirectResponse
from os import path
import yaml
import jinja2
2024-07-01 22:26:39 +00:00
from datetime import datetime
2024-06-09 07:05:57 +00:00
2024-07-01 22:26:39 +00:00
from core.frontend import FE_Base, UserSession, PermissionOffset
2024-06-09 07:05:57 +00:00
from core.config import CoreConfig
from .database import Mai2Data
from .config import Mai2Config
from .const import Mai2Constants
class Mai2Frontend(FE_Base):
def __init__(
self, cfg: CoreConfig, environment: jinja2.Environment, cfg_dir: str
) -> None:
super().__init__(cfg, environment)
self.data = Mai2Data(cfg)
self.game_cfg = Mai2Config()
if path.exists(f"{cfg_dir}/{Mai2Constants.CONFIG_NAME}"):
self.game_cfg.update(
yaml.safe_load(open(f"{cfg_dir}/{Mai2Constants.CONFIG_NAME}"))
)
self.nav_name = "maimai"
def get_routes(self) -> List[Route]:
return [
Route("/", self.render_GET, methods=['GET']),
Mount("/playlog", routes=[
Route("/", self.render_GET_playlog, methods=['GET']),
Route("/{index}", self.render_GET_playlog, methods=['GET']),
]),
2024-07-01 22:26:39 +00:00
Mount("/events", routes=[
Route("/", self.render_events, methods=['GET']),
Route("/{event_id:int}", self.render_event_edit, methods=['GET']),
Route("/update", self.update_event, methods=['POST']),
Route("/version.change", self.version_change, methods=['POST']),
]),
2024-06-09 07:05:57 +00:00
Route("/update.name", self.update_name, methods=['POST']),
Route("/version.change", self.version_change, methods=['POST']),
]
async def render_GET(self, request: Request) -> bytes:
template = self.environment.get_template(
"titles/mai2/templates/mai2_index.jinja"
)
usr_sesh = self.validate_session(request)
if not usr_sesh:
usr_sesh = UserSession()
2024-07-01 22:26:39 +00:00
incoming_ver = usr_sesh.maimai_version
2024-06-09 07:05:57 +00:00
if usr_sesh.user_id > 0:
versions = await self.data.profile.get_all_profile_versions(usr_sesh.user_id)
profile = []
if versions:
# maimai_version is -1 means it is not initialized yet, select a default version from existing.
2024-07-01 22:26:39 +00:00
if incoming_ver < 0:
2024-06-09 07:05:57 +00:00
usr_sesh.maimai_version = versions[0]['version']
profile = await self.data.profile.get_profile_detail(usr_sesh.user_id, usr_sesh.maimai_version)
versions = [x['version'] for x in versions]
resp = Response(template.render(
title=f"{self.core_config.server.name} | {self.nav_name}",
game_list=self.environment.globals["game_list"],
sesh=vars(usr_sesh),
user_id=usr_sesh.user_id,
profile=profile,
version_list=Mai2Constants.VERSION_STRING,
versions=versions,
cur_version=usr_sesh.maimai_version
), media_type="text/html; charset=utf-8")
2024-07-01 22:26:39 +00:00
if incoming_ver < 0:
2024-06-09 07:05:57 +00:00
encoded_sesh = self.encode_session(usr_sesh)
2024-06-09 07:14:43 +00:00
resp.delete_cookie("ARTEMIS_SESH")
resp.set_cookie("ARTEMIS_SESH", encoded_sesh)
2024-06-09 07:05:57 +00:00
return resp
else:
return RedirectResponse("/gate/", 303)
async def render_GET_playlog(self, request: Request) -> bytes:
template = self.environment.get_template(
"titles/mai2/templates/mai2_playlog.jinja"
)
usr_sesh = self.validate_session(request)
if not usr_sesh:
usr_sesh = UserSession()
if usr_sesh.user_id > 0:
if usr_sesh.maimai_version < 0:
return RedirectResponse("/game/mai2/", 303)
path_index = request.path_params.get('index')
if not path_index or int(path_index) < 1:
index = 0
else:
index = int(path_index) - 1 # 0 and 1 are 1st page
user_id = usr_sesh.user_id
playlog_count = await self.data.score.get_user_playlogs_count(user_id)
if playlog_count < index * 20 :
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),
playlog_count=0
), media_type="text/html; charset=utf-8")
playlog = await self.data.score.get_playlogs(user_id, index, 20)
playlog_with_title = []
for record in playlog:
music_chart = await self.data.static.get_music_chart(usr_sesh.maimai_version, record.musicId, record.level)
if music_chart:
difficultyNum=music_chart.chartId
difficulty=music_chart.difficulty
artist=music_chart.artist
title=music_chart.title
else:
difficultyNum=0
difficulty=0
artist="unknown"
title="musicid: " + str(record.musicId)
playlog_with_title.append({
"raw": record,
"title": title,
"difficultyNum": difficultyNum,
"difficulty": difficulty,
"artist": artist,
})
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),
user_id=usr_sesh.user_id,
playlog=playlog_with_title,
playlog_count=playlog_count
), media_type="text/html; charset=utf-8")
else:
return RedirectResponse("/gate/", 303)
async def update_name(self, request: Request) -> bytes:
usr_sesh = self.validate_session(request)
if not usr_sesh:
return RedirectResponse("/gate/", 303)
form_data = await request.form()
new_name: str = form_data.get("new_name")
new_name_full = ""
if not new_name:
return RedirectResponse("/gate/?e=4", 303)
if len(new_name) > 8:
return RedirectResponse("/gate/?e=8", 303)
for x in new_name: # FIXME: This will let some invalid characters through atm
o = ord(x)
try:
if o == 0x20:
new_name_full += chr(0x3000)
elif o < 0x7F and o > 0x20:
new_name_full += chr(o + 0xFEE0)
elif o <= 0x7F:
self.logger.warn(f"Invalid ascii character {o:02X}")
return RedirectResponse("/gate/?e=4", 303)
else:
new_name_full += x
except Exception as e:
self.logger.error(f"Something went wrong parsing character {o:04X} - {e}")
return RedirectResponse("/gate/?e=4", 303)
if not await self.data.profile.update_name(usr_sesh.user_id, new_name_full):
return RedirectResponse("/gate/?e=999", 303)
return RedirectResponse("/game/mai2/?s=1", 303)
async def version_change(self, request: Request):
usr_sesh = self.validate_session(request)
if not usr_sesh:
usr_sesh = UserSession()
2024-07-01 22:26:39 +00:00
if "/events/" in request.url.path:
resp = RedirectResponse("/game/mai2/events/", 303)
else:
resp = RedirectResponse("/game/mai2/", 303)
2024-06-09 07:05:57 +00:00
if usr_sesh.user_id > 0:
form_data = await request.form()
maimai_version = form_data.get("version")
self.logger.info(f"version change to: {maimai_version}")
if(maimai_version.isdigit()):
usr_sesh.maimai_version=int(maimai_version)
encoded_sesh = self.encode_session(usr_sesh)
2024-07-01 22:26:39 +00:00
self.logger.debug(f"Created session with JWT {encoded_sesh}")
2024-06-09 07:14:43 +00:00
resp.set_cookie("ARTEMIS_SESH", encoded_sesh)
2024-06-09 07:05:57 +00:00
return resp
else:
2024-07-01 22:26:39 +00:00
return RedirectResponse("/gate/", 303)
async def render_events(self, request: Request) -> Response:
usr_sesh = self.validate_session(request)
if not usr_sesh:
return RedirectResponse("/gate/", 303)
if not self.test_perm(usr_sesh.permissions, PermissionOffset.SYSADMIN):
return RedirectResponse("/game/mai2/", 303)
template = self.environment.get_template(
"titles/mai2/templates/events/mai2_events.jinja"
)
incoming_ver = usr_sesh.maimai_version
evts = []
if incoming_ver < 0:
usr_sesh.maimai_version = Mai2Constants.VER_MAIMAI_DX
event_list = await self.data.static.get_game_events(usr_sesh.maimai_version)
self.logger.info(f"Get events for v{usr_sesh.maimai_version}")
for event in event_list:
evts.append({
"id": event['id'],
"version": event['version'],
"eventId": event['eventId'],
"eventType": event['type'],
"name": event['name'],
"startDate": event['startDate'].strftime("%x %X"),
"enabled": "true" if event['enabled'] else "false",
})
resp = Response(template.render(
title=f"{self.core_config.server.name} | {self.nav_name} Events",
game_list=self.environment.globals["game_list"],
sesh=vars(usr_sesh),
version_list=Mai2Constants.VERSION_STRING,
events=evts
), media_type="text/html; charset=utf-8")
if incoming_ver < 0:
encoded_sesh = self.encode_session(usr_sesh)
resp.delete_cookie("ARTEMIS_SESH")
resp.set_cookie("ARTEMIS_SESH", encoded_sesh)
return resp
async def render_event_edit(self, request: Request) -> Response:
usr_sesh = self.validate_session(request)
if not usr_sesh:
return RedirectResponse("/gate/", 303)
if not self.test_perm(usr_sesh.permissions, PermissionOffset.SYSADMIN):
return RedirectResponse("/game/mai2/", 303)
template = self.environment.get_template(
"titles/mai2/templates/events/mai2_event_edit.jinja"
)
evt_id = request.path_params.get("event_id")
event_id = await self.data.static.get_event_by_id(evt_id)
if not event_id:
return RedirectResponse("/game/mai2/events/", 303)
return Response(template.render(
title=f"{self.core_config.server.name} | {self.nav_name} Edit Event {evt_id}",
game_list=self.environment.globals["game_list"],
sesh=vars(usr_sesh),
user_id=usr_sesh.user_id,
version_list=Mai2Constants.VERSION_STRING,
cur_version=usr_sesh.maimai_version,
event=event_id._asdict()
), media_type="text/html; charset=utf-8")
async def update_event(self, request: Request) -> RedirectResponse:
usr_sesh = self.validate_session(request)
if not usr_sesh:
return RedirectResponse("/gate/", 303)
if not self.test_perm(usr_sesh.permissions, PermissionOffset.SYSADMIN):
return RedirectResponse("/game/mai2/", 303)
form_data = await request.form()
print(form_data)
event_id: int = form_data.get("evtId", None)
new_enabled: bool = bool(form_data.get("evtEnabled", False))
try:
new_start_date: datetime = datetime.strptime(form_data.get("evtStart", None), "%Y-%m-%dT%H:%M:%S")
except:
new_start_date = None
print(f"{event_id} {new_enabled} {new_start_date}")
if event_id is None or new_start_date is None:
return RedirectResponse("/game/mai2/events/?e=4", 303)
await self.data.static.update_event_by_id(int(event_id), new_enabled, new_start_date)
return RedirectResponse("/game/mai2/events/?s=1", 303)