artemis/titles/mai2/schema/score.py

from typing import Dict, List, Optional
from sqlalchemy import Table, Column, UniqueConstraint, PrimaryKeyConstraint, and_
from sqlalchemy.types import Integer, String, TIMESTAMP, Boolean, JSON, BigInteger
from sqlalchemy.schema import ForeignKey
from sqlalchemy.sql import func, select
from sqlalchemy.engine import Row
from sqlalchemy.dialects.mysql import insert

from core.data.schema import BaseData, metadata
from core.data import cached

best_score = Table(
    "mai2_score_best",
    metadata,
    Column("id", Integer, primary_key=True, nullable=False),
    Column(
        "user",
        ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"),
        nullable=False,
    ),
    Column("musicId", Integer),
    Column("level", Integer),
    Column("playCount", Integer),
    Column("achievement", Integer),
    Column("comboStatus", Integer),
    Column("syncStatus", Integer),
    Column("deluxscoreMax", Integer),
    Column("scoreRank", Integer),
    Column("extNum1", Integer, server_default="0"),
    UniqueConstraint("user", "musicId", "level", name="mai2_score_best_uk"),
    mysql_charset="utf8mb4",
)

playlog = Table(
    "mai2_playlog",
    metadata,
    Column("id", Integer, primary_key=True, nullable=False),
    Column(
        "user",
        ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"),
        nullable=False,
    ),
    Column("userId", BigInteger),
    Column("orderId", Integer),
    Column("playlogId", BigInteger),
    Column("version", Integer),
    Column("placeId", Integer),
    Column("placeName", String(255)),
    Column("loginDate", BigInteger),
    Column("playDate", String(255)),
    Column("userPlayDate", String(255)),
    Column("type", Integer),
    Column("musicId", Integer),
    Column("level", Integer),
    Column("trackNo", Integer),
    Column("vsMode", Integer),
    Column("vsUserName", String(255)),
    Column("vsStatus", Integer),
    Column("vsUserRating", Integer),
    Column("vsUserAchievement", Integer),
    Column("vsUserGradeRank", Integer),
    Column("vsRank", Integer),
    Column("playerNum", Integer),
    Column("playedUserId1", BigInteger),
    Column("playedUserName1", String(255)),
    Column("playedMusicLevel1", Integer),
    Column("playedUserId2", BigInteger),
    Column("playedUserName2", String(255)),
    Column("playedMusicLevel2", Integer),
    Column("playedUserId3", BigInteger),
    Column("playedUserName3", String(255)),
    Column("playedMusicLevel3", Integer),
    Column("characterId1", Integer),
    Column("characterLevel1", Integer),
    Column("characterAwakening1", Integer),
    Column("characterId2", Integer),
    Column("characterLevel2", Integer),
    Column("characterAwakening2", Integer),
    Column("characterId3", Integer),
    Column("characterLevel3", Integer),
    Column("characterAwakening3", Integer),
    Column("characterId4", Integer),
    Column("characterLevel4", Integer),
    Column("characterAwakening4", Integer),
    Column("characterId5", Integer),
    Column("characterLevel5", Integer),
    Column("characterAwakening5", Integer),
    Column("achievement", Integer),
    Column("deluxscore", Integer),
    Column("scoreRank", Integer),
    Column("maxCombo", Integer),
    Column("totalCombo", Integer),
    Column("maxSync", Integer),
    Column("totalSync", Integer),
    Column("tapCriticalPerfect", Integer),
    Column("tapPerfect", Integer),
    Column("tapGreat", Integer),
    Column("tapGood", Integer),
    Column("tapMiss", Integer),
    Column("holdCriticalPerfect", Integer),
    Column("holdPerfect", Integer),
    Column("holdGreat", Integer),
    Column("holdGood", Integer),
    Column("holdMiss", Integer),
    Column("slideCriticalPerfect", Integer),
    Column("slidePerfect", Integer),
    Column("slideGreat", Integer),
    Column("slideGood", Integer),
    Column("slideMiss", Integer),
    Column("touchCriticalPerfect", Integer),
    Column("touchPerfect", Integer),
    Column("touchGreat", Integer),
    Column("touchGood", Integer),
    Column("touchMiss", Integer),
    Column("breakCriticalPerfect", Integer),
    Column("breakPerfect", Integer),
    Column("breakGreat", Integer),
    Column("breakGood", Integer),
    Column("breakMiss", Integer),
    Column("isTap", Boolean),
    Column("isHold", Boolean),
    Column("isSlide", Boolean),
    Column("isTouch", Boolean),
    Column("isBreak", Boolean),
    Column("isCriticalDisp", Boolean),
    Column("isFastLateDisp", Boolean),
    Column("fastCount", Integer),
    Column("lateCount", Integer),
    Column("isAchieveNewRecord", Boolean),
    Column("isDeluxscoreNewRecord", Boolean),
    Column("comboStatus", Integer),
    Column("syncStatus", Integer),
    Column("isClear", Boolean),
    Column("beforeRating", Integer),
    Column("afterRating", Integer),
    Column("beforeGrade", Integer),
    Column("afterGrade", Integer),
    Column("afterGradeRank", Integer),
    Column("beforeDeluxRating", Integer),
    Column("afterDeluxRating", Integer),
    Column("isPlayTutorial", Boolean),
    Column("isEventMode", Boolean),
    Column("isFreedomMode", Boolean),
    Column("playMode", Integer),
    Column("isNewFree", Boolean),
    Column("extNum1", Integer),
    Column("extNum2", Integer),
    Column("extNum4", Integer, server_default="0"),
    Column("trialPlayAchievement", Integer),
    mysql_charset="utf8mb4",
)

course = Table(
    "mai2_score_course",
    metadata,
    Column("id", Integer, primary_key=True, nullable=False),
    Column(
        "user",
        ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"),
        nullable=False,
    ),
    Column("courseId", Integer),
    Column("isLastClear", Boolean),
    Column("totalRestlife", Integer),
    Column("totalAchievement", Integer),
    Column("totalDeluxscore", Integer),
    Column("playCount", Integer),
    Column("clearDate", String(25)),
    Column("lastPlayDate", String(25)),
    Column("bestAchievement", Integer),
    Column("bestAchievementDate", String(25)),
    Column("bestDeluxscore", Integer),
    Column("bestDeluxscoreDate", String(25)),
    UniqueConstraint("user", "courseId", name="mai2_score_best_uk"),
    mysql_charset="utf8mb4",
)

playlog_old = Table(
    "maimai_playlog",
    metadata,
    Column("id", Integer, primary_key=True, nullable=False),
    Column(
        "user",
        ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"),
        nullable=False,
    ),
    Column("version", Integer),
    # Pop access code
    Column("orderId", Integer),
    Column("sortNumber", Integer),
    Column("placeId", Integer),
    Column("placeName", String(255)),
    Column("country", String(255)),
    Column("regionId", Integer),
    Column("playDate", String(255)),
    Column("userPlayDate", String(255)),
    Column("musicId", Integer),
    Column("level", Integer),
    Column("gameMode", Integer),
    Column("rivalNum", Integer),
    Column("track", Integer),
    Column("eventId", Integer),
    Column("isFreeToPlay", Boolean),
    Column("playerRating", Integer),
    Column("playedUserId1", Integer),
    Column("playedUserId2", Integer),
    Column("playedUserId3", Integer),
    Column("playedUserName1", String(255)),
    Column("playedUserName2", String(255)),
    Column("playedUserName3", String(255)),
    Column("playedMusicLevel1", Integer),
    Column("playedMusicLevel2", Integer),
    Column("playedMusicLevel3", Integer),
    Column("achievement", Integer),
    Column("score", Integer),
    Column("tapScore", Integer),
    Column("holdScore", Integer),
    Column("slideScore", Integer),
    Column("breakScore", Integer),
    Column("syncRate", Integer),
    Column("vsWin", Integer),
    Column("isAllPerfect", Boolean),
    Column("fullCombo", Integer),
    Column("maxFever", Integer),
    Column("maxCombo", Integer),
    Column("tapPerfect", Integer),
    Column("tapGreat", Integer),
    Column("tapGood", Integer),
    Column("tapBad", Integer),
    Column("holdPerfect", Integer),
    Column("holdGreat", Integer),
    Column("holdGood", Integer),
    Column("holdBad", Integer),
    Column("slidePerfect", Integer),
    Column("slideGreat", Integer),
    Column("slideGood", Integer),
    Column("slideBad", Integer),
    Column("breakPerfect", Integer),
    Column("breakGreat", Integer),
    Column("breakGood", Integer),
    Column("breakBad", Integer),
    Column("judgeStyle", Integer),
    Column("isTrackSkip", Boolean),
    Column("isHighScore", Boolean),
    Column("isChallengeTrack", Boolean),
    Column("challengeLife", Integer),
    Column("challengeRemain", Integer),
    Column("isAllPerfectPlus", Integer),
    mysql_charset="utf8mb4",
)

best_score_old = Table(
    "maimai_score_best",
    metadata,
    Column("id", Integer, primary_key=True, nullable=False),
    Column(
        "user",
        ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"),
        nullable=False,
    ),
    Column("musicId", Integer),
    Column("level", Integer),
    Column("playCount", Integer),
    Column("achievement", Integer),
    Column("scoreMax", Integer),
    Column("syncRateMax", Integer),
    Column("isAllPerfect", Boolean),
    Column("isAllPerfectPlus", Integer),
    Column("fullCombo", Integer),
    Column("maxFever", Integer),
    UniqueConstraint("user", "musicId", "level", name="maimai_score_best_uk"),
    mysql_charset="utf8mb4",
)

class Mai2ScoreData(BaseData):
    def put_best_score(self, user_id: int, score_data: Dict, is_dx: bool = True) -> Optional[int]:
        score_data["user"] = user_id

        if is_dx:
            sql = insert(best_score).values(**score_data)
        else:
            sql = insert(best_score_old).values(**score_data)
        conflict = sql.on_duplicate_key_update(**score_data)

        result = self.execute(conflict)
        if result is None:
            self.logger.error(
                f"put_best_score:  Failed to insert best score! user_id {user_id} is_dx {is_dx}"
            )
            return None
        return result.lastrowid

    @cached(2)
    def get_best_scores(self, user_id: int, song_id: int = None, is_dx: bool = True) -> Optional[List[Row]]:
        if is_dx:
            sql = best_score.select(
                and_(
                    best_score.c.user == user_id,
                    (best_score.c.song_id == song_id) if song_id is not None else True,
                )
            )
        else:
            sql = best_score_old.select(
                and_(
                    best_score_old.c.user == user_id,
                    (best_score_old.c.song_id == song_id) if song_id is not None else True,
                )
            )

        result = self.execute(sql)
        if result is None:
            return None
        return result.fetchall()

    def get_best_score(
        self, user_id: int, song_id: int, chart_id: int
    ) -> Optional[Row]:
        sql = best_score.select(
            and_(
                best_score.c.user == user_id,
                best_score.c.song_id == song_id,
                best_score.c.chart_id == chart_id,
            )
        )

        result = self.execute(sql)
        if result is None:
            return None
        return result.fetchone()

    def put_playlog(self, user_id: int, playlog_data: Dict, is_dx: bool = True) -> Optional[int]:
        playlog_data["user"] = user_id

        if is_dx:
            sql = insert(playlog).values(**playlog_data)
        else:
            sql = insert(playlog_old).values(**playlog_data)

        conflict = sql.on_duplicate_key_update(**playlog_data)

        result = self.execute(conflict)
        if result is None:
            self.logger.error(f"put_playlog:  Failed to insert! user_id {user_id} is_dx {is_dx}")
            return None
        return result.lastrowid

    def put_course(self, user_id: int, course_data: Dict) -> Optional[int]:
        course_data["user"] = user_id
        sql = insert(course).values(**course_data)

        conflict = sql.on_duplicate_key_update(**course_data)

        result = self.execute(conflict)
        if result is None:
            self.logger.error(f"put_course:  Failed to insert! user_id {user_id}")
            return None
        return result.lastrowid

    def get_courses(self, user_id: int) -> Optional[List[Row]]:
        sql = course.select(course.c.user == user_id)

        result = self.execute(sql)
        if result is None:
            return None
        return result.fetchone()