maimaiDX-Api/Best50_To_Diving_Fish.py

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)