Merge branch 'master' of https://github.com/Remik1r3n/maimaiDX-Api

Best50_To_Diving_Fish.py

@@ -0,0 +1,82 @@
+# 非常 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 loguru import logger
+
+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))
+
+def maimaiUserMusicDetailToDivingFish(userMusicDetailList: list) -> list:
+    '''舞萌的 UserMusicDetail 成绩格式转换成水鱼的格式'''
+    divingFishList = []
+    for currentMusicDetail in userMusicDetailList:
+        # 每一个乐曲都判断下是 DX 还是标准
+        if currentMusicDetail['musicId'] >= 10000:
+            notesType = 'DX'
+        else:
+            notesType = 'SD'
+
+        divingFishList.append({
+            'achievements': (currentMusicDetail['achievement'] / 10000), # 水鱼的成绩是 float 而非舞萌的 int
+            'title': (getMusicTitle(currentMusicDetail['musicId'])), # 水鱼用的是歌名而不是 ID（导致不得不用数据库处理转换
+            'type': notesType, # 我不理解这为什么不能在后端判断
+            'level_index': currentMusicDetail['level'],
+            'fc': currentMusicDetail['comboStatus'],
+            'fs': currentMusicDetail['syncStatus'],
+            'dxScore': currentMusicDetail['deluxscoreMax'],
+        })
+    return divingFishList
+
+if __name__ == '__main__':
+    userId = testUid
+    currentLoginTimestamp = generateTimestamp()
+    loginResult = apiLogin(currentLoginTimestamp, userId)
+
+    if loginResult['returnCode'] != 1:
+        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)
+    finally:
+        #logger.error(f"Error: {e}")
+        logger.info(apiLogout(currentLoginTimestamp, userId))

ChangeVersionNumber.py

@@ -0,0 +1,50 @@
+# 改变版本号，实现伪封号和解封号
+
+from loguru import logger
+from Static_Settings 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,
+    "level": 4,
+    "playCount": 1,
+    "achievement": 0,
+    "comboStatus": 0,
+    "syncStatus": 0,
+    "deluxscoreMax": 0,
+    "scoreRank": 0,
+    "extNum1": 0
+})
+    userAllPatches = {
+    "upsertUserAll": {
+        "userData": [{
+            "lastRomVersion": romVersion,
+            "lastDataVersion": dataVersion
+        }],
+        "userMusicDetailList": [musicData],
+        "isNewMusicDetailList": "1"  #1避免覆盖
+    }}
+    logger.info("Changing version number to " + dataVersion + " and " + romVersion)
+    result = implFullPlayAction(userId, currentLoginTimestamp, currentLoginResult, musicData, userAllPatches, True)
+    return result
+
+if __name__ == "__main__":
+    userId = testUid
+    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(apiLogout(currentLoginTimestamp, userId))
+    finally:
+        logger.info(apiLogout(currentLoginTimestamp, userId))
+        #logger.warning("Error")

ForceLogout.py

@@ -0,0 +1,50 @@
+from Static_Settings import *
+from API_TitleServer import *
+from GetPreview import apiGetUserPreview
+from HelperLogInOut import apiLogout
+import json
+from loguru import logger
+import time
+import datetime
+
+def isUserLoggedIn(userId):
+    return json.loads(apiGetUserPreview(userId))['isLogin']
+
+def logOut(userId, Timestamp):
+    try:
+        result = apiLogout(Timestamp, userId)
+        loadedResult = json.loads(result)
+        if loadedResult['returnCode'] == 1:
+            return True
+    except:
+        return False
+
+def forceLogout(userId, beginTimestamp):
+    # 将人类可读的时间转换为 Unix 时间戳
+    original_timestamp = beginTimestamp
+    
+    # 定义时间范围（前后 20 分钟）
+    time_range = 20 * 60  # 20 分钟，以秒为单位
+    
+    # 尝试从原始时间戳开始，逐步向两边扩展
+    for offset in range(0, time_range + 1):
+        # 尝试往前 offset 秒
+        timestamp = original_timestamp - offset
+        logger.info(f"尝试时间戳: {timestamp}")
+        if logout_user(timestamp, userid):
+            if not is_user_logged_in(userid):
+                print(f"用户 {userid} 已成功登出，使用的时间戳: {timestamp}")
+                return
+        # 尝试往后 offset 秒
+        timestamp = original_timestamp + offset
+        if logout_user(timestamp, userid):
+            if not is_user_logged_in(userid):
+                print(f"用户 {userid} 已成功登出，使用的时间戳: {timestamp}")
+                return
+    
+    print(f"无法在前后 20 分钟内登出用户 {userid}")
+
+# 示例使用
+userId = testUid
+human_time = "2023-10-01 12:00:00"  # 用户输入的人类可读时间
+force_logout(userId, human_time)

GenerateMusicDB.py

@@ -0,0 +1,51 @@
+# 感谢伟大的 Diving-Fish 让我被迫直面恐惧写这个逼玩意
+
+import xml.dom.minidom as minidom
+from pathlib import Path
+import json
+from loguru import logger
+
+def makeMusicDBJson():
+    '''
+    从 HDD 的文件来生成 music_db.json
+    推荐的是如果要国服用 那就用国际服的文件来生成
+    免得国服每次更新还要重新生成太麻烦
+    '''
+    # 记得改
+    A000_DIR = Path('/run/media/remik1r3n/软件/maimaiDX_SDGB/Sinmai_Data/StreamingAssets/A000')
+    OPTION_DIR = Path('/run/media/remik1r3n/软件/maimaiDX_SDGB/Sinmai_Data/StreamingAssets')
+
+    music_db: dict[str, dict[str, str | int]] = {}
+    DEST_PATH = Path('./musicDB.json')
+
+    music_folders = [f for f in (A000_DIR / 'music').iterdir() if f.is_dir()]
+    for option_dir in OPTION_DIR.iterdir():
+        if (option_dir / 'music').exists():
+            music_folders.extend([f for f in (option_dir / 'music').iterdir() if f.is_dir()])
+
+    for folder in music_folders:
+        xml_path = (folder / 'Music.xml')
+        if xml_path.exists():
+            xml = minidom.parse(xml_path.as_posix())
+            data = xml.getElementsByTagName('MusicData')[0]
+            music_id = data.getElementsByTagName('name')[0].getElementsByTagName('id')[0].firstChild.data
+            music_name = data.getElementsByTagName('name')[0].getElementsByTagName('str')[0].firstChild.data
+            music_version = data.getElementsByTagName('AddVersion')[0].getElementsByTagName('id')[0].firstChild.data
+            music_db[music_id] = {
+                "name": music_name,
+                "version": int(music_version)
+            }
+    logger.debug(f'Found {len(music_db)} music data')
+    serialized = '{\n'
+    sorted_keys = sorted(music_db.keys(), key=lambda x: int(x))
+    for key in sorted_keys:
+        value = music_db[key]
+        serialized += f'  "{key}": {json.dumps(value, ensure_ascii=False)},\n'
+    serialized = serialized[:-2] + '\n}'
+
+    with open(DEST_PATH, 'w') as f:
+        f.write(serialized)
+
+if __name__ == '__main__':
+    makeMusicDBJson()
+    print('Done.')

HelperFullPlay.py

@@ -43,7 +43,7 @@ def applyUserAllPatches(userAll, patches):
             # 否则直接更新或添加key
             userAll[key] = value
 
-def implFullPlayAction(userId: int, currentLoginTimestamp:int, currentLoginResult, musicData, userAllPatches) -> str:
+def implFullPlayAction(userId: int, currentLoginTimestamp:int, currentLoginResult, musicData, userAllPatches, debugMode=False) -> str:
     '''
     一份完整的上机实现，可以打 patch 来实现各种功能
     需要在外部先登录并传入登录结果
@@ -78,6 +78,12 @@ def implFullPlayAction(userId: int, currentLoginTimestamp:int, currentLoginResul
         currentUserAll = generateFullUserAll(userId, currentLoginResult, currentLoginTimestamp, currentUserData2, currentSpecialNumber)
         # 应用参数里的补丁
         applyUserAllPatches(currentUserAll, userAllPatches)
+        
+        # 调试模式下直接输出数据
+        if debugMode:
+            logger.debug("调试模式：当前 UserAll 数据：" + json.dumps(currentUserAll, indent=4))
+            return
+
         # 建构 Json 数据
         data = json.dumps(currentUserAll)
         # 开始上传 UserAll

MusicDB.py

@@ -0,0 +1,20 @@
+# 感谢伟大的 Diving-Fish 让我被迫直面恐惧写这个逼玩意
+
+import xml.dom.minidom as minidom
+from pathlib import Path
+import rapidjson as json
+from typing import Dict, Union
+
+# 定义音乐数据库的类型注解
+MusicDBType = Dict[int, Dict[str, Union[int, str]]]
+
+# 将 '__all__' 用于模块导出声明
+__all__ = ['musicDB']
+
+# 读取并解析 JSON 文件
+with open('musicDB.json', 'r', encoding='utf-8') as f:
+    # 使用 json.load 直接从文件对象读取 JSON 数据
+    data = json.load(f)
+
+# 将 JSON 数据转换为指定格式的字典
+musicDB: MusicDBType = {int(k): v for k, v in data.items()}