MonkeyBusiness/modules/iidx/music.py

import time
from enum import IntEnum

from fastapi import APIRouter, Request, Response
from tinydb import where

from core_common import core_process_request, core_prepare_response, E
from core_database import get_db

import config

router = APIRouter(prefix="/local", tags=["local"])
router.model_whitelist = ["LDJ", "KDZ", "JDZ"]


class ClearFlags(IntEnum):
    NO_PLAY = 0
    FAILED = 1
    ASSIST_CLEAR = 2
    EASY_CLEAR = 3
    CLEAR = 4
    HARD_CLEAR = 5
    EX_HARD_CLEAR = 6
    FULL_COMBO = 7


@router.post("/{gameinfo}/music/getrank")
async def music_getrank(request: Request):
    request_info = await core_process_request(request)
    game_version = request_info["game_version"]

    iidxid = int(request_info["root"][0].attrib["iidxid"])
    play_style = int(request_info["root"][0].attrib["cltype"])

    all_scores = {}
    db = get_db()
    for record in db.table("iidx_scores_best").search(
        (where("music_id") < (game_version + 1) * 1000)
        & (where("iidx_id") == iidxid)
        & (where("play_style") == play_style)
    ):
        music_id = record["music_id"]
        clear_flg = record["clear_flg"]
        if game_version < 20:
            m = str(music_id)
            music_id = int("".join([m[: len(m) - 3], m[-2:]]))
            if clear_flg == ClearFlags.FULL_COMBO and game_version < 19:
                clear_flg = 6
        ex_score = record["ex_score"]
        miss_count = record["miss_count"]
        cid = record["chart_id"]
        if cid in (0, 4, 5, 9):
            continue
        chart_id = cid - 1

        if music_id not in all_scores:
            all_scores[music_id] = {
                0: {"clear_flg": -1, "ex_score": -1, "miss_count": -1},
                1: {"clear_flg": -1, "ex_score": -1, "miss_count": -1},
                2: {"clear_flg": -1, "ex_score": -1, "miss_count": -1},
            }

        all_scores[music_id][chart_id]["clear_flg"] = clear_flg
        all_scores[music_id][chart_id]["ex_score"] = ex_score
        all_scores[music_id][chart_id]["miss_count"] = miss_count

    response = E.response(
        E.music(
            E.style(type=play_style),
            *[
                E.m(
                    [
                        -1,
                        k,
                        *[all_scores[k][d]["clear_flg"] for d in range(3)],
                        *[all_scores[k][d]["ex_score"] for d in range(3)],
                        *[all_scores[k][d]["miss_count"] for d in range(3)],
                    ],
                    __type="s16",
                )
                for k in all_scores
            ]
        )
    )

    assert response is not None

    response_body, response_headers = await core_prepare_response(request, response)
    return Response(content=response_body, headers=response_headers)


@router.post("/{gameinfo}/music/crate")
async def music_crate(request: Request):
    request_info = await core_process_request(request)
    game_version = request_info["game_version"]

    db = get_db()
    all_score_stats = db.table("iidx_score_stats").search(
        (where("music_id") < (game_version + 1) * 1000)
    )

    crate = {}
    fcrate = {}
    for stat in all_score_stats:
        if game_version < 20:
            m = str(stat["music_id"])
            stat["music_id"] = int("".join([m[: len(m) - 3], m[-2:]]))

        if stat["music_id"] not in crate:
            crate[stat["music_id"]] = [101] * 6
        if stat["music_id"] not in fcrate:
            fcrate[stat["music_id"]] = [101] * 6

        if stat["play_style"] == 0:
            old_to_new_adjust = -1
        elif stat["play_style"] == 1:
            old_to_new_adjust = 2

        crate[stat["music_id"]][stat["chart_id"] + old_to_new_adjust] = (
            int(stat["clear_rate"]) // 10
        )
        fcrate[stat["music_id"]][stat["chart_id"] + old_to_new_adjust] = (
            int(stat["fc_rate"]) // 10
        )

    response = E.response(
        E.music(*[E.c(crate[k] + fcrate[k], mid=k, __type="u8") for k in crate])
    )

    response_body, response_headers = await core_prepare_response(request, response)
    return Response(content=response_body, headers=response_headers)


@router.post("/{gameinfo}/music/reg")
async def music_reg(request: Request):
    request_info = await core_process_request(request)
    game_version = request_info["game_version"]

    timestamp = time.time()

    root = request_info["root"][0]

    clear_flg = int(root.attrib["cflg"])
    clid = int(root.attrib["clid"])
    great_num = int(root.attrib["gnum"])
    iidx_id = int(root.attrib["iidxid"])
    miss_num = int(root.attrib["mnum"])
    pgreat_num = int(root.attrib["pgnum"])
    pid = int(root.attrib["pid"])
    ex_score = (pgreat_num * 2) + great_num
    if game_version == 20:
        is_death = int(root.attrib["is_death"])
        music_id = int(root.attrib["mid"])
    else:
        is_death = 1 if clear_flg < ClearFlags.ASSIST_CLEAR else 0
        m = str(root.attrib["mid"])
        music_id = int("0".join([m[: len(m) - 2], m[-2:]]))
        if clear_flg == 6 and game_version < 19:
            clear_flg = ClearFlags.FULL_COMBO

    if clid < 3:
        note_id = clid + 1
        play_style = 0
    else:
        note_id = clid - 2
        play_style = 1

    ghost = root.find("ghost").text

    db = get_db()
    db.table("iidx_scores").insert(
        {
            "timestamp": timestamp,
            "game_version": game_version,
            "iidx_id": iidx_id,
            "pid": pid,
            "clear_flg": clear_flg,
            "is_death": is_death,
            "music_id": music_id,
            "play_style": play_style,
            "chart_id": note_id,
            "pgreat_num": pgreat_num,
            "great_num": great_num,
            "ex_score": ex_score,
            "miss_count": miss_num,
            "ghost": ghost,
        },
    )

    best_score = db.table("iidx_scores_best").get(
        (where("iidx_id") == iidx_id)
        & (where("play_style") == play_style)
        & (where("music_id") == music_id)
        & (where("chart_id") == note_id)
    )
    best_score = {} if best_score is None else best_score

    if clear_flg < ClearFlags.EASY_CLEAR:
        miss_num = -1
    best_miss_count = best_score.get("miss_count", miss_num)
    if best_miss_count == -1:
        miss_count = max(miss_num, best_miss_count)
    elif clear_flg > ClearFlags.ASSIST_CLEAR:
        miss_count = min(miss_num, best_miss_count)
    else:
        miss_count = best_miss_count
    best_ex_score = best_score.get("ex_score", ex_score)
    best_score_data = {
        "game_version": game_version,
        "iidx_id": iidx_id,
        "pid": pid,
        "play_style": play_style,
        "music_id": music_id,
        "chart_id": note_id,
        "miss_count": miss_count,
        "ex_score": max(ex_score, best_ex_score),
        "ghost": ghost if ex_score >= best_ex_score else best_score.get("ghost", ghost),
        "ghost_gauge": best_score.get("ghost_gauge", 0),
        "clear_flg": max(clear_flg, best_score.get("clear_flg", clear_flg)),
        "gauge_type": best_score.get("gauge_type", 0),
    }

    db.table("iidx_scores_best").upsert(
        best_score_data,
        (where("iidx_id") == iidx_id)
        & (where("play_style") == play_style)
        & (where("music_id") == music_id)
        & (where("chart_id") == note_id),
    )

    score_stats = db.table("iidx_score_stats").get(
        (where("music_id") == music_id)
        & (where("play_style") == play_style)
        & (where("chart_id") == note_id)
    )
    score_stats = {} if score_stats is None else score_stats

    score_stats["game_version"] = game_version
    score_stats["play_style"] = play_style
    score_stats["music_id"] = music_id
    score_stats["chart_id"] = note_id
    score_stats["play_count"] = score_stats.get("play_count", 0) + 1
    score_stats["fc_count"] = score_stats.get("fc_count", 0) + (
        1 if clear_flg == ClearFlags.FULL_COMBO else 0
    )
    score_stats["clear_count"] = score_stats.get("clear_count", 0) + (
        1 if clear_flg >= ClearFlags.EASY_CLEAR else 0
    )
    score_stats["fc_rate"] = int(
        (score_stats["fc_count"] / score_stats["play_count"]) * 1000
    )
    score_stats["clear_rate"] = int(
        (score_stats["clear_count"] / score_stats["play_count"]) * 1000
    )

    db.table("iidx_score_stats").upsert(
        score_stats,
        (where("music_id") == music_id)
        & (where("play_style") == play_style)
        & (where("chart_id") == note_id),
    )

    ranklist_data = []
    ranklist_scores = db.table("iidx_scores_best").search(
        (where("play_style") == play_style)
        & (where("music_id") == music_id)
        & (where("chart_id") == note_id)
    )
    ranklist_scores = [] if ranklist_scores is None else ranklist_scores

    ranklist_scores_ranked = []

    for score in ranklist_scores:
        profile = db.table("iidx_profile").get(where("iidx_id") == score["iidx_id"])

        if profile is None or str(game_version) not in profile["version"]:
            continue

        game_profile = profile["version"][str(game_version)]

        ranklist_scores_ranked.append(
            {
                "opname": config.arcade,
                "name": game_profile["djname"],
                "pid": game_profile["region"],
                "body": game_profile.get("body", 0),
                "face": game_profile.get("face", 0),
                "hair": game_profile.get("hair", 0),
                "hand": game_profile.get("hand", 0),
                "head": game_profile.get("head", 0),
                "dgrade": game_profile["grade_double"],
                "sgrade": game_profile["grade_single"],
                "score": score["ex_score"],
                "iidx_id": score["iidx_id"],
                "clflg": score["clear_flg"],
                "myFlg": score["iidx_id"] == iidx_id,
            }
        )

    ranklist_scores_ranked = sorted(
        ranklist_scores_ranked, key=lambda x: (x["clflg"], x["score"]), reverse=True
    )

    myRank = 0
    for rnum, score in enumerate(ranklist_scores_ranked):
        r = E.data(
            rnum=rnum + 1,
            opname=score["opname"],
            name=score["name"],
            pid=score["pid"],
            body=score["body"],
            face=score["face"],
            hair=score["hair"],
            hand=score["hand"],
            head=score["head"],
            dgrade=score["dgrade"],
            sgrade=score["sgrade"],
            score=score["score"],
            iidx_id=score["iidx_id"],
            clflg=score["clflg"],
            myFlg=score["myFlg"],
            achieve=0,
        )
        ranklist_data.append(r)

        if score["myFlg"]:
            myRank = rnum + 1

    response = E.response(
        E.music(
            E.ranklist(*ranklist_data, total_user_num=len(ranklist_data)),
            E.shopdata(rank=myRank),
            clid=clid,
            crate=score_stats["clear_rate"] // 10,
            frate=score_stats["fc_rate"] // 10,
            mid=music_id,
        )
    )

    response_body, response_headers = await core_prepare_response(request, response)
    return Response(content=response_body, headers=response_headers)


@router.post("/{gameinfo}/music/appoint")
async def music_appoint(request: Request):
    request_info = await core_process_request(request)

    iidxid = int(request_info["root"][0].attrib["iidxid"])
    music_id = int(request_info["root"][0].attrib["mid"])
    chart_id = int(request_info["root"][0].attrib["clid"])

    db = get_db()
    record = db.table("iidx_scores_best").get(
        (where("iidx_id") == iidxid)
        & (where("music_id") == music_id)
        & (where("chart_id") == chart_id)
    )

    vals = []
    if record is not None:
        vals.append(
            E.mydata(
                record["ghost"],
                score=record["ex_score"],
                __type="bin",
                __size=len(record["ghost"]) // 2,
            )
        )

    response = E.response(E.music(*vals))

    response_body, response_headers = await core_prepare_response(request, response)
    return Response(content=response_body, headers=response_headers)