Initial commit: Add maimaiDX API web application with AimeDB scanning and logging features

backend/Best50_To_Diving_Fish.py

@@ -0,0 +1,199 @@
+import requests
+from loguru import logger
+
+from HelperGetUserMusicDetail import getUserFullMusicDetail
+from HelperMusicDB import getMusicTitle
+
+
+class divingFishAuthFailError(Exception):
+    pass
+
+
+class divingFishCommError(Exception):
+    pass
+
+
+# 水鱼查分器的 API 地址
+BASE_URL = "https://www.diving-fish.com/api/maimaidxprober"
+
+# 水鱼查分器的成绩状态转换
+COMBO_ID_TO_NAME = ["", "fc", "fcp", "ap", "app"]
+SYNC_ID_TO_NAME = ["", "fs", "fsp", "fsd", "fsdp", "sync"]
+
+
+def apiDivingFish(method: str, apiPath: str, importToken: str, data=None):
+    """水鱼查分器的 API 通讯实现"""
+    headers = {"Import-Token": importToken}
+    if method == "POST":
+        headers["Content-Type"] = "application/json"
+    logger.info(f"水鱼查分器 API 请求：{method} {BASE_URL + apiPath}")
+    if method == "POST":
+        response = requests.post(
+            url=BASE_URL + apiPath,
+            json=data,
+            headers=headers,
+        )
+    elif method == "GET":
+        response = requests.get(
+            url=BASE_URL + apiPath,
+            headers=headers,
+        )
+    elif method == "DELETE":
+        response = requests.delete(
+            url=BASE_URL + apiPath,
+            headers=headers,
+        )
+    else:
+        logger.error(f"未知的请求方法：{method}")
+        raise ValueError(f"未知的请求方法：{method}")
+
+    logger.info(f"水鱼查分器请求结果：{response.status_code}")
+    logger.debug(f"水鱼查分器回应：{response.text}")
+    finalResponseTextDecode = response.text.encode("utf-8").decode("unicode_escape")
+    logger.debug(f"水鱼查分器回应解码后：{finalResponseTextDecode}")
+    match response.status_code:
+        case 200:
+            return response.json()
+        case 500:
+            raise divingFishAuthFailError
+        case _:
+            raise divingFishCommError
+
+
+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):
+    """重置水鱼查分器的用户数据"""
+    return apiDivingFish("DELETE", "/player/delete_records", fishImportToken)
+
+
+def getFishUserInfo(userQQ: int):
+    """按QQ获取水鱼查分器的用户信息"""
+    return apiDivingFish("POST", "/query/player", "", {"qq": userQQ})
+
+
+def maimaiUserMusicDetailToDivingFishFormat(userMusicDetailList) -> list:
+    """舞萌的 UserMusicDetail 成绩格式转换成水鱼的格式"""
+    divingFishList = []
+    for currentMusicDetail in userMusicDetailList:
+        # musicId 大于 100000 属于宴谱，不计入
+        if currentMusicDetail["musicId"] >= 100000:
+            continue
+        # 获得歌名
+        currentMusicTitle = getMusicTitle(currentMusicDetail["musicId"])
+        # 如果数据库里未找到此歌曲
+        if currentMusicTitle == "R_ERR_MUSIC_ID_NOT_IN_DATABASE":
+            logger.warning(f"数据库无此歌曲 跳过: {currentMusicDetail['musicId']}")
+            continue
+        # 每一个乐曲都判断下是 DX 还是标准
+        if currentMusicDetail["musicId"] >= 10000:
+            notesType = "DX"
+        else:
+            notesType = "SD"
+        # 追加进列表
+        try:
+            divingFishList.append(
+                {
+                    "achievements": (
+                        currentMusicDetail["achievement"] / 10000
+                    ),  # 水鱼的成绩是 float 而非舞萌的 int
+                    "title": currentMusicTitle,
+                    "type": notesType,
+                    "level_index": currentMusicDetail["level"],
+                    "fc": COMBO_ID_TO_NAME[currentMusicDetail["comboStatus"]],
+                    "fs": SYNC_ID_TO_NAME[currentMusicDetail["syncStatus"]],
+                    "dxScore": currentMusicDetail["deluxscoreMax"],
+                }
+            )
+        except Exception:
+            logger.error(f"无法将 UserMusic 翻译成水鱼格式: {currentMusicDetail}")
+
+    return divingFishList
+
+
+def isVaildFishToken(importToken: str):
+    """通过尝试获取一次成绩，检查水鱼查分器的 Token 是否有效
+    有效返回 True，无效返回 False"""
+    result = apiDivingFish("GET", "/player/records", importToken)
+    logger.debug(f"水鱼查分器 Token 检查结果：{result}")
+    if result:
+        return True
+    return False
+
+
+def implGetUserCurrentDXRating(userQQ: int):
+    """获取用户当前的 DX RATING"""
+    try:
+        playerData = getFishUserInfo(userQQ)
+        playerRating = playerData["rating"]
+        logger.info(f"用户 {userQQ} 的 DX RATING 是 {playerRating}")
+    except Exception as e:
+        logger.warning(f"无法获取用户 {userQQ} 的 DX RATING: {e}")
+        return False
+    return playerRating
+
+
+def implUserMusicToDivingFish(userId: int, fishImportToken: str):
+    """上传所有成绩到水鱼的参考实现。
+    返回一个 int 的 ErrorCode。
+    0: Success
+    1: Get User Music Fail
+    2: Auth Fail
+    3: Comm Error
+    """
+    logger.info("开始尝试上传舞萌成绩到水鱼查分器！")
+    try:
+        userFullMusicDetailList = getUserFullMusicDetail(userId)
+        logger.info("成功得到成绩！转换成水鱼格式..")
+        divingFishData = maimaiUserMusicDetailToDivingFishFormat(
+            userFullMusicDetailList
+        )
+        logger.info("转换成功！开始上传水鱼..")
+    except Exception as e:
+        logger.error(f"获取成绩失败！{e}")
+        return 1
+    try:
+        updateFishRecords(fishImportToken, divingFishData)
+    except divingFishAuthFailError:
+        logger.error("水鱼查分器认证失败！")
+        return 2
+    except divingFishCommError:
+        logger.error("水鱼查分器通讯失败！")
+        return 3
+
+
+def generateDebugTestScore():
+    """生成测试成绩"""
+    return [
+        {
+            "achievement": 1010000,
+            "comboStatus": 4,
+            "deluxscoreMax": 4026,
+            "level": 4,
+            "musicId": 834,
+            "syncStatus": 4,
+        },
+        {
+            "achievement": 1010000,
+            "comboStatus": 4,
+            "deluxscoreMax": 4200,
+            "level": 4,
+            "musicId": 11663,
+            "syncStatus": 4,
+        },
+    ]
+
+
+if __name__ == "__main__":
+    userId = int(input("userId: "))
+    fishImportToken = input("DivingFish Token: ")
+
+    implUserMusicToDivingFish(userId, fishImportToken)