1
0
forked from Hay1tsme/artemis
artemis/titles/cxb/base.py

548 lines
20 KiB
Python
Raw Normal View History

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
2023-03-09 16:38:58 +00:00
class CxbBase:
def __init__(self, cfg: CoreConfig, game_cfg: CxbConfig) -> None:
2023-03-09 16:38:58 +00:00
self.config = cfg # Config file
self.game_config = game_cfg
2023-03-09 16:38:58 +00:00
self.data = CxbData(cfg) # Database
self.game = CxbConstants.GAME_CODE
self.logger = logging.getLogger("cxb")
self.version = CxbConstants.VER_CROSSBEATS_REV
2023-03-09 16:38:58 +00:00
def handle_action_rpreq_request(self, data: Dict) -> Dict:
2023-03-09 16:38:58 +00:00
return {}
def handle_action_hitreq_request(self, data: Dict) -> Dict:
2023-03-09 16:38:58 +00:00
return {"data": []}
def handle_auth_usercheck_request(self, data: Dict) -> Dict:
2023-03-09 16:38:58 +00:00
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")
2023-03-09 16:38:58 +00:00
return {"exist": "true", "logout": "true"}
self.logger.info(f"No profile for aime id {data['usercheck']['authid']}")
2023-03-09 16:38:58 +00:00
return {"exist": "false", "logout": "true"}
def handle_auth_entry_request(self, data: Dict) -> Dict:
self.logger.info(f"New profile for {data['entry']['authid']}")
2023-03-09 16:38:58 +00:00
return {"token": data["entry"]["authid"], "uid": data["entry"]["authid"]}
def handle_auth_login_request(self, data: Dict) -> Dict:
2023-03-09 16:38:58 +00:00
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']}")
2023-03-09 16:38:58 +00:00
return {"token": data["login"]["authid"], "uid": data["login"]["authid"]}
self.logger.warn(f"User {data['login']['authid']} does not have a profile")
2023-03-09 16:38:58 +00:00
return {}
def handle_action_loadrange_request(self, data: Dict) -> Dict:
2023-03-09 16:38:58 +00:00
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)
2023-03-09 16:38:58 +00:00
songs = self.data.score.get_best_scores(uid)
data1 = []
index = []
versionindex = []
2023-03-09 16:38:58 +00:00
for profile_index in profile:
profile_data = profile_index["data"]
if int(range_start) == 800000:
2023-03-09 16:38:58 +00:00
return {"index": range_start, "data": [], "version": 10400}
if not (int(range_start) <= int(profile_index[3]) <= int(range_end)):
continue
2023-03-09 16:38:58 +00:00
# Prevent loading of the coupons within the profile to use the force unlock instead
elif 500 <= int(profile_index[3]) <= 510:
continue
2023-03-09 16:38:58 +00:00
# Prevent loading of songs saved in the profile
elif 100000 <= int(profile_index[3]) <= 110000:
continue
2023-03-09 16:38:58 +00:00
# Prevent loading of the shop list / unlocked titles & icons saved in the profile
elif 200000 <= int(profile_index[3]) <= 210000:
continue
2023-03-09 16:38:58 +00:00
# Prevent loading of stories in the profile
elif 900000 <= int(profile_index[3]) <= 900200:
continue
else:
index.append(profile_index[3])
2023-03-09 16:38:58 +00:00
data1.append(
b64encode(
bytes(json.dumps(profile_data, separators=(",", ":")), "utf-8")
).decode("utf-8")
)
2023-03-09 16:38:58 +00:00
"""
100000 = Songs
200000 = Shop
300000 = Courses
400000 = Events
500000 = Challenges
600000 = Bonuses
700000 = rcLog
800000 = Partners
900000 = Stories
2023-03-09 16:38:58 +00:00
"""
# Coupons
2023-03-09 16:38:58 +00:00
for i in range(500, 510):
index.append(str(i))
couponid = int(i) - 500
2023-03-09 16:38:58 +00:00
dataValue = [
{
"couponId": str(couponid),
"couponNum": "1",
"couponLog": [],
}
]
data1.append(
b64encode(
bytes(json.dumps(dataValue[0], separators=(",", ":")), "utf-8")
).decode("utf-8")
)
# ShopList_Title
2023-03-09 16:38:58 +00:00
for i in range(200000, 201451):
index.append(str(i))
shopid = int(i) - 200000
2023-03-09 16:38:58 +00:00
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
2023-03-09 16:38:58 +00:00
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
2023-03-09 16:38:58 +00:00
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 = []
2023-03-09 16:38:58 +00:00
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:
2023-03-09 16:38:58 +00:00
versionindex.append("10400")
2023-03-09 16:38:58 +00:00
return {"index": index, "data": data1, "version": versionindex}
def handle_action_saveindex_request(self, data: Dict) -> Dict:
2023-03-09 16:38:58 +00:00
save_data = data["saveindex"]
try:
2023-03-09 16:38:58 +00:00
# REV Omnimix Version Fetcher
gameversion = data["saveindex"]["data"][0][2]
self.logger.warning(f"Game Version is {gameversion}")
except:
pass
2023-03-09 16:38:58 +00:00
if "10205" in gameversion:
2023-03-09 16:38:58 +00:00
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
2023-03-09 16:38:58 +00:00
if "mcode" in value[1] and "musicState" in value[1]:
song_json = json.loads(value[1])
2023-03-09 16:38:58 +00:00
songCode = []
2023-03-09 16:38:58 +00:00
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:
2023-03-09 16:38:58 +00:00
self.logger.info(
f"Saving CrossBeats REV Sunrise profile for {data['saveindex']['uid']}"
)
2023-03-09 16:38:58 +00:00
# Sunrise
try:
2023-03-09 16:38:58 +00:00
profileIndex = save_data["index"].index("0")
except:
2023-03-09 16:38:58 +00:00
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:
2023-03-09 16:38:58 +00:00
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"]:
2023-03-09 16:38:58 +00:00
if int(value) in range(100000, 110000):
indexSongList.append(value)
2023-03-09 16:38:58 +00:00
for index, value in enumerate(data["saveindex"]["data"]):
2023-03-09 16:38:58 +00:00
if "mcode" not in value:
continue
2023-03-09 16:38:58 +00:00
if "playedUserId" in value:
continue
2023-03-09 16:38:58 +00:00
data1 = json.loads(value)
songCode = []
2023-03-09 16:38:58 +00:00
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
2023-03-09 16:38:58 +00:00
return {}
def handle_action_sprankreq_request(self, data: Dict) -> Dict:
2023-03-09 16:38:58 +00:00
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:
2023-03-09 16:38:58 +00:00
rankList.append(
{
"sc": [rank["score"], rank["song_id"]],
"rid": rank["rev_id"],
"clear": rank["clear"],
}
)
else:
2023-03-09 16:38:58 +00:00
rankList.append(
{
"sc": [rank["score"]],
"rid": rank["rev_id"],
"clear": rank["clear"],
}
)
return {
"uid": data["sprankreq"]["uid"],
"aid": data["sprankreq"]["aid"],
"rank": rankList,
2023-03-09 16:38:58 +00:00
"rankx": [1, 1, 1],
}
def handle_action_getadv_request(self, data: Dict) -> Dict:
2023-03-09 16:38:58 +00:00
return {"data": [{"r": "1", "i": "100300", "c": "20"}]}
def handle_action_getmsg_request(self, data: Dict) -> Dict:
2023-03-09 16:38:58 +00:00
return {"msgs": []}
def handle_auth_logout_request(self, data: Dict) -> Dict:
2023-03-09 16:38:58 +00:00
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}")
2023-03-09 16:38:58 +00:00
for rid in data["rankreg"]["data"]:
# REV S2
if "clear" in rid:
try:
2023-03-09 16:38:58 +00:00
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:
2023-03-09 16:38:58 +00:00
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:
2023-03-09 16:38:58 +00:00
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:
2023-03-09 16:38:58 +00:00
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:
2023-03-09 16:38:58 +00:00
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"]
2023-03-09 16:38:58 +00:00
if not p:
self.data.item.put_energy(uid, 5)
2023-03-09 16:38:58 +00:00
return {
"class": data1["myClass"],
"granted": "5",
"total": "5",
2023-03-09 16:38:58 +00:00
"threshold": "1000",
}
array = []
2023-03-09 16:38:58 +00:00
newenergy = int(energy) + 5
self.data.item.put_energy(uid, newenergy)
if int(energy) <= 995:
2023-03-09 16:38:58 +00:00
array.append(
{
"class": data1["myClass"],
"granted": "5",
"total": str(energy),
"threshold": "1000",
}
)
else:
2023-03-09 16:38:58 +00:00
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": ""}