artemis/titles/ongeki/schema/profile.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.engine.base import Connection
from sqlalchemy.schema import ForeignKey
from sqlalchemy.sql import func, select, delete
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"),
    Column("lastEmoneyCredit", Integer, server_default="0"),
    Column("isDialogWatchedSuggestMemory", Boolean, 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"

    async 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 = await self.execute(sql)
        if result is None:
            return None

        row = result.fetchone()
        if row is None:
            return None

        return row["userName"]
    
    async 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 = await self.execute(sql)
        if result is None:
            return None
        return result.fetchone()

    async 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 = await self.execute(sql)
        if result is None:
            return None
        return result.fetchone()

    async def get_profile_options(self, aime_id: int) -> Optional[Row]:
        sql = select(option).where(
            and_(
                option.c.user == aime_id,
            )
        )

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

    async def get_profile_recent_rating(self, aime_id: int) -> Optional[List[Row]]:
        sql = select(recent_rating).where(recent_rating.c.user == aime_id)

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

    async def get_profile_rating_log(self, aime_id: int) -> Optional[List[Row]]:
        sql = select(rating_log).where(rating_log.c.user == aime_id)

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

    async 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 = await self.execute(sql)
        if result is None:
            return None
        return result.fetchall()

    async def get_kop(self, aime_id: int) -> Optional[List[Row]]:
        sql = select(kop).where(kop.c.user == aime_id)

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

    async def get_rivals(self, aime_id: int) -> Optional[List[Row]]:
        sql = select(rival.c.rivalUserId).where(rival.c.user == aime_id)

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

    async 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 = await self.execute(conflict)

        if result is None:
            self.logger.warning(f"put_profile_data: Failed to update! aime_id: {aime_id}")
            return None
        return result.lastrowid

    async 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 = await self.execute(conflict)

        if result is None:
            self.logger.warning(
                f"put_profile_options: Failed to update! aime_id: {aime_id}"
            )
            return None
        return result.lastrowid

    async 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 = await self.execute(conflict)
        if result is None:
            self.logger.warning(
                f"put_profile_recent_rating: failed to update recent rating! aime_id {aime_id}"
            )
            return None
        return result.lastrowid

    async def put_profile_bp_list(
        self, aime_id: int, bp_base_list: List[Dict]
    ) -> Optional[int]:
        pass

    async 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 = await self.execute(conflict)
        if result is None:
            self.logger.warning(
                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

    async 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 = await self.execute(conflict)
        if result is None:
            self.logger.warning(
                f"put_profile_activity: failed to put activity! aime_id {aime_id} kind {kind} activity_id {activity_id}"
            )
            return None
        return result.lastrowid

    async 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 = await self.execute(conflict)
        if result is None:
            self.logger.warning(
                f"put_profile_region: failed to update! aime_id {aime_id} region {region}"
            )
            return None
        return result.lastrowid

    async 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 = await self.execute(conflict)

        if result is None:
            self.logger.warning(f"put_best_score: Failed to add score! aime_id: {aime_id}")
            return None
        return result.lastrowid

    async 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 = await self.execute(conflict)

        if result is None:
            self.logger.warning(f"put_kop: Failed to add score! aime_id: {aime_id}")
            return None
        return result.lastrowid

    async 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(rivalUserId=rival_id)

        result = await self.execute(conflict)
        if result is None:
            self.logger.warning(
                f"put_rival: failed to update! aime_id: {aime_id}, rival_id: {rival_id}"
            )
            return None
        return result.lastrowid
    async def delete_rival(self, aime_id: int, rival_id: int) -> Optional[int]:
        sql = delete(rival).where(rival.c.user==aime_id, rival.c.rivalUserId==rival_id)
        result = await self.execute(sql)
        if result is None:
            self.logger.error(f"delete_rival: failed to delete! aime_id: {aime_id}, rival_id: {rival_id}")
        else:
            return result.rowcount