import logging import json from decimal import Decimal from base64 import b64encode from typing import Any, Dict from hashlib import md5 from datetime import datetime from core.config import CoreConfig from titles.cxb.config import CxbConfig from titles.cxb.const import CxbConstants from titles.cxb.database import CxbData class CxbBase: def __init__(self, cfg: CoreConfig, game_cfg: CxbConfig) -> None: self.config = cfg # Config file self.game_config = game_cfg self.data = CxbData(cfg) # Database self.game = CxbConstants.GAME_CODE self.logger = logging.getLogger("cxb") self.version = CxbConstants.VER_CROSSBEATS_REV def handle_action_rpreq_request(self, data: Dict) -> Dict: return {} def handle_action_hitreq_request(self, data: Dict) -> Dict: return {"data": []} def handle_auth_usercheck_request(self, data: Dict) -> Dict: profile = self.data.profile.get_profile_index( 0, data["usercheck"]["authid"], self.version ) if profile is not None: self.logger.info(f"User {data['usercheck']['authid']} has CXB profile") return {"exist": "true", "logout": "true"} self.logger.info(f"No profile for aime id {data['usercheck']['authid']}") return {"exist": "false", "logout": "true"} def handle_auth_entry_request(self, data: Dict) -> Dict: self.logger.info(f"New profile for {data['entry']['authid']}") return {"token": data["entry"]["authid"], "uid": data["entry"]["authid"]} def handle_auth_login_request(self, data: Dict) -> Dict: profile = self.data.profile.get_profile_index( 0, data["login"]["authid"], self.version ) if profile is not None: self.logger.info(f"Login user {data['login']['authid']}") return {"token": data["login"]["authid"], "uid": data["login"]["authid"]} self.logger.warn(f"User {data['login']['authid']} does not have a profile") return {} def handle_action_loadrange_request(self, data: Dict) -> Dict: range_start = data["loadrange"]["range"][0] range_end = data["loadrange"]["range"][1] uid = data["loadrange"]["uid"] self.logger.info(f"Load data for {uid}") profile = self.data.profile.get_profile(uid, self.version) songs = self.data.score.get_best_scores(uid) data1 = [] index = [] versionindex = [] for profile_index in profile: profile_data = profile_index["data"] if int(range_start) == 800000: return {"index": range_start, "data": [], "version": 10400} if not (int(range_start) <= int(profile_index[3]) <= int(range_end)): continue # Prevent loading of the coupons within the profile to use the force unlock instead elif 500 <= int(profile_index[3]) <= 510: continue # Prevent loading of songs saved in the profile elif 100000 <= int(profile_index[3]) <= 110000: continue # Prevent loading of the shop list / unlocked titles & icons saved in the profile elif 200000 <= int(profile_index[3]) <= 210000: continue # Prevent loading of stories in the profile elif 900000 <= int(profile_index[3]) <= 900200: continue else: index.append(profile_index[3]) data1.append( b64encode( bytes(json.dumps(profile_data, separators=(",", ":")), "utf-8") ).decode("utf-8") ) """ 100000 = Songs 200000 = Shop 300000 = Courses 400000 = Events 500000 = Challenges 600000 = Bonuses 700000 = rcLog 800000 = Partners 900000 = Stories """ # Coupons for i in range(500, 510): index.append(str(i)) couponid = int(i) - 500 dataValue = [ { "couponId": str(couponid), "couponNum": "1", "couponLog": [], } ] data1.append( b64encode( bytes(json.dumps(dataValue[0], separators=(",", ":")), "utf-8") ).decode("utf-8") ) # ShopList_Title for i in range(200000, 201451): index.append(str(i)) shopid = int(i) - 200000 dataValue = [ { "shopId": shopid, "shopState": "2", "isDisable": "t", "isDeleted": "f", "isSpecialFlag": "f", } ] data1.append( b64encode( bytes(json.dumps(dataValue[0], separators=(",", ":")), "utf-8") ).decode("utf-8") ) # ShopList_Icon for i in range(202000, 202264): index.append(str(i)) shopid = int(i) - 200000 dataValue = [ { "shopId": shopid, "shopState": "2", "isDisable": "t", "isDeleted": "f", "isSpecialFlag": "f", } ] data1.append( b64encode( bytes(json.dumps(dataValue[0], separators=(",", ":")), "utf-8") ).decode("utf-8") ) # Stories for i in range(900000, 900003): index.append(str(i)) storyid = int(i) - 900000 dataValue = [ { "storyId": storyid, "unlockState1": ["t"] * 10, "unlockState2": ["t"] * 10, "unlockState3": ["t"] * 10, "unlockState4": ["t"] * 10, "unlockState5": ["t"] * 10, "unlockState6": ["t"] * 10, "unlockState7": ["t"] * 10, "unlockState8": ["t"] * 10, "unlockState9": ["t"] * 10, "unlockState10": ["t"] * 10, "unlockState11": ["t"] * 10, "unlockState12": ["t"] * 10, "unlockState13": ["t"] * 10, "unlockState14": ["t"] * 10, "unlockState15": ["t"] * 10, "unlockState16": ["t"] * 10, } ] data1.append( b64encode( bytes(json.dumps(dataValue[0], separators=(",", ":")), "utf-8") ).decode("utf-8") ) for song in songs: song_data = song["data"] songCode = [] songCode.append( { "mcode": song_data["mcode"], "musicState": song_data["musicState"], "playCount": song_data["playCount"], "totalScore": song_data["totalScore"], "highScore": song_data["highScore"], "everHighScore": song_data["everHighScore"] if "everHighScore" in song_data else ["0", "0", "0", "0", "0"], "clearRate": song_data["clearRate"], "rankPoint": song_data["rankPoint"], "normalCR": song_data["normalCR"] if "normalCR" in song_data else ["0", "0", "0", "0", "0"], "survivalCR": song_data["survivalCR"] if "survivalCR" in song_data else ["0", "0", "0", "0", "0"], "ultimateCR": song_data["ultimateCR"] if "ultimateCR" in song_data else ["0", "0", "0", "0", "0"], "nohopeCR": song_data["nohopeCR"] if "nohopeCR" in song_data else ["0", "0", "0", "0", "0"], "combo": song_data["combo"], "coupleUserId": song_data["coupleUserId"], "difficulty": song_data["difficulty"], "isFullCombo": song_data["isFullCombo"], "clearGaugeType": song_data["clearGaugeType"], "fieldType": song_data["fieldType"], "gameType": song_data["gameType"], "grade": song_data["grade"], "unlockState": song_data["unlockState"], "extraState": song_data["extraState"], } ) index.append(song_data["index"]) data1.append( b64encode( bytes(json.dumps(songCode[0], separators=(",", ":")), "utf-8") ).decode("utf-8") ) for v in index: try: v_profile = self.data.profile.get_profile_index(0, uid, self.version) v_profile_data = v_profile["data"] versionindex.append(int(v_profile_data["appVersion"])) except: versionindex.append("10400") return {"index": index, "data": data1, "version": versionindex} def handle_action_saveindex_request(self, data: Dict) -> Dict: save_data = data["saveindex"] try: # REV Omnimix Version Fetcher gameversion = data["saveindex"]["data"][0][2] self.logger.warning(f"Game Version is {gameversion}") except: pass if "10205" in gameversion: self.logger.info( f"Saving CrossBeats REV profile for {data['saveindex']['uid']}" ) # Alright.... time to bring the jank code for value in data["saveindex"]["data"]: if "playedUserId" in value[1]: self.data.profile.put_profile( data["saveindex"]["uid"], self.version, value[0], value[1] ) if "mcode" not in value[1]: self.data.profile.put_profile( data["saveindex"]["uid"], self.version, value[0], value[1] ) if "shopId" in value: continue if "mcode" in value[1] and "musicState" in value[1]: song_json = json.loads(value[1]) songCode = [] songCode.append( { "mcode": song_json["mcode"], "musicState": song_json["musicState"], "playCount": song_json["playCount"], "totalScore": song_json["totalScore"], "highScore": song_json["highScore"], "clearRate": song_json["clearRate"], "rankPoint": song_json["rankPoint"], "combo": song_json["combo"], "coupleUserId": song_json["coupleUserId"], "difficulty": song_json["difficulty"], "isFullCombo": song_json["isFullCombo"], "clearGaugeType": song_json["clearGaugeType"], "fieldType": song_json["fieldType"], "gameType": song_json["gameType"], "grade": song_json["grade"], "unlockState": song_json["unlockState"], "extraState": song_json["extraState"], "index": value[0], } ) self.data.score.put_best_score( data["saveindex"]["uid"], song_json["mcode"], self.version, value[0], songCode[0], ) return {} else: self.logger.info( f"Saving CrossBeats REV Sunrise profile for {data['saveindex']['uid']}" ) # Sunrise try: profileIndex = save_data["index"].index("0") except: return {"data": ""} # Maybe profile = json.loads(save_data["data"][profileIndex]) aimeId = profile["aimeId"] i = 0 for index, value in enumerate(data["saveindex"]["data"]): if int(data["saveindex"]["index"][index]) == 101: self.data.profile.put_profile( aimeId, self.version, data["saveindex"]["index"][index], value ) if ( int(data["saveindex"]["index"][index]) >= 700000 and int(data["saveindex"]["index"][index]) <= 701000 ): self.data.profile.put_profile( aimeId, self.version, data["saveindex"]["index"][index], value ) if ( int(data["saveindex"]["index"][index]) >= 500 and int(data["saveindex"]["index"][index]) <= 510 ): self.data.profile.put_profile( aimeId, self.version, data["saveindex"]["index"][index], value ) if "playedUserId" in value: self.data.profile.put_profile( aimeId, self.version, data["saveindex"]["index"][index], json.loads(value), ) if "mcode" not in value and "normalCR" not in value: self.data.profile.put_profile( aimeId, self.version, data["saveindex"]["index"][index], json.loads(value), ) if "shopId" in value: continue # MusicList Index for the profile indexSongList = [] for value in data["saveindex"]["index"]: if int(value) in range(100000, 110000): indexSongList.append(value) for index, value in enumerate(data["saveindex"]["data"]): if "mcode" not in value: continue if "playedUserId" in value: continue data1 = json.loads(value) songCode = [] songCode.append( { "mcode": data1["mcode"], "musicState": data1["musicState"], "playCount": data1["playCount"], "totalScore": data1["totalScore"], "highScore": data1["highScore"], "everHighScore": data1["everHighScore"], "clearRate": data1["clearRate"], "rankPoint": data1["rankPoint"], "normalCR": data1["normalCR"], "survivalCR": data1["survivalCR"], "ultimateCR": data1["ultimateCR"], "nohopeCR": data1["nohopeCR"], "combo": data1["combo"], "coupleUserId": data1["coupleUserId"], "difficulty": data1["difficulty"], "isFullCombo": data1["isFullCombo"], "clearGaugeType": data1["clearGaugeType"], "fieldType": data1["fieldType"], "gameType": data1["gameType"], "grade": data1["grade"], "unlockState": data1["unlockState"], "extraState": data1["extraState"], "index": indexSongList[i], } ) self.data.score.put_best_score( aimeId, data1["mcode"], self.version, indexSongList[i], songCode[0] ) i += 1 return {} def handle_action_sprankreq_request(self, data: Dict) -> Dict: uid = data["sprankreq"]["uid"] self.logger.info(f"Get best rankings for {uid}") p = self.data.score.get_best_rankings(uid) rankList: list[Dict[str, Any]] = [] for rank in p: if rank["song_id"] is not None: rankList.append( { "sc": [rank["score"], rank["song_id"]], "rid": rank["rev_id"], "clear": rank["clear"], } ) else: rankList.append( { "sc": [rank["score"]], "rid": rank["rev_id"], "clear": rank["clear"], } ) return { "uid": data["sprankreq"]["uid"], "aid": data["sprankreq"]["aid"], "rank": rankList, "rankx": [1, 1, 1], } def handle_action_getadv_request(self, data: Dict) -> Dict: return {"data": [{"r": "1", "i": "100300", "c": "20"}]} def handle_action_getmsg_request(self, data: Dict) -> Dict: return {"msgs": []} def handle_auth_logout_request(self, data: Dict) -> Dict: return {"auth": True} def handle_action_rankreg_request(self, data: Dict) -> Dict: uid = data["rankreg"]["uid"] self.logger.info(f"Put {len(data['rankreg']['data'])} rankings for {uid}") for rid in data["rankreg"]["data"]: # REV S2 if "clear" in rid: try: self.data.score.put_ranking( user_id=uid, rev_id=int(rid["rid"]), song_id=int(rid["sc"][1]), score=int(rid["sc"][0]), clear=rid["clear"], ) except: self.data.score.put_ranking( user_id=uid, rev_id=int(rid["rid"]), song_id=0, score=int(rid["sc"][0]), clear=rid["clear"], ) # REV else: try: self.data.score.put_ranking( user_id=uid, rev_id=int(rid["rid"]), song_id=int(rid["sc"][1]), score=int(rid["sc"][0]), clear=0, ) except: self.data.score.put_ranking( user_id=uid, rev_id=int(rid["rid"]), song_id=0, score=int(rid["sc"][0]), clear=0, ) return {} def handle_action_addenergy_request(self, data: Dict) -> Dict: uid = data["addenergy"]["uid"] self.logger.info(f"Add energy to user {uid}") profile = self.data.profile.get_profile_index(0, uid, self.version) data1 = profile["data"] p = self.data.item.get_energy(uid) energy = p["energy"] if not p: self.data.item.put_energy(uid, 5) return { "class": data1["myClass"], "granted": "5", "total": "5", "threshold": "1000", } array = [] newenergy = int(energy) + 5 self.data.item.put_energy(uid, newenergy) if int(energy) <= 995: array.append( { "class": data1["myClass"], "granted": "5", "total": str(energy), "threshold": "1000", } ) else: array.append( { "class": data1["myClass"], "granted": "0", "total": str(energy), "threshold": "1000", } ) return array[0] def handle_action_eventreq_request(self, data: Dict) -> Dict: self.logger.info(data) return {"eventreq": ""} def handle_action_stampreq_request(self, data: Dict) -> Dict: self.logger.info(data) return {"stampreq": ""}