forked from Fragrance/eaquira
139 lines
3.7 KiB
Python
139 lines
3.7 KiB
Python
import json
|
|
import asyncio
|
|
import logging
|
|
|
|
import httpx
|
|
|
|
from sdgb import MaimaiClient
|
|
from sdgb.settings import userId, musicData
|
|
from sdgb.payload import (
|
|
requestData_UserLogin,
|
|
requestData_UserLogout,
|
|
requestData_UserPreview,
|
|
requestData_UserData,
|
|
UserAll_payload,
|
|
)
|
|
|
|
maimai = MaimaiClient()
|
|
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
async def run_workflow(
|
|
self: MaimaiClient,
|
|
musicId: int,
|
|
unlock_music: bool = True,
|
|
unlock_master: bool = False,
|
|
unlock_remaster: bool = False,
|
|
):
|
|
|
|
async with httpx.AsyncClient(verify=False) as client:
|
|
# Preview 探测
|
|
|
|
PreviewResponse = json.loads(
|
|
await self.call_api(
|
|
client, "GetUserPreviewApi", requestData_UserPreview, userId
|
|
)
|
|
)
|
|
if PreviewResponse["isLogin"]:
|
|
logger.error("已在他处登录。")
|
|
return
|
|
|
|
# UserLogin
|
|
|
|
LoginResponse = json.loads(
|
|
await self.call_api(client, "UserLoginApi", requestData_UserLogin, userId)
|
|
)
|
|
if LoginResponse["returnCode"] != 1:
|
|
logger.error("login failed.")
|
|
return
|
|
|
|
loginId = LoginResponse["loginId"]
|
|
loginDate = LoginResponse["lastLoginDate"]
|
|
|
|
# UserData 等
|
|
|
|
tasks = [
|
|
self.call_api(client, "GetUserDataApi", requestData_UserData, userId),
|
|
self.call_api(client, "GetUserExtendApi", requestData_UserData, userId),
|
|
self.call_api(client, "GetUserOptionApi", requestData_UserData, userId),
|
|
self.call_api(client, "GetUserRatingApi", requestData_UserData, userId),
|
|
self.call_api(client, "GetUserChargeApi", requestData_UserData, userId),
|
|
self.call_api(client, "GetUserActivityApi", requestData_UserData, userId),
|
|
self.call_api(
|
|
client, "GetUserMissionDataApi", requestData_UserData, userId
|
|
),
|
|
]
|
|
|
|
GeneralUserInfo = await asyncio.gather(*tasks)
|
|
await asyncio.sleep(60) # 模拟游戏时间
|
|
|
|
# UserAll
|
|
|
|
requestData_UserAll = UserAll_payload(
|
|
loginId, loginDate, musicData, GeneralUserInfo
|
|
)
|
|
|
|
userItemList = []
|
|
|
|
if unlock_music:
|
|
userItemList.append(
|
|
0,
|
|
{
|
|
"itemKind": 5, # MUSIC
|
|
"itemId": musicId,
|
|
"stock": 1,
|
|
"isValid": True,
|
|
},
|
|
)
|
|
|
|
if unlock_master:
|
|
userItemList.append(
|
|
0,
|
|
{
|
|
"itemKind": 6,
|
|
"itemId": musicId, # MASTER
|
|
"stock": 1,
|
|
"isValid": True,
|
|
},
|
|
)
|
|
|
|
if unlock_remaster:
|
|
userItemList.append(
|
|
{
|
|
"itemKind": 7,
|
|
"itemId": musicId, # RE: MASTER
|
|
"stock": 1,
|
|
"isValid": True,
|
|
}
|
|
)
|
|
|
|
requestData_UserAll |= {
|
|
"userMusicDetailList": [musicData],
|
|
"isNewMusicDetailList": "1",
|
|
"userItemList": userItemList,
|
|
"isNewItemList": len(userItemList) * "1",
|
|
}
|
|
|
|
await self.call_api(client, "UpsertUserAllApi", requestData_UserAll, userId)
|
|
|
|
# UserLogout
|
|
|
|
await self.call_api(client, "UserLogoutApi", requestData_UserLogout, userId)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(
|
|
run_workflow(
|
|
maimai,
|
|
musicId=834,
|
|
unlock_music=False,
|
|
# unlock_master=True,
|
|
unlock_remaster=True,
|
|
)
|
|
)
|