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_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.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["courseId1"], unlock_challenge["courseId2"], unlock_challenge["courseId3"], unlock_challenge["courseId4"], unlock_challenge["courseId5"], ] 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.ALWAYS_UNLOCKED.value, # always unlocked "conditionId": -1, }, ) 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)