forked from Dniel97/artemis
548 lines
20 KiB
Python
548 lines
20 KiB
Python
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": ""}
|