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

API_TitleServer.py

@@ -9,7 +9,7 @@ import random
 import time
 from ctypes import c_int32
 from Crypto.Cipher import AES
-from Crypto.Util.Padding import unpad
+from Crypto.Util.Padding import pad, unpad
 from Config import *
 import certifi
 
@@ -28,35 +28,24 @@ class SDGBRequestError(SDGBApiError):
 class SDGBResponseError(SDGBApiError):
     pass
 
-class AES_PKCS7(object):
+class AESPKCS7:
+    # 实现了 maimai 通讯所用的 AES 加密的类
     def __init__(self, key: str, iv: str):
         self.key = key.encode('utf-8')
         self.iv = iv.encode('utf-8')
         self.mode = AES.MODE_CBC
-
-    def encrypt(self, content):
-        cipher = AES.new(self.key, AES.MODE_CBC, self.iv)
-        content_padding = self.pkcs7padding(content)
-        encrypt_bytes = cipher.encrypt(content_padding.encode('utf-8'))
-        return encrypt_bytes
-
-    def decrypt(self, content):
-        cipher = AES.new(self.key, AES.MODE_CBC, self.iv)
-        return cipher.decrypt(content)
-    
-    def pkcs7unpadding(self, text):
-        length = len(text)
-        unpadding = ord(text[length - 1])
-        return text[0:length - unpadding]
-
-    def pkcs7padding(self, text):
-        bs = 16
-        length = len(text)
-        bytes_length = len(text.encode('utf-8'))
-        padding_size = length if (bytes_length == length) else bytes_length
-        padding = bs - padding_size % bs
-        padding_text = chr(padding) * padding
-        return text + padding_text
+    # 加密
+    def encrypt(self, content: bytes) -> bytes:
+        cipher = AES.new(self.key, self.mode, self.iv)
+        content_padded = pad(content, AES.block_size)
+        encrypted_bytes = cipher.encrypt(content_padded)
+        return encrypted_bytes
+    # 解密
+    def decrypt(self, encrypted_content: bytes) -> str:
+        cipher = AES.new(self.key, self.mode, self.iv)
+        decrypted_padded = cipher.decrypt(encrypted_content)
+        decrypted = unpad(decrypted_padded, AES.block_size)
+        return decrypted
 
 def getSDGBApiHash(api):
     # API 的 Hash 的生成
@@ -73,7 +62,7 @@ def apiSDGB(data:str, targetApi:str, userAgentExtraData:str, noLog:bool=False, t
     """
     maxRetries = 3
     agentExtra = str(userAgentExtraData)
-    aes = AES_PKCS7(AesKey, AesIV)
+    aes = AESPKCS7(AesKey, AesIV)
     reqData_encrypted = aes.encrypt(data)
     reqData_deflated = zlib.compress(reqData_encrypted)
     endpoint = "https://maimai-gm.wahlap.com:42081/Maimai2Servlet/"
@@ -134,12 +123,12 @@ def apiSDGB(data:str, targetApi:str, userAgentExtraData:str, noLog:bool=False, t
             # 请求格式错误，不需要重试
             raise SDGBRequestError("请求格式错误")
         except SDGBResponseError as e:
-            # 响应解析错误，重试但是只一次
+            # 响应解析错误，这种有一定可能是我们的问题，所以只重试一次
             logger.warning(f"将重试一次 Resp Err: {e}")
             retries += 2
             time.sleep(2)
         except Exception as e:
-            # 其他错误，重试
+            # 其他错误，重试多次
             logger.warning(f"将开始重试请求. {e}")
             retries += 1
             time.sleep(2)

ActionLoginBonus.py

@@ -147,7 +147,7 @@ def generateLoginBonusList(UserLoginBonusList, generateMode=1):
 
 if __name__ == "__main__":
     # ログインボーナスデータをアップロードする
-    userId = testUid
+    userId = testUid8
     currentLoginTimestamp = generateTimestamp()
     currentLoginResult = apiLogin(currentLoginTimestamp, userId)
     implLoginBonus(userId, currentLoginTimestamp, currentLoginResult, 2)

Best50_To_Diving_Fish.py

@@ -177,9 +177,3 @@ def generateDebugTestScore():
         }
     ]
 
-if __name__ == '__main__':
-    if True:
-        importToken = "b230686d65c90fae594162707304bc66b5322a4591e56870a7ac6fd8e1b5244404e4973d194c4a87a31a8177d2f69150674c2924484808b69a2388ab3d0db7b7"
-        #currentLoginTimestamp = generateTimestamp()
-        #implUserMusicToDivingFish(userId, importToken)
-        implGetUserCurrentDXRating(1723464362)

GetPreview.py

@@ -17,7 +17,7 @@ def apiGetUserPreview(userId, noLog:bool=False) -> str:
 # CLI 示例
 if __name__ == "__main__":
     #userId = input("请输入用户 ID：")
-    userId = testUid2
+    userId = testUid8
     print(apiGetUserPreview(userId))
 
 ###

HelperGetUserThing.py

@@ -3,17 +3,18 @@ from loguru import logger
 import rapidjson as json
 from API_TitleServer import apiSDGB
 
+def implGetUser_(thing:str, userId:int, noLog=False) -> dict:
+    """获取用户某些数据的 API 实现，返回 Dict"""
+    # 获取 Json String
+    result = apiGetUserThing(userId, thing, noLog)
+    # 转换为 Dict
+    userthingDict = json.loads(result)
+    # 返回 Dict
+    return userthingDict
+
 def apiGetUserData(userId:int) -> str:
-    """已弃用，将逐步淘汰"""
-    #logger.info("apiGetUserData 已弃用，将逐步淘汰。")
-    # 构建 Payload
-    data = json.dumps({
-        "userId": userId
-    })
-    # 发送请求
-    userdata_result = apiSDGB(data, "GetUserDataApi", userId)
-    # 返回响应
-    return userdata_result
+    """Now aka of implGetUser_(Data)"""
+    return implGetUser_("Data", userId)
 
 def apiGetUserThing(userId:int, thing:str, noLog=False) -> str:
     """获取用户数据的 API 请求器，返回 Json String"""
@@ -26,11 +27,3 @@ def apiGetUserThing(userId:int, thing:str, noLog=False) -> str:
     # 返回响应
     return userthing_result
 
-def implGetUser_(thing:str, userId:int, noLog=False) -> dict:
-    """获取用户某些数据的 API 实现，返回 Dict"""
-    # 获取 Json String
-    userthing_result = apiGetUserThing(userId, thing, noLog)
-    # 转换为 Dict
-    userthing_dict = json.loads(userthing_result)
-    # 返回 Dict
-    return userthing_dict

HelperMisc.py

@@ -27,11 +27,13 @@ def getHumanReadableLoginErrorCode(loginResult) -> str:
             return "❌ 用户正在上机游玩，请下机后再试，或等待 15 分钟。"
         case 102:
             return "⚠️ 请在微信公众号内点击一次获取新的二维码，然后再试。"
+        case 103:
+            return "❌ 试图登录的账号 UID 无效，请检查账号是否正确。"
         case _:
-            return "❌ 登录失败！请反馈给作者。错误详情："+ loginResult
+            return "❌ 登录失败！这不应该发生，请反馈此问题。错误详情："+ loginResult
 
 def getFriendlyUserData(userId:int) -> str:
-    '''生成一个人类可读的用户数据'''
+    '''生成一个(相对)友好的UserData的人话'''
     userData1 = implGetUser_("Data", userId)
     userData = userData1.get("userData", {})
     userRegion = implGetUser_("Region", userId)
@@ -115,6 +117,7 @@ def getHumanReadableTicketList(jsonString: str):
     return result
 
 def getHumanReadableUserData(userDataJson:str) -> str:
+    '''生成一个人类可读的 UserData 的数据（比较详细）'''
     loadedUserData = json.loads(userDataJson)
     userId = loadedUserData.get("userId")
     userData = loadedUserData.get("userData", {})
@@ -197,7 +200,7 @@ WAHLAP_REGIONS = {
     22: '山东',
     23: '山西',
     24: '四川',
-    25: '（未知）',
+    25: '（未知25）',
     26: '云南',
     27: '浙江',
     28: '广西',

HelperUserAll.py

@@ -89,7 +89,8 @@ def generateUserAllData(userId, currentLoginResult, currentLoginTimestamp, curre
                     "lastGameId": "SDGB",
                     "lastRomVersion": currentUserData2['lastRomVersion'],
                     "lastDataVersion": currentUserData2['lastDataVersion'],
-                    "lastLoginDate": currentLoginResult['lastLoginDate'], # sb
+                    #"lastLoginDate": currentLoginResult['lastLoginDate'], # sb
+                    "lastLoginDate": currentUserData2['lastLoginDate'], # 等待测试
                     "lastPlayDate": datetime.now(pytz.timezone('Asia/Shanghai')).strftime('%Y-%m-%d %H:%M:%S') + '.0',
                     "lastPlayCredit": 1,
                     "lastPlayMode": 0,

Standalone/DummyAuthLiteServer.py

@@ -1,6 +1,9 @@
 # 舞萌DX Auth-Lite 服务器模拟实现
 # 仅实现了 /net/initialize 接口，用来处理 PowerOn
 
+# NOT FINISHED: ONLY A DUMMY IMPLEMENTATION FOR TESTING
+# Contact me if you have more information about this
+
 from fastapi import (
     FastAPI,
     Request

Standalone/Script_GenerateLoginBonusDB.py

@@ -0,0 +1,6 @@
+# Never Gonna Give You Up
+# Never Gonna Let You Down
+# Never Gonna Run Around And Desert You
+# Never Gonna Make You Cry
+# Never Gonna Say Goodbye
+# Never Gonna Tell A Lie And Hurt You

_Special.py

@@ -12,6 +12,7 @@ def implChangeVersionNumber(userId: int, currentLoginTimestamp:int, currentLogin
         "userData": [{
             "lastRomVersion": romVersion,
             "lastDataVersion": dataVersion,
+            "playerRating": 114514
         }],
         "userMusicDetailList": [musicData],
         "isNewMusicDetailList": "1"  #1避免覆盖
@@ -20,7 +21,7 @@ def implChangeVersionNumber(userId: int, currentLoginTimestamp:int, currentLogin
     return result
 
 if __name__ == "__main__":
-    userId = testUid
+    userId = testUid8
     currentLoginTimestamp = generateTimestamp()
     loginResult = apiLogin(currentLoginTimestamp, userId)