8 Commits

14 changed files with 1101 additions and 154 deletions

View File

@ -349,12 +349,22 @@ class AllnetServlet:
not self.config.allnet.allow_online_updates
or not self.config.allnet.update_cfg_folder
):
return PlainTextResponse(urllib.parse.unquote(urllib.parse.urlencode(vars(resp))) + "\n")
resp = urllib.parse.unquote(urllib.parse.urlencode(vars(resp))) + "\n"
if is_dfi:
return PlainTextResponse(
self.to_dfi(resp) + b"\r\n", headers={ "Pragma": "DFI" }
)
return PlainTextResponse(resp)
else:
machine = await self.data.arcade.get_machine(req.serial)
if not machine or not machine['ota_enable'] or not machine['is_cab'] or machine['is_blacklisted']:
return PlainTextResponse(urllib.parse.unquote(urllib.parse.urlencode(vars(resp))) + "\n")
resp = urllib.parse.unquote(urllib.parse.urlencode(vars(resp))) + "\n"
if is_dfi:
return PlainTextResponse(
self.to_dfi(resp) + b"\r\n", headers={ "Pragma": "DFI" }
)
return PlainTextResponse(resp)
if path.exists(
f"{self.config.allnet.update_cfg_folder}/{req.game_id}-{req.ver.replace('.', '')}-app.ini"
@ -744,7 +754,7 @@ class AllnetDownloadOrderRequest:
self.encode = req.get("encode", "")
class AllnetDownloadOrderResponse:
def __init__(self, stat: int = 1, serial: str = "", uri: str = "") -> None:
def __init__(self, stat: int = 1, serial: str = "", uri: str = "null") -> None:
self.stat = stat
self.serial = serial
self.uri = uri

View File

@ -60,13 +60,14 @@ Games listed below have been tested and confirmed working.
### SDHD/SDBT
| Version ID | Version Name |
| ---------- | ------------------- |
| 11 | CHUNITHM NEW!! |
| 12 | CHUNITHM NEW PLUS!! |
| 13 | CHUNITHM SUN |
| 14 | CHUNITHM SUN PLUS |
| 15 | CHUNITHM LUMINOUS |
| Version ID | Version Name |
| ---------- | ---------------------- |
| 11 | CHUNITHM NEW!! |
| 12 | CHUNITHM NEW PLUS!! |
| 13 | CHUNITHM SUN |
| 14 | CHUNITHM SUN PLUS |
| 15 | CHUNITHM LUMINOUS |
| 16 | CHUNITHM LUMINOUS PLUS |
### Importer

View File

@ -40,6 +40,9 @@ version:
15:
rom: 2.20.00
data: 2.20.00
16:
rom: 2.25.00
data: 2.25.00
crypto:
encrypted_only: False

View File

@ -30,6 +30,7 @@ Games listed below have been tested and confirmed working. Only game versions ol
+ SUN
+ SUN PLUS
+ LUMINOUS
+ LUMINOUS PLUS
+ crossbeats REV.
+ Crossbeats REV.
@ -84,3 +85,6 @@ Read [Games specific info](docs/game_specific_info.md) for all supported games,
## Production guide
See the [production guide](docs/prod.md) for running a production server.
## Text User Interface
Invoke `tui.py` (with optional `-c <command dir>` parameter) for an interactive TUI to perform management actions (add, edit or delete users, cards, arcades and machines) without needing to spin up the frontend. Requires installing asciimatics via `pip install asciimatics`

View File

@ -8,7 +8,7 @@ import pytz
from core.config import CoreConfig
from titles.chuni.config import ChuniConfig
from titles.chuni.const import ChuniConstants, ItemKind
from titles.chuni.const import ChuniConstants, FavoriteItemKind, ItemKind
from titles.chuni.database import ChuniData
@ -1014,6 +1014,28 @@ class ChuniBase:
)
await self.data.profile.put_net_battle(user_id, net_battle)
# New in LUMINOUS PLUS
if "userFavoriteMusicList" in upsert:
# musicId, orderId
music_ids = set(int(m["musicId"]) for m in upsert["userFavoriteMusicList"])
current_favorites = await self.data.item.get_all_favorites(
user_id, self.version, fav_kind=FavoriteItemKind.MUSIC
)
if current_favorites is None:
current_favorites = []
current_favorite_ids = set(x.favId for x in current_favorites)
keep_ids = current_favorite_ids.intersection(music_ids)
deleted_ids = current_favorite_ids - keep_ids
added_ids = music_ids - keep_ids
for fav_id in deleted_ids:
await self.data.item.delete_favorite_music(user_id, self.version, fav_id)
for fav_id in added_ids:
await self.data.item.put_favorite_music(user_id, self.version, fav_id)
return {"returnCode": "1"}
async def handle_upsert_user_chargelog_api_request(self, data: Dict) -> Dict:

View File

@ -25,6 +25,7 @@ class ChuniConstants:
VER_CHUNITHM_SUN = 13
VER_CHUNITHM_SUN_PLUS = 14
VER_CHUNITHM_LUMINOUS = 15
VER_CHUNITHM_LUMINOUS_PLUS = 16
VERSION_NAMES = [
"CHUNITHM",
@ -43,6 +44,7 @@ class ChuniConstants:
"CHUNITHM SUN",
"CHUNITHM SUN PLUS",
"CHUNITHM LUMINOUS",
"CHUNITHM LUMINOUS PLUS",
]
SCORE_RANK_INTERVALS_OLD = [
@ -98,6 +100,7 @@ class MapAreaConditionType(IntEnum):
TROPHY_OBTAINED = 3
RANK_SSSP = 18
RANK_SSS = 19
RANK_SSP = 20
RANK_SS = 21
@ -127,7 +130,7 @@ class ItemKind(IntEnum):
FRAME = 2
"""
"Frame" is the background for the gauge/score/max combo display
shown during gameplay. This item cannot be equipped (as of LUMINOUS)
shown during gameplay. This item cannot be equipped (as of LUMINOUS PLUS)
and is hardcoded to the current game's version.
"""
@ -146,7 +149,7 @@ class ItemKind(IntEnum):
ULTIMA_UNLOCK = 12
"""This only applies to ULTIMA difficulties that are *not* unlocked by
SS-ing EXPERT+MASTER.
reaching S rank on EXPERT difficulty or above.
"""

View File

@ -36,6 +36,7 @@ from .newplus import ChuniNewPlus
from .sun import ChuniSun
from .sunplus import ChuniSunPlus
from .luminous import ChuniLuminous
from .luminousplus import ChuniLuminousPlus
class ChuniServlet(BaseServlet):
def __init__(self, core_cfg: CoreConfig, cfg_dir: str) -> None:
@ -64,6 +65,7 @@ class ChuniServlet(BaseServlet):
ChuniSun,
ChuniSunPlus,
ChuniLuminous,
ChuniLuminousPlus,
]
self.logger = logging.getLogger("chuni")
@ -107,6 +109,7 @@ class ChuniServlet(BaseServlet):
f"{ChuniConstants.VER_CHUNITHM_SUN_PLUS}_int": 36,
ChuniConstants.VER_CHUNITHM_LUMINOUS: 8,
f"{ChuniConstants.VER_CHUNITHM_LUMINOUS}_int": 8,
ChuniConstants.VER_CHUNITHM_LUMINOUS_PLUS: 56,
}
for version, keys in self.game_cfg.crypto.keys.items():
@ -235,8 +238,10 @@ class ChuniServlet(BaseServlet):
internal_ver = ChuniConstants.VER_CHUNITHM_SUN
elif version >= 215 and version < 220: # SUN PLUS
internal_ver = ChuniConstants.VER_CHUNITHM_SUN_PLUS
elif version >= 220: # LUMINOUS
elif version >= 220 and version < 225: # LUMINOUS
internal_ver = ChuniConstants.VER_CHUNITHM_LUMINOUS
elif version >= 225: # LUMINOUS PLUS
internal_ver = ChuniConstants.VER_CHUNITHM_LUMINOUS_PLUS
elif game_code == "SDGS": # Int
if version < 105: # SUPERSTAR
internal_ver = ChuniConstants.VER_CHUNITHM_CRYSTAL_PLUS
@ -250,8 +255,10 @@ class ChuniServlet(BaseServlet):
internal_ver = ChuniConstants.VER_CHUNITHM_SUN
elif version >= 125 and version < 130: # SUN PLUS
internal_ver = ChuniConstants.VER_CHUNITHM_SUN_PLUS
elif version >= 130: # LUMINOUS
elif version >= 130 and version < 135: # LUMINOUS
internal_ver = ChuniConstants.VER_CHUNITHM_LUMINOUS
elif version >= 135: # LUMINOUS PLUS
internal_ver = ChuniConstants.VER_CHUNITHM_LUMINOUS_PLUS
if all(c in string.hexdigits for c in endpoint) and len(endpoint) == 32:
# If we get a 32 character long hex string, it's a hash and we're
@ -311,8 +318,10 @@ class ChuniServlet(BaseServlet):
return Response(zlib.compress(b'{"stat": "0"}'))
try:
unzip = zlib.decompress(req_raw)
if request.headers.get("x-debug") is not None:
unzip = req_raw
else:
unzip = zlib.decompress(req_raw)
except zlib.error as e:
self.logger.error(
f"Failed to decompress v{version} {endpoint} request -> {e}"
@ -352,6 +361,9 @@ class ChuniServlet(BaseServlet):
self.logger.debug(f"Response {resp}")
if request.headers.get("x-debug") is not None:
return Response(json.dumps(resp, ensure_ascii=False).encode("utf-8"))
zipped = zlib.compress(json.dumps(resp, ensure_ascii=False).encode("utf-8"))
if not encrtped:

View File

@ -2,9 +2,13 @@ from datetime import timedelta
from typing import Dict
from core.config import CoreConfig
from titles.chuni.sunplus import ChuniSunPlus
from titles.chuni.const import ChuniConstants, MapAreaConditionLogicalOperator, MapAreaConditionType
from titles.chuni.config import ChuniConfig
from titles.chuni.const import (
ChuniConstants,
MapAreaConditionLogicalOperator,
MapAreaConditionType,
)
from titles.chuni.sunplus import ChuniSunPlus
class ChuniLuminous(ChuniSunPlus):
@ -18,7 +22,7 @@ class ChuniLuminous(ChuniSunPlus):
# Does CARD MAKER 1.35 work this far up?
user_data["lastDataVersion"] = "2.20.00"
return user_data
async def handle_get_user_c_mission_api_request(self, data: Dict) -> Dict:
user_id = data["userId"]
mission_id = data["missionId"]
@ -28,7 +32,7 @@ class ChuniLuminous(ChuniSunPlus):
mission_data = await self.data.item.get_cmission(user_id, mission_id)
progress_data = await self.data.item.get_cmission_progress(user_id, mission_id)
if mission_data and progress_data:
point = mission_data["point"]
@ -48,12 +52,14 @@ class ChuniLuminous(ChuniSunPlus):
"userCMissionProgressList": progress_list,
}
async def handle_get_user_net_battle_ranking_info_api_request(self, data: Dict) -> Dict:
async def handle_get_user_net_battle_ranking_info_api_request(
self, data: Dict
) -> Dict:
user_id = data["userId"]
net_battle = {}
net_battle_data = await self.data.profile.get_net_battle(user_id)
if net_battle_data:
net_battle = {
"isRankUpChallengeFailed": net_battle_data["isRankUpChallengeFailed"],
@ -94,131 +100,135 @@ class ChuniLuminous(ChuniSunPlus):
# (event ID 14214) was imported into ARTEMiS, we disable the requirement
# for this trophy.
if 14214 in event_by_id:
mission_in_progress_end_date = (event_by_id[14214]["startDate"] - timedelta(hours=2)).strftime(self.date_time_format)
conditions.extend([
{
"mapAreaId": 2206201, # BlythE ULTIMA
"length": 1,
# Obtain the trophy "MISSION in progress".
"mapAreaConditionList": [
{
"type": MapAreaConditionType.TROPHY_OBTAINED.value,
"conditionId": 6832,
"logicalOpe": MapAreaConditionLogicalOperator.AND.value,
"startDate": start_date,
"endDate": mission_in_progress_end_date,
}
],
},
{
"mapAreaId": 2206202, # PRIVATE SERVICE ULTIMA
"length": 1,
# Obtain the trophy "MISSION in progress".
"mapAreaConditionList": [
{
"type": MapAreaConditionType.TROPHY_OBTAINED.value,
"conditionId": 6832,
"logicalOpe": MapAreaConditionLogicalOperator.AND.value,
"startDate": start_date,
"endDate": mission_in_progress_end_date,
}
],
},
{
"mapAreaId": 2206203, # New York Back Raise
"length": 1,
# SS NightTheater's EXPERT chart and get the title
# "今宵、劇場に映し出される景色とは――――。"
"mapAreaConditionList": [
{
"type": MapAreaConditionType.TROPHY_OBTAINED.value,
"conditionId": 6833,
"logicalOpe": MapAreaConditionLogicalOperator.AND.value,
"startDate": start_date,
"endDate": "2099-12-31 00:00:00.0",
},
],
},
{
"mapAreaId": 2206204, # Spasmodic
"length": 2,
# - Get 1 miss on Random (any difficulty) and get the title "当たり待ち"
# - Get 1 miss on 花たちに希望を (any difficulty) and get the title "花たちに希望を"
"mapAreaConditionList": [
{
"type": MapAreaConditionType.TROPHY_OBTAINED.value,
"conditionId": 6834,
"logicalOpe": MapAreaConditionLogicalOperator.AND.value,
"startDate": start_date,
"endDate": "2099-12-31 00:00:00.0",
},
{
"type": MapAreaConditionType.TROPHY_OBTAINED.value,
"conditionId": 6835,
"logicalOpe": MapAreaConditionLogicalOperator.AND.value,
"startDate": start_date,
"endDate": "2099-12-31 00:00:00.0",
},
],
},
{
"mapAreaId": 2206205, # ΩΩPARTS
"length": 2,
# - S Sage EXPERT to get the title "マターリ進行キボンヌ"
# - Equip this title and play cab-to-cab with another person with this title
# to get "マターリしようよ". Disabled because it is difficult to play cab2cab
# on data setups. A network operator may consider re-enabling it by uncommenting
# the second condition.
"mapAreaConditionList": [
{
"type": MapAreaConditionType.TROPHY_OBTAINED.value,
"conditionId": 6836,
"logicalOpe": MapAreaConditionLogicalOperator.AND.value,
"startDate": start_date,
"endDate": "2099-12-31 00:00:00.0",
},
# {
# "type": MapAreaConditionType.TROPHY_OBTAINED.value,
# "conditionId": 6837,
# "logicalOpe": MapAreaConditionLogicalOperator.AND.value,
# "startDate": start_date,
# "endDate": "2099-12-31 00:00:00.0",
# },
],
},
{
"mapAreaId": 2206206, # Blow My Mind
"length": 1,
# SS on CHAOS EXPERT, Hydra EXPERT, Surive EXPERT and Jakarta PROGRESSION EXPERT
# to get the title "Can you hear me?"
"mapAreaConditionList": [
{
"type": MapAreaConditionType.TROPHY_OBTAINED.value,
"conditionId": 6838,
"logicalOpe": MapAreaConditionLogicalOperator.AND.value,
"startDate": start_date,
"endDate": "2099-12-31 00:00:00.0",
},
],
},
{
"mapAreaId": 2206207, # VALLIS-NERIA
"length": 6,
# Finish the 6 other areas
"mapAreaConditionList": [
{
"type": MapAreaConditionType.MAP_AREA_CLEARED.value,
"conditionId": x,
"logicalOpe": MapAreaConditionLogicalOperator.AND.value,
"startDate": start_date,
"endDate": "2099-12-31 00:00:00.0",
}
for x in range(2206201, 2206207)
],
},
])
mission_in_progress_end_date = (
event_by_id[14214]["startDate"] - timedelta(hours=2)
).strftime(self.date_time_format)
conditions.extend(
[
{
"mapAreaId": 2206201, # BlythE ULTIMA
"length": 1,
# Obtain the trophy "MISSION in progress".
"mapAreaConditionList": [
{
"type": MapAreaConditionType.TROPHY_OBTAINED.value,
"conditionId": 6832,
"logicalOpe": MapAreaConditionLogicalOperator.AND.value,
"startDate": start_date,
"endDate": mission_in_progress_end_date,
}
],
},
{
"mapAreaId": 2206202, # PRIVATE SERVICE ULTIMA
"length": 1,
# Obtain the trophy "MISSION in progress".
"mapAreaConditionList": [
{
"type": MapAreaConditionType.TROPHY_OBTAINED.value,
"conditionId": 6832,
"logicalOpe": MapAreaConditionLogicalOperator.AND.value,
"startDate": start_date,
"endDate": mission_in_progress_end_date,
}
],
},
{
"mapAreaId": 2206203, # New York Back Raise
"length": 1,
# SS NightTheater's EXPERT chart and get the title
# "今宵、劇場に映し出される景色とは――――。"
"mapAreaConditionList": [
{
"type": MapAreaConditionType.TROPHY_OBTAINED.value,
"conditionId": 6833,
"logicalOpe": MapAreaConditionLogicalOperator.AND.value,
"startDate": start_date,
"endDate": "2099-12-31 00:00:00.0",
},
],
},
{
"mapAreaId": 2206204, # Spasmodic
"length": 2,
# - Get 1 miss on Random (any difficulty) and get the title "当たり待ち"
# - Get 1 miss on 花たちに希望を (any difficulty) and get the title "花たちに希望を"
"mapAreaConditionList": [
{
"type": MapAreaConditionType.TROPHY_OBTAINED.value,
"conditionId": 6834,
"logicalOpe": MapAreaConditionLogicalOperator.AND.value,
"startDate": start_date,
"endDate": "2099-12-31 00:00:00.0",
},
{
"type": MapAreaConditionType.TROPHY_OBTAINED.value,
"conditionId": 6835,
"logicalOpe": MapAreaConditionLogicalOperator.AND.value,
"startDate": start_date,
"endDate": "2099-12-31 00:00:00.0",
},
],
},
{
"mapAreaId": 2206205, # ΩΩPARTS
"length": 2,
# - S Sage EXPERT to get the title "マターリ進行キボンヌ"
# - Equip this title and play cab-to-cab with another person with this title
# to get "マターリしようよ". Disabled because it is difficult to play cab2cab
# on data setups. A network operator may consider re-enabling it by uncommenting
# the second condition.
"mapAreaConditionList": [
{
"type": MapAreaConditionType.TROPHY_OBTAINED.value,
"conditionId": 6836,
"logicalOpe": MapAreaConditionLogicalOperator.AND.value,
"startDate": start_date,
"endDate": "2099-12-31 00:00:00.0",
},
# {
# "type": MapAreaConditionType.TROPHY_OBTAINED.value,
# "conditionId": 6837,
# "logicalOpe": MapAreaConditionLogicalOperator.AND.value,
# "startDate": start_date,
# "endDate": "2099-12-31 00:00:00.0",
# },
],
},
{
"mapAreaId": 2206206, # Blow My Mind
"length": 1,
# SS on CHAOS EXPERT, Hydra EXPERT, Surive EXPERT and Jakarta PROGRESSION EXPERT
# to get the title "Can you hear me?"
"mapAreaConditionList": [
{
"type": MapAreaConditionType.TROPHY_OBTAINED.value,
"conditionId": 6838,
"logicalOpe": MapAreaConditionLogicalOperator.AND.value,
"startDate": start_date,
"endDate": "2099-12-31 00:00:00.0",
},
],
},
{
"mapAreaId": 2206207, # VALLIS-NERIA
"length": 6,
# Finish the 6 other areas
"mapAreaConditionList": [
{
"type": MapAreaConditionType.MAP_AREA_CLEARED.value,
"conditionId": x,
"logicalOpe": MapAreaConditionLogicalOperator.AND.value,
"startDate": start_date,
"endDate": "2099-12-31 00:00:00.0",
}
for x in range(2206201, 2206207)
],
},
]
)
# LUMINOUS ep. I
if 14005 in event_by_id:
start_date = event_by_id[14005]["startDate"].strftime(self.date_time_format)
@ -226,7 +236,7 @@ class ChuniLuminous(ChuniSunPlus):
if not mystic_area_1_added:
conditions.append(mystic_area_1_conditions)
mystic_area_1_added = True
mystic_area_1_conditions["length"] += 1
mystic_area_1_conditions["mapAreaConditionList"].append(
{
@ -254,15 +264,15 @@ class ChuniLuminous(ChuniSunPlus):
],
}
)
# LUMINOUS ep. II
if 14251 in event_by_id:
start_date = event_by_id[14251]["startDate"].strftime(self.date_time_format)
if not mystic_area_1_added:
conditions.append(mystic_area_1_conditions)
mystic_area_1_added = True
mystic_area_1_conditions["length"] += 1
mystic_area_1_conditions["mapAreaConditionList"].append(
{
@ -291,6 +301,203 @@ class ChuniLuminous(ChuniSunPlus):
}
)
# LUMINOUS ep. III
if 14481 in event_by_id:
start_date = event_by_id[14481]["startDate"].strftime(self.date_time_format)
if not mystic_area_1_added:
conditions.append(mystic_area_1_conditions)
mystic_area_1_added = True
mystic_area_1_conditions["length"] += 1
mystic_area_1_conditions["mapAreaConditionList"].append(
{
"type": MapAreaConditionType.MAP_CLEARED.value,
"conditionId": 3020703,
"logicalOpe": MapAreaConditionLogicalOperator.OR.value,
"startDate": start_date,
"endDate": "2099-12-31 00:00:00.0",
}
)
conditions.append(
{
"mapAreaId": 3229304, # Mystic Rainbow of LUMINOUS Area 4,
"length": 1,
# Unlocks when LUMINOUS ep. III is completed.
"mapAreaConditionList": [
{
"type": MapAreaConditionType.MAP_CLEARED.value,
"conditionId": 3020703,
"logicalOpe": MapAreaConditionLogicalOperator.AND.value,
"startDate": start_date,
"endDate": "2099-12-31 00:00:00.0",
},
],
}
)
# 1UM1N0U5 ep. 111
if 14483 in event_by_id:
start_date = event_by_id[14483]["startDate"].replace(
hour=0, minute=0, second=0
)
# conditions to unlock the 6 "Key of ..." area in the map
# for the first 14 days: Defandour MASTER AJ, crazy (about you) MASTER AJ, Halcyon ULTIMA SSS
title_conditions = [
{
"type": MapAreaConditionType.ALL_JUSTICE.value,
"conditionId": 258103, # Defandour MASTER
"logicalOpe": MapAreaConditionLogicalOperator.AND.value,
"startDate": start_date.strftime(self.date_time_format),
"endDate": (
start_date + timedelta(days=14) - timedelta(seconds=1)
).strftime(self.date_time_format),
},
{
"type": MapAreaConditionType.ALL_JUSTICE.value,
"conditionId": 258003, # crazy (about you) MASTER
"logicalOpe": MapAreaConditionLogicalOperator.AND.value,
"startDate": start_date.strftime(self.date_time_format),
"endDate": (
start_date + timedelta(days=14) - timedelta(seconds=1)
).strftime(self.date_time_format),
},
{
"type": MapAreaConditionType.RANK_SSS.value,
"conditionId": 17304, # Halcyon ULTIMA
"logicalOpe": MapAreaConditionLogicalOperator.AND.value,
"startDate": start_date.strftime(self.date_time_format),
"endDate": (
start_date + timedelta(days=14) - timedelta(seconds=1)
).strftime(self.date_time_format),
},
]
# For each next 14 days, the conditions are lowered to SS+, S+, S, and then always unlocked
for i, typ in enumerate(
[
MapAreaConditionType.RANK_SSP.value,
MapAreaConditionType.RANK_SP.value,
MapAreaConditionType.RANK_S.value,
MapAreaConditionType.ALWAYS_UNLOCKED.value,
]
):
start = (start_date + timedelta(days=14 * (i + 1))).strftime(
self.date_time_format
)
if typ != MapAreaConditionType.ALWAYS_UNLOCKED.value:
end = (
start_date + timedelta(days=14 * (i + 2)) - timedelta(seconds=1)
).strftime(self.date_time_format)
title_conditions.extend(
[
{
"type": typ,
"conditionId": condition_id,
"logicalOpe": MapAreaConditionLogicalOperator.AND.value,
"startDate": start,
"endDate": end,
}
for condition_id in {17304, 258003, 258103}
]
)
else:
end = "2099-12-31 00:00:00"
title_conditions.append(
{
"type": typ,
"conditionId": 0,
"logicalOpe": MapAreaConditionLogicalOperator.AND.value,
"startDate": start,
"endDate": end,
}
)
# actually add all the conditions
for map_area_id in range(3229201, 3229207):
conditions.append(
{
"mapAreaId": map_area_id,
"length": len(title_conditions),
"mapAreaConditionList": title_conditions,
}
)
# Ultimate Force
# For the first 14 days, the condition is to obtain all 9 "Key of ..." titles
# Afterwards, the condition is the 6 "Key of ..." titles that you can obtain
# by playing the 6 areas, as well as obtaining specific ranks on
# [CRYSTAL_ACCESS] / Strange Love / βlαnoir
ultimate_force_conditions = []
# Trophies obtained by playing the 6 areas
for trophy_id in {6851, 6853, 6855, 6857, 6858, 6860}:
ultimate_force_conditions.append(
{
"type": MapAreaConditionType.TROPHY_OBTAINED.value,
"conditionId": trophy_id,
"logicalOpe": MapAreaConditionLogicalOperator.AND.value,
"startDate": start_date.strftime(self.date_time_format),
"endDate": "2099-12-31 00:00:00",
}
)
# βlαnoir MASTER SSS+ / Strange Love MASTER SSS+ / [CRYSTAL_ACCESS] MASTER SSS+
for trophy_id in {6852, 6854, 6856}:
ultimate_force_conditions.append(
{
"type": MapAreaConditionType.TROPHY_OBTAINED.value,
"conditionId": trophy_id,
"logicalOpe": MapAreaConditionLogicalOperator.AND.value,
"startDate": start_date.strftime(self.date_time_format),
"endDate": (
start_date + timedelta(days=14) - timedelta(seconds=1)
).strftime(self.date_time_format),
}
)
# For each next 14 days, the rank conditions for the 3 songs lowers
# Finally, the Ultimate Force area is unlocked as soon as you finish the 6 other areas.
for i, typ in enumerate(
[
MapAreaConditionType.RANK_SSS.value,
MapAreaConditionType.RANK_SS.value,
MapAreaConditionType.RANK_S.value,
]
):
start = (start_date + timedelta(days=14 * (i + 1))).strftime(
self.date_time_format
)
end = (
start_date + timedelta(days=14 * (i + 2)) - timedelta(seconds=1)
).strftime(self.date_time_format)
ultimate_force_conditions.extend(
[
{
"type": typ,
"conditionId": condition_id,
"logicalOpe": MapAreaConditionLogicalOperator.AND.value,
"startDate": start,
"endDate": end,
}
for condition_id in {109403, 212103, 244203}
]
)
conditions.append(
{
"mapAreaId": 3229207,
"length": len(ultimate_force_conditions),
"mapAreaConditionList": ultimate_force_conditions,
}
)
return {
"length": len(conditions),

View File

@ -0,0 +1,170 @@
from datetime import timedelta
from typing import Dict
from core.config import CoreConfig
from titles.chuni.config import ChuniConfig
from titles.chuni.const import ChuniConstants, MapAreaConditionLogicalOperator, MapAreaConditionType
from titles.chuni.luminous import ChuniLuminous
class ChuniLuminousPlus(ChuniLuminous):
def __init__(self, core_cfg: CoreConfig, game_cfg: ChuniConfig) -> None:
super().__init__(core_cfg, game_cfg)
self.version = ChuniConstants.VER_CHUNITHM_LUMINOUS_PLUS
async def handle_cm_get_user_preview_api_request(self, data: Dict) -> Dict:
user_data = await super().handle_cm_get_user_preview_api_request(data)
# Does CARD MAKER 1.35 work this far up?
user_data["lastDataVersion"] = "2.25.00"
return user_data
async def handle_get_user_c_mission_list_api_request(self, data: Dict) -> Dict:
user_id = int(data["userId"])
user_mission_list_request = data["userCMissionList"]
user_mission_list = []
for request in user_mission_list_request:
user_id = int(request["userId"])
mission_id = int(request["missionId"])
point = int(request["point"])
mission_data = await self.data.item.get_cmission(user_id, mission_id)
progress_data = await self.data.item.get_cmission_progress(user_id, mission_id)
if mission_data is None or progress_data is None:
continue
point = mission_data.point
user_mission_progress_list = [
{
"order": progress.order,
"stage": progress.stage,
"progress": progress.progress,
}
for progress in progress_data
]
user_mission_list.append(
{
"userId": user_id,
"missionId": mission_id,
"point": point,
"userCMissionProgressList": user_mission_progress_list,
},
)
return {
"userId": user_id,
"userCMissionList": user_mission_list,
}
async def handle_get_game_map_area_condition_api_request(self, data: Dict) -> Dict:
# There is no game data for this, everything is server side.
# However, we can selectively show/hide events as data is imported into the server.
events = await self.data.static.get_enabled_events(self.version)
event_by_id = {evt["eventId"]: evt for evt in events}
conditions = []
# LUMINOUS ep. Ascension
if ep_ascension := event_by_id.get(15512):
start_date = ep_ascension["startDate"].replace(hour=0, minute=0, second=0)
# Finish LUMINOUS ep. VII to unlock LUMINOUS ep. Ascension.
task_track_map_conditions = [
{
"type": MapAreaConditionType.MAP_CLEARED.value,
"conditionId": 3020707,
"logicalOpe": MapAreaConditionLogicalOperator.AND.value,
"startDate": start_date.strftime(self.date_time_format),
"endDate": "2099-12-31 00:00:00",
}
]
# You also need to reach a specific rank on Acid God MASTER.
# This condition lowers every 7 days.
# After the first 4 weeks, you only need to finish ep. VII.
for i, typ in enumerate([
MapAreaConditionType.RANK_SSSP.value,
MapAreaConditionType.RANK_SSS.value,
MapAreaConditionType.RANK_SS.value,
MapAreaConditionType.RANK_S.value,
]):
start = start_date + timedelta(days=7 * i)
end = start_date + timedelta(days=7 * (i + 1)) - timedelta(seconds=1)
task_track_map_conditions.append(
{
"type": typ,
"conditionId": 265103,
"logicalOpe": MapAreaConditionLogicalOperator.AND.value,
"startDate": start.strftime(self.date_time_format),
"endDate": end.strftime(self.date_time_format),
}
)
conditions.extend(
[
{
"mapAreaId": map_area_id,
"length": len(task_track_map_conditions),
"mapAreaConditionList": task_track_map_conditions,
}
for map_area_id in {3220801, 3220802, 3220803, 3220804}
]
)
# To unlock the final map area (Forsaken Tale), achieve a specific rank
# on the 4 task tracks in the previous map areas. This condition also lowers
# every 7 days, similar to Acid God.
# After 28 days, you only need to finish the other 4 areas in ep. Ascension.
forsaken_tale_conditions = []
for i, typ in enumerate([
MapAreaConditionType.RANK_SSSP.value,
MapAreaConditionType.RANK_SSS.value,
MapAreaConditionType.RANK_SS.value,
MapAreaConditionType.RANK_S.value,
]):
start = start_date + timedelta(days=7 * i)
end = start_date + timedelta(days=7 * (i + 1)) - timedelta(seconds=1)
forsaken_tale_conditions.extend(
[
{
"type": typ,
"conditionId": condition_id,
"logicalOpe": MapAreaConditionLogicalOperator.AND.value,
"startDate": start.strftime(self.date_time_format),
"endDate": end.strftime(self.date_time_format),
}
for condition_id in {98203, 108603, 247503, 233903}
]
)
forsaken_tale_conditions.extend(
[
{
"type": MapAreaConditionType.MAP_AREA_CLEARED.value,
"conditionId": map_area_id,
"logicalOpe": MapAreaConditionLogicalOperator.AND.value,
"startDate": (start_date + timedelta(days=28)).strftime(self.date_time_format),
"endDate": "2099-12-31 00:00:00",
}
for map_area_id in {3220801, 3220802, 3220803, 3220804}
]
)
conditions.append(
{
"mapAreaId": 3220805,
"length": len(forsaken_tale_conditions),
"mapAreaConditionList": forsaken_tale_conditions,
}
)
return {
"length": len(conditions),
"gameMapAreaConditionList": conditions,
}

View File

@ -36,6 +36,8 @@ class ChuniNew(ChuniBase):
return "215"
if self.version == ChuniConstants.VER_CHUNITHM_LUMINOUS:
return "220"
if self.version == ChuniConstants.VER_CHUNITHM_LUMINOUS_PLUS:
return "225"
async def handle_get_game_setting_api_request(self, data: Dict) -> Dict:
# use UTC time and convert it to JST time by adding +9

View File

@ -179,7 +179,14 @@ class ChuniRomVersion():
# sort it by version number for easy iteration
ChuniRomVersion.Versions = dict(sorted(all_versions.items()))
def __init__(self, rom_version: str) -> None:
def __init__(self, rom_version: Optional[str] = None) -> None:
if rom_version is None:
self.major = 0
self.minor = 0
self.maint = 0
self.version = "0.00.00"
return
(major, minor, maint) = rom_version.split('.')
self.major = int(major)
self.minor = int(minor)
@ -343,6 +350,10 @@ class ChuniScoreData(BaseData):
# for each romVersion recorded, check if it maps back the current version we are operating on
matching_rom_versions = []
for v in record_versions:
# Do this to prevent null romVersion from causing an error in ChuniRomVersion.__init__()
if v[0] is None:
continue
if ChuniRomVersion(v[0]).get_int_version() == version:
matching_rom_versions += [v[0]]

View File

@ -155,7 +155,7 @@ class CxbServlet(BaseServlet):
filename = filetype_split[len(filetype_split) - 1]
match = re.match(
"^([A-Za-z]*)(\d\d\d\d)$", filetype_split[len(filetype_split) - 1]
r"^([A-Za-z]*)(\d\d\d\d)$", filetype_split[len(filetype_split) - 1]
)
if match:
func_to_find += f"{inflection.underscore(match.group(1))}xxxx"

View File

@ -212,6 +212,7 @@ class Mai2Base:
lastLoginDate = "2017-12-05 07:00:00.0"
if consec is None or not consec:
await self.data.profile.add_consec_login(data["userId"], self.version)
consec_ct = 1
else:

501
tui.py Normal file
View File

@ -0,0 +1,501 @@
#!/usr/bin/env
from typing import Optional, List
import asyncio
import argparse
from os import path, mkdir, W_OK, access
import yaml
import bcrypt
import secrets
import string
from sqlalchemy.engine import Row
from core.data import Data
from core.config import CoreConfig
try:
from asciimatics.widgets import Frame, Layout, Text, Button, RadioButtons, CheckBox, Divider, Label
from asciimatics.scene import Scene
from asciimatics.screen import Screen
from asciimatics.exceptions import ResizeScreenError, NextScene, StopApplication
except:
print("Artemis TUI requires asciimatics, please install it using pip")
exit(1)
class State:
class SelectedUser:
def __init__(self, id: Optional[int] = None, name: Optional[str] = None):
self.id = id
self.name = name
def __str__(self):
if self.id is not None:
return f"{self.name} ({self.id})" if self.name else f"User{self.id:04d}"
return "None"
def __int__(self):
return self.id if self.id else 0
class SelectedCard:
def __init__(self, id: Optional[int] = None, access_code: Optional[str] = None):
self.id = id
self.access_code = access_code
def __str__(self):
if self.id is not None and self.access_code:
return f"{self.access_code} ({self.id})"
return "None"
def __int__(self):
return self.id if self.id else 0
class SelectedArcade:
def __init__(self, id: Optional[int] = None, country: Optional[str] = None, name: Optional[str] = None):
self.id = id
self.country = country
self.name = name
def __str__(self):
if self.id is not None:
return f"{self.name} ({self.country}{self.id:05d})" if self.name else f"{self.country}{self.id:05d}"
return "None"
def __int__(self):
return self.id if self.id else 0
class SelectedMachine:
def __init__(self, id: Optional[int] = None, serial: Optional[str] = None):
self.id = id
self.serial = serial
def __str__(self):
if self.id is not None:
return f"{self.serial} ({self.id})"
return "None"
def __int__(self):
return self.id if self.id else 0
def __init__(self):
self.selected_user: self.SelectedUser = self.SelectedUser()
self.selected_card: self.SelectedCard = self.SelectedCard()
self.selected_arcade: self.SelectedArcade = self.SelectedArcade()
self.selected_machine: self.SelectedMachine = self.SelectedMachine()
self.last_err: str = ""
self.search_results: List[Row] = []
self.search_type: str = ""
def set_user(self, id: int, username: Optional[str]) -> None:
self.selected_user = self.SelectedUser(id, username)
def clear_user(self) -> None:
self.selected_user = self.SelectedUser()
def set_card(self, id: int, access_code: Optional[str]) -> None:
self.selected_card = self.SelectedCard(id, access_code)
def clear_card(self) -> None:
self.selected_card = self.SelectedCard()
def set_arcade(self, id: int, country: str = "JPN", name: Optional[str] = None) -> None:
self.selected_arcade = self.SelectedArcade(id, country, name)
def clear_arcade(self) -> None:
self.selected_arcade = self.SelectedArcade()
def set_machine(self, id: int, serial: Optional[str]) -> None:
self.selected_machine = self.SelectedMachine(id, serial)
def clear_machine(self) -> None:
self.selected_machine = self.SelectedMachine()
def set_last_err(self, err: str) -> None:
self.last_err = err
def clear_last_err(self) -> None:
self.last_err = ""
def clear_search_results(self) -> None:
self.search_results = []
state = State()
data: Data = None
loop: asyncio.AbstractEventLoop = asyncio.new_event_loop()
class MainView(Frame):
def __init__(self, screen: Screen):
super(MainView, self).__init__(
screen,
screen.height * 2 // 3,
screen.width * 2 // 3,
hover_focus=True,
can_scroll=False,
title="ARTEMiS TUI"
)
layout = Layout([100], True)
self.add_layout(layout)
layout.add_widget(Button("User Management", self._user_mgmt))
layout.add_widget(Button("Card Management", self._card_mgmt))
layout.add_widget(Button("Arcade Management", self._arcade_mgmt))
layout.add_widget(Button("Machine Management", self._mech_mgmt))
layout.add_widget(Button("Quit", self._quit))
self.fix()
def _user_mgmt(self):
self.save()
raise NextScene("User Management")
def _card_mgmt(self):
self.save()
raise NextScene("Card Management")
def _arcade_mgmt(self):
self.save()
raise NextScene("Arcade Management")
def _mech_mgmt(self):
self.save()
raise NextScene("Mech Management")
@staticmethod
def _quit():
raise StopApplication("User pressed quit")
class ManageUser(Frame):
def __init__(self, screen: Screen):
super(ManageUser, self).__init__(
screen,
screen.height * 2 // 3,
screen.width * 2 // 3,
hover_focus=True,
can_scroll=False,
title="User Management",
on_load=self._redraw
)
layout = Layout([3])
self.add_layout(layout)
layout.add_widget(Button("Create User", self._create_user))
layout.add_widget(Button("Lookup User", self._lookup))
def _redraw(self):
self._layouts = [self._layouts[0]]
layout = Layout([3])
self.add_layout(layout)
layout.add_widget(Button("Edit User", self._edit_user, disabled=state.selected_user.id == 0 or state.selected_user.id is None))
layout.add_widget(Button("Delete User", self._del_user, disabled=state.selected_user.id == 0 or state.selected_user.id is None))
layout.add_widget((Divider()))
usr_cards = []
#if state.selected_user.id != 0:
#cards = data.card.get_user_cards(state.selected_user.id)
#for card in cards:
#usr_cards.append(card._asdict())
if len(usr_cards) > 0:
layout3 = Layout([100], True)
self.add_layout(layout3)
layout3.add_widget(RadioButtons(
[(f"{card['id']}\t{card['access_code']}\t{card['status']}", card['id']) for card in usr_cards],
"Cards:",
"usr_cards"
))
layout3.add_widget(Divider())
layout2 = Layout([1, 1, 1])
self.add_layout(layout2)
a = Text("", f"status", readonly=True, disabled=True)
a.value = f"Selected User: {state.selected_user}"
layout2.add_widget(a)
layout2.add_widget(Button("Back", self._back), 2)
self.fix()
def _create_user(self):
self.save()
raise NextScene("Create User")
def _lookup(self):
self.save()
raise NextScene("Lookup User")
def _edit_user(self):
self.save()
raise NextScene("Edit User")
def _del_user(self):
self.save()
raise NextScene("Delete User")
def _back(self):
self.save()
raise NextScene("Main")
class CreateUserView(Frame):
def __init__(self, screen: Screen):
super(CreateUserView, self).__init__(
screen,
screen.height * 2 // 3,
screen.width * 2 // 3,
hover_focus=True,
can_scroll=False,
title="Create User"
)
layout = Layout([100], fill_frame=True)
self.add_layout(layout)
layout.add_widget(Text("Username:", "username"))
layout.add_widget(Text("Email:", "email"))
layout.add_widget(Text("Password:", "passwd"))
layout.add_widget(CheckBox("", "Add Card:", "is_add_card", ))
layout.add_widget(RadioButtons([
("User", "1"),
("User Manager", "2"),
("Arcde Manager", "4"),
("Sysadmin", "8"),
("Owner", "255"),
], "Role:", "role"))
layout3 = Layout([100])
self.add_layout(layout3)
layout3.add_widget(Text("", f"status", readonly=True, disabled=True))
layout2 = Layout([1, 1, 1, 1])
self.add_layout(layout2)
layout2.add_widget(Button("OK", self._ok), 0)
layout2.add_widget(Button("Cancel", self._cancel), 3)
self.fix()
def _ok(self):
self.save()
if not self.data.get("username"):
state.set_last_err("Username cannot be blank")
self.find_widget('status').value = state.last_err
self.screen.reset()
return
state.clear_last_err()
self.find_widget('status').value = state.last_err
if not self.data.get("passwd"):
pw = "".join(
secrets.choice(string.ascii_letters + string.digits) for i in range(20)
)
else:
pw = self.data.get("passwd")
hash = bcrypt.hashpw(pw.encode(), bcrypt.gensalt())
loop.run_until_complete(self._create_user_async(self.data.get("username"), hash.decode(), self.data.get("email"), self.data.get('role')))
raise NextScene("User Management")
async def _create_user_async(self, username: str, password: str, email: Optional[str], role: str):
usr_id = await data.user.create_user(
username=username,
email=email if email else None,
password=password,
permission=int(role)
)
state.set_user(usr_id, username)
def _cancel(self):
state.clear_last_err()
self.find_widget('status').value = state.last_err
raise NextScene("User Management")
class SearchResultsView(Frame):
def __init__(self, screen: Screen):
super(SearchResultsView, self).__init__(
screen,
screen.height * 2 // 3,
screen.width * 2 // 3,
hover_focus=True,
can_scroll=False,
title="Search Results",
on_load=self._redraw
)
layout2 = Layout([1, 1, 1, 1])
self.add_layout(layout2)
layout2.add_widget(Button("Select", self._select_current), 2)
layout2.add_widget(Button("Cancel", self._cancel), 2)
def _redraw(self):
self._layouts = [self._layouts[0]]
layout = Layout([100], fill_frame=True)
self.add_layout(layout)
opts = []
if state.search_type == "user":
layout.add_widget(Label(" ID | Username | Role | Email"))
layout.add_widget(Divider())
for usr in state.search_results:
name = str(usr['username'])
if len(name) < 8:
name = str(usr['username']) + ' ' * (8 - len(name))
elif len(name) > 8:
name = usr['username'][:5] + "..."
opts.append((f"{usr['id']:05d} | {name} | {usr['permissions']:08b} | {usr['email']}", state.SelectedUser(usr["id"], str(usr['username']))))
layout.add_widget(RadioButtons(opts, "", "selopt"))
self.fix()
def _select_current(self):
self.save()
a = self.data.get('selopt')
state.set_user(a.id, a.name)
raise NextScene("User Management")
def _cancel(self):
state.clear_last_err()
raise NextScene("User Management")
class LookupUserView(Frame):
def __init__(self, screen):
super(LookupUserView, self).__init__(
screen,
screen.height * 2 // 3,
screen.width * 2 // 3,
hover_focus=True,
can_scroll=False,
title="Lookup User"
)
layout = Layout([1, 1], fill_frame=True)
self.add_layout(layout)
layout.add_widget(RadioButtons([
("Username", "1"),
("Email", "2"),
("Access Code", "3"),
("User ID", "4"),
], "Search By:", "search_type"))
layout.add_widget(Text("Search:", "search_str"), 1)
layout3 = Layout([100])
self.add_layout(layout3)
layout3.add_widget(Text("", f"status", readonly=True, disabled=True))
layout2 = Layout([1, 1, 1, 1])
self.add_layout(layout2)
layout2.add_widget(Button("Search", self._lookup), 0)
layout2.add_widget(Button("Cancel", self._cancel), 3)
self.fix()
def _lookup(self):
self.save()
if not self.data.get("search_str"):
state.set_last_err("Search cannot be blank")
self.find_widget('status').value = state.last_err
self.screen.reset()
return
state.clear_last_err()
self.find_widget('status').value = state.last_err
search_type = self.data.get("search_type")
if search_type == "1":
loop.run_until_complete(self._lookup_user_by_username(self.data.get("search_str")))
elif search_type == "2":
loop.run_until_complete(self._lookup_user_by_email(self.data.get("search_str")))
elif search_type == "3":
loop.run_until_complete(self._lookup_user_by_access_code(self.data.get("search_str")))
elif search_type == "4":
loop.run_until_complete(self._lookup_user_by_id(self.data.get("search_str")))
else:
state.set_last_err("Unknown search type")
self.find_widget('status').value = state.last_err
self.screen.reset()
return
if len(state.search_results) < 1:
state.set_last_err("Search returned no results")
self.find_widget('status').value = state.last_err
self.screen.reset()
return
state.search_type = "user"
raise NextScene("Search Results")
async def _lookup_user_by_id(self, user_id: str):
usr = await data.user.get_user(user_id)
if usr is not None:
state.search_results = [usr]
async def _lookup_user_by_username(self, username: str):
usr = await data.user.find_user_by_username(username)
if usr is not None:
state.search_results = usr
async def _lookup_user_by_email(self, email: str):
usr = await data.user.find_user_by_email(email)
if usr is not None:
state.search_results = usr
async def _lookup_user_by_access_code(self, access_code: str):
card = await data.card.get_card_by_access_code(access_code)
if card is not None:
usr = await data.user.get_user(card['user'])
if usr is not None:
state.search_results = [usr]
def _cancel(self):
state.clear_last_err()
self.find_widget('status').value = state.last_err
raise NextScene("User Management")
def demo(screen:Screen, scene: Scene):
scenes = [
Scene([MainView(screen)], -1, name="Main"),
Scene([ManageUser(screen)], -1, name="User Management"),
Scene([CreateUserView(screen)], -1, name="Create User"),
Scene([LookupUserView(screen)], -1, name="Lookup User"),
Scene([SearchResultsView(screen)], -1, name="Search Results"),
]
screen.play(scenes, stop_on_resize=False, start_scene=scene, allow_int=True)
last_scene = None
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Database utilities")
parser.add_argument(
"--config", "-c", type=str, help="Config folder to use", default="config"
)
args = parser.parse_args()
cfg = CoreConfig()
if path.exists(f"{args.config}/core.yaml"):
cfg_dict = yaml.safe_load(open(f"{args.config}/core.yaml"))
cfg_dict.get("database", {})["loglevel"] = "info"
cfg.update(cfg_dict)
if not path.exists(cfg.server.log_dir):
mkdir(cfg.server.log_dir)
if not access(cfg.server.log_dir, W_OK):
print(
f"Log directory {cfg.server.log_dir} NOT writable, please check permissions"
)
exit(1)
data = Data(cfg)
while True:
try:
Screen.wrapper(demo, catch_interrupt=True, arguments=[last_scene])
exit(0)
except ResizeScreenError as e:
last_scene = e.scene