Merge pull request 'CHUNITHM VERSE support' (#224) from feature/chuni_verse_support into develop

Reviewed-on: https://gitea.tendokyu.moe/Hay1tsme/artemis/pulls/224
This commit is contained in:
Hay1tsme
2025-09-27 20:20:21 +00:00
22 changed files with 1015 additions and 241 deletions

View File

@@ -1,6 +1,9 @@
# Changelog
Documenting updates to ARTEMiS, to be updated every time the master branch is pushed to.
## 20250803
+ CHUNITHM VERSE support added
## 20250327
+ O.N.G.E.K.I. bright MEMORY Act.3 support added
+ CardMaker support updated

View File

@@ -0,0 +1,85 @@
"""CHUNITHM VERSE support
Revision ID: 49c295e89cd4
Revises: f6007bbf057d
Create Date: 2025-03-09 14:10:03.067328
"""
from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import mysql
from sqlalchemy.sql import func
# revision identifiers, used by Alembic.
revision = "49c295e89cd4"
down_revision = "f6007bbf057d"
branch_labels = None
depends_on = None
def upgrade():
### commands auto generated by Alembic - please adjust! ###
op.add_column("chuni_profile_data", sa.Column("trophyIdSub1", sa.Integer()))
op.add_column("chuni_profile_data", sa.Column("trophyIdSub2", sa.Integer()))
op.add_column("chuni_score_playlog", sa.Column("monthPoint", sa.Integer()))
op.add_column("chuni_score_playlog", sa.Column("eventPoint", sa.Integer()))
op.create_table(
"chuni_static_unlock_challenge",
sa.Column("id", sa.Integer(), primary_key=True, nullable=False),
sa.Column("version", sa.Integer(), nullable=False),
sa.Column("unlockChallengeId", sa.Integer(), nullable=False),
sa.Column("name", sa.String(length=255)),
sa.Column("isEnabled", sa.Boolean(), server_default="1"),
sa.Column("startDate", sa.TIMESTAMP(), server_default=func.now()),
sa.Column("courseId1", sa.Integer()),
sa.Column("courseId2", sa.Integer()),
sa.Column("courseId3", sa.Integer()),
sa.Column("courseId4", sa.Integer()),
sa.Column("courseId5", sa.Integer()),
sa.UniqueConstraint(
"version", "unlockChallengeId", name="chuni_static_unlock_challenge_uk"
),
mysql_charset="utf8mb4",
)
op.create_table(
"chuni_item_unlock_challenge",
sa.Column("id", sa.Integer(), primary_key=True, nullable=False),
sa.Column("version", sa.Integer(), nullable=False),
sa.Column(
"user",
sa.Integer(),
sa.ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"),
nullable=False,
),
sa.Column("unlockChallengeId", sa.Integer(), nullable=False),
sa.Column("status", sa.Integer()),
sa.Column("clearCourseId", sa.Integer()),
sa.Column("conditionType", sa.Integer()),
sa.Column("score", sa.Integer()),
sa.Column("life", sa.Integer()),
sa.Column("clearDate", sa.TIMESTAMP(), server_defaul=func.now()),
sa.UniqueConstraint(
"version",
"user",
"unlockChallengeId",
name="chuni_item_unlock_challenge_uk",
),
mysql_charset="utf8mb4",
)
# ### end Alembic commands ###
def downgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.drop_column("chuni_score_playlog", "eventPoint")
op.drop_column("chuni_score_playlog", "monthPoint")
op.drop_column("chuni_profile_data", "trophyIdSub2")
op.drop_column("chuni_profile_data", "trophyIdSub1")
op.drop_table("chuni_static_unlock_challenge")
op.drop_table("chuni_item_unlock_challenge")
# ### end Alembic commands ###

View File

@@ -68,6 +68,7 @@ Games listed below have been tested and confirmed working.
| 14 | CHUNITHM SUN PLUS |
| 15 | CHUNITHM LUMINOUS |
| 16 | CHUNITHM LUMINOUS PLUS |
| 17 | CHUNITHM VERSE |
### Importer

View File

@@ -44,6 +44,9 @@ version:
16:
rom: 2.25.00
data: 2.25.00
17:
rom: 2.30.00
data: 2.30.00
crypto:
encrypted_only: False

View File

@@ -38,6 +38,7 @@ Games listed below have been tested and confirmed working. Only game versions ol
+ SUN PLUS
+ LUMINOUS
+ LUMINOUS PLUS
+ VERSE
+ crossbeats REV.
+ Crossbeats REV.

View File

@@ -53,7 +53,9 @@ class ChuniBase:
if not self.game_cfg.mods.use_login_bonus:
return {"returnCode": 1}
login_bonus_presets = await self.data.static.get_login_bonus_presets(self.version)
login_bonus_presets = await self.data.static.get_login_bonus_presets(
self.version
)
for preset in login_bonus_presets:
# check if a user already has some pogress and if not add the
@@ -197,15 +199,21 @@ class ChuniBase:
async def handle_get_game_message_api_request(self, data: Dict) -> Dict:
return {
"type": data["type"],
"length": 1,
"gameMessageList": [{
"id": 1,
"type": 1,
"message": f"Welcome to {self.core_cfg.server.name} network!" if not self.game_cfg.server.news_msg else self.game_cfg.server.news_msg,
"startDate": "2017-12-05 07:00:00.0",
"endDate": "2099-12-31 00:00:00.0"
}]
"type": data["type"],
"length": 1,
"gameMessageList": [
{
"id": 1,
"type": 1,
"message": (
f"Welcome to {self.core_cfg.server.name} network!"
if not self.game_cfg.server.news_msg
else self.game_cfg.server.news_msg
),
"startDate": "2017-12-05 07:00:00.0",
"endDate": "2099-12-31 00:00:00.0",
}
],
}
async def handle_get_game_ranking_api_request(self, data: Dict) -> Dict:
@@ -217,7 +225,10 @@ class ChuniBase:
async def handle_get_game_setting_api_request(self, data: Dict) -> Dict:
# if reboot start/end time is not defined use the default behavior of being a few hours ago
if self.core_cfg.title.reboot_start_time == "" or self.core_cfg.title.reboot_end_time == "":
if (
self.core_cfg.title.reboot_start_time == ""
or self.core_cfg.title.reboot_end_time == ""
):
reboot_start = datetime.strftime(
datetime.utcnow() + timedelta(hours=6), self.date_time_format
)
@@ -226,15 +237,29 @@ class ChuniBase:
)
else:
# get current datetime in JST
current_jst = datetime.now(pytz.timezone('Asia/Tokyo')).date()
current_jst = datetime.now(pytz.timezone("Asia/Tokyo")).date()
# parse config start/end times into datetime
reboot_start_time = datetime.strptime(self.core_cfg.title.reboot_start_time, "%H:%M")
reboot_end_time = datetime.strptime(self.core_cfg.title.reboot_end_time, "%H:%M")
reboot_start_time = datetime.strptime(
self.core_cfg.title.reboot_start_time, "%H:%M"
)
reboot_end_time = datetime.strptime(
self.core_cfg.title.reboot_end_time, "%H:%M"
)
# offset datetimes with current date/time
reboot_start_time = reboot_start_time.replace(year=current_jst.year, month=current_jst.month, day=current_jst.day, tzinfo=pytz.timezone('Asia/Tokyo'))
reboot_end_time = reboot_end_time.replace(year=current_jst.year, month=current_jst.month, day=current_jst.day, tzinfo=pytz.timezone('Asia/Tokyo'))
reboot_start_time = reboot_start_time.replace(
year=current_jst.year,
month=current_jst.month,
day=current_jst.day,
tzinfo=pytz.timezone("Asia/Tokyo"),
)
reboot_end_time = reboot_end_time.replace(
year=current_jst.year,
month=current_jst.month,
day=current_jst.day,
tzinfo=pytz.timezone("Asia/Tokyo"),
)
# create strings for use in gameSetting
reboot_start = reboot_start_time.strftime(self.date_time_format)
@@ -255,6 +280,7 @@ class ChuniBase:
"isDumpUpload": "false",
"isAou": "false",
}
async def handle_get_user_activity_api_request(self, data: Dict) -> Dict:
user_activity_list = await self.data.profile.get_profile_activity(
data["userId"], data["kind"]
@@ -285,7 +311,7 @@ class ChuniBase:
rows = await self.data.item.get_characters(
user_id, limit=max_ct + 1, offset=next_idx
)
if rows is None or len(rows) == 0:
return {
"userId": user_id,
@@ -335,7 +361,7 @@ class ChuniBase:
return {
"userId": data["userId"],
"length": 0,
"userRecentPlayerList": [], # playUserId, playUserName, playDate, friendPoint
"userRecentPlayerList": [], # playUserId, playUserName, playDate, friendPoint
}
async def handle_get_user_course_api_request(self, data: Dict) -> Dict:
@@ -421,15 +447,9 @@ class ChuniBase:
p = await self.data.profile.get_rival(data["rivalId"])
if p is None:
return {}
userRivalData = {
"rivalId": p.user,
"rivalName": p.userName
}
return {
"userId": data["userId"],
"userRivalData": userRivalData
}
userRivalData = {"rivalId": p.user, "rivalName": p.userName}
return {"userId": data["userId"], "userRivalData": userRivalData}
async def handle_get_user_rival_music_api_request(self, data: Dict) -> Dict:
user_id = int(data["userId"])
rival_id = int(data["rivalId"])
@@ -459,18 +479,25 @@ class ChuniBase:
# note that itertools.groupby will only work on sorted keys, which is already sorted by
# the query in get_scores
for music_id, details_iter in itertools.groupby(music_details, key=lambda x: x["musicId"]):
for music_id, details_iter in itertools.groupby(
music_details, key=lambda x: x["musicId"]
):
details: list[dict[Any, Any]] = [
{"level": d["level"], "scoreMax": d["scoreMax"]}
for d in details_iter
{"level": d["level"], "scoreMax": d["scoreMax"]} for d in details_iter
]
music_list.append({"musicId": music_id, "length": len(details), "userRivalMusicDetailList": details})
music_list.append(
{
"musicId": music_id,
"length": len(details),
"userRivalMusicDetailList": details,
}
)
returned_music_details_count += len(details)
if len(music_list) >= max_ct:
break
# if we returned fewer PBs than we originally asked for from the database, that means
# we queried for the PBs of max_ct + 1 songs.
if returned_music_details_count < len(rows):
@@ -485,7 +512,7 @@ class ChuniBase:
"nextIndex": next_idx,
"userRivalMusicList": music_list,
}
async def handle_get_user_favorite_item_api_request(self, data: Dict) -> Dict:
user_id = int(data["userId"])
next_idx = int(data["nextIndex"])
@@ -571,7 +598,9 @@ class ChuniBase:
async def handle_get_user_login_bonus_api_request(self, data: Dict) -> Dict:
user_id = data["userId"]
user_login_bonus = await self.data.item.get_all_login_bonus(user_id, self.version)
user_login_bonus = await self.data.item.get_all_login_bonus(
user_id, self.version
)
# ignore the loginBonus request if its disabled in config
if user_login_bonus is None or not self.game_cfg.mods.use_login_bonus:
return {"userId": user_id, "length": 0, "userLoginBonusList": []}
@@ -621,7 +650,7 @@ class ChuniBase:
rows = await self.data.score.get_scores(
user_id, limit=max_ct + 1, offset=next_idx
)
if rows is None or len(rows) == 0:
return {
"userId": user_id,
@@ -636,7 +665,9 @@ class ChuniBase:
# note that itertools.groupby will only work on sorted keys, which is already sorted by
# the query in get_scores
for _music_id, details_iter in itertools.groupby(music_details, key=lambda x: x["musicId"]):
for _music_id, details_iter in itertools.groupby(
music_details, key=lambda x: x["musicId"]
):
details: list[dict[Any, Any]] = []
for d in details_iter:
@@ -650,14 +681,14 @@ class ChuniBase:
if len(music_list) >= max_ct:
break
# if we returned fewer PBs than we originally asked for from the database, that means
# we queried for the PBs of max_ct + 1 songs.
if returned_music_details_count < len(rows):
next_idx += max_ct
else:
next_idx = -1
return {
"userId": user_id,
"length": len(music_list),
@@ -687,7 +718,9 @@ class ChuniBase:
return bytes([ord(c) for c in src]).decode("utf-8")
async def handle_get_user_preview_api_request(self, data: Dict) -> Dict:
profile = await self.data.profile.get_profile_preview(data["userId"], self.version)
profile = await self.data.profile.get_profile_preview(
data["userId"], self.version
)
if profile is None:
return None
profile_character = await self.data.item.get_character(
@@ -729,7 +762,9 @@ class ChuniBase:
}
async def handle_get_user_recent_rating_api_request(self, data: Dict) -> Dict:
recent_rating_list = await self.data.profile.get_profile_recent_rating(data["userId"])
recent_rating_list = await self.data.profile.get_profile_recent_rating(
data["userId"]
)
if recent_rating_list is None:
return {
"userId": data["userId"],
@@ -762,7 +797,7 @@ class ChuniBase:
profile = await self.data.profile.get_profile_data(data["userId"], self.version)
if profile is None:
return {"userId": data["userId"], "teamId": 0}
return {"userId": data["userId"], "teamId": 0}
if profile and profile["teamId"]:
# Get team by id
@@ -787,7 +822,7 @@ class ChuniBase:
"teamId": team_id,
"teamRank": team_rank,
"teamName": team_name,
"assaultTimeRate": 1, # TODO: Figure out assaultTime, which might be team point boost?
"assaultTimeRate": 1, # TODO: Figure out assaultTime, which might be team point boost?
"userTeamPoint": {
"userId": data["userId"],
"teamId": team_id,
@@ -796,7 +831,7 @@ class ChuniBase:
"aggrDate": data["playDate"],
},
}
async def handle_get_team_course_setting_api_request(self, data: Dict) -> Dict:
return {
"userId": data["userId"],
@@ -805,7 +840,9 @@ class ChuniBase:
"teamCourseSettingList": [],
}
async def handle_get_team_course_setting_api_request_proto(self, data: Dict) -> Dict:
async def handle_get_team_course_setting_api_request_proto(
self, data: Dict
) -> Dict:
return {
"userId": data["userId"],
"length": 1,
@@ -820,11 +857,11 @@ class ChuniBase:
"teamCourseMusicList": [
{"track": 184, "type": 1, "level": 3, "selectLevel": -1},
{"track": 184, "type": 1, "level": 3, "selectLevel": -1},
{"track": 184, "type": 1, "level": 3, "selectLevel": -1}
{"track": 184, "type": 1, "level": 3, "selectLevel": -1},
],
"teamCourseRankingInfoList": [],
"recodeDate": "2099-12-31 11:59:99.0",
"isPlayed": False
"isPlayed": False,
}
],
}
@@ -834,7 +871,7 @@ class ChuniBase:
"userId": data["userId"],
"length": 0,
"nextIndex": -1,
"teamCourseRuleList": []
"teamCourseRuleList": [],
}
async def handle_get_team_course_rule_api_request_proto(self, data: Dict) -> Dict:
@@ -849,7 +886,7 @@ class ChuniBase:
"damageMiss": 1,
"damageAttack": 1,
"damageJustice": 1,
"damageJusticeC": 1
"damageJusticeC": 1,
}
],
}
@@ -860,7 +897,7 @@ class ChuniBase:
if int(user_id) & 0x1000000000001 == 0x1000000000001:
place_id = int(user_id) & 0xFFFC00000000
self.logger.info("Guest play from place ID %d, ignoring.", place_id)
return {"returnCode": "1"}
@@ -882,7 +919,9 @@ class ChuniBase:
)
if "userGameOption" in upsert:
await self.data.profile.put_profile_option(user_id, upsert["userGameOption"][0])
await self.data.profile.put_profile_option(
user_id, upsert["userGameOption"][0]
)
if "userGameOptionEx" in upsert:
await self.data.profile.put_profile_option_ex(
@@ -929,33 +968,41 @@ class ChuniBase:
for playlog in upsert["userPlaylogList"]:
# convert the player names to utf-8
if playlog["playedUserName1"] is not None:
playlog["playedUserName1"] = self.read_wtf8(playlog["playedUserName1"])
playlog["playedUserName1"] = self.read_wtf8(
playlog["playedUserName1"]
)
if playlog["playedUserName2"] is not None:
playlog["playedUserName2"] = self.read_wtf8(playlog["playedUserName2"])
playlog["playedUserName2"] = self.read_wtf8(
playlog["playedUserName2"]
)
if playlog["playedUserName3"] is not None:
playlog["playedUserName3"] = self.read_wtf8(playlog["playedUserName3"])
playlog["playedUserName3"] = self.read_wtf8(
playlog["playedUserName3"]
)
await self.data.score.put_playlog(user_id, playlog, self.version)
if "userTeamPoint" in upsert:
team_points = upsert["userTeamPoint"]
try:
for tp in team_points:
if tp["teamId"] != '65535':
if tp["teamId"] != "65535":
# Fetch the current team data
current_team = await self.data.profile.get_team_by_id(tp["teamId"])
current_team = await self.data.profile.get_team_by_id(
tp["teamId"]
)
# Calculate the new teamPoint
new_team_point = int(tp["teamPoint"]) + current_team["teamPoint"]
new_team_point = (
int(tp["teamPoint"]) + current_team["teamPoint"]
)
# Prepare the data to update
team_data = {
"teamPoint": new_team_point
}
team_data = {"teamPoint": new_team_point}
# Update the team data
await self.data.profile.update_team(tp["teamId"], team_data)
except:
pass # Probably a better way to catch if the team is not set yet (new profiles), but let's just pass
pass # Probably a better way to catch if the team is not set yet (new profiles), but let's just pass
if "userMapAreaList" in upsert:
for map_area in upsert["userMapAreaList"]:
await self.data.item.put_map_area(user_id, map_area)
@@ -973,22 +1020,28 @@ class ChuniBase:
await self.data.item.put_login_bonus(
user_id, self.version, login["presetId"], isWatched=True
)
if "userRecentPlayerList" in upsert: # TODO: Seen in Air, maybe implement sometime
if (
"userRecentPlayerList" in upsert
): # TODO: Seen in Air, maybe implement sometime
for rp in upsert["userRecentPlayerList"]:
pass
for rating_type in {"userRatingBaseList", "userRatingBaseHotList", "userRatingBaseNextList"}:
for rating_type in {
"userRatingBaseList",
"userRatingBaseHotList",
"userRatingBaseNextList",
}:
if rating_type not in upsert:
continue
await self.data.profile.put_profile_rating(
user_id,
self.version,
rating_type,
upsert[rating_type],
)
# added in LUMINOUS
if "userCMissionList" in upsert:
for cmission in upsert["userCMissionList"]:
@@ -1003,7 +1056,9 @@ class ChuniBase:
)
for progress in cmission["userCMissionProgressList"]:
await self.data.item.put_cmission_progress(user_id, mission_id, progress)
await self.data.item.put_cmission_progress(
user_id, mission_id, progress
)
if "userNetBattleData" in upsert:
net_battle = upsert["userNetBattleData"][0]
@@ -1035,10 +1090,20 @@ class ChuniBase:
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)
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)
# added in CHUNITHM VERSE
if "userUnlockChallengeList" in upsert:
for unlock_challenge in upsert["userUnlockChallengeList"]:
await self.data.item.put_unlock_challenge(
user_id, self.version, unlock_challenge
)
return {"returnCode": "1"}

View File

@@ -28,6 +28,7 @@ class ChuniConstants:
VER_CHUNITHM_SUN_PLUS = 14
VER_CHUNITHM_LUMINOUS = 15
VER_CHUNITHM_LUMINOUS_PLUS = 16
VER_CHUNITHM_VERSE = 17
VERSION_NAMES = [
"CHUNITHM",
@@ -47,6 +48,7 @@ class ChuniConstants:
"CHUNITHM SUN PLUS",
"CHUNITHM LUMINOUS",
"CHUNITHM LUMINOUS PLUS",
"CHUNITHM VERSE"
]
SCORE_RANK_INTERVALS_OLD = [

View File

@@ -655,14 +655,18 @@ class ChuniFrontend(FE_Base):
form_data = await request.form()
new_nameplate: str = form_data.get("nameplate")
new_trophy: str = form_data.get("trophy")
new_trophy_sub_1: str = form_data.get("trophySub1")
new_trophy_sub_2: str = form_data.get("trophySub2")
new_character: str = form_data.get("character")
if not new_nameplate or \
not new_trophy or \
not new_trophy_sub_1 or \
not new_trophy_sub_2 or \
not new_character:
return RedirectResponse("/game/chuni/userbox?e=4", 303)
if not await self.data.profile.update_userbox(usr_sesh.user_id, usr_sesh.chunithm_version, new_nameplate, new_trophy, new_character):
if not await self.data.profile.update_userbox(usr_sesh.user_id, usr_sesh.chunithm_version, new_nameplate, new_trophy, new_trophy_sub_1, new_trophy_sub_2, new_character):
return RedirectResponse("/gate/?e=999", 303)
return RedirectResponse("/game/chuni/userbox", 303)

View File

@@ -1,20 +1,22 @@
from starlette.requests import Request
from starlette.routing import Route
from starlette.responses import Response
import asyncio
import re
import logging
import coloredlogs
from logging.handlers import TimedRotatingFileHandler
import zlib
import yaml
import json
import inflection
import string
from os import path
from typing import Tuple, Dict, List
from logging.handlers import TimedRotatingFileHandler
from starlette.requests import Request
from starlette.routing import Route
from starlette.responses import Response
from Crypto.Cipher import AES
from Crypto.Util.Padding import pad
from Crypto.Protocol.KDF import PBKDF2
from Crypto.Hash import SHA1
from os import path
from typing import Tuple, Dict, List
from core import CoreConfig, Utils
from core.title import BaseServlet
@@ -37,6 +39,8 @@ from .sun import ChuniSun
from .sunplus import ChuniSunPlus
from .luminous import ChuniLuminous
from .luminousplus import ChuniLuminousPlus
from .verse import ChuniVerse
class ChuniServlet(BaseServlet):
def __init__(self, core_cfg: CoreConfig, cfg_dir: str) -> None:
@@ -66,6 +70,7 @@ class ChuniServlet(BaseServlet):
ChuniSunPlus,
ChuniLuminous,
ChuniLuminousPlus,
ChuniVerse
]
self.logger = logging.getLogger("chuni")
@@ -96,15 +101,15 @@ class ChuniServlet(BaseServlet):
known_iter_counts = {
ChuniConstants.VER_CHUNITHM_CRYSTAL_PLUS: 67,
f"{ChuniConstants.VER_CHUNITHM_CRYSTAL_PLUS}_int": 25, # SUPERSTAR
f"{ChuniConstants.VER_CHUNITHM_CRYSTAL_PLUS}_int": 25, # SUPERSTAR
ChuniConstants.VER_CHUNITHM_PARADISE: 44,
f"{ChuniConstants.VER_CHUNITHM_PARADISE}_int": 51, # SUPERSTAR PLUS
f"{ChuniConstants.VER_CHUNITHM_PARADISE}_int": 51, # SUPERSTAR PLUS
ChuniConstants.VER_CHUNITHM_NEW: 54,
f"{ChuniConstants.VER_CHUNITHM_NEW}_int": 49,
f"{ChuniConstants.VER_CHUNITHM_NEW}_chn": 37,
ChuniConstants.VER_CHUNITHM_NEW_PLUS: 25,
f"{ChuniConstants.VER_CHUNITHM_NEW_PLUS}_int": 31,
f"{ChuniConstants.VER_CHUNITHM_NEW_PLUS}_chn": 35, # NEW
f"{ChuniConstants.VER_CHUNITHM_NEW_PLUS}_chn": 35, # NEW
ChuniConstants.VER_CHUNITHM_SUN: 70,
f"{ChuniConstants.VER_CHUNITHM_SUN}_int": 35,
ChuniConstants.VER_CHUNITHM_SUN_PLUS: 36,
@@ -113,6 +118,7 @@ class ChuniServlet(BaseServlet):
f"{ChuniConstants.VER_CHUNITHM_LUMINOUS}_int": 8,
f"{ChuniConstants.VER_CHUNITHM_LUMINOUS}_chn": 8,
ChuniConstants.VER_CHUNITHM_LUMINOUS_PLUS: 56,
ChuniConstants.VER_CHUNITHM_VERSE: 42,
}
for version, keys in self.game_cfg.crypto.keys.items():
@@ -123,7 +129,7 @@ class ChuniServlet(BaseServlet):
version_idx = version
else:
version_idx = int(version.split("_")[0])
salt = bytes.fromhex(keys[2])
if len(keys) >= 4:
@@ -153,12 +159,9 @@ class ChuniServlet(BaseServlet):
and version_idx >= ChuniConstants.VER_CHUNITHM_NEW
):
method_fixed += "C3Exp"
elif (
isinstance(version, str)
and version.endswith("_chn")
):
elif isinstance(version, str) and version.endswith("_chn"):
method_fixed += "Chn"
hash = PBKDF2(
method_fixed,
salt,
@@ -167,7 +170,8 @@ class ChuniServlet(BaseServlet):
hmac_hash_module=SHA1,
)
hashed_name = hash.hex()[:32] # truncate unused bytes like the game does
# truncate unused bytes like the game does
hashed_name = hash.hex()[:32]
self.hash_table[version][hashed_name] = method_fixed
self.logger.debug(
@@ -189,7 +193,9 @@ class ChuniServlet(BaseServlet):
return True
def get_allnet_info(self, game_code: str, game_ver: int, keychip: str) -> Tuple[str, str]:
def get_allnet_info(
self, game_code: str, game_ver: int, keychip: str
) -> Tuple[str, str]:
title_port_int = Utils.get_title_port(self.core_cfg)
title_port_ssl_int = Utils.get_title_port_ssl(self.core_cfg)
@@ -203,7 +209,7 @@ class ChuniServlet(BaseServlet):
if proto == "https":
t_port = f":{title_port_ssl_int}" if title_port_ssl_int != 443 else ""
else:
else:
t_port = f":{title_port_int}" if title_port_int != 80 else ""
return (
@@ -213,14 +219,22 @@ class ChuniServlet(BaseServlet):
def get_routes(self) -> List[Route]:
return [
Route("/{game:str}/{version:int}/ChuniServlet/{endpoint:str}", self.render_POST, methods=['POST']),
Route("/{game:str}/{version:int}/ChuniServlet/MatchingServer/{endpoint:str}", self.render_POST, methods=['POST']),
Route(
"/{game:str}/{version:int}/ChuniServlet/{endpoint:str}",
self.render_POST,
methods=["POST"],
),
Route(
"/{game:str}/{version:int}/ChuniServlet/MatchingServer/{endpoint:str}",
self.render_POST,
methods=["POST"],
),
]
async def render_POST(self, request: Request) -> bytes:
endpoint: str = request.path_params.get('endpoint')
version: int = request.path_params.get('version')
game_code: str = request.path_params.get('game')
endpoint: str = request.path_params.get("endpoint")
version: int = request.path_params.get("version")
game_code: str = request.path_params.get("game")
if endpoint.lower() == "ping":
return Response(zlib.compress(b'{"returnCode": "1"}'))
@@ -231,65 +245,71 @@ class ChuniServlet(BaseServlet):
internal_ver = 0
client_ip = Utils.get_ip_addr(request)
if game_code == "SDHD" or game_code == "SDBT": # JP
if version < 105: # 1.0
internal_ver = ChuniConstants.VER_CHUNITHM
elif version >= 105 and version < 110: # PLUS
internal_ver = ChuniConstants.VER_CHUNITHM_PLUS
elif version >= 110 and version < 115: # AIR
internal_ver = ChuniConstants.VER_CHUNITHM_AIR
elif version >= 115 and version < 120: # AIR PLUS
internal_ver = ChuniConstants.VER_CHUNITHM_AIR_PLUS
elif version >= 120 and version < 125: # STAR
internal_ver = ChuniConstants.VER_CHUNITHM_STAR
elif version >= 125 and version < 130: # STAR PLUS
internal_ver = ChuniConstants.VER_CHUNITHM_STAR_PLUS
elif version >= 130 and version < 135: # AMAZON
internal_ver = ChuniConstants.VER_CHUNITHM_AMAZON
elif version >= 135 and version < 140: # AMAZON PLUS
internal_ver = ChuniConstants.VER_CHUNITHM_AMAZON_PLUS
elif version >= 140 and version < 145: # CRYSTAL
internal_ver = ChuniConstants.VER_CHUNITHM_CRYSTAL
elif version >= 145 and version < 150: # CRYSTAL PLUS
internal_ver = ChuniConstants.VER_CHUNITHM_CRYSTAL_PLUS
elif version >= 150 and version < 200: # PARADISE
internal_ver = ChuniConstants.VER_CHUNITHM_PARADISE
elif version >= 200 and version < 205: # NEW!!
internal_ver = ChuniConstants.VER_CHUNITHM_NEW
elif version >= 205 and version < 210: # NEW PLUS!!
internal_ver = ChuniConstants.VER_CHUNITHM_NEW_PLUS
elif version >= 210 and version < 215: # SUN
internal_ver = ChuniConstants.VER_CHUNITHM_SUN
elif version >= 215 and version < 220: # SUN PLUS
internal_ver = ChuniConstants.VER_CHUNITHM_SUN_PLUS
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
elif version >= 105 and version < 110: # SUPERSTAR PLUS *Cursed but needed due to different encryption key
internal_ver = ChuniConstants.VER_CHUNITHM_PARADISE
elif version >= 110 and version < 115: # NEW
internal_ver = ChuniConstants.VER_CHUNITHM_NEW
elif version >= 115 and version < 120: # NEW PLUS!!
internal_ver = ChuniConstants.VER_CHUNITHM_NEW_PLUS
elif version >= 120 and version < 125: # SUN
internal_ver = ChuniConstants.VER_CHUNITHM_SUN
elif version >= 125 and version < 130: # SUN PLUS
internal_ver = ChuniConstants.VER_CHUNITHM_SUN_PLUS
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
elif game_code == "SDHJ": # Chn
if version < 110: # NEW
internal_ver = ChuniConstants.VER_CHUNITHM_NEW
elif version >= 110 and version < 120: # NEW *Cursed but needed due to different encryption key
internal_ver = ChuniConstants.VER_CHUNITHM_NEW_PLUS
elif version >= 120: # LUMINOUS
internal_ver = ChuniConstants.VER_CHUNITHM_LUMINOUS
if game_code == "SDHD" or game_code == "SDBT": # JP
if version < 105: # 1.0
internal_ver = ChuniConstants.VER_CHUNITHM
elif version >= 105 and version < 110: # PLUS
internal_ver = ChuniConstants.VER_CHUNITHM_PLUS
elif version >= 110 and version < 115: # AIR
internal_ver = ChuniConstants.VER_CHUNITHM_AIR
elif version >= 115 and version < 120: # AIR PLUS
internal_ver = ChuniConstants.VER_CHUNITHM_AIR_PLUS
elif version >= 120 and version < 125: # STAR
internal_ver = ChuniConstants.VER_CHUNITHM_STAR
elif version >= 125 and version < 130: # STAR PLUS
internal_ver = ChuniConstants.VER_CHUNITHM_STAR_PLUS
elif version >= 130 and version < 135: # AMAZON
internal_ver = ChuniConstants.VER_CHUNITHM_AMAZON
elif version >= 135 and version < 140: # AMAZON PLUS
internal_ver = ChuniConstants.VER_CHUNITHM_AMAZON_PLUS
elif version >= 140 and version < 145: # CRYSTAL
internal_ver = ChuniConstants.VER_CHUNITHM_CRYSTAL
elif version >= 145 and version < 150: # CRYSTAL PLUS
internal_ver = ChuniConstants.VER_CHUNITHM_CRYSTAL_PLUS
elif version >= 150 and version < 200: # PARADISE
internal_ver = ChuniConstants.VER_CHUNITHM_PARADISE
elif version >= 200 and version < 205: # NEW!!
internal_ver = ChuniConstants.VER_CHUNITHM_NEW
elif version >= 205 and version < 210: # NEW PLUS!!
internal_ver = ChuniConstants.VER_CHUNITHM_NEW_PLUS
elif version >= 210 and version < 215: # SUN
internal_ver = ChuniConstants.VER_CHUNITHM_SUN
elif version >= 215 and version < 220: # SUN PLUS
internal_ver = ChuniConstants.VER_CHUNITHM_SUN_PLUS
elif version >= 220 and version < 225: # LUMINOUS
internal_ver = ChuniConstants.VER_CHUNITHM_LUMINOUS
elif version >= 225 and version < 230: # LUMINOUS PLUS
internal_ver = ChuniConstants.VER_CHUNITHM_LUMINOUS_PLUS
elif version >= 230: # VERSE
internal_ver = ChuniConstants.VER_CHUNITHM_VERSE
elif game_code == "SDGS": # Int
if version < 105: # SUPERSTAR
internal_ver = ChuniConstants.VER_CHUNITHM_CRYSTAL_PLUS
elif (
version >= 105 and version < 110
): # SUPERSTAR PLUS *Cursed but needed due to different encryption key
internal_ver = ChuniConstants.VER_CHUNITHM_PARADISE
elif version >= 110 and version < 115: # NEW
internal_ver = ChuniConstants.VER_CHUNITHM_NEW
elif version >= 115 and version < 120: # NEW PLUS!!
internal_ver = ChuniConstants.VER_CHUNITHM_NEW_PLUS
elif version >= 120 and version < 125: # SUN
internal_ver = ChuniConstants.VER_CHUNITHM_SUN
elif version >= 125 and version < 130: # SUN PLUS
internal_ver = ChuniConstants.VER_CHUNITHM_SUN_PLUS
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
elif game_code == "SDHJ": # Chn
if version < 110: # NEW
internal_ver = ChuniConstants.VER_CHUNITHM_NEW
elif (
version >= 110 and version < 120
): # NEW *Cursed but needed due to different encryption key
internal_ver = ChuniConstants.VER_CHUNITHM_NEW_PLUS
elif version >= 120: # LUMINOUS
internal_ver = ChuniConstants.VER_CHUNITHM_LUMINOUS
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
@@ -376,11 +396,11 @@ class ChuniServlet(BaseServlet):
else:
endpoint = endpoint
func_to_find = "handle_" + inflection.underscore(endpoint) + "_request"
func_to_find = "handle_" + self.strict_underscore(endpoint) + "_request"
handler_cls = self.versions[internal_ver](self.core_cfg, self.game_cfg)
if not hasattr(handler_cls, func_to_find):
self.logger.warning(f"Unhandled v{version} request {endpoint}")
self.logger.warning(f"Unhandled v{version} request {func_to_find}")
resp = {"returnCode": 1}
else:
@@ -414,3 +434,9 @@ class ChuniServlet(BaseServlet):
)
return Response(crypt.encrypt(padded))
def strict_underscore(self, name: str) -> str:
# Insert underscores between *all* capital letters
name = re.sub(r"([A-Z])([A-Z])", r"\1_\2", name)
return inflection.underscore(name)

View File

@@ -16,8 +16,8 @@ class ChuniLuminous(ChuniSunPlus):
super().__init__(core_cfg, game_cfg)
self.version = ChuniConstants.VER_CHUNITHM_LUMINOUS
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)
async def handle_c_m_get_user_preview_api_request(self, data: Dict) -> Dict:
user_data = await super().handle_c_m_get_user_preview_api_request(data)
# Does CARD MAKER 1.35 work this far up?
user_data["lastDataVersion"] = "2.20.00"

View File

@@ -12,8 +12,8 @@ class ChuniLuminousPlus(ChuniLuminous):
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)
async def handle_c_m_get_user_preview_api_request(self, data: Dict) -> Dict:
user_data = await super().handle_c_m_get_user_preview_api_request(data)
# Does CARD MAKER 1.35 work this far up?
user_data["lastDataVersion"] = "2.25.00"

View File

@@ -28,16 +28,18 @@ class ChuniNew(ChuniBase):
def _interal_ver_to_intver(self) -> str:
if self.version == ChuniConstants.VER_CHUNITHM_NEW:
return "200"
if self.version == ChuniConstants.VER_CHUNITHM_NEW_PLUS:
elif self.version == ChuniConstants.VER_CHUNITHM_NEW_PLUS:
return "205"
if self.version == ChuniConstants.VER_CHUNITHM_SUN:
elif self.version == ChuniConstants.VER_CHUNITHM_SUN:
return "210"
if self.version == ChuniConstants.VER_CHUNITHM_SUN_PLUS:
elif self.version == ChuniConstants.VER_CHUNITHM_SUN_PLUS:
return "215"
if self.version == ChuniConstants.VER_CHUNITHM_LUMINOUS:
elif self.version == ChuniConstants.VER_CHUNITHM_LUMINOUS:
return "220"
if self.version == ChuniConstants.VER_CHUNITHM_LUMINOUS_PLUS:
elif self.version == ChuniConstants.VER_CHUNITHM_LUMINOUS_PLUS:
return "225"
elif self.version == ChuniConstants.VER_CHUNITHM_VERSE:
return "230"
async def handle_get_game_setting_api_request(self, data: Dict) -> Dict:
# use UTC time and convert it to JST time by adding +9
@@ -171,7 +173,7 @@ class ChuniNew(ChuniBase):
}
return data1
async def handle_cm_get_user_preview_api_request(self, data: Dict) -> Dict:
async def handle_c_m_get_user_preview_api_request(self, data: Dict) -> Dict:
p = await self.data.profile.get_profile_data(data["userId"], self.version)
if p is None:
return {}
@@ -244,7 +246,7 @@ class ChuniNew(ChuniBase):
"ssrBookCalcList": [],
}
async def handle_cm_get_user_data_api_request(self, data: Dict) -> Dict:
async def handle_c_m_get_user_data_api_request(self, data: Dict) -> Dict:
p = await self.data.profile.get_profile_data(data["userId"], self.version)
if p is None:
return {}
@@ -347,10 +349,10 @@ class ChuniNew(ChuniBase):
"userCardPrintStateList": card_print_state_list,
}
async def handle_cm_get_user_character_api_request(self, data: Dict) -> Dict:
async def handle_c_m_get_user_character_api_request(self, data: Dict) -> Dict:
return await super().handle_get_user_character_api_request(data)
async def handle_cm_get_user_item_api_request(self, data: Dict) -> Dict:
async def handle_c_m_get_user_item_api_request(self, data: Dict) -> Dict:
return await super().handle_get_user_item_api_request(data)
async def handle_roll_gacha_api_request(self, data: Dict) -> Dict:
@@ -395,7 +397,7 @@ class ChuniNew(ChuniBase):
return {"length": len(rolled_cards), "gameGachaCardList": rolled_cards}
async def handle_cm_upsert_user_gacha_api_request(self, data: Dict) -> Dict:
async def handle_c_m_upsert_user_gacha_api_request(self, data: Dict) -> Dict:
upsert = data["cmUpsertUserGacha"]
user_id = data["userId"]
place_id = data["placeId"]
@@ -450,7 +452,7 @@ class ChuniNew(ChuniBase):
"userCardPrintStateList": card_print_state_list,
}
async def handle_cm_upsert_user_printlog_api_request(self, data: Dict) -> Dict:
async def handle_c_m_upsert_user_printlog_api_request(self, data: Dict) -> Dict:
return {
"returnCode": 1,
"orderId": 0,
@@ -458,7 +460,7 @@ class ChuniNew(ChuniBase):
"apiName": "CMUpsertUserPrintlogApi",
}
async def handle_cm_upsert_user_print_api_request(self, data: Dict) -> Dict:
async def handle_c_m_upsert_user_print_api_request(self, data: Dict) -> Dict:
user_print_detail = data["userPrintDetail"]
user_id = data["userId"]
@@ -483,7 +485,7 @@ class ChuniNew(ChuniBase):
"apiName": "CMUpsertUserPrintApi",
}
async def handle_cm_upsert_user_print_subtract_api_request(self, data: Dict) -> Dict:
async def handle_c_m_upsert_user_print_subtract_api_request(self, data: Dict) -> Dict:
upsert = data["userCardPrintState"]
user_id = data["userId"]
place_id = data["placeId"]
@@ -500,7 +502,7 @@ class ChuniNew(ChuniBase):
return {"returnCode": "1", "apiName": "CMUpsertUserPrintSubtractApi"}
async def handle_cm_upsert_user_print_cancel_api_request(self, data: Dict) -> Dict:
async def handle_c_m_upsert_user_print_cancel_api_request(self, data: Dict) -> Dict:
order_ids = data["orderIdList"]
user_id = data["userId"]

View File

@@ -11,8 +11,8 @@ class ChuniNewPlus(ChuniNew):
super().__init__(core_cfg, game_cfg)
self.version = ChuniConstants.VER_CHUNITHM_NEW_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)
async def handle_c_m_get_user_preview_api_request(self, data: Dict) -> Dict:
user_data = await super().handle_c_m_get_user_preview_api_request(data)
# hardcode lastDataVersion for CardMaker 1.35 A028
user_data["lastDataVersion"] = "2.05.00"

View File

@@ -62,6 +62,7 @@ class ChuniReader(BaseReader):
await self.read_character(f"{dir}/chara", dds_images, this_opt_id)
await self.read_map_icon(f"{dir}/mapIcon", this_opt_id)
await self.read_system_voice(f"{dir}/systemVoice", this_opt_id)
await self.read_unlock_challenge(f"{dir}/unlockChallenge")
async def read_login_bonus(self, root_dir: str, opt_id: Optional[int] = None) -> None:
for root, dirs, files in walk(f"{root_dir}loginBonusPreset"):
@@ -499,6 +500,38 @@ class ChuniReader(BaseReader):
self.logger.info(f"Opt folder {opt_folder} (Database ID {opt_id}) contains {data_config['Version']['Name']} v{data_config['Version']['VerMajor']}.{data_config['Version']['VerMinor']}.{opt_seq}")
return opt_id
async def read_unlock_challenge(self, uc_dir: str) -> None:
for root, dirs, files in walk(uc_dir):
for dir in dirs:
if path.exists(f"{root}/{dir}/UnlockChallenge.xml"):
with open(f"{root}/{dir}/UnlockChallenge.xml", "r", encoding="utf-8") as fp:
strdata = fp.read()
xml_root = ET.fromstring(strdata)
for name in xml_root.findall("name"):
id = name.find("id").text
name = name.find("str").text
course_ids = []
for course in xml_root.find("musicList/list/UnlockChallengeMusicListSubData/unlockChallengeMusicData/courseList/list").findall("UnlockChallengeCourseListSubData"):
course_id = course.find("unlockChallengeCourseData/courseName").find("id").text
course_ids.append(course_id)
# Build keyword arguments dynamically for up to 5 course IDs
course_kwargs = {
f"course_id{i+1}": course_ids[i]
for i in range(min(5, len(course_ids)))
}
result = await self.data.static.put_unlock_challenge(
self.version, id, name,
**course_kwargs
)
if result is not None:
self.logger.info(f"Inserted unlock challenge {id}")
else:
self.logger.warning(f"Failed to unlock challenge {id}")
def copy_image(self, filename: str, src_dir: str, dst_dir: str) -> None:
# Convert the image to png so we can easily display it in the frontend
file_src = path.join(src_dir, filename)

View File

@@ -22,6 +22,7 @@ character: Table = Table(
Column("id", Integer, primary_key=True, nullable=False),
Column(
"user",
Integer,
ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"),
nullable=False,
),
@@ -46,6 +47,7 @@ item: Table = Table(
Column("id", Integer, primary_key=True, nullable=False),
Column(
"user",
Integer,
ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"),
nullable=False,
),
@@ -63,6 +65,7 @@ duel = Table(
Column("id", Integer, primary_key=True, nullable=False),
Column(
"user",
Integer,
ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"),
nullable=False,
),
@@ -85,6 +88,7 @@ map = Table(
Column("id", Integer, primary_key=True, nullable=False),
Column(
"user",
Integer,
ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"),
nullable=False,
),
@@ -107,6 +111,7 @@ map_area = Table(
Column("id", Integer, primary_key=True, nullable=False),
Column(
"user",
Integer,
ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"),
nullable=False,
),
@@ -127,6 +132,7 @@ gacha = Table(
Column("id", Integer, primary_key=True, nullable=False),
Column(
"user",
Integer,
ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"),
nullable=False,
),
@@ -147,6 +153,7 @@ print_state: Table = Table(
Column("id", Integer, primary_key=True, nullable=False),
Column(
"user",
Integer,
ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"),
nullable=False,
),
@@ -167,6 +174,7 @@ print_detail = Table(
Column("id", Integer, primary_key=True, nullable=False),
Column(
"user",
Integer,
ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"),
nullable=False,
),
@@ -197,6 +205,7 @@ login_bonus = Table(
Column("id", Integer, primary_key=True, nullable=False),
Column(
"user",
Integer,
ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"),
nullable=False,
),
@@ -216,6 +225,7 @@ favorite: Table = Table(
Column("id", Integer, primary_key=True, nullable=False),
Column(
"user",
Integer,
ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"),
nullable=False,
),
@@ -232,6 +242,7 @@ matching = Table(
Column("roomId", Integer, nullable=False),
Column(
"user",
Integer,
ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"),
nullable=False,
),
@@ -249,6 +260,7 @@ cmission = Table(
Column("id", Integer, primary_key=True, nullable=False),
Column(
"user",
Integer,
ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"),
nullable=False,
),
@@ -262,7 +274,12 @@ cmission_progress = Table(
"chuni_item_cmission_progress",
metadata,
Column("id", Integer, primary_key=True, nullable=False),
Column("user", ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"), nullable=False),
Column(
"user",
Integer,
ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"),
nullable=False,
),
Column("missionId", Integer, nullable=False),
Column("order", Integer),
Column("stage", Integer),
@@ -273,14 +290,35 @@ cmission_progress = Table(
mysql_charset="utf8mb4",
)
unlock_challenge = Table(
"chuni_item_unlock_challenge",
metadata,
Column("id", Integer, primary_key=True, nullable=False),
Column("version", Integer, nullable=False),
Column(
"user",
Integer,
ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"),
nullable=False,
),
Column("unlockChallengeId", Integer, nullable=False),
Column("status", Integer),
Column("clearCourseId", Integer),
Column("conditionType", Integer),
Column("score", Integer),
Column("life", Integer),
Column("clearDate", TIMESTAMP, server_default=func.now()),
UniqueConstraint(
"version", "user", "unlockChallengeId", name="chuni_item_unlock_challenge_uk"
),
mysql_charset="utf8mb4",
)
class ChuniItemData(BaseData):
async def get_oldest_free_matching(self, version: int) -> Optional[Row]:
sql = matching.select(
and_(
matching.c.version == version,
matching.c.isFull == False
)
and_(matching.c.version == version, matching.c.isFull == False)
).order_by(matching.c.roomId.asc())
result = await self.execute(sql)
@@ -289,11 +327,9 @@ class ChuniItemData(BaseData):
return result.fetchone()
async def get_newest_matching(self, version: int) -> Optional[Row]:
sql = matching.select(
and_(
matching.c.version == version
)
).order_by(matching.c.roomId.desc())
sql = matching.select(and_(matching.c.version == version)).order_by(
matching.c.roomId.desc()
)
result = await self.execute(sql)
if result is None:
@@ -301,11 +337,7 @@ class ChuniItemData(BaseData):
return result.fetchone()
async def get_all_matchings(self, version: int) -> Optional[List[Row]]:
sql = matching.select(
and_(
matching.c.version == version
)
)
sql = matching.select(and_(matching.c.version == version))
result = await self.execute(sql)
if result is None:
@@ -329,7 +361,7 @@ class ChuniItemData(BaseData):
matching_member_info_list: List,
user_id: int = None,
rest_sec: int = 60,
is_full: bool = False
is_full: bool = False,
) -> Optional[int]:
sql = insert(matching).values(
roomId=room_id,
@@ -452,23 +484,31 @@ class ChuniItemData(BaseData):
return None
return result.fetchone()
async def put_favorite_music(self, user_id: int, version: int, music_id: int) -> Optional[int]:
sql = insert(favorite).values(user=user_id, version=version, favId=music_id, favKind=1)
async def put_favorite_music(
self, user_id: int, version: int, music_id: int
) -> Optional[int]:
sql = insert(favorite).values(
user=user_id, version=version, favId=music_id, favKind=1
)
conflict = sql.on_duplicate_key_update(user=user_id, version=version, favId=music_id, favKind=1)
conflict = sql.on_duplicate_key_update(
user=user_id, version=version, favId=music_id, favKind=1
)
result = await self.execute(conflict)
if result is None:
return None
return result.lastrowid
async def delete_favorite_music(self, user_id: int, version: int, music_id: int) -> Optional[int]:
async def delete_favorite_music(
self, user_id: int, version: int, music_id: int
) -> Optional[int]:
sql = delete(favorite).where(
and_(
favorite.c.user==user_id,
favorite.c.version==version,
favorite.c.favId==music_id,
favorite.c.favKind==1
favorite.c.user == user_id,
favorite.c.version == version,
favorite.c.favId == music_id,
favorite.c.favKind == 1,
)
)
@@ -611,8 +651,12 @@ class ChuniItemData(BaseData):
return None
return result.lastrowid
async def get_map_areas(self, user_id: int, map_area_ids: List[int]) -> Optional[List[Row]]:
sql = select(map_area).where(map_area.c.user == user_id, map_area.c.mapAreaId.in_(map_area_ids))
async def get_map_areas(
self, user_id: int, map_area_ids: List[int]
) -> Optional[List[Row]]:
sql = select(map_area).where(
map_area.c.user == user_id, map_area.c.mapAreaId.in_(map_area_ids)
)
result = await self.execute(sql)
if result is None:
@@ -713,7 +757,7 @@ class ChuniItemData(BaseData):
)
return None
return result.lastrowid
async def put_cmission_progress(
self, user_id: int, mission_id: int, progress_data: Dict
) -> Optional[int]:
@@ -723,10 +767,10 @@ class ChuniItemData(BaseData):
sql = insert(cmission_progress).values(**progress_data)
conflict = sql.on_duplicate_key_update(**progress_data)
result = await self.execute(conflict)
if result is None:
return None
return result.lastrowid
async def get_cmission_progress(
@@ -739,21 +783,21 @@ class ChuniItemData(BaseData):
)
).order_by(cmission_progress.c.order.asc())
result = await self.execute(sql)
if result is None:
return None
return result.fetchall()
async def get_cmission(self, user_id: int, mission_id: int) -> Optional[Row]:
sql = cmission.select(
and_(cmission.c.user == user_id, cmission.c.missionId == mission_id)
)
result = await self.execute(sql)
if result is None:
return None
return result.fetchone()
async def put_cmission(self, user_id: int, mission_data: Dict) -> Optional[int]:
@@ -762,17 +806,46 @@ class ChuniItemData(BaseData):
sql = insert(cmission).values(**mission_data)
conflict = sql.on_duplicate_key_update(**mission_data)
result = await self.execute(conflict)
if result is None:
return None
return result.lastrowid
async def get_cmissions(self, user_id: int) -> Optional[List[Row]]:
sql = cmission.select(cmission.c.user == user_id)
result = await self.execute(sql)
if result is None:
return None
return result.fetchall()
async def put_unlock_challenge(
self, user_id: int, version: int, unlock_challenge_data: Dict
) -> Optional[int]:
unlock_challenge_data["user"] = user_id
unlock_challenge_data["version"] = version
sql = insert(unlock_challenge).values(**unlock_challenge_data)
conflict = sql.on_duplicate_key_update(**unlock_challenge_data)
result = await self.execute(conflict)
if result is None:
return None
return result.lastrowid
async def get_unlock_challenges(
self, user_id: int, version: int
) -> Optional[List[Row]]:
sql = unlock_challenge.select(
and_(
unlock_challenge.c.user == user_id,
unlock_challenge.c.version == version,
)
)
result = await self.execute(sql)
if result is None:
return None
return result.fetchall()

View File

@@ -15,6 +15,7 @@ profile = Table(
Column("id", Integer, primary_key=True, nullable=False),
Column(
"user",
Integer,
ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"),
nullable=False,
),
@@ -25,6 +26,8 @@ profile = Table(
Column("frameId", Integer),
Column("isMaimai", Boolean),
Column("trophyId", Integer),
Column("trophyIdSub1", Integer),
Column("trophyIdSub2", Integer),
Column("userName", String(25)),
Column("isWebJoin", Boolean),
Column("playCount", Integer),
@@ -139,6 +142,7 @@ profile_ex = Table(
Column("id", Integer, primary_key=True, nullable=False),
Column(
"user",
Integer,
ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"),
nullable=False,
),
@@ -187,6 +191,7 @@ option = Table(
Column("id", Integer, primary_key=True, nullable=False),
Column(
"user",
Integer,
ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"),
nullable=False,
),
@@ -250,6 +255,7 @@ option_ex = Table(
Column("id", Integer, primary_key=True, nullable=False),
Column(
"user",
Integer,
ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"),
nullable=False,
),
@@ -283,6 +289,7 @@ recent_rating = Table(
Column("id", Integer, primary_key=True, nullable=False),
Column(
"user",
Integer,
ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"),
nullable=False,
),
@@ -297,6 +304,7 @@ region = Table(
Column("id", Integer, primary_key=True, nullable=False),
Column(
"user",
Integer,
ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"),
nullable=False,
),
@@ -312,6 +320,7 @@ activity = Table(
Column("id", Integer, primary_key=True, nullable=False),
Column(
"user",
Integer,
ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"),
nullable=False,
),
@@ -334,6 +343,7 @@ charge = Table(
Column("id", Integer, primary_key=True, nullable=False),
Column(
"user",
Integer,
ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"),
nullable=False,
),
@@ -354,6 +364,7 @@ emoney = Table(
Column("id", Integer, primary_key=True, nullable=False),
Column(
"user",
Integer,
ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"),
nullable=False,
),
@@ -373,6 +384,7 @@ overpower = Table(
Column("id", Integer, primary_key=True, nullable=False),
Column(
"user",
Integer,
ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"),
nullable=False,
),
@@ -400,6 +412,7 @@ rating = Table(
Column("id", Integer, primary_key=True, nullable=False),
Column(
"user",
Integer,
ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"),
nullable=False,
),
@@ -461,10 +474,12 @@ class ChuniProfileData(BaseData):
return False
return True
async def update_userbox(self, user_id: int, version: int, new_nameplate: int, new_trophy: int, new_character: int) -> bool:
async def update_userbox(self, user_id: int, version: int, new_nameplate: int, new_trophy: int, new_trophy_sub_1: int, new_trophy_sub_2: int, new_character: int) -> bool:
sql = profile.update((profile.c.user == user_id) & (profile.c.version == version)).values(
nameplateId=new_nameplate,
trophyId=new_trophy,
trophyIdSub1=new_trophy_sub_1,
trophyIdSub2=new_trophy_sub_2,
charaIllustId=new_character
)
result = await self.execute(sql)
@@ -899,4 +914,4 @@ class ChuniProfileData(BaseData):
async def get_net_battle(self, aime_id: int) -> Optional[Row]:
result = await self.execute(net_battle.select(net_battle.c.user == aime_id))
if result:
return result.fetchone()
return result.fetchone()

View File

@@ -17,6 +17,7 @@ course: Table = Table(
Column("id", Integer, primary_key=True, nullable=False),
Column(
"user",
Integer,
ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"),
nullable=False,
),
@@ -48,6 +49,7 @@ best_score: Table = Table(
Column("id", Integer, primary_key=True, nullable=False),
Column(
"user",
Integer,
ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"),
nullable=False,
),
@@ -79,6 +81,7 @@ playlog = Table(
Column("id", Integer, primary_key=True, nullable=False),
Column(
"user",
Integer,
ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"),
nullable=False,
),
@@ -139,6 +142,8 @@ playlog = Table(
Column("regionId", Integer),
Column("machineType", Integer),
Column("ticketId", Integer),
Column("monthPoint", Integer),
Column("eventPoint", Integer),
mysql_charset="utf8mb4"
)
@@ -420,4 +425,4 @@ class ChuniScoreData(BaseData):
return None
rows = result.fetchall()
return [dict(row) for row in rows]
return [dict(row) for row in rows]

View File

@@ -289,6 +289,27 @@ login_bonus = Table(
mysql_charset="utf8mb4",
)
unlock_challenge = Table(
"chuni_static_unlock_challenge",
metadata,
Column("id", Integer, primary_key=True, nullable=False),
Column("version", Integer, nullable=False),
Column("unlockChallengeId", Integer, nullable=False),
Column("name", String(255)),
Column("isEnabled", Boolean, server_default="1"),
Column("startDate", TIMESTAMP, server_default=func.now()),
Column("courseId1", Integer),
Column("courseId2", Integer),
Column("courseId3", Integer),
Column("courseId4", Integer),
Column("courseId5", Integer),
UniqueConstraint(
"version", "unlockChallengeId", name="chuni_static_unlock_challenge_pk"
),
mysql_charset="utf8mb4",
)
class ChuniStaticData(BaseData):
async def put_login_bonus(
self,
@@ -556,7 +577,7 @@ class ChuniStaticData(BaseData):
return result.fetchall()
async def get_music(self, version: int) -> Optional[List[Row]]:
sql = music.select(music.c.version <= version)
sql = music.select(music.c.version == version)
result = await self.execute(sql)
if result is None:
@@ -586,6 +607,28 @@ class ChuniStaticData(BaseData):
if result is None:
return None
return result.fetchone()
async def get_music_by_metadata(
self, title: Optional[str] = None, artist: Optional[str] = None, genre: Optional[str] = None
) -> Optional[List[Row]]:
# all conditions should use like for partial matches
conditions = []
if title:
conditions.append(music.c.title.like(f"%{title}%"))
if artist:
conditions.append(music.c.artist.like(f"%{artist}%"))
if genre:
conditions.append(music.c.genre.like(f"%{genre}%"))
if not conditions:
return None
sql = select(music).where(and_(*conditions))
result = await self.execute(sql)
if result is None:
return None
return result.fetchall()
async def put_avatar(
self,
@@ -629,11 +672,25 @@ class ChuniStaticData(BaseData):
return None
return result.lastrowid
async def get_avatar_items(self, version: int, category: int, enabled_only: bool = True) -> Optional[List[Dict]]:
async def get_avatar_items(
self, version: int, category: int, enabled_only: bool = True
) -> Optional[List[Dict]]:
if enabled_only:
sql = select(avatar).where((avatar.c.version == version) & (avatar.c.category == category) & (avatar.c.isEnabled)).order_by(avatar.c.sortName)
sql = (
select(avatar)
.where(
(avatar.c.version == version)
& (avatar.c.category == category)
& (avatar.c.isEnabled)
)
.order_by(avatar.c.sortName)
)
else:
sql = select(avatar).where((avatar.c.version == version) & (avatar.c.category == category)).order_by(avatar.c.sortName)
sql = (
select(avatar)
.where((avatar.c.version == version) & (avatar.c.category == category))
.order_by(avatar.c.sortName)
)
result = await self.execute(sql)
if result is None:
@@ -676,11 +733,21 @@ class ChuniStaticData(BaseData):
return None
return result.lastrowid
async def get_nameplates(self, version: int, enabled_only: bool = True) -> Optional[List[Dict]]:
async def get_nameplates(
self, version: int, enabled_only: bool = True
) -> Optional[List[Dict]]:
if enabled_only:
sql = select(nameplate).where((nameplate.c.version == version) & (nameplate.c.isEnabled)).order_by(nameplate.c.sortName)
sql = (
select(nameplate)
.where((nameplate.c.version == version) & (nameplate.c.isEnabled))
.order_by(nameplate.c.sortName)
)
else:
sql = select(nameplate).where(nameplate.c.version == version).order_by(nameplate.c.sortName)
sql = (
select(nameplate)
.where(nameplate.c.version == version)
.order_by(nameplate.c.sortName)
)
result = await self.execute(sql)
if result is None:
@@ -720,11 +787,21 @@ class ChuniStaticData(BaseData):
return None
return result.lastrowid
async def get_trophies(self, version: int, enabled_only: bool = True) -> Optional[List[Dict]]:
async def get_trophies(
self, version: int, enabled_only: bool = True
) -> Optional[List[Dict]]:
if enabled_only:
sql = select(trophy).where((trophy.c.version == version) & (trophy.c.isEnabled)).order_by(trophy.c.name)
sql = (
select(trophy)
.where((trophy.c.version == version) & (trophy.c.isEnabled))
.order_by(trophy.c.name)
)
else:
sql = select(trophy).where(trophy.c.version == version).order_by(trophy.c.name)
sql = (
select(trophy)
.where(trophy.c.version == version)
.order_by(trophy.c.name)
)
result = await self.execute(sql)
if result is None:
@@ -767,11 +844,21 @@ class ChuniStaticData(BaseData):
return None
return result.lastrowid
async def get_map_icons(self, version: int, enabled_only: bool = True) -> Optional[List[Dict]]:
async def get_map_icons(
self, version: int, enabled_only: bool = True
) -> Optional[List[Dict]]:
if enabled_only:
sql = select(map_icon).where((map_icon.c.version == version) & (map_icon.c.isEnabled)).order_by(map_icon.c.sortName)
sql = (
select(map_icon)
.where((map_icon.c.version == version) & (map_icon.c.isEnabled))
.order_by(map_icon.c.sortName)
)
else:
sql = select(map_icon).where(map_icon.c.version == version).order_by(map_icon.c.sortName)
sql = (
select(map_icon)
.where(map_icon.c.version == version)
.order_by(map_icon.c.sortName)
)
result = await self.execute(sql)
if result is None:
@@ -814,11 +901,21 @@ class ChuniStaticData(BaseData):
return None
return result.lastrowid
async def get_system_voices(self, version: int, enabled_only: bool = True) -> Optional[List[Dict]]:
async def get_system_voices(
self, version: int, enabled_only: bool = True
) -> Optional[List[Dict]]:
if enabled_only:
sql = select(system_voice).where((system_voice.c.version == version) & (system_voice.c.isEnabled)).order_by(system_voice.c.sortName)
sql = (
select(system_voice)
.where((system_voice.c.version == version) & (system_voice.c.isEnabled))
.order_by(system_voice.c.sortName)
)
else:
sql = select(system_voice).where(system_voice.c.version == version).order_by(system_voice.c.sortName)
sql = (
select(system_voice)
.where(system_voice.c.version == version)
.order_by(system_voice.c.sortName)
)
result = await self.execute(sql)
if result is None:
@@ -873,11 +970,21 @@ class ChuniStaticData(BaseData):
return None
return result.lastrowid
async def get_characters(self, version: int, enabled_only: bool = True) -> Optional[List[Dict]]:
async def get_characters(
self, version: int, enabled_only: bool = True
) -> Optional[List[Dict]]:
if enabled_only:
sql = select(character).where((character.c.version == version) & (character.c.isEnabled)).order_by(character.c.sortName)
sql = (
select(character)
.where((character.c.version == version) & (character.c.isEnabled))
.order_by(character.c.sortName)
)
else:
sql = select(character).where(character.c.version == version).order_by(character.c.sortName)
sql = (
select(character)
.where(character.c.version == version)
.order_by(character.c.sortName)
)
result = await self.execute(sql)
if result is None:
@@ -1074,3 +1181,54 @@ class ChuniStaticData(BaseData):
self.logger.error(f"Failed to set opt enabled status to {enabled} for opt {opt_id}")
return False
return True
async def put_unlock_challenge(
self,
version: int,
unlock_challenge_id: int,
name: str,
course_id1: Optional[int] = None,
course_id2: Optional[int] = None,
course_id3: Optional[int] = None,
course_id4: Optional[int] = None,
course_id5: Optional[int] = None,
) -> Optional[int]:
sql = insert(unlock_challenge).values(
version=version,
unlockChallengeId=unlock_challenge_id,
name=name,
courseId1=course_id1,
courseId2=course_id2,
courseId3=course_id3,
courseId4=course_id4,
courseId5=course_id5,
)
conflict = sql.on_duplicate_key_update(
name=name,
courseId1=course_id1,
courseId2=course_id2,
courseId3=course_id3,
courseId4=course_id4,
courseId5=course_id5,
)
result = await self.execute(conflict)
if result is None:
return None
return result.lastrowid
async def get_unlock_challenges(self, version: int) -> Optional[List[Dict]]:
sql = unlock_challenge.select(
and_(
unlock_challenge.c.version == version,
unlock_challenge.c.isEnabled == True,
)
).order_by(unlock_challenge.c.startDate.asc())
result = await self.execute(sql)
if result is None:
return None
return result.fetchall()

View File

@@ -11,8 +11,8 @@ class ChuniSun(ChuniNewPlus):
super().__init__(core_cfg, game_cfg)
self.version = ChuniConstants.VER_CHUNITHM_SUN
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)
async def handle_c_m_get_user_preview_api_request(self, data: Dict) -> Dict:
user_data = await super().handle_c_m_get_user_preview_api_request(data)
# hardcode lastDataVersion for CardMaker 1.35 A032
user_data["lastDataVersion"] = "2.10.00"

View File

@@ -11,8 +11,8 @@ class ChuniSunPlus(ChuniSun):
super().__init__(core_cfg, game_cfg)
self.version = ChuniConstants.VER_CHUNITHM_SUN_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)
async def handle_c_m_get_user_preview_api_request(self, data: Dict) -> Dict:
user_data = await super().handle_c_m_get_user_preview_api_request(data)
# I don't know if lastDataVersion is going to matter, I don't think CardMaker 1.35 works this far up
user_data["lastDataVersion"] = "2.15.00"

View File

@@ -44,7 +44,25 @@
<tr><td>Nameplate:</td><td style="width: 80%;"><div id="name_nameplate"></div></td></tr>
<tr><td>Trophy:</td><td><div id="name_trophy">
<select name="trophy" id="trophy" onchange="changeTrophy()" style="width:100%;">
<select name="trophy" id="trophy" onclick="changeTrophy()" style="width:100%;">
{% for item in trophies.values() %}
<option value="{{ item["id"] }}" class="trophy-rank{{ item["rarity"] }}">{{ item["name"] }}</option>
{% endfor %}
</select>
</div></td></tr>
<tr><td>Trophy Sub 1:</td><td><div id="name_trophy">
<select name="trophy-sub-1" id="trophy-sub-1" onclick="changeTrophySub1()" style="width:100%;">
<option value="-1"></option>
{% for item in trophies.values() %}
<option value="{{ item["id"] }}" class="trophy-rank{{ item["rarity"] }}">{{ item["name"] }}</option>
{% endfor %}
</select>
</div></td></tr>
<tr><td>Trophy Sub 2:</td><td><div id="name_trophy">
<select name="trophy-sub-2" id="trophy-sub-2" onclick="changeTrophySub2()" style="width:100%;">
<option value="-1"></option>
{% for item in trophies.values() %}
<option value="{{ item["id"] }}" class="trophy-rank{{ item["rarity"] }}">{{ item["name"] }}</option>
{% endfor %}
@@ -124,7 +142,10 @@ userbox_components = {
};
types = Object.keys(userbox_components);
orig_trophy = curr_trophy = "{{ profile.trophyId }}";
orig_trophy_sub_1 = curr_trophy_sub_1 = "{{ profile.trophyIdSub1 }}";
orig_trophy_sub_2 = curr_trophy_sub_2 = "{{ profile.trophyIdSub2 }}";
curr_trophy_img = "";
curr_trophy_name = "";
function enableButtons(enabled) {
document.getElementById("reset-btn").disabled = !enabled;
@@ -169,6 +190,7 @@ function changeTrophy() {
var trophy_element = document.getElementById("trophy");
curr_trophy = trophy_element.value;
curr_trophy_name = trophy_element[trophy_element.selectedIndex].innerText
curr_trophy_img = getRankImage(trophy_element[trophy_element.selectedIndex]);
updatePreview();
if (curr_trophy != orig_trophy) {
@@ -176,12 +198,38 @@ function changeTrophy() {
}
}
function changeTrophySub1() {
var trophy_element = document.getElementById("trophy-sub-1");
curr_trophy_sub_1 = trophy_element.value;
curr_trophy_img = getRankImage(trophy_element[trophy_element.selectedIndex]);
curr_trophy_name = trophy_element[trophy_element.selectedIndex].innerText
updatePreview();
if (curr_trophy_sub_1 != orig_trophy_sub_1) {
enableButtons(true);
}
}
function changeTrophySub2() {
var trophy_element = document.getElementById("trophy-sub-2");
curr_trophy_sub_2 = trophy_element.value;
curr_trophy_img = getRankImage(trophy_element[trophy_element.selectedIndex]);
curr_trophy_name = trophy_element[trophy_element.selectedIndex].innerText
updatePreview();
if (curr_trophy_sub_2 != orig_trophy_sub_2) {
enableButtons(true);
}
}
function resetUserbox() {
for (const type of types) {
changeItem(type, userbox_components[type][orig_id], userbox_components[type][orig_name], userbox_components[type][orig_img]);
}
// reset trophy
document.getElementById("trophy").value = orig_trophy;
document.getElementById("trophy-sub-1").value = orig_trophy_sub_1;
document.getElementById("trophy-sub-2").value = orig_trophy_sub_2;
changeTrophy();
// disable the save/reset buttons until something changes
enableButtons(false);
@@ -193,12 +241,14 @@ function updatePreview() {
document.getElementById("name_" + type).innerHTML = userbox_components[type][curr_name];
}
document.getElementById("preview_trophy_rank").src = "img/rank/" + curr_trophy_img;
document.getElementById("preview_trophy_name").innerHTML = document.getElementById("trophy")[document.getElementById("trophy").selectedIndex].innerText;
document.getElementById("preview_trophy_name").innerHTML = curr_trophy_name;
}
function saveUserbox() {
$.post("/game/chuni/update.userbox", { nameplate: userbox_components["nameplate"][curr_id],
trophy: curr_trophy,
trophy: curr_trophy,
trophySub1: curr_trophy_sub_1,
trophySub2: curr_trophy_sub_2,
character: userbox_components["character"][curr_id] })
.done(function (data) {
// set the current as the original and disable buttons
@@ -207,7 +257,9 @@ function saveUserbox() {
userbox_components[type][orig_name] = userbox_components[type][orig_name];
userbox_components[type][orig_img] = userbox_components[type][curr_img];
}
orig_trophy = curr_trophy
orig_trophy = curr_trophy;
orig_trophy_sub_1 = curr_trophy_sub_1;
orig_trophy_sub_2 = curr_trophy_sub_2;
enableButtons(false);
})
.fail(function () {

246
titles/chuni/verse.py Normal file
View File

@@ -0,0 +1,246 @@
from datetime import datetime, timedelta
from typing import Dict, List, Set
from core.config import CoreConfig
from titles.chuni.config import ChuniConfig
from titles.chuni.const import (
ChuniConstants,
MapAreaConditionLogicalOperator,
MapAreaConditionType,
)
from titles.chuni.luminousplus import ChuniLuminousPlus
class ChuniVerse(ChuniLuminousPlus):
def __init__(self, core_cfg: CoreConfig, game_cfg: ChuniConfig) -> None:
super().__init__(core_cfg, game_cfg)
self.version = ChuniConstants.VER_CHUNITHM_VERSE
async def handle_c_m_get_user_preview_api_request(self, data: Dict) -> Dict:
user_data = await super().handle_c_m_get_user_preview_api_request(data)
# Does CARD MAKER 1.35 work this far up?
user_data["lastDataVersion"] = "2.30.00"
return user_data
async def handle_get_game_course_level_api_request(self, data: Dict) -> Dict:
unlock_challenges = await self.data.static.get_unlock_challenges(self.version)
game_course_level_list = []
for unlock_challenge in unlock_challenges:
course_ids = [
unlock_challenge[f"courseId{i}"]
for i in range(1, 6)
if unlock_challenge[f"courseId{i}"] is not None
]
start_date = unlock_challenge["startDate"].replace(
hour=0, minute=0, second=0
)
for i, course_id in enumerate(course_ids):
start = start_date + timedelta(days=7 * i)
end = start_date + timedelta(days=7 * (i + 1)) - timedelta(seconds=1)
if i == len(course_ids) - 1:
# If this is the last course, set end date to a far future date
end = datetime(2099, 1, 1)
game_course_level_list.append(
{
"courseId": course_id,
"startDate": start.strftime(self.date_time_format),
"endDate": end.strftime(self.date_time_format),
}
)
return {
"length": len(game_course_level_list),
"gameCourseLevelList": game_course_level_list,
}
async def handle_get_game_u_c_condition_api_request(self, data: Dict) -> Dict:
unlock_challenges = await self.data.static.get_unlock_challenges(self.version)
game_unlock_challenge_condition_list = []
conditions = {
# unlock Theatore Creatore (ULTIMA) after clearing map VERSE ep. I
10001: {
"type": MapAreaConditionType.MAP_CLEARED.value,
"conditionId": 3020798,
},
# unlock Crossmythos Rhapsodia after clearing map VERSE ep. IV
10006: {
"type": MapAreaConditionType.MAP_CLEARED.value,
"conditionId": 3020802,
},
}
for unlock_challenge in unlock_challenges:
unlock_challenge_id = unlock_challenge["unlockChallengeId"]
unlock_condition = conditions.get(
unlock_challenge_id,
{
"type": MapAreaConditionType.TROPHY_OBTAINED.value, # always unlocked
"conditionId": 0,
},
)
game_unlock_challenge_condition_list.append(
{
"unlockChallengeId": unlock_challenge_id,
"length": 1,
"conditionList": [
{
"type": unlock_condition["type"],
"conditionId": unlock_condition["conditionId"],
"logicalOpe": MapAreaConditionLogicalOperator.AND.value,
"startDate": unlock_challenge["startDate"].strftime(
self.date_time_format
),
"endDate": datetime(2099, 1, 1).strftime(
self.date_time_format
),
}
],
}
)
return {
"length": len(game_unlock_challenge_condition_list),
"gameUnlockChallengeConditionList": game_unlock_challenge_condition_list,
}
async def handle_get_user_u_c_api_request(self, data: Dict) -> Dict:
user_id = data["userId"]
user_unlock_challenges = await self.data.item.get_unlock_challenges(
user_id, self.version
)
user_unlock_challenge_list = [
{
"unlockChallengeId": user_uc["unlockChallengeId"],
"status": user_uc["status"],
"clearCourseId": user_uc["clearCourseId"],
"conditionType": user_uc["conditionType"],
"score": user_uc["score"],
"life": user_uc["life"],
"clearDate": user_uc["clearDate"].strftime(self.date_time_format),
}
for user_uc in user_unlock_challenges
]
return {
"userId": user_id,
"userUnlockChallengeList": user_unlock_challenge_list,
}
async def handle_get_user_rec_music_api_request(self, data: Dict) -> Dict:
rec_limit = 25 # limit for recommendations
user_id = data["userId"]
user_rec_music_set = set()
recent_rating = await self.data.profile.get_profile_recent_rating(user_id)
if not recent_rating:
# If no recent ratings, return an empty list
return {
"length": 0,
"userRecMusicList": [],
}
recent_ratings = recent_rating["recentRating"]
# cache music info
music_info_list = []
for recent_rating in recent_ratings:
music_id = recent_rating["musicId"]
music_info = await self.data.static.get_song(music_id)
if music_info:
music_info_list.append(music_info)
# use a set to avoid duplicates
user_rec_music_set = set()
# try adding recommendations in order of: title → artist → genre
for field in ("title", "artist", "genre"):
await self._add_recommendations(
field, user_rec_music_set, music_info_list, rec_limit
)
if len(user_rec_music_set) >= rec_limit:
break
user_rec_music_list = [
{
"musicId": 1, # no idea
# recMusicList is a semi colon-separated list of music IDs and their order comma separated
# for some reason, not all music ids are shown in game?!
"recMusicList": ";".join(
f"{music_id},{index + 1}"
for index, music_id in enumerate(user_rec_music_set)
),
},
]
return {
"length": len(user_rec_music_list),
"userRecMusicList": user_rec_music_list,
}
async def handle_get_user_rec_rating_api_request(self, data: Dict) -> Dict:
class GetUserRecRatingApi:
class UserRecRating:
ratingMin: int
ratingMax: int
# same as recMusicList in get_user_rec_music_api_request
recMusicList: str
length: int
userRecRatingList: list[UserRecRating]
user_id = data["userId"]
user_rec_rating_list = []
return {
"length": len(user_rec_rating_list),
"userRecRatingList": user_rec_rating_list,
}
async def _add_recommendations(
self,
field: str,
user_rec_music_set: Set[int],
music_info_list: List[Dict],
limit: int = 25,
) -> None:
"""
Adds music recommendations based on a specific metadata field (title/artist/genre),
excluding music IDs already in the user's recent ratings and recommendations.
"""
# Collect all existing songId to exclude from recommendations
existing_music_ids = {info["songId"] for info in music_info_list}
for music_info in music_info_list:
if len(user_rec_music_set) >= limit:
break
metadata_value = music_info[field]
if not metadata_value:
continue
recs = await self.data.static.get_music_by_metadata(
**{field: metadata_value}
)
for rec in recs or []:
song_id = rec["songId"]
# skip if the song is already in the user's recent ratings
# or if the song is already in the user's recommendations
if (
len(user_rec_music_set) >= limit
or song_id in existing_music_ids
or song_id in user_rec_music_set
):
continue
user_rec_music_set.add(song_id)