Initial restarted commit

API_TitleServer.py

@@ -0,0 +1,170 @@
+# 舞萌DX 2024
+# 标题服务器通讯实现
+
+import zlib
+import hashlib
+import requests
+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 Static_Settings import *
+
+# 舞萌DX 2024
+AesKey = "n7bx6:@Fg_:2;5E89Phy7AyIcpxEQ:R@"
+AesIV = ";;KjR1C3hgB1ovXa"
+ObfuscateParam = "BEs2D5vW"
+
+class WahlapServerBoomedError(Exception):
+    pass
+    
+class Request500Error(Exception):
+    pass
+
+class aes_pkcs7(object):
+    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 SDGBApiHash(api):
+    return hashlib.md5((api+"MaimaiChn"+ObfuscateParam).encode()).hexdigest()
+
+def apiSDGB(data:str, useApi, agentExtraData, maxRetries=5):
+    '''
+    舞萌DX 2024 API 通讯用函数
+    :param data: 请求数据
+    :param useApi: 使用的 API
+    :param agentExtraData: UA 附加信息，机台相关则为狗号（如A63E01E9564），用户相关则为 UID
+    :param maxRetry: 最大重试次数, 默认为 3
+    '''
+
+    # 历史遗留代码有时候会传入 int，故先全部转 str
+    agentExtra = str(agentExtraData)
+
+    # 编码好请求，准备发送
+    aes = aes_pkcs7(AesKey,AesIV)
+    data = data
+    data_enc = aes.encrypt(data)
+    data_def = zlib.compress(data_enc)
+    requests.packages.urllib3.disable_warnings()
+    endpoint = "https://maimai-gm.wahlap.com:42081/Maimai2Servlet/"
+
+    logger.debug("TitleServer Request Start: "+ str(useApi)+" , Data: "+str(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.")
+            
+            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
+            # 解压成功，解密请求并返回
+            resultResponse = unpad(aes.decrypt(responseDecompressed), 16).decode()
+            logger.info("TitleServer Response OK!")
+            logger.debug("TitleServer Response: " + str(resultResponse))
+            return resultResponse
+        
+        # 除了 404 和 500 之外的错误重试
+        except Request500Error:
+            raise Request500Error("500，请求格式错误")
+        except NotImplementedError:
+            raise NotImplementedError("请求未知错误")
+        except Exception as e:
+            logger.warning(f"Request Failed! Will now retry.. {e}")
+            retries += 1
+            time.sleep(4)
+    else:
+        # 重试次数用尽，WahlapServerBoomedError
+        raise WahlapServerBoomedError("重试多次仍然不能成功请求")
+    
+
+def calcSpecialNumber():
+    '''使用 c_int32 实现的 SpecialNumber 算法'''
+    rng = random.SystemRandom()
+    num2 = rng.randint(1, 1037933) * 2069
+    num2 += 0x400
+    num2 = c_int32(num2).value
+    result = c_int32(0)
+    for _ in range(32):
+        result.value <<= 1
+        result.value += num2 & 1
+        num2 >>= 1
+    return c_int32(result.value).value
+
+
+def calcSpecialNumber2():
+    '''实验性替代 SpecialNumber 算法'''
+    max = 1037933
+    num2 = random.randint(1, max) * 2069
+
+    num2 += 1024  # specialnum
+    num3 = 0
+    for i in range(0, 32):
+        num3 <<= 1
+        num3 += num2 % 2
+        num2 >>= 1
+
+    return num3