mirror of
https://github.com/Remik1r3n/maimaiDX-Api.git
synced 2025-05-20 01:07:28 +08:00
UNSTABLE:一大堆细节改进 警告:在测试稳定前可以先不要跟进此commit改动
This commit is contained in:
parent
a87a459ac3
commit
c7716ed04d
@ -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)
|
||||
|
@ -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)
|
||||
|
@ -49,9 +49,11 @@ def apiDivingFish(method:str, apiPath:str, importToken:str, data=None):
|
||||
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 resetFishRecords(fishImportToken:str):
|
||||
@ -141,4 +143,4 @@ if __name__ == '__main__':
|
||||
importToken = testImportToken
|
||||
#currentLoginTimestamp = generateTimestamp()
|
||||
#implUserMusicToDivingFish(userId, importToken)
|
||||
implResetFishUser(importToken)
|
||||
resetFishRecords(importToken)
|
||||
|
@ -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))
|
||||
|
||||
###
|
||||
|
@ -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
|
@ -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: '广西',
|
||||
|
@ -68,7 +68,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,
|
||||
|
@ -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
|
||||
|
6
Standalone/Script_GenerateLoginBonusDB.py
Normal file
6
Standalone/Script_GenerateLoginBonusDB.py
Normal file
@ -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
|
@ -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)
|
||||
|
||||
|
Loading…
x
Reference in New Issue
Block a user