Vanilla/artemis
1
0
Fork 0
forked from Hay1tsme/artemis
Code Pull Requests Activity

frontend: arcade management rework

Browse Source
This commit is contained in:
Hay1tsme
2025-03-25 10:43:26 -04:00
parent 20d9a2da9c
commit b6d22ef41d
8 changed files with 556 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files

241
core/frontend.py
View File

@ -12,7 +12,6 @@ import jwt
import yaml
import secrets
import string
import random
from base64 import b64decode
from enum import Enum
from datetime import datetime, timezone
@ -20,6 +19,7 @@ from os import path, environ, mkdir, W_OK, access
from core import CoreConfig, Utils
from core.data import Data
from core.const import AllnetCountryCode
# A-HJ-NP-Z
SERIAL_LETTERS = ['A', 'B', 'C', 'D', 'E', 'F', 'G', 'H', 'J', 'K', 'L', 'M', 'N', 'P', 'Q', 'R', 'S', 'T', 'U', 'V', 'W', 'X', 'Y', 'Z']
@ -37,8 +37,7 @@ class ShopPermissionOffset(Enum):
VIEW = 0 # View info and cabs
BOOKKEEP = 1 # View bookeeping info
EDITOR = 2 # Can edit name, settings
REGISTRAR = 3 # Can add cabs
# 4 - 6 reserved for future use
# 3 - 6 reserved for future use
OWNER = 7 # Can do anything
class ShopOwner():
@ -149,10 +148,13 @@ class FrontendServlet():
Mount("/shop", routes=[
Route("/", self.arcade.render_GET, methods=['GET']),
Route("/{shop_id:int}", self.arcade.render_GET, methods=['GET']),
Route("/{shop_id:int}/info.update", self.arcade.update_shop, methods=['POST']),
]),
Mount("/cab", routes=[
Route("/", self.machine.render_GET, methods=['GET']),
Route("/{machine_id:int}", self.machine.render_GET, methods=['GET']),
Route("/{machine_id:int}/info.update", self.machine.update_cab, methods=['POST']),
Route("/{machine_id:int}/reassign", self.machine.reassign_cab, methods=['POST']),
]),
Mount("/game", routes=g_routes),
Route("/robots.txt", self.robots)
@ -455,6 +457,16 @@ class FE_User(FE_Base):
card_data = []
arcade_data = []
managed_arcades = await self.data.arcade.get_arcades_managed_by_user(user_id)
if managed_arcades:
for arcade in managed_arcades:
ac = await self.data.arcade.get_arcade(arcade['id'])
if ac:
arcade_data.append({
"id": ac['id'],
"name": ac['name'],
})
for c in cards:
if c['is_locked']:
status = 'Locked'
@ -861,14 +873,16 @@ class FE_System(FE_Base):
name = frm.get("shopName", None)
country = frm.get("shopCountry", "JPN")
ip = frm.get("shopIp", None)
owner = frm.get("shopOwner", None)
acid = await self.data.arcade.create_arcade(name if name else None, name if name else None, country)
if not acid:
return RedirectResponse("/sys/?e=99", 303)
if ip:
# TODO: set IP
pass
await self.data.arcade.set_arcade_vpn_ip(acid, ip if ip else None)
if owner:
await self.data.arcade.add_arcade_owner(acid, int(owner), 255)
return Response(template.render(
title=f"{self.core_config.server.name} | System",
@ -947,15 +961,20 @@ class FE_Arcade(FE_Base):
shop_id = request.path_params.get('shop_id', None)
usr_sesh = self.validate_session(request)
if not usr_sesh or not self.test_perm(usr_sesh.permissions, PermissionOffset.ACMOD):
self.logger.warning(f"User {usr_sesh.user_id} does not have permission to view shops!")
if not usr_sesh:
return RedirectResponse("/gate/", 303)
if not shop_id:
return Response(template.render(
title=f"{self.core_config.server.name} | Arcade",
sesh=vars(usr_sesh),
), media_type="text/html; charset=utf-8")
return Response('Not Found', status_code=404)
is_acmod = self.test_perm(usr_sesh.permissions, PermissionOffset.ACMOD)
if not is_acmod:
usr_shop_perm = await self.data.arcade.get_manager_permissions(usr_sesh.user_id, shop_id)
if usr_shop_perm is None or usr_shop_perm == 0:
self.logger.warning(f"User {usr_sesh.user_id} does not have permission to view shop {shop_id}!")
return RedirectResponse("/", 303)
else:
usr_shop_perm = 15 # view, bookeep, edit
sinfo = await self.data.arcade.get_arcade(shop_id)
if not sinfo:
@ -974,38 +993,204 @@ class FE_Arcade(FE_Base):
"game": x['game'],
})
managers = []
if (usr_shop_perm & 1 << ShopPermissionOffset.OWNER.value) or is_acmod:
mgrs = await self.data.arcade.get_arcade_owners(sinfo['id'])
if mgrs:
for mgr in mgrs:
usr = await self.data.user.get_user(mgr['user'])
managers.append({
'user': mgr['user'],
'name': usr['username'] if usr['username'] else 'No Name Set',
'is_view': bool(mgr['permissions'] & 1 << ShopPermissionOffset.VIEW.value),
'is_bookkeep': bool(mgr['permissions'] & 1 << ShopPermissionOffset.BOOKKEEP.value),
'is_edit': bool(mgr['permissions'] & 1 << ShopPermissionOffset.EDITOR.value),
'is_owner': bool(mgr['permissions'] & 1 << ShopPermissionOffset.OWNER.value),
})
if request.query_params.get("e", None):
err = int(request.query_params.get("e"))
else:
err = 0
if request.query_params.get("s", None):
suc = int(request.query_params.get("s"))
else:
suc = 0
return Response(template.render(
title=f"{self.core_config.server.name} | Arcade",
sesh=vars(usr_sesh),
arcade={
"name": sinfo['name'],
"id": sinfo['id'],
"cabs": cablst
}
cablst=cablst,
arcade=sinfo._asdict(),
can_bookkeep=bool(usr_shop_perm & 1 << ShopPermissionOffset.BOOKKEEP.value) or is_acmod,
can_edit=bool(usr_shop_perm & 1 << ShopPermissionOffset.EDITOR.value) or is_acmod,
is_owner=usr_shop_perm & 1 << ShopPermissionOffset.OWNER.value,
is_acmod=is_acmod,
managers=managers,
error=err,
success=suc
), media_type="text/html; charset=utf-8")
async def update_shop(self, request: Request):
shop_id = request.path_params.get('shop_id', None)
usr_sesh = self.validate_session(request)
if not usr_sesh:
return RedirectResponse("/gate/", 303)
sinfo = await self.data.arcade.get_arcade(shop_id)
if not shop_id or not sinfo:
return RedirectResponse("/", 303)
if not self.test_perm(usr_sesh.permissions, PermissionOffset.ACMOD):
usr_shop_perm = await self.data.arcade.get_manager_permissions(usr_sesh.user_id, sinfo['id'])
if usr_shop_perm is None or usr_shop_perm == 0:
self.logger.warning(f"User {usr_sesh.user_id} does not have permission to view shop {sinfo['id']}!")
return RedirectResponse("/", 303)
frm = await request.form()
new_name = frm.get('name', None)
new_nickname = frm.get('nickname', None)
new_country = frm.get('country', None)
new_region1 = frm.get('region1', None)
new_region2 = frm.get('region2', None)
new_tz = frm.get('tz', None)
new_ip = frm.get('ip', None)
try:
AllnetCountryCode(new_country)
except ValueError:
new_country = 'JPN'
did_name = await self.data.arcade.set_arcade_name_nickname(sinfo['id'], new_name if new_name else f'Arcade{sinfo["id"]}', new_nickname if new_nickname else None)
did_region = await self.data.arcade.set_arcade_region_info(sinfo['id'], new_country, new_region1 if new_region1 else None, new_region2 if new_region2 else None, None, None)
did_timezone = await self.data.arcade.set_arcade_timezone(sinfo['id'], new_tz if new_tz else None)
did_vpn = await self.data.arcade.set_arcade_vpn_ip(sinfo['id'], new_ip if new_ip else None)
if not did_name or not did_region or not did_timezone or not did_vpn:
self.logger.error(f"Failed to update some shop into: Name: {did_name} Region: {did_region} TZ: {did_timezone} VPN: {did_vpn}")
return RedirectResponse(f"/shop/{shop_id}?e=15", 303)
return RedirectResponse(f"/shop/{shop_id}?s=1", 303)
class FE_Machine(FE_Base):
async def render_GET(self, request: Request):
template = self.environment.get_template("core/templates/machine/index.jinja")
cab_id = request.path_params.get('cab_id', None)
cab_id = request.path_params.get('machine_id', None)
usr_sesh = self.validate_session(request)
if not usr_sesh or not self.test_perm(usr_sesh.permissions, PermissionOffset.ACMOD):
self.logger.warning(f"User {usr_sesh.user_id} does not have permission to view shops!")
if not usr_sesh:
return RedirectResponse("/gate/", 303)
cab = await self.data.arcade.get_machine(id=cab_id)
if not cab_id:
return Response(template.render(
title=f"{self.core_config.server.name} | Machine",
sesh=vars(usr_sesh),
), media_type="text/html; charset=utf-8")
if not cab_id or not cab:
return Response('Not Found', status_code=404)
shop = await self.data.arcade.get_arcade(cab['arcade'])
is_acmod = self.test_perm(usr_sesh.permissions, PermissionOffset.ACMOD)
if not is_acmod:
usr_shop_perm = await self.data.arcade.get_manager_permissions(usr_sesh.user_id, shop['id'])
if usr_shop_perm is None or usr_shop_perm == 0:
self.logger.warning(f"User {usr_sesh.user_id} does not have permission to view shop {shop['id']}!")
return RedirectResponse("/", 303)
else:
usr_shop_perm = 15 # view, bookeep, edit
if request.query_params.get("e", None):
err = int(request.query_params.get("e"))
else:
err = 0
if request.query_params.get("s", None):
suc = int(request.query_params.get("s"))
else:
suc = 0
return Response(template.render(
title=f"{self.core_config.server.name} | Machine",
sesh=vars(usr_sesh),
arcade={}
arcade=shop._asdict(),
machine=cab._asdict(),
can_bookkeep=bool(usr_shop_perm & 1 << ShopPermissionOffset.BOOKKEEP.value) or is_acmod,
can_edit=bool(usr_shop_perm & 1 << ShopPermissionOffset.EDITOR.value) or is_acmod,
is_owner=usr_shop_perm & 1 << ShopPermissionOffset.OWNER.value,
is_acmod=is_acmod,
error=err,
success=suc
), media_type="text/html; charset=utf-8")
async def update_cab(self, request: Request):
cab_id = request.path_params.get('machine_id', None)
usr_sesh = self.validate_session(request)
if not usr_sesh:
return RedirectResponse("/gate/", 303)
cab = await self.data.arcade.get_machine(id=cab_id)
if not cab_id or not cab:
return RedirectResponse("/", 303)
if not self.test_perm(usr_sesh.permissions, PermissionOffset.ACMOD):
usr_shop_perm = await self.data.arcade.get_manager_permissions(usr_sesh.user_id, cab['arcade'])
if usr_shop_perm is None or usr_shop_perm == 0:
self.logger.warning(f"User {usr_sesh.user_id} does not have permission to view shop {cab['arcade']}!")
return RedirectResponse("/", 303)
frm = await request.form()
new_game = frm.get('game', None)
new_country = frm.get('country', None)
new_tz = frm.get('tz', None)
new_is_cab = frm.get('is_cab', False) == 'on'
new_is_ota = frm.get('is_ota', False) == 'on'
new_memo = frm.get('memo', None)
try:
AllnetCountryCode(new_country)
except ValueError:
new_country = None
did_game = await self.data.arcade.set_machine_game(cab['id'], new_game if new_game else None)
did_country = await self.data.arcade.set_machine_country(cab['id'], new_country if new_country else None)
did_timezone = await self.data.arcade.set_machine_timezone(cab['id'], new_tz if new_tz else None)
did_real_cab = await self.data.arcade.set_machine_real_cabinet(cab['id'], new_is_cab)
did_ota = await self.data.arcade.set_machine_can_ota(cab['id'], new_is_ota)
did_memo = await self.data.arcade.set_machine_memo(cab['id'], new_memo if new_memo else None)
if not did_game or not did_country or not did_timezone or not did_real_cab or not did_ota or not did_memo:
self.logger.error(f"Failed to update some shop into: Game: {did_game} Country: {did_country} TZ: {did_timezone} Real: {did_real_cab} OTA: {did_ota} Memo: {did_memo}")
return RedirectResponse(f"/cab/{cab['id']}?e=15", 303)
return RedirectResponse(f"/cab/{cab_id}?s=1", 303)
async def reassign_cab(self, request: Request):
cab_id = request.path_params.get('machine_id', None)
usr_sesh = self.validate_session(request)
if not usr_sesh:
return RedirectResponse("/gate/", 303)
cab = await self.data.arcade.get_machine(id=cab_id)
if not cab_id or not cab:
return RedirectResponse("/", 303)
frm = await request.form()
new_shop = frm.get('new_arcade', None)
if not self.test_perm(usr_sesh.permissions, PermissionOffset.ACMOD):
self.logger.warning(f"User {usr_sesh.user_id} does not have permission to reassign cab {cab['id']} to arcade !")
return RedirectResponse(f"/cab/{cab_id}?e=11", 303)
new_sinfo = await self.data.arcade.get_arcade(new_shop)
if not new_sinfo:
return RedirectResponse(f"/cab/{cab_id}?e=14", 303)
if not await self.data.arcade.set_machine_arcade(cab['id'], new_sinfo['id']):
return RedirectResponse(f"/cab/{cab_id}?e=99", 303)
return RedirectResponse(f"/cab/{cab_id}?s=2", 303)
cfg_dir = environ.get("ARTEMIS_CFG_DIR", "config")
cfg: CoreConfig = CoreConfig()