Alicew/maimaiDX-Api
1
0
Fork 0
forked from Kohaku/maimaiDX-Api
Code Pull Requests Activity

Huge Rewrite!

Browse Source
This commit is contained in:
Remik1r3n 2025-02-02 03:17:13 +08:00
parent 3638de178d
commit 90d8b74c45
28 changed files with 1038 additions and 420 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored
View File

@ -1,5 +1,6 @@
# Anti-leak
Private_Static_Settings.py
MyConfig.py
# Byte-compiled / optimized / DLL files
__pycache__/

56
API_AimeDB.py
View File

@ -5,35 +5,39 @@ import json
import re
# 计算 SHA256
def compute_sha256(input_str):
def getSHA256(input_str):
"""SHA256计算"""
return hashlib.sha256(input_str.encode('utf-8')).hexdigest().upper()
# 生成时间戳
def get_timestamp():
def generateSEGATimestamp():
"""SEGA格式的 YYMMDDHHMMSS 时间戳（sb玩意）"""
return time.strftime("%y%m%d%H%M%S", time.localtime())
# 计算认证 key
def calculate_auth_key(time_stamp: str, chip_id: str, auth_key_param: str) -> str:
"""计算 Key"""
return hashlib.sha256((chip_id + time_stamp + auth_key_param).encode("utf-8")).hexdigest().upper()
def calcSEGAAimeDBAuthKey(varString:str, timestamp:str, commonKey:str="XcW5FW4cPArBXEk4vzKz3CIrMuA5EVVW") -> str:
"""计算 SEGA AimeDB 的认证 key"""
return hashlib.sha256((varString + timestamp + commonKey).encode("utf-8")).hexdigest().upper()
def apiAimeDB(qrCode):
"""AimeDB 扫码 API 实现"""
CHIP_ID = "A63E-01E68606624"
COMMON_KEY = "XcW5FW4cPArBXEk4vzKz3CIrMuA5EVVW"
API_URL = "http://ai.sys-allnet.cn/wc_aime/api/get_data"
def apiAimeDB(qr_code, chip_id, auth_key_param, game_id, api_url):
"""AimeDB API 实现"""
# 生成一个时间戳
time_stamp = get_timestamp()
timestamp = generateSEGATimestamp()
# 使用时间戳计算 key
auth_key = calculate_auth_key(time_stamp, chip_id, auth_key_param)
currentKey = calcSEGAAimeDBAuthKey(qrCode, timestamp, COMMON_KEY)
# 构造请求数据
payload = {
"chipID": chip_id,
"openGameID": game_id,
"key": auth_key,
"qrCode": qr_code,
"timestamp": time_stamp
"chipID": CHIP_ID,
"openGameID": "MAID",
"key": currentKey,
"qrCode": qrCode,
"timestamp": timestamp
}
# 输出准备好的请求数据
@ -42,11 +46,11 @@ def apiAimeDB(qr_code, chip_id, auth_key_param, game_id, api_url):
# 发送 POST 请求
headers = {
"Connection": "Keep-Alive",
"Host": api_url.split("//")[-1].split("/")[0],
"Host": API_URL.split("//")[-1].split("/")[0],
"User-Agent": "WC_AIME_LIB",
"Content-Type": "application/json",
}
response = requests.post(api_url, data=json.dumps(payload, separators=(',', ':')), headers=headers)
response = requests.post(API_URL, data=json.dumps(payload, separators=(',', ':')), headers=headers)
# 返回服务器的响应
return response
@ -64,20 +68,19 @@ def isSGWCFormat(input_string: str) -> bool:
return True
def implAimeDB(qrcode_content_full:str) -> str:
def implAimeDB(qrCode:str, isAlreadyFinal:bool=False) -> str:
'''
Aime DB 的请求的参考实现。
提供完整 QRCode 内容，返回响应的字符串（Json格式）
'''
CHIP_ID = "A63E-01E68606624"
AUTH_KEY_PARAM = "XcW5FW4cPArBXEk4vzKz3CIrMuA5EVVW"
GAME_ID = "MAID"
API_URL = "http://ai.sys-allnet.cn/wc_aime/api/get_data"
qr_code_final = qrcode_content_full[20:]
if isAlreadyFinal:
qr_code_final = qrCode
else:
# 提取有效部分(Hash)
qr_code_final = qrCode[20:]
# 发送请求
response = apiAimeDB(qr_code_final, CHIP_ID, AUTH_KEY_PARAM, GAME_ID, API_URL)
response = apiAimeDB(qr_code_final)
# 获得结果
print("implAimeDB: StatusCode is ", response.status_code)
@ -97,13 +100,12 @@ def implGetUID(qr_content:str) -> dict:
# 发送请求并处理响应
try:
result_string = implAimeDB(qr_content)
result_dict = json.loads(result_string)
result = json.loads(implAimeDB(qr_content))
except:
return {'errorID': 60002} # 无法解码 Response 的内容
# 返回结果
return result_dict
return result
if __name__ == "__main__":
userInputQR = input("QRCode: ")

9
API_TitleServer.py
View File

@ -13,7 +13,7 @@ from ctypes import c_int32
from Crypto.Cipher import AES
from Crypto.Util.Padding import unpad
from Static_Settings import *
from Config import *
# 舞萌DX 2024
AesKey = "n7bx6:@Fg_:2;5E89Phy7AyIcpxEQ:R@"
@ -59,14 +59,15 @@ class aes_pkcs7(object):
def SDGBApiHash(api):
return hashlib.md5((api+"MaimaiChn"+ObfuscateParam).encode()).hexdigest()
def apiSDGB(data:str, useApi, agentExtraData, maxRetries=3):
def apiSDGB(data:str, useApi, agentExtraData, noLog=False):
'''
舞萌DX 2024 API 通讯用函数
:param data: 请求数据
:param useApi: 使用的 API
:param agentExtraData: UA 附加信息，机台相关则为狗号（如A63E01E9564），用户相关则为 UID
:param maxRetry: 最大重试次数,
:param noLog: 是否不记录日志
'''
maxRetries = 3
# 历史遗留代码有时候会传入 int，故先全部转 str
agentExtra = str(agentExtraData)
@ -79,6 +80,7 @@ def apiSDGB(data:str, useApi, agentExtraData, maxRetries=3):
requests.packages.urllib3.disable_warnings()
endpoint = "https://maimai-gm.wahlap.com:42081/Maimai2Servlet/"
if not noLog:
logger.debug("TitleServer Request Start: "+ str(useApi)+" , Data: "+str(data))
retries = 0
@ -124,6 +126,7 @@ def apiSDGB(data:str, useApi, agentExtraData, maxRetries=3):
# 解压成功，解密请求并返回
resultResponse = unpad(aes.decrypt(responseDecompressed), 16).decode()
logger.info("TitleServer:" + useApi + " Response: " + str(responseRaw.status_code))
if not noLog:
logger.debug("TitleServer Response: " + str(resultResponse))
return resultResponse

14
ChangeVersionNumber.py → ActionChangeVersion.py
View File

@ -1,13 +1,13 @@
# 改变版本号，实现伪封号和解封号
# 改变版本号，实现伪封号和解封号之类
from loguru import logger
from Static_Settings import *
from Config import *
from HelperLogInOut import apiLogin, apiLogout, generateTimestamp
from HelperFullPlay import implFullPlayAction
def implChangeVersionNumber(userId: int, currentLoginTimestamp:int, currentLoginResult, dataVersion="1.40.09", romVersion="1.41.00") -> str:
musicData= ({
"musicId": 834,
"musicId": 834, # PANDORA PARADOXXX
"level": 4,
"playCount": 1,
"achievement": 0,
@ -27,7 +27,7 @@ def implChangeVersionNumber(userId: int, currentLoginTimestamp:int, currentLogin
"isNewMusicDetailList": "1" #1避免覆盖
}}
logger.info("Changing version number to " + dataVersion + " and " + romVersion)
result = implFullPlayAction(userId, currentLoginTimestamp, currentLoginResult, musicData, userAllPatches, True)
result = implFullPlayAction(userId, currentLoginTimestamp, currentLoginResult, musicData, userAllPatches)
return result
if __name__ == "__main__":
@ -35,15 +35,11 @@ if __name__ == "__main__":
currentLoginTimestamp = generateTimestamp()
loginResult = apiLogin(currentLoginTimestamp, userId)
musicId = 852 #229 is guruguru wash
levelId = 3 #3 is MASTER
if loginResult['returnCode'] != 1:
logger.info("登录失败")
exit()
try:
#logger.info(implDeleteMusicRecord(userId, currentLoginTimestamp, loginResult, musicId, levelId))
logger.info(implChangeVersionNumber(userId, currentLoginTimestamp, loginResult, "1.30.00", "1.30.00"))
logger.info(implChangeVersionNumber(userId, currentLoginTimestamp, loginResult, "1.00.00", "1.00.00"))
logger.info(apiLogout(currentLoginTimestamp, userId))
finally:
logger.info(apiLogout(currentLoginTimestamp, userId))

6
ActionLoginBonus.py
View File

@ -4,13 +4,13 @@
import json
from loguru import logger
from Static_Settings import *
from Config import *
from API_TitleServer import apiSDGB
from HelperLogInOut import apiLogin, apiLogout, generateTimestamp
from HelperFullPlay import implFullPlayAction
def apiQueryLoginBonus(userId:int) -> str:
'''ログインボーナスを取得する API Requestor'''
'''ログインボーナスを取得する API'''
data = json.dumps({
"userId": int(userId),
"nextIndex": 0,
@ -70,7 +70,7 @@ def generateLoginBonusList(UserLoginBonusList, generateMode=1):
# HDDから、ログインボーナスデータを読み込む
# アップデートがある場合、このファイルを更新する必要があります
# 必ず最新のデータを使用してください
with open('loginBonus.json', encoding='utf-8') as file:
with open('./Data/loginBonus.json', encoding='utf-8') as file:
cache = json.load(file)
loginBonusIdList = [item['id'] for item in cache]
logger.debug(f"ログインボーナスIDリスト: {loginBonusIdList}")

16
ActionScoreRecord.py
View File

@ -1,7 +1,7 @@
# 删除和上传成绩
from loguru import logger
from Static_Settings import *
from Config import *
from HelperLogInOut import apiLogin, apiLogout, generateTimestamp
from HelperFullPlay import implFullPlayAction
@ -26,8 +26,12 @@ def implDeleteMusicRecord(userId: int, currentLoginTimestamp:int, currentLoginRe
return result
def implUploadMusicRecord(userId: int, currentLoginTimestamp:int, currentLoginResult, musicId:int, levelId:int, achievement:int, dxScore:int) -> str:
'''VERY EARLY STAGE OF UPLOADING SCORES. DO NOT USE THIS FUNCTION.'''
'''
VERY EARLY STAGE OF UPLOADING SCORES!!!! DO NOT USE THIS!!!!
上传成绩的参考实现。
'''
# 要上传的数据
musicData= ({
"musicId": musicId,
"level": levelId,
@ -42,7 +46,7 @@ def implUploadMusicRecord(userId: int, currentLoginTimestamp:int, currentLoginRe
userAllPatches = {
"upsertUserAll": {
"userMusicDetailList": [musicData],
"isNewMusicDetailList": "0" # 0为编辑，即可删除掉成绩
"isNewMusicDetailList": "1" # 0编辑 1插入
}}
result = implFullPlayAction(userId, currentLoginTimestamp, currentLoginResult, musicData, userAllPatches)
return result
@ -52,15 +56,15 @@ if __name__ == "__main__":
currentLoginTimestamp = generateTimestamp()
loginResult = apiLogin(currentLoginTimestamp, userId)
musicId = 852 #229 is guruguru wash
musicId = 852 #852 is tiamat
levelId = 3 #3 is MASTER
if loginResult['returnCode'] != 1:
logger.info("登录失败")
exit()
try:
#logger.info(implDeleteMusicRecord(userId, currentLoginTimestamp, loginResult, musicId, levelId))
logger.info(implUploadMusicRecord(userId, currentLoginTimestamp, loginResult, musicId, levelId, 1000000, 100))
logger.info(implDeleteMusicRecord(userId, currentLoginTimestamp, loginResult, musicId, levelId))
#logger.info(implUploadMusicRecord(userId, currentLoginTimestamp, loginResult, musicId, levelId, 1000000, 100))
logger.info(apiLogout(currentLoginTimestamp, userId))
finally:
logger.info(apiLogout(currentLoginTimestamp, userId))

4
ActionUnlockVarious.py → ActionUnlockItem.py
View File

@ -2,7 +2,7 @@
from loguru import logger
from Static_Settings import *
from Config import *
from HelperLogInOut import apiLogin, apiLogout, generateTimestamp
from HelperUnlockThing import implUnlockThing
@ -48,7 +48,7 @@ if __name__ == "__main__":
loginResult = apiLogin(currentLoginTimestamp, userId)
# Change you want item ID
wantToUnlockItemId = 11624
wantToUnlockItemId = 11538
if loginResult['returnCode'] != 1:
logger.info("登录失败")

124
Best50_To_Diving_Fish.py
View File

@ -1,55 +1,98 @@
# 非常 All-in Boom 的 B50 更新实现。
from API_TitleServer import *
from HelperLogInOut import apiLogin, apiLogout, generateTimestamp
from Static_Settings import *
import json
from MusicDB import musicDB
from Config import *
from loguru import logger
from HelperGetUserMusicDetail import getUserFullMusicDetail
from HelperMusicDB import getMusicTitle
import requests
def getMusicTitle(musicId: int) -> str:
'''从数据库获取音乐的标题'''
logger.debug(f"查询歌名: {musicId}")
musicInfo = musicDB.get(musicId)
if not musicInfo:
logger.warning(f"数据库里未找到此歌曲: {musicId}")
return "ERR_R_MUSIC_ID_NOT_IN_DATABASE"
musicName = musicInfo.get("name")
logger.debug(f"成功查询到歌名: {musicName}")
return musicName
def getUserMusicDetail(userId:int, nextIndex:int=0, maxCount:int=50) -> dict:
'''获取用户的成绩的API'''
data = json.dumps({
"userId": int(userId),
"nextIndex": nextIndex,
"maxCount": maxCount
})
return json.loads(apiSDGB(data, "GetUserMusicApi", userId))
# 日志设置
if False:
import sys
log_level = "DEBUG"
log_format = "<green>{time:YYYY-MM-DD HH:mm:ss.SSS zz}</green> | <level>{level: <8}</level> | <yellow>Line {line: >4} ({file}):</yellow> <b>{message}</b>"
logger.add(sys.stderr, level=log_level, format=log_format, colorize=True, backtrace=True, diagnose=True)
logger.add("file.log", level=log_level, format=log_format, colorize=False, backtrace=True, diagnose=True)
# 水鱼查分器的 API 地址
BASE_URL = 'https://www.diving-fish.com/api/maimaidxprober'
# 水鱼查分器的成绩状态转换
COMBO_ID_TO_NAME = ['', 'fc', 'fcp', 'ap', 'app']
SYNC_ID_TO_NAME = ['', 'fs', 'fsp', 'fsd', 'fsdp', 'sync']
def apiDivingFish(method:str, apiPath:str, importToken:str, data:dict=None) -> dict:
'''水鱼查分器的 API 通讯实现'''
headers = {
"Import-Token": importToken
}
if method == 'POST':
headers['Content-Type'] = 'application/json'
logger.info(f'水鱼查分器 API 请求：{method} {BASE_URL + apiPath}')
if method == 'POST':
response = requests.post(
url=BASE_URL + apiPath,
json=data,
headers=headers,
)
elif method == 'GET':
response = requests.get(
url=BASE_URL + apiPath,
headers=headers,
)
else:
raise NotImplementedError
logger.info(f'水鱼查分器请求结果：{response.status_code} {response.text}')
if response.status_code != 200:
return False
return response.json()
def getFishRecords(importToken: str) -> dict:
return apiDivingFish('GET', '/player/records', importToken)
def updateFishRecords(importToken: str, records: list[dict]) -> dict:
return apiDivingFish('POST', '/player/update_records', importToken, records)
def maimaiUserMusicDetailToDivingFish(userMusicDetailList: list) -> list:
'''舞萌的 UserMusicDetail 成绩格式转换成水鱼的格式'''
divingFishList = []
for currentMusicDetail in userMusicDetailList:
# musicId 大于 100000 属于宴谱，不计入
if currentMusicDetail['musicId'] >= 100000:
continue
# 获得歌名
currentMusicTitle = getMusicTitle(currentMusicDetail['musicId'])
# 如果数据库里未找到此歌曲
if currentMusicTitle == "R_ERR_MUSIC_ID_NOT_IN_DATABASE":
logger.warning(f"数据库无此歌曲 跳过: {currentMusicDetail['musicId']}")
continue
# 每一个乐曲都判断下是 DX 还是标准
if currentMusicDetail['musicId'] >= 10000:
notesType = 'DX'
else:
notesType = 'SD'
# 追加进列表
try:
divingFishList.append({
'achievements': (currentMusicDetail['achievement'] / 10000), # 水鱼的成绩是 float 而非舞萌的 int
'title': (getMusicTitle(currentMusicDetail['musicId'])), # 水鱼用的是歌名而不是 ID（导致不得不用数据库处理转换
'title': currentMusicTitle,
'type': notesType, # 我不理解这为什么不能在后端判断
'level_index': currentMusicDetail['level'],
'fc': currentMusicDetail['comboStatus'],
'fs': currentMusicDetail['syncStatus'],
'fc': COMBO_ID_TO_NAME[currentMusicDetail['comboStatus']],
'fs': SYNC_ID_TO_NAME[currentMusicDetail['syncStatus']],
'dxScore': currentMusicDetail['deluxscoreMax'],
})
except:
print(currentMusicDetail)
logger.error(f"Error: {currentMusicDetail}")
return divingFishList
if __name__ == '__main__':
if True:
userId = testUid
importToken = testImportToken
currentLoginTimestamp = generateTimestamp()
loginResult = apiLogin(currentLoginTimestamp, userId)
@ -57,26 +100,13 @@ if __name__ == '__main__':
logger.info("登录失败")
exit()
try:
## Begin
userMusicDetailList_current = []
nextIndex:int|None = None # 初始化 nextIndex
while nextIndex != 0 or nextIndex is None: #只要还有nextIndex就一直获取获取
userMusicResponse = getUserMusicDetail(userId, nextIndex or 0)
nextIndex = userMusicResponse['nextIndex']
logger.info(f"NextIndex: {nextIndex}")
for currentMusic in userMusicResponse['userMusicList']:
for currentMusicDetail in currentMusic['userMusicDetailList']:
if not currentMusicDetail['playCount'] > 0:
continue
userMusicDetailList_current.append(currentMusicDetail)
print("---------------")
print(str(userMusicDetailList_current))
## End
#logger.info(apiLogout(currentLoginTimestamp, userId))
logger.warning("Now We Begin To Build DiveFish Data")
divingFishData = maimaiUserMusicDetailToDivingFish(userMusicDetailList_current)
logger.info(divingFishData)
userFullMusicDetailList = getUserFullMusicDetail(userId)
logger.warning("Now We Begin To Build DivingFish Data")
divingFishData = maimaiUserMusicDetailToDivingFish(userFullMusicDetailList)
logger.debug(divingFishData)
logger.warning("Now We Begin To Update DivingFish Data")
updateFishRecords(importToken, divingFishData)
finally:
#logger.error(f"Error: {e}")
logger.info(apiLogout(currentLoginTimestamp, userId))

2
ChargeTicket.py
View File

@ -3,7 +3,7 @@ import json
import pytz
from datetime import datetime, timedelta
from Static_Settings import *
from Config import *
from API_TitleServer import apiSDGB
from HelperGetUserThing import apiGetUserData

2
Static_Settings.py → Config.py
View File

@ -5,4 +5,4 @@ placeName = "赛博时空枣庄市中店"
clientId = "A63E01E9564"
# 日本精工，安全防漏
#from Private_Static_Settings import *
#from MyConfig import *

0
loginBonus.json → Data/loginBonus.json
View File

1014
musicDB.json → Data/musicDB.json
View File

File diff suppressed because it is too large Load Diff

2
ForceLogout.py
View File

@ -1,4 +1,4 @@
from Static_Settings import *
from Config import *
from API_TitleServer import *
from GetPreview import apiGetUserPreview
from HelperLogInOut import apiLogout

0
GenericCLI.py
View File

4
GetPreview.py
View File

@ -2,6 +2,7 @@
import json
from API_TitleServer import apiSDGB
from Config import *
def apiGetUserPreview(userId) -> str:
data = json.dumps({
@ -12,5 +13,6 @@ def apiGetUserPreview(userId) -> str:
# CLI 示例
if __name__ == "__main__":
userId = input("请输入用户 ID：")
#userId = input("请输入用户 ID：")
userId = testUid
print(apiGetUserPreview(userId))

16
HelperFullPlay.py
View File

@ -1,7 +1,7 @@
import json
from loguru import logger
from Static_Settings import *
from Config import *
from API_TitleServer import apiSDGB, calcSpecialNumber, WahlapServerBoomedError, Request500Error
from HelperGetUserThing import implGetUser_
from HelperUploadUserPlayLog import apiUploadUserPlaylog
@ -39,6 +39,15 @@ def applyUserAllPatches(userAll, patches):
if isinstance(value, dict) and key in userAll and isinstance(userAll[key], dict):
# 如果patch的值是字典，并且userAll中对应的key也是字典，递归处理
applyUserAllPatches(userAll[key], value)
elif isinstance(value, list) and key in userAll and isinstance(userAll[key], list):
# 如果值是列表，进行详细的更新处理
for i, patch_item in enumerate(value):
if i < len(userAll[key]) and isinstance(patch_item, dict) and isinstance(userAll[key][i], dict):
# 如果列表项是字典，更新字典中的字段
applyUserAllPatches(userAll[key][i], patch_item)
elif i >= len(userAll[key]):
# 如果patch的列表比userAll的列表长，追加新的元素
userAll[key].append(patch_item)
else:
# 否则直接更新或添加key
userAll[key] = value
@ -81,7 +90,8 @@ def implFullPlayAction(userId: int, currentLoginTimestamp:int, currentLoginResul
# 调试模式下直接输出数据
if debugMode:
logger.debug("调试模式：当前 UserAll 数据：" + json.dumps(currentUserAll, indent=4))
logger.debug("调试模式：构建出的 UserAll 数据：" + json.dumps(currentUserAll, indent=4))
logger.info("Bye!")
return
# 建构 Json 数据
@ -90,7 +100,7 @@ def implFullPlayAction(userId: int, currentLoginTimestamp:int, currentLoginResul
try:
currentUserAllResult = json.loads(apiSDGB(data, "UpsertUserAllApi", userId))
except Request500Error:
logger.warning("500 Error Triggered. Rebuilding data.")
logger.warning("上传 UserAll 出现 500. 重建数据.")
retries += 1
continue
except Exception:

66
HelperGetUserMusicDetail.py Normal file
View File

@ -0,0 +1,66 @@
# 获取用户成绩的各种实现
from API_TitleServer import *
from HelperLogInOut import apiLogin, apiLogout, generateTimestamp
from Config import *
import json
from HelperMusicDB import getMusicTitle
from loguru import logger
import sys
# 日志设置
#log_level = "DEBUG"
#log_format = "<green>{time:YYYY-MM-DD HH:mm:ss.SSS zz}</green> | <level>{level: <8}</level> | <yellow>Line {line: >4} ({file}):</yellow> <b>{message}</b>"
#logger.add(sys.stderr, level=log_level, format=log_format, colorize=True, backtrace=True, diagnose=True)
#logger.add("file.log", level=log_level, format=log_format, colorize=False, backtrace=True, diagnose=True)
def getUserMusicDetail(userId:int, nextIndex:int=0, maxCount:int=50) -> dict:
'''获取用户的成绩的API'''
data = json.dumps({
"userId": int(userId),
"nextIndex": nextIndex,
"maxCount": maxCount
})
return json.loads(apiSDGB(data, "GetUserMusicApi", userId))
def getUserFullMusicDetail(userId: int) -> dict:
'''获取用户的全部成绩'''
userMusicDetailList_current = []
nextIndex:int|None = None # 初始化 nextIndex
while nextIndex != 0 or nextIndex is None: #只要还有nextIndex就一直获取获取
userMusicResponse = getUserMusicDetail(userId, nextIndex or 0)
nextIndex = userMusicResponse['nextIndex']
logger.info(f"NextIndex: {nextIndex}")
for currentMusic in userMusicResponse['userMusicList']:
for currentMusicDetail in currentMusic['userMusicDetailList']:
if not currentMusicDetail['playCount'] > 0:
continue
userMusicDetailList_current.append(currentMusicDetail)
return userMusicDetailList_current
def parseUserFullMusicDetail(userFullMusicDetailList: list) -> dict:
'''解析用户的全部成绩'''
musicDetailList = []
for currentMusicDetail in userFullMusicDetailList:
musicDetailList.append({
'歌名': getMusicTitle(currentMusicDetail['musicId']),
'难度': currentMusicDetail['level'],
'分数': currentMusicDetail['achievement'] / 10000,
'DX分数': currentMusicDetail['deluxscoreMax']
})
return musicDetailList
if __name__ == '__main__':
userId = testUid5
currentLoginTimestamp = generateTimestamp()
loginResult = apiLogin(currentLoginTimestamp, userId)
if loginResult['returnCode'] != 1:
logger.info("登录失败")
exit()
try:
userFullMusicDetailList = getUserFullMusicDetail(userId)
parsedUserFullMusicDetail = parseUserFullMusicDetail(userFullMusicDetailList)
logger.info(parsedUserFullMusicDetail)
finally:
#logger.error(f"Error: {e}")
logger.info(apiLogout(currentLoginTimestamp, userId))

12
HelperGetUserThing.py
View File

@ -5,7 +5,7 @@ from API_TitleServer import apiSDGB
def apiGetUserData(userId:int) -> str:
'''已弃用，将逐步淘汰'''
logger.warning("apiGetUserData 已弃用，将逐步淘汰。")
logger.info("apiGetUserData 已弃用，将逐步淘汰。")
# 构建 Payload
data = json.dumps({
"userId": userId
@ -15,21 +15,21 @@ def apiGetUserData(userId:int) -> str:
# 返回响应
return userdata_result
def apiGetUserThing(userId:int, thing:str) -> str:
def apiGetUserThing(userId:int, thing:str, noLog=False) -> str:
'''获取用户数据的 API 请求器，返回 Json String'''
# 构建 Payload
data = json.dumps({
"userId": userId
})
# 发送请求
userthing_result = apiSDGB(data, "GetUser" + thing + "Api", userId)
userthing_result = apiSDGB(data, "GetUser" + thing + "Api", userId, noLog)
# 返回响应
return userthing_result
def implGetUser_(thing:str, userId:int) -> dict:
'''获取用户数据的 API 实现，返回 Dict'''
def implGetUser_(thing:str, userId:int, noLog=False) -> dict:
'''获取用户某些数据的 API 实现，返回 Dict'''
# 获取 Json String
userthing_result = apiGetUserThing(userId, thing)
userthing_result = apiGetUserThing(userId, thing, noLog)
# 转换为 Dict
userthing_dict = json.loads(userthing_result)
# 返回 Dict

2
HelperLogInOut.py
View File

@ -5,7 +5,7 @@ import json
import time
from loguru import logger
from Static_Settings import *
from Config import *
from API_TitleServer import apiSDGB
def apiLogin(timestamp:int, userId:int) -> dict:

14
HelperMusicDB.py Normal file
View File

@ -0,0 +1,14 @@
from MusicDB import musicDB
from loguru import logger
def getMusicTitle(musicId: int) -> str:
'''从数据库获取音乐的标题'''
#logger.debug(f"查询歌名: {musicId}")
musicInfo = musicDB.get(musicId)
if not musicInfo:
logger.warning(f"数据库里未找到此歌曲: {musicId}")
return "R_ERR_MUSIC_ID_NOT_IN_DATABASE"
musicName = musicInfo.get("name")
#logger.debug(f"成功查询到歌名: {musicName}")
return musicName

4
HelperUnlockThing.py
View File

@ -1,11 +1,11 @@
# 解锁东西的一个通用的助手，不可独立使用
from loguru import logger
from Static_Settings import *
from Config import *
from HelperFullPlay import implFullPlayAction
def implUnlockThing(newUserItemList, userId: int, currentLoginTimestamp:int, currentLoginResult) -> str:
musicData= ({
"musicId": 566, #天火明命
"musicId": 11538, # Amber Chronicle
"level": 0,
"playCount": 1,
"achievement": 0,

2
HelperUploadUserPlayLog.py
View File

@ -8,7 +8,7 @@ from datetime import datetime
from loguru import logger
from API_TitleServer import apiSDGB
from Static_Settings import *
from Config import *
def apiUploadUserPlaylog(userId:int, musicDataToBeUploaded, currentUserData2, loginId:int) -> str:
'''返回 Json String。'''

12
HelperUserAll.py
View File

@ -2,7 +2,7 @@
import pytz
from datetime import datetime
from Static_Settings import *
from Config import *
from HelperGetUserThing import implGetUser_
def generateFullUserAll(userId, currentLoginResult, currentLoginTimestamp, currentUserData2, currentSpecialNumber):
@ -12,11 +12,11 @@ def generateFullUserAll(userId, currentLoginResult, currentLoginTimestamp, curre
currentUserAll = generateUserAllData(userId, currentLoginResult, currentLoginTimestamp, currentUserData2, currentSpecialNumber)
# 然后从服务器取得必要的数据
currentUserExtend = implGetUser_("Extend", userId)
currentUserOption = implGetUser_("Option", userId)
currentUserRating = implGetUser_("Rating", userId)
currentUserActivity = implGetUser_("Activity", userId)
currentUserCharge = implGetUser_("Charge", userId)
currentUserExtend = implGetUser_("Extend", userId, True)
currentUserOption = implGetUser_("Option", userId, True)
currentUserRating = implGetUser_("Rating", userId, True)
currentUserActivity = implGetUser_("Activity", userId, True)
currentUserCharge = implGetUser_("Charge", userId, True)
# 把这些数据都追加进去
currentUserAll['upsertUserAll']['userExtend'] = [currentUserExtend['userExtend']]

6
MusicDB.py
View File

@ -1,7 +1,3 @@
# 感谢伟大的 Diving-Fish 让我被迫直面恐惧写这个逼玩意
import xml.dom.minidom as minidom
from pathlib import Path
import rapidjson as json
from typing import Dict, Union
@ -12,7 +8,7 @@ MusicDBType = Dict[int, Dict[str, Union[int, str]]]
__all__ = ['musicDB']
# 读取并解析 JSON 文件
with open('musicDB.json', 'r', encoding='utf-8') as f:
with open('./Data/musicDB.json', 'r', encoding='utf-8') as f:
# 使用 json.load 直接从文件对象读取 JSON 数据
data = json.load(f)

3
DecryptHDD.py → Standalone/DecryptHDD.py
View File

@ -1,6 +1,7 @@
# 解密从 HDD 抓包得到的数据
# 兼容 PRiSM 和 CN 2024
# 仅用于分析
# 完全 Standalone，不依赖于其他文件
import base64
import zlib

9
GenerateMusicDB.py → Standalone/Script_GenerateMusicDB.py
View File

@ -12,11 +12,11 @@ def makeMusicDBJson():
免得国服每次更新还要重新生成太麻烦
'''
# 记得改
A000_DIR = Path('/run/media/remik1r3n/软件/maimaiDX_SDGB/Sinmai_Data/StreamingAssets/A000')
OPTION_DIR = Path('/run/media/remik1r3n/软件/maimaiDX_SDGB/Sinmai_Data/StreamingAssets')
A000_DIR = Path('H:\PRiSM\Package\Sinmai_Data\StreamingAssets\A000')
OPTION_DIR = Path('H:\PRiSM\Package\Sinmai_Data\StreamingAssets')
music_db: dict[str, dict[str, str | int]] = {}
DEST_PATH = Path('./musicDB.json')
DEST_PATH = Path('../Data/musicDB.json')
music_folders = [f for f in (A000_DIR / 'music').iterdir() if f.is_dir()]
for option_dir in OPTION_DIR.iterdir():
@ -43,9 +43,10 @@ def makeMusicDBJson():
serialized += f' "{key}": {json.dumps(value, ensure_ascii=False)},\n'
serialized = serialized[:-2] + '\n}'
with open(DEST_PATH, 'w') as f:
with open(DEST_PATH, 'w', encoding='utf-8') as f:
f.write(serialized)
if __name__ == '__main__':
makeMusicDBJson()
print('Done.')

10
UI.py → Standalone/UI.py
View File

@ -6,6 +6,12 @@ from PyQt6.QtWidgets import (
)
from PyQt6.QtCore import Qt
# 将当前目录的父目录加入到 sys.path 中
from pathlib import Path
current_dir = Path(__file__).resolve().parent
parent_dir = current_dir.parent
sys.path.append(str(parent_dir))
from API_TitleServer import *
def sendRequest(requestText:str, apiNameText:str, uid:int) -> str:
@ -14,8 +20,10 @@ def sendRequest(requestText:str, apiNameText:str, uid:int) -> str:
data = json.dumps(data)
except:
return "给出的输入不是有效的 JSON"
try:
result = apiSDGB(data, apiNameText, uid)
except Exception as e:
return "请求失败：" + str(e)
return result

0
__init__.py
View File