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.sql import func, select from sqlalchemy.engine import Row from sqlalchemy.dialects.mysql import insert from core.data.schema import BaseData, metadata from core.config import CoreConfig # Cammel case column names technically don't follow the other games but # it makes it way easier on me to not fuck with what the games has profile = Table( "ongeki_profile_data", metadata, Column("id", Integer, primary_key=True, nullable=False), Column("user", ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"), nullable=False), Column("version", Integer, nullable=False), Column("userName", String(8)), Column("level", Integer), Column("reincarnationNum", Integer), Column("exp", Integer), Column("point", Integer), Column("totalPoint", Integer), Column("playCount", Integer), Column("jewelCount", Integer), Column("totalJewelCount", Integer), Column("medalCount", Integer), Column("playerRating", Integer), Column("highestRating", Integer), Column("battlePoint", Integer), Column("nameplateId", Integer), Column("trophyId", Integer), Column("cardId", Integer), Column("characterId", Integer), Column("characterVoiceNo", Integer), Column("tabSetting", Integer), Column("tabSortSetting", Integer), Column("cardCategorySetting", Integer), Column("cardSortSetting", Integer), Column("playedTutorialBit", Integer), Column("firstTutorialCancelNum", Integer), Column("sumTechHighScore", BigInteger), Column("sumTechBasicHighScore", BigInteger), Column("sumTechAdvancedHighScore", BigInteger), Column("sumTechExpertHighScore", BigInteger), Column("sumTechMasterHighScore", BigInteger), Column("sumTechLunaticHighScore", BigInteger), Column("sumBattleHighScore", BigInteger), Column("sumBattleBasicHighScore", BigInteger), Column("sumBattleAdvancedHighScore", BigInteger), Column("sumBattleExpertHighScore", BigInteger), Column("sumBattleMasterHighScore", BigInteger), Column("sumBattleLunaticHighScore", BigInteger), Column("eventWatchedDate", String(255)), Column("cmEventWatchedDate", String(255)), Column("firstGameId", String(8)), Column("firstRomVersion", String(8)), Column("firstDataVersion", String(8)), Column("firstPlayDate", String(255)), Column("lastGameId", String(8)), Column("lastRomVersion", String(8)), Column("lastDataVersion", String(8)), Column("compatibleCmVersion", String(8)), Column("lastPlayDate", String(255)), Column("lastPlaceId", Integer), Column("lastPlaceName", String(255)), Column("lastRegionId", Integer), Column("lastRegionName", String(255)), Column("lastAllNetId", Integer), Column("lastClientId", String(16)), Column("lastUsedDeckId", Integer), Column("lastPlayMusicLevel", Integer), Column("banStatus", Integer, server_default="0"), Column("rivalScoreCategorySetting", Integer, server_default="0"), Column("overDamageBattlePoint", Integer, server_default="0"), Column("bestBattlePoint", Integer, server_default="0"), Column("lastEmoneyBrand", Integer, server_default="0"), UniqueConstraint("user", "version", name="ongeki_profile_profile_uk"), mysql_charset='utf8mb4' ) # No point setting defaults since the game sends everything on profile creation anyway option = Table( "ongeki_profile_option", metadata, Column("id", Integer, primary_key=True, nullable=False), Column("user", ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"), nullable=False), Column("optionSet", Integer), Column("speed", Integer), Column("mirror", Integer), Column("judgeTiming", Integer), Column("judgeAdjustment", Integer), Column("abort", Integer), Column("tapSound", Integer), Column("volGuide", Integer), Column("volAll", Integer), Column("volTap", Integer), Column("volCrTap", Integer), Column("volHold", Integer), Column("volSide", Integer), Column("volFlick", Integer), Column("volBell", Integer), Column("volEnemy", Integer), Column("volSkill", Integer), Column("volDamage", Integer), Column("colorField", Integer), Column("colorLaneBright", Integer), Column("colorLane", Integer), Column("colorSide", Integer), Column("effectDamage", Integer), Column("effectPos", Integer), Column("judgeDisp", Integer), Column("judgePos", Integer), Column("judgeBreak", Integer), Column("judgeHit", Integer), Column("platinumBreakDisp", Integer), Column("judgeCriticalBreak", Integer), Column("matching", Integer), Column("dispPlayerLv", Integer), Column("dispRating", Integer), Column("dispBP", Integer), Column("headphone", Integer), Column("stealthField", Integer), Column("colorWallBright", Integer), UniqueConstraint("user", name="ongeki_profile_option_uk"), mysql_charset='utf8mb4' ) activity = Table( "ongeki_profile_activity", metadata, Column("id", Integer, primary_key=True, nullable=False), Column("user", ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"), nullable=False), Column("kind", Integer), Column("activityId", Integer), Column("sortNumber", Integer), Column("param1", Integer), Column("param2", Integer), Column("param3", Integer), Column("param4", Integer), UniqueConstraint("user", "kind", "activityId", name="ongeki_profile_activity_uk"), mysql_charset='utf8mb4' ) recent_rating = Table( "ongeki_profile_recent_rating", metadata, Column("id", Integer, primary_key=True, nullable=False), Column("user", ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"), nullable=False), Column("recentRating", JSON), UniqueConstraint("user", name="ongeki_profile_recent_rating_uk"), mysql_charset='utf8mb4' ) rating_log = Table( "ongeki_profile_rating_log", metadata, Column("id", Integer, primary_key=True, nullable=False), Column("user", ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"), nullable=False), Column("highestRating", Integer), Column("dataVersion", String(10)), UniqueConstraint("user", "dataVersion", name="ongeki_profile_rating_log_uk"), mysql_charset='utf8mb4' ) region = Table( "ongeki_profile_region", metadata, Column("id", Integer, primary_key=True, nullable=False), Column("user", ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"), nullable=False), Column("regionId", Integer), Column("playCount", Integer), Column("created", String(25)), UniqueConstraint("user", "regionId", name="ongeki_profile_region_uk"), mysql_charset='utf8mb4' ) training_room = Table ( "ongeki_profile_training_room", metadata, Column("id", Integer, primary_key=True, nullable=False), Column("user", ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade")), Column("roomId", Integer), Column("authKey", Integer), Column("cardId", Integer), Column("valueDate", String(25)), UniqueConstraint("user", "roomId", name="ongeki_profile_training_room_uk"), mysql_charset='utf8mb4' ) kop = Table ( "ongeki_profile_kop", metadata, Column("id", Integer, primary_key=True, nullable=False), Column("user", ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade")), Column("authKey", Integer), Column("kopId", Integer), Column("areaId", Integer), Column("totalTechScore", Integer), Column("totalPlatinumScore", Integer), Column("techRecordDate", String(25)), Column("isTotalTechNewRecord", Boolean), UniqueConstraint("user", "kopId", name="ongeki_profile_kop_uk"), mysql_charset='utf8mb4' ) rival = Table( "ongeki_profile_rival", metadata, Column("id", Integer, primary_key=True, nullable=False), Column("user", ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade")), Column("rivalUserId", ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade")), UniqueConstraint("user", "rivalUserId", name="ongeki_profile_rival_uk"), mysql_charset='utf8mb4' ) class OngekiProfileData(BaseData): def __init__(self, cfg: CoreConfig, conn: Connection) -> None: super().__init__(cfg, conn) self.date_time_format_ext = "%Y-%m-%d %H:%M:%S.%f" # needs to be lopped off at [:-5] self.date_time_format_short = "%Y-%m-%d" def get_profile_name(self, aime_id: int, version: int) -> Optional[str]: sql = select(profile.c.userName).where(and_(profile.c.user == aime_id, profile.c.version == version)) result = self.execute(sql) if result is None: return None row = result.fetchone() if row is None: return None return row["userName"] def get_profile_preview(self, aime_id: int, version: int) -> Optional[Row]: sql = select([profile, option]).join(option, profile.c.user == option.c.user).filter( and_(profile.c.user == aime_id, profile.c.version == version) ) result = self.execute(sql) if result is None: return None return result.fetchone() def get_profile_data(self, aime_id: int, version: int) -> Optional[Row]: sql = select(profile).where(and_( profile.c.user == aime_id, profile.c.version == version, )) result = self.execute(sql) if result is None: return None return result.fetchone() def get_profile_options(self, aime_id: int) -> Optional[Row]: sql = select(option).where(and_( option.c.user == aime_id, )) result = self.execute(sql) if result is None: return None return result.fetchone() def get_profile_recent_rating(self, aime_id: int) -> Optional[List[Row]]: sql = select(recent_rating).where(recent_rating.c.user == aime_id) result = self.execute(sql) if result is None: return None return result.fetchone() def get_profile_rating_log(self, aime_id: int) -> Optional[List[Row]]: sql = select(rating_log).where(recent_rating.c.user == aime_id) result = self.execute(sql) if result is None: return None return result.fetchall() def get_profile_activity(self, aime_id: int, kind: int = None) -> Optional[List[Row]]: sql = select(activity).where(and_( activity.c.user == aime_id, (activity.c.kind == kind) if kind is not None else True )) result = self.execute(sql) if result is None: return None return result.fetchall() def get_kop(self, aime_id: int) -> Optional[List[Row]]: sql = select(kop).where(kop.c.user == aime_id) result = self.execute(sql) if result is None: return None return result.fetchall() def get_rivals(self, aime_id: int) -> Optional[List[Row]]: sql = select(rival.c.rivalUserId).where(rival.c.user == aime_id) result = self.execute(sql) if result is None: return None return result.fetchall() def put_profile_data(self, aime_id: int, version: int, data: Dict) -> Optional[int]: data["user"] = aime_id data["version"] = version data.pop("accessCode") sql = insert(profile).values(**data) conflict = sql.on_duplicate_key_update(**data) result = self.execute(conflict) if result is None: self.logger.warn(f"put_profile_data: Failed to update! aime_id: {aime_id}") return None return result.lastrowid def put_profile_options(self, aime_id: int, options_data: Dict) -> Optional[int]: options_data["user"] = aime_id sql = insert(option).values(**options_data) conflict = sql.on_duplicate_key_update(**options_data) result = self.execute(conflict) if result is None: self.logger.warn(f"put_profile_options: Failed to update! aime_id: {aime_id}") return None return result.lastrowid def put_profile_recent_rating(self, aime_id: int, recent_rating_data: List[Dict]) -> Optional[int]: sql = insert(recent_rating).values( user=aime_id, recentRating=recent_rating_data ) conflict = sql.on_duplicate_key_update( recentRating=recent_rating_data ) result = self.execute(conflict) if result is None: self.logger.warn(f"put_profile_recent_rating: failed to update recent rating! aime_id {aime_id}") return None return result.lastrowid def put_profile_bp_list(self, aime_id: int, bp_base_list: List[Dict]) -> Optional[int]: pass def put_profile_rating_log(self, aime_id: int, data_version: str, highest_rating: int) -> Optional[int]: sql = insert(rating_log).values( user=aime_id, dataVersion=data_version, highestRating=highest_rating ) conflict = sql.on_duplicate_key_update( highestRating=highest_rating ) result = self.execute(conflict) if result is None: self.logger.warn(f"put_profile_rating_log: failed to update rating log! aime_id {aime_id} data_version {data_version} highest_rating {highest_rating}") return None return result.lastrowid def put_profile_activity(self, aime_id: int, kind: int, activity_id: int, sort_num: int, p1: int, p2: int, p3: int, p4: int) -> Optional[int]: sql = insert(activity).values( user=aime_id, kind=kind, activityId=activity_id, sortNumber=sort_num, param1=p1, param2=p2, param3=p3, param4=p4 ) conflict = sql.on_duplicate_key_update( sortNumber=sort_num, param1=p1, param2=p2, param3=p3, param4=p4 ) result = self.execute(conflict) if result is None: self.logger.warn(f"put_profile_activity: failed to put activity! aime_id {aime_id} kind {kind} activity_id {activity_id}") return None return result.lastrowid def put_profile_region(self, aime_id: int, region: int, date: str) -> Optional[int]: sql = insert(activity).values( user=aime_id, region=region, playCount=1, created=date ) conflict = sql.on_duplicate_key_update( playCount=activity.c.playCount + 1, ) result = self.execute(conflict) if result is None: self.logger.warn(f"put_profile_region: failed to update! aime_id {aime_id} region {region}") return None return result.lastrowid def put_training_room(self, aime_id: int, room_detail: Dict) -> Optional[int]: room_detail["user"] = aime_id sql = insert(training_room).values(**room_detail) conflict = sql.on_duplicate_key_update(**room_detail) result = self.execute(conflict) if result is None: self.logger.warn(f"put_best_score: Failed to add score! aime_id: {aime_id}") return None return result.lastrowid def put_kop(self, aime_id: int, kop_data: Dict) -> Optional[int]: kop_data["user"] = aime_id sql = insert(kop).values(**kop_data) conflict = sql.on_duplicate_key_update(**kop_data) result = self.execute(conflict) if result is None: self.logger.warn(f"put_kop: Failed to add score! aime_id: {aime_id}") return None return result.lastrowid def put_rival(self, aime_id: int, rival_id: int) -> Optional[int]: sql = insert(rival).values( user = aime_id, rivalUserId = rival_id ) conflict = sql.on_duplicate_key_update(rival = rival_id) result = self.execute(conflict) if result is None: self.logger.warn(f"put_rival: failed to update! aime_id: {aime_id}, rival_id: {rival_id}") return None return result.lastrowid