forked from Hay1tsme/artemis
247 lines
8.8 KiB
Python
247 lines
8.8 KiB
Python
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)
|