refactor: patcher + fullplay workflow

.gitignore

@@ -1,3 +1,5 @@
 **/__pycache__
 **/settings.py
 **/.DS_Store
+
+*.egg-info

.python-version

@@ -0,0 +1 @@
+3.14

action/GetUserPreviewApi.py

@@ -1,23 +1,29 @@
 import asyncio
 
-from chime import *
+import httpx
+
 from sdgb import MaimaiClient
-from settings import *
+from sdgb.chime import qr_api
+from sdgb.settings import clientId, qrCode
 
 maimai = MaimaiClient()
 
-userId = qr_api(qrCode)['userID']
+qrResp = qr_api(qrCode)
-token = qr_api(qrCode)['token']
+userId = qrResp["userID"]
+token = qrResp["token"]
 
-async def run_workflow(maimai):
+
+async def run_workflow(maimai: MaimaiClient):
     async with httpx.AsyncClient(verify=False) as client:
         data = {
             "userId": userId,
-            "segaIdAuthKey":"",
+            "segaIdAuthKey": "",
             "token": token,
-            "clientId": clientId
+            "clientId": clientId,
         }
         result = await maimai.call_api(client, "GetUserPreviewApi", data, userId)
+    print(result)
+
 
 # 执行入口
 if __name__ == "__main__":

action/PowerOn.py

@@ -1,46 +1,45 @@
+from urllib.parse import unquote
+
 import httpx
 from Crypto.Cipher import AES
 from Crypto.Util.Padding import pad
-from urllib.parse import unquote
+
 
 def enc(key, iv, data):
     cipher = AES.new(key, AES.MODE_CBC, iv)
     encrypted = cipher.encrypt(data)
     return encrypted
 
+
 def dec(key, iv, data):
     de_cipher = AES.new(key, AES.MODE_CBC, iv)
     decrypted = de_cipher.decrypt(data)
     return decrypted
 
+
 def hello():
-    key = bytes([ 47, 63, 106, 111, 43, 34, 76, 38, 92, 67, 114, 57, 40, 61, 107, 71 ])
+    key = bytes([47, 63, 106, 111, 43, 34, 76, 38, 92, 67, 114, 57, 40, 61, 107, 71])
-    #key = bytes([ 45, 97, 53, 55, 85, 88, 52, 121, 57, 47, 104, 40, 73, 109, 65, 81 ])
+    # key = bytes([ 45, 97, 53, 55, 85, 88, 52, 121, 57, 47, 104, 40, 73, 109, 65, 81 ])
-    iv = bytes.fromhex('00000000000000000000000000000000')
+    iv = bytes.fromhex("00000000000000000000000000000000")
-    ua = 'SDGB;Windows/Lite'  
+    ua = "SDGB;Windows/Lite"
-    #ua = 'SDHJ;Windows/Lite'
+    # ua = 'SDHJ;Windows/Lite'
 
     # 构建 payload
-    content = bytes([0] * 16) + b'title_id=SDGB&title_ver=1.52&client_id=A63E01E6149'
+    content = bytes([0] * 16) + b"title_id=SDGB&title_ver=1.52&client_id=A63E01E6149"
     print(f"Content: {content}")
 
-    header = bytes.fromhex('00000000000000000000000000000000')
+    header = bytes.fromhex("00000000000000000000000000000000")
     bytes_data = pad(header + content, 16)
     encrypted = enc(key, iv, bytes_data)
 
     # --- HTTPX 修改部分 ---
-    headers = {
+    headers = {"User-Agent": ua, "Pragma": "DFI"}
-        'User-Agent': ua,
-        'Pragma': 'DFI'
-    }
 
     try:
         # 发送 POST 请求
         # urllib3 的 body 参数在 httpx 中对应 content (用于二进制数据)
         r = httpx.post(
-            'http://at.sys-allnet.cn/net/initialize',
+            "https://at.sys-allnet.cn/net/initialize", content=encrypted, headers=headers
-            content=encrypted, 
-            headers=headers
         )
 
         # 检查响应状态码 (可选，但在 httpx 中推荐)
@@ -54,7 +53,7 @@ def hello():
         if len(resp_data) >= 16:
             decrypted = dec(key, resp_data[:16], resp_data)
             decrypted_bytes = decrypted[16:]
-            decrypted_str = unquote(decrypted_bytes.decode('UTF-8'), 'utf-8')
+            decrypted_str = unquote(decrypted_bytes.decode("UTF-8"), "utf-8")
             print(f"Decrypted: {decrypted_str}")
         else:
             print("Response data too short.")
@@ -62,5 +61,6 @@ def hello():
     except httpx.RequestError as e:
         print(f"An error occurred while requesting: {e}")
 
-if __name__ == '__main__':
+
+if __name__ == "__main__":
     hello()

action/UnlockMusic.py

@@ -0,0 +1,154 @@
+import json
+import asyncio
+import logging
+from typing import Callable
+
+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__)
+
+
+def music_user_all_patcher(
+    musicId: int,
+    unlock_music: bool = True,
+    unlock_master: bool = False,
+    unlock_remaster: bool = False,
+) -> Callable[[dict], None]:
+    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,
+            }
+        )
+
+    def merge(d: dict):
+        d.update(
+            {
+                "userMusicDetailList": [musicData],
+                "isNewMusicDetailList": "1",
+                "userItemList": userItemList,
+                "isNewItemList": len(userItemList) * "1",
+            }
+        )
+
+    return merge
+
+
+async def run_workflow(
+    self: MaimaiClient,
+    user_all_patcher: Callable[[dict], None],
+):
+
+    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
+        )
+
+        user_all_patcher(requestData_UserAll)
+
+        await self.call_api(client, "UpsertUserAllApi", requestData_UserAll, userId)
+
+        # UserLogout
+
+        await self.call_api(client, "UserLogoutApi", requestData_UserLogout, userId)
+
+
+if __name__ == "__main__":
+    patcher = music_user_all_patcher(
+        musicId=834,
+        unlock_music=False,
+        # unlock_master=True,
+        unlock_remaster=True,
+    )
+
+    asyncio.run(
+        run_workflow(
+            maimai,
+            user_all_patcher=patcher,
+        )
+    )

action/UpsertMusic.py

@@ -1,43 +1,53 @@
 import json
 import asyncio
-import httpx
-import time
 import logging
-from sdgb import MaimaiClient
 
-from settings import userId, musicData
+import httpx
-from chime import *
+
-from payload import *
+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,
+    level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
-    format='%(asctime)s - %(levelname)s - %(message)s'
 )
 logger = logging.getLogger(__name__)
 
-async def run_workflow(self):
+
+async def run_workflow(self: MaimaiClient):
 
     async with httpx.AsyncClient(verify=False) as client:
-
         # Preview 探测
 
-        PreviewResponse = json.loads(await self.call_api(client, "GetUserPreviewApi", requestData_UserPreview, userId))
+        PreviewResponse = json.loads(
-        if PreviewResponse["isLogin"] == True:
+            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))
+        LoginResponse = json.loads(
+            await self.call_api(client, "UserLoginApi", requestData_UserLogin, userId)
+        )
         if LoginResponse["returnCode"] != 1:
             logger.error("login failed.")
             return
 
-        loginId = LoginResponse['loginId']
+        loginId = LoginResponse["loginId"]
-        loginDate = LoginResponse['lastLoginDate']
+        loginDate = LoginResponse["lastLoginDate"]
 
         # UserData 等
 
@@ -48,15 +58,19 @@ async def run_workflow(self):
             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),
+            self.call_api(
+                client, "GetUserMissionDataApi", requestData_UserData, userId
+            ),
         ]
 
         GeneralUserInfo = await asyncio.gather(*tasks)
-        await asyncio.sleep(60) # 模拟游戏时间
+        await asyncio.sleep(60)  # 模拟游戏时间
 
         # UserAll
 
-        requestData_UserAll = UserAll_payload(loginId, loginDate, musicData, GeneralUserInfo)
+        requestData_UserAll = UserAll_payload(
+            loginId, loginDate, musicData, GeneralUserInfo
+        )
 
         await self.call_api(client, "UpsertUserAllApi", requestData_UserAll, userId)
 
@@ -64,5 +78,6 @@ async def run_workflow(self):
 
         await self.call_api(client, "UserLogoutApi", requestData_UserLogout, userId)
 
+
 if __name__ == "__main__":
     asyncio.run(run_workflow(maimai))

action/UserLogoutApi.py

@@ -1,12 +1,19 @@
-import json
 import asyncio
+
 import httpx
+
 from sdgb import MaimaiClient
-from settings import *
+from sdgb.settings import (
+    userId,
+    regionId,
+    placeId,
+    clientId,
+)
 
 maimai = MaimaiClient()
 
-async def run_workflow(maimai):
+
+async def run_workflow(maimai: MaimaiClient):
     async with httpx.AsyncClient(verify=False) as client:
         data = {
             "userId": userId,
@@ -15,9 +22,11 @@ async def run_workflow(maimai):
             "placeId": placeId,
             "clientId": clientId,
             "dateTime": 1767000000,
-            "type": 4
+            "type": 4,
         }
         await maimai.call_api(client, "UserLogoutApi", data, userId)
+
+
 # 执行入口
 if __name__ == "__main__":
     asyncio.run(run_workflow(maimai))