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

from Requests to HTTPX (Higher performance, I guess!)

Browse Source
This commit is contained in:
Remik1r3n 2025-02-06 23:30:49 +08:00
parent e73bdcda66
commit a16525c52e
8 changed files with 117 additions and 81 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

131
API_TitleServer.py
View File

@ -3,16 +3,13 @@
import zlib
import hashlib
import requests
import httpx
from loguru import logger
import random
import time
from ctypes import c_int32
from Crypto.Cipher import AES
from Crypto.Util.Padding import unpad
from Config import *
# 舞萌DX 2024
@ -20,10 +17,16 @@ AesKey = "n7bx6:@Fg_:2;5E89Phy7AyIcpxEQ:R@"
AesIV = ";;KjR1C3hgB1ovXa"
ObfuscateParam = "BEs2D5vW"
class WahlapServerBoomedError(Exception):
class SDGBApiError(Exception):
pass
class Request500Error(Exception):
class SDGBMaxRetriesError(SDGBApiError):
pass
class SDGBRequestError(SDGBApiError):
pass
class SDGBResponseError(SDGBApiError):
pass
class AES_PKCS7(object):
@ -56,93 +59,84 @@ class AES_PKCS7(object):
padding_text = chr(padding) * padding
return text + padding_text
def SDGBApiHash(api):
def getSDGBApiHash(api):
return hashlib.md5((api+"MaimaiChn"+ObfuscateParam).encode()).hexdigest()
def apiSDGB(data:str, useApi, agentExtraData, noLog=False):
def apiSDGB(data:str, targetApi:str, userAgentExtraData:str, noLog:bool=False, timeout:int=5):
"""
舞萌DX 2024 API 通讯用函数
:param data: 请求数据
:param useApi: 使用的 API
:param agentExtraData: UA 附加信息，机台相关则为狗号（如A63E01E9564），用户相关则为 UID
:param targetApi: 使用的 API
:param userAgentExtraData: UA 附加信息，机台相关则为狗号（如A63E01E9564），用户相关则为 UID
:param noLog: 是否不记录日志
"""
maxRetries = 3
# 历史遗留代码有时候会传入 int，故先全部转 str
agentExtra = str(agentExtraData)
# 编码好请求，准备发送
agentExtra = str(userAgentExtraData)
aes = AES_PKCS7(AesKey, AesIV)
data = data
data_enc = aes.encrypt(data)
data_def = zlib.compress(data_enc)
requests.packages.urllib3.disable_warnings()
reqData_encrypted = aes.encrypt(data)
reqData_deflated = zlib.compress(reqData_encrypted)
endpoint = "https://maimai-gm.wahlap.com:42081/Maimai2Servlet/"
if not noLog:
logger.debug("TitleServer Request Start: "+ str(useApi)+" , Data: "+str(data))
logger.debug(f"开始请求 {targetApi}，以 {data}")
retries = 0
while retries < maxRetries:
try:
# 发送请求
responseRaw = requests.post(endpoint + SDGBApiHash(useApi), headers={
"User-Agent": f"{SDGBApiHash(useApi)}#{agentExtra}",
"Content-Type": "application/json",
"Mai-Encoding": "1.40",
"Accept-Encoding": "",
"Charset": "UTF-8",
"Content-Encoding": "deflate",
"Expect": "100-continue"
}, data=data_def, verify=False)
logger.debug("TitleServer Request Sent.")
response = httpx.post(
url=endpoint + getSDGBApiHash(targetApi),
headers={
"User-Agent": f"{getSDGBApiHash(targetApi)}#{agentExtra}",
"Content-Type": "application/json",
"Mai-Encoding": "1.40",
"Accept-Encoding": "",
"Charset": "UTF-8",
"Content-Encoding": "deflate",
"Expect": "100-continue"
},
content=reqData_deflated,
verify=False,
timeout=timeout
)
logger.info(f"{targetApi} 请求结果: {response.status_code}")
if response.status_code == 200:
logger.debug("200 OK!")
else:
errorMessage = f"请求失败: {response.status_code}"
logger.error(errorMessage)
raise SDGBRequestError(errorMessage)
responseRAWContent = response.content
logger.debug("TitleServer Response Code: " + str(responseRaw.status_code))
# 如果是 404 或 500，直接抛出异常，不再继续重试
match responseRaw.status_code:
case 200:
logger.debug("Request 200 OK!")
case 404:
logger.error(f"Request 404! ")
raise NotImplementedError
case 500:
logger.error(f"Request Failed! 500!!!! ")
raise Request500Error
case _:
logger.error(f"Request Failed! {responseRaw.status_code}")
raise NotImplementedError
responseContent = responseRaw.content
# 尝试解压请求
try:
responseDecompressed = zlib.decompress(responseContent)
logger.debug("Successfully decompressed response.")
except zlib.error as e:
logger.warning(f"RAW Response: {responseContent}")
logger.warning(f"Wahlap Server Boomed! Will now retry.{e}")
retries += 1
time.sleep(4) # 休眠4秒后重试
continue
# 解压成功，解密请求并返回
responseDecompressed = zlib.decompress(responseRAWContent)
logger.debug("成功解压响应！")
except zlib.error:
logger.warning(f"无法解压，得到的原始响应: {responseRAWContent}")
raise SDGBResponseError("Decompression failed")
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))
logger.debug(f"响应: {resultResponse}")
return resultResponse
# 除了 404 和 500 之外的错误重试
except Request500Error:
raise Request500Error("500，请求格式错误")
except NotImplementedError:
raise NotImplementedError("请求未知错误")
# 异常处理
except SDGBRequestError as e:
# 请求格式错误，不需要重试
raise SDGBRequestError("请求格式错误")
except SDGBResponseError as e:
# 响应解析错误，重试但是只一次
logger.warning(f"Will now retry. {e}")
retries += 2
time.sleep(2)
except Exception as e:
logger.warning(f"Request Failed! Will now retry.. {e}")
# 其他错误，重试
logger.warning(f"Will now retry. {e}")
retries += 1
time.sleep(3)
else:
# 重试次数用尽，WahlapServerBoomedError
raise WahlapServerBoomedError("重试多次仍然不能成功请求")
raise SDGBApiError("Multiple retries failed to make a successful request")
def calcSpecialNumber():
"""使用 c_int32 实现的 SpecialNumber 算法"""
@ -157,7 +151,6 @@ def calcSpecialNumber():
num2 >>= 1
return c_int32(result.value).value
def calcSpecialNumber2():
"""实验性替代 SpecialNumber 算法"""
max = 1037933

1
ActionLoginBonus.py
View File

@ -12,7 +12,6 @@ from HelperFullPlay import implFullPlayAction
class NoSelectedBonusError(Exception):
pass
def apiQueryLoginBonus(userId:int) -> str:
"""ログインボーナスを取得する API"""
data = json.dumps({

8
Best50_To_Diving_Fish.py
View File

@ -102,12 +102,12 @@ def isVaildFishToken(importToken:str):
def implUserMusicToDivingFish(userId:int, fishImportToken:str):
'''上传所有成绩到水鱼的参考实现'''
logger.info("Start to upload user music detail to DivingFish")
logger.info("开始上传舞萌成绩到水鱼查分器！")
userFullMusicDetailList = getUserFullMusicDetail(userId)
logger.info("Got UserData, Convert to Fish Format")
logger.info("成功得到成绩！转换成水鱼格式..")
divingFishData = maimaiUserMusicDetailToDivingFishFormat(userFullMusicDetailList)
logger.info("Convert OK. Start to Update Fish Records")
updateFishRecords(fishImportToken, divingFishData)
logger.info("转换成功！开始上传水鱼..")
return updateFishRecords(fishImportToken, divingFishData)
if __name__ == '__main__':
if True:

7
ChargeTicket.py
View File

@ -60,3 +60,10 @@ def implBuyTicket(userId:int, ticketType:int):
getTicketResponseStr = apiBuyTicket(userId, ticketType, ticketType-1, playerRating, playCount)
# 返回结果
return getTicketResponseStr
if __name__ == "__main__":
userId = testUid2
ticketType = 3
print(implBuyTicket(userId, ticketType))
print(apiQueryTicket(userId))

8
HelperFullPlay.py
View File

@ -2,7 +2,7 @@ import json
from loguru import logger
from Config import *
from API_TitleServer import apiSDGB, calcSpecialNumber, WahlapServerBoomedError, Request500Error
from API_TitleServer import *
from HelperGetUserThing import implGetUser_
from HelperUploadUserPlayLog import apiUploadUserPlaylog
from HelperUserAll import generateFullUserAll
@ -80,16 +80,16 @@ def implFullPlayAction(userId: int, currentLoginTimestamp:int, currentLoginResul
# 开始上传 UserAll
try:
currentUserAllResult = json.loads(apiSDGB(data, "UpsertUserAllApi", userId))
except Request500Error:
except SDGBRequestError:
logger.warning("上传 UserAll 出现 500. 重建数据.")
retries += 1
continue
except Exception:
raise WahlapServerBoomedError("邪门错误")
raise SDGBApiError("邪门错误")
# 成功上传后退出循环
break
else: # 重试次数超过3次
raise Request500Error("多次尝试后仍无法成功上传 UserAll")
raise SDGBRequestError
logger.info("上机：结果："+ str(currentUserAllResult))
return currentUserAllResult

2
HelperGetUserThing.py
View File

@ -5,7 +5,7 @@ from API_TitleServer import apiSDGB
def apiGetUserData(userId:int) -> str:
"""已弃用，将逐步淘汰"""
logger.info("apiGetUserData 已弃用，将逐步淘汰。")
#logger.info("apiGetUserData 已弃用，将逐步淘汰。")
# 构建 Payload
data = json.dumps({
"userId": userId

2
Standalone/DummyAimeDBServer.py
View File

@ -2,6 +2,8 @@
# 适用于舞萌DX 2024
# 理论可用于 HDD 登号等（这种情况下自行修改 hosts
# SGWCMAID111111111111AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA
## 配置
# 0 返回本地生成的假结果
# 1 原样返回官方服务器的结果

35
_Special.py Normal file
View File

@ -0,0 +1,35 @@
# 纯纯测试用
from loguru import logger
from Config import *
from HelperLogInOut import apiLogin, apiLogout, generateTimestamp
from HelperFullPlay import implFullPlayAction, generateMusicData
def implChangeVersionNumber(userId: int, currentLoginTimestamp:int, currentLoginResult, dataVersion="1.40.09", romVersion="1.41.00") -> str:
musicData = generateMusicData()
userAllPatches = {
"upsertUserAll": {
"userData": [{
"playerRating": 114514,
}],
"userMusicDetailList": [musicData],
"isNewMusicDetailList": "1" #1避免覆盖
}}
logger.info("Changing version number to " + dataVersion + " and " + romVersion)
result = implFullPlayAction(userId, currentLoginTimestamp, currentLoginResult, musicData, userAllPatches)
return result
if __name__ == "__main__":
userId = testUid
currentLoginTimestamp = generateTimestamp()
loginResult = apiLogin(currentLoginTimestamp, userId)
if loginResult['returnCode'] != 1:
logger.info("登录失败")
exit()
try:
logger.info(implChangeVersionNumber(userId, currentLoginTimestamp, loginResult, "1.00.00", "1.00.00"))
logger.info(apiLogout(currentLoginTimestamp, userId))
finally:
logger.info(apiLogout(currentLoginTimestamp, userId))
#logger.warning("Error")