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.engine.base import Connection from sqlalchemy.schema import ForeignKey from sqlalchemy.engine import Row from sqlalchemy.sql import func, select from sqlalchemy.dialects.mysql import insert from sqlalchemy.sql.expression import exists from core.data.schema import BaseData, metadata course = Table( "chuni_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("classId", Integer), Column("playCount", Integer), Column("scoreMax", Integer), Column("isFullCombo", Boolean), Column("isAllJustice", Boolean), Column("isSuccess", Boolean), Column("scoreRank", Integer), Column("eventId", Integer), Column("lastPlayDate", String(25)), Column("param1", Integer), Column("param2", Integer), Column("param3", Integer), Column("param4", Integer), Column("isClear", Boolean), Column("theoryCount", Integer), Column("orderId", Integer), Column("playerRating", Integer), UniqueConstraint("user", "courseId", name="chuni_score_course_uk"), mysql_charset="utf8mb4", ) best_score = Table( "chuni_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("scoreMax", Integer), Column("resRequestCount", Integer), Column("resAcceptCount", Integer), Column("resSuccessCount", Integer), Column("missCount", Integer), Column("maxComboCount", Integer), Column("isFullCombo", Boolean), Column("isAllJustice", Boolean), Column("isSuccess", Boolean), Column("fullChain", Integer), Column("maxChain", Integer), Column("scoreRank", Integer), Column("isLock", Boolean), Column("ext1", Integer), Column("theoryCount", Integer), UniqueConstraint("user", "musicId", "level", name="chuni_score_best_uk"), mysql_charset="utf8mb4", ) playlog = Table( "chuni_score_playlog", metadata, Column("id", Integer, primary_key=True, nullable=False), Column( "user", ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"), nullable=False, ), Column("orderId", Integer), Column("sortNumber", Integer), Column("placeId", Integer), Column("playDate", String(20)), Column("userPlayDate", String(20)), Column("musicId", Integer), Column("level", Integer), Column("customId", Integer), Column("playedUserId1", Integer), Column("playedUserId2", Integer), Column("playedUserId3", Integer), Column("playedUserName1", String(20)), Column("playedUserName2", String(20)), Column("playedUserName3", String(20)), Column("playedMusicLevel1", Integer), Column("playedMusicLevel2", Integer), Column("playedMusicLevel3", Integer), Column("playedCustom1", Integer), Column("playedCustom2", Integer), Column("playedCustom3", Integer), Column("track", Integer), Column("score", Integer), Column("rank", Integer), Column("maxCombo", Integer), Column("maxChain", Integer), Column("rateTap", Integer), Column("rateHold", Integer), Column("rateSlide", Integer), Column("rateAir", Integer), Column("rateFlick", Integer), Column("judgeGuilty", Integer), Column("judgeAttack", Integer), Column("judgeJustice", Integer), Column("judgeCritical", Integer), Column("eventId", Integer), Column("playerRating", Integer), Column("isNewRecord", Boolean), Column("isFullCombo", Boolean), Column("fullChainKind", Integer), Column("isAllJustice", Boolean), Column("isContinue", Boolean), Column("isFreeToPlay", Boolean), Column("characterId", Integer), Column("skillId", Integer), Column("playKind", Integer), Column("isClear", Boolean), Column("skillLevel", Integer), Column("skillEffect", Integer), Column("placeName", String(255)), Column("isMaimai", Boolean), Column("commonId", Integer), Column("charaIllustId", Integer), Column("romVersion", String(255)), Column("judgeHeaven", Integer), Column("regionId", Integer), Column("machineType", Integer), mysql_charset="utf8mb4" ) class ChuniScoreData(BaseData): def get_courses(self, aime_id: int) -> Optional[Row]: sql = select(course).where(course.c.user == aime_id) result = self.execute(sql) if result is None: return None return result.fetchall() def put_course(self, aime_id: int, course_data: Dict) -> Optional[int]: course_data["user"] = aime_id course_data = self.fix_bools(course_data) sql = insert(course).values(**course_data) conflict = sql.on_duplicate_key_update(**course_data) result = self.execute(conflict) if result is None: return None return result.lastrowid def get_scores(self, aime_id: int) -> Optional[Row]: sql = select(best_score).where(best_score.c.user == aime_id) result = self.execute(sql) if result is None: return None return result.fetchall() def put_score(self, aime_id: int, score_data: Dict) -> Optional[int]: score_data["user"] = aime_id score_data = self.fix_bools(score_data) sql = insert(best_score).values(**score_data) conflict = sql.on_duplicate_key_update(**score_data) result = self.execute(conflict) if result is None: return None return result.lastrowid def get_playlogs(self, aime_id: int) -> Optional[Row]: sql = select(playlog).where(playlog.c.user == aime_id) result = self.execute(sql) if result is None: return None return result.fetchall() def put_playlog(self, aime_id: int, playlog_data: Dict, version: int) -> Optional[int]: # Calculate the ROM version that should be inserted into the DB, based on the version of the ggame being inserted # We only need from Version 10 (Plost) and back, as newer versions include romVersion in their upsert # This matters both for gameRankings, as well as a future DB update to keep version data separate romVer = { 10: "1.50.0", 9: "1.45.0", 8: "1.40.0", 7: "1.35.0", 6: "1.30.0", 5: "1.25.0", 4: "1.20.0", 3: "1.15.0", 2: "1.10.0", 1: "1.05.0", 0: "1.00.0" } playlog_data["user"] = aime_id playlog_data = self.fix_bools(playlog_data) if "romVersion" not in playlog_data: playlog_data["romVersion"] = romVer.get(version, "1.00.0") sql = insert(playlog).values(**playlog_data) conflict = sql.on_duplicate_key_update(**playlog_data) result = self.execute(conflict) if result is None: return None return result.lastrowid def get_rankings(self, version: int) -> Optional[List[Dict]]: # Calculates the ROM version that should be fetched for rankings, based on the game version being retrieved # This prevents tracks that are not accessible in your version from counting towards the 10 results romVer = { 13: "2.10%", 12: "2.05%", 11: "2.00%", 10: "1.50%", 9: "1.45%", 8: "1.40%", 7: "1.35%", 6: "1.30%", 5: "1.25%", 4: "1.20%", 3: "1.15%", 2: "1.10%", 1: "1.05%", 0: "1.00%" } sql = select([playlog.c.musicId.label('id'), func.count(playlog.c.musicId).label('point')]).where((playlog.c.level != 4) & (playlog.c.romVersion.like(romVer.get(version, "%")))).group_by(playlog.c.musicId).order_by(func.count(playlog.c.musicId).desc()).limit(10) result = self.execute(sql) if result is None: return None rows = result.fetchall() return [dict(row) for row in rows] def get_rival_music(self, rival_id: int) -> Optional[List[Dict]]: sql = select(best_score).where(best_score.c.user == rival_id) result = self.execute(sql) if result is None: return None return result.fetchall()