maimaiDX-Api/Best50_To_Diving_Fish.py

from API_TitleServer import *
from HelperLogInOut import apiLogin, apiLogout, generateTimestamp
from Config import *
from loguru import logger
from HelperGetUserMusicDetail import getUserFullMusicDetail
from HelperMusicDB import getMusicTitle
import requests
import rapidjson as json

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:
            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
        }
    ]