artemis/titles/idac/schema/item.py

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

from core.data.schema import BaseData, metadata

car = Table(
    "idac_user_car",
    metadata,
    Column("id", Integer, primary_key=True, nullable=False),
    Column("user", ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade")),
    Column("version", Integer, nullable=False),
    Column("car_id", Integer),
    Column("style_car_id", Integer),
    Column("color", Integer),
    Column("bureau", Integer),
    Column("kana", Integer),
    Column("s_no", Integer),
    Column("l_no", Integer),
    Column("car_flag", Integer),
    Column("tune_point", Integer),
    Column("tune_level", Integer, server_default="1"),
    Column("tune_parts", Integer),
    Column("infinity_tune", Integer, server_default="0"),
    Column("online_vs_win", Integer, server_default="0"),
    Column(
        "pickup_seq", Integer, server_default="1"
    ),  # the order in which the car was picked up
    Column(
        "purchase_seq", Integer, server_default="1"
    ),  # the order in which the car was purchased
    Column("color_stock_list", String(32)),
    Column("color_stock_new_list", String(32)),
    Column("parts_stock_list", String(48)),
    Column("parts_stock_new_list", String(48)),
    Column("parts_set_equip_list", String(48)),
    Column("parts_list", JSON),
    Column("equip_parts_count", Integer, server_default="0"),
    Column("total_car_parts_count", Integer, server_default="0"),
    Column("use_count", Integer, server_default="0"),
    Column("story_use_count", Integer, server_default="0"),
    Column("timetrial_use_count", Integer, server_default="0"),
    Column("vs_use_count", Integer, server_default="0"),
    Column("net_vs_use_count", Integer, server_default="0"),
    Column("theory_use_count", Integer, server_default="0"),
    Column("car_mileage", Integer, server_default="0"),
    UniqueConstraint("user", "version", "style_car_id", name="idac_user_car_uk"),
    mysql_charset="utf8mb4",
)

ticket = Table(
    "idac_user_ticket",
    metadata,
    Column("id", Integer, primary_key=True, nullable=False),
    Column("user", ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade")),
    Column("ticket_id", Integer),
    Column("ticket_cnt", Integer),
    UniqueConstraint("user", "ticket_id", name="idac_user_ticket_uk"),
    mysql_charset="utf8mb4",
)

story = Table(
    "idac_user_story",
    metadata,
    Column("id", Integer, primary_key=True, nullable=False),
    Column("user", ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade")),
    Column("story_type", Integer),
    Column("chapter", Integer),
    Column("loop_count", Integer, server_default="1"),
    UniqueConstraint("user", "chapter", name="idac_user_story_uk"),
    mysql_charset="utf8mb4",
)

episode = Table(
    "idac_user_story_episode",
    metadata,
    Column("id", Integer, primary_key=True, nullable=False),
    Column("user", ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade")),
    Column("chapter", Integer),
    Column("episode", Integer),
    Column("play_status", Integer),
    UniqueConstraint("user", "chapter", "episode", name="idac_user_story_episode_uk"),
    mysql_charset="utf8mb4",
)

difficulty = Table(
    "idac_user_story_episode_difficulty",
    metadata,
    Column("id", Integer, primary_key=True, nullable=False),
    Column("user", ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade")),
    Column("episode", Integer),
    Column("difficulty", Integer),
    Column("play_count", Integer),
    Column("clear_count", Integer),
    Column("play_status", Integer),
    Column("play_score", Integer),
    UniqueConstraint(
        "user", "episode", "difficulty", name="idac_user_story_episode_difficulty_uk"
    ),
    mysql_charset="utf8mb4",
)

course = Table(
    "idac_user_course",
    metadata,
    Column("id", Integer, primary_key=True, nullable=False),
    Column("user", ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade")),
    Column("course_id", Integer),
    Column("run_counts", Integer, server_default="1"),
    Column("skill_level_exp", Integer, server_default="0"),
    UniqueConstraint("user", "course_id", name="idac_user_course_uk"),
    mysql_charset="utf8mb4",
)

trial = Table(
    "idac_user_time_trial",
    metadata,
    Column("id", Integer, primary_key=True, nullable=False),
    Column("user", ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade")),
    Column("version", Integer, nullable=False),
    Column("style_car_id", Integer),
    Column("course_id", Integer),
    Column("eval_id", Integer, server_default="0"),
    Column("goal_time", Integer),
    Column("section_time_1", Integer),
    Column("section_time_2", Integer),
    Column("section_time_3", Integer),
    Column("section_time_4", Integer),
    Column("mission", Integer),
    Column("play_dt", TIMESTAMP, server_default=func.now()),
    UniqueConstraint(
        "user", "version", "course_id", "style_car_id", name="idac_user_time_trial_uk"
    ),
    mysql_charset="utf8mb4",
)

challenge = Table(
    "idac_user_challenge",
    metadata,
    Column("id", Integer, primary_key=True, nullable=False),
    Column("user", ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade")),
    Column("vs_type", Integer),
    Column("play_difficulty", Integer),
    Column("cleared_difficulty", Integer),
    Column("story_type", Integer),
    Column("play_count", Integer, server_default="1"),
    Column("weak_difficulty", Integer, server_default="0"),
    Column("eval_id", Integer),
    Column("advantage", Integer),
    Column("sec1_advantage_avg", Integer),
    Column("sec2_advantage_avg", Integer),
    Column("sec3_advantage_avg", Integer),
    Column("sec4_advantage_avg", Integer),
    Column("nearby_advantage_rate", Integer),
    Column("win_flag", Integer),
    Column("result", Integer),
    Column("record", Integer),
    Column("course_id", Integer),
    Column("last_play_course_id", Integer),
    Column("style_car_id", Integer),
    Column("course_day", Integer),
    UniqueConstraint(
        "user", "vs_type", "play_difficulty", name="idac_user_challenge_uk"
    ),
    mysql_charset="utf8mb4",
)

theory_course = Table(
    "idac_user_theory_course",
    metadata,
    Column("id", Integer, primary_key=True, nullable=False),
    Column("user", ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade")),
    Column("course_id", Integer),
    Column("max_victory_grade", Integer, server_default="0"),
    Column("run_count", Integer, server_default="1"),
    Column("powerhouse_lv", Integer),
    Column("powerhouse_exp", Integer),
    Column("played_powerhouse_lv", Integer),
    Column("update_dt", TIMESTAMP, server_default=func.now()),
    UniqueConstraint("user", "course_id", name="idac_user_theory_course_uk"),
    mysql_charset="utf8mb4",
)

theory_partner = Table(
    "idac_user_theory_partner",
    metadata,
    Column("id", Integer, primary_key=True, nullable=False),
    Column("user", ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade")),
    Column("partner_id", Integer),
    Column("fellowship_lv", Integer),
    Column("fellowship_exp", Integer),
    UniqueConstraint("user", "partner_id", name="idac_user_theory_partner_uk"),
    mysql_charset="utf8mb4",
)

theory_running = Table(
    "idac_user_theory_running",
    metadata,
    Column("id", Integer, primary_key=True, nullable=False),
    Column("user", ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade")),
    Column("course_id", Integer),
    Column("attack", Integer),
    Column("defense", Integer),
    Column("safety", Integer),
    Column("runaway", Integer),
    Column("trick_flag", Integer),
    UniqueConstraint("user", "course_id", name="idac_user_theory_running_uk"),
    mysql_charset="utf8mb4",
)

vs_info = Table(
    "idac_user_vs_info",
    metadata,
    Column("id", Integer, primary_key=True, nullable=False),
    Column("user", ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade")),
    Column("battle_mode", Integer),
    Column("invalid", Integer),
    Column("str", Integer),
    Column("str_now", Integer),
    Column("lose_now", Integer),
    Column("break_count", Integer),
    Column("break_penalty_flag", Boolean),
    UniqueConstraint("user", "battle_mode", name="idac_user_vs_info_uk"),
    mysql_charset="utf8mb4",
)

vs_course_info = Table(
    "idac_user_vs_course_info",
    metadata,
    Column("id", Integer, primary_key=True, nullable=False),
    Column("user", ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade")),
    Column("battle_mode", Integer),
    Column("course_id", Integer),
    Column("vs_cnt", Integer),
    Column("vs_win", Integer),
    UniqueConstraint("user", "battle_mode", "course_id", name="idac_user_vs_course_info_uk"),
    mysql_charset="utf8mb4",
)

round_infos = Table(
	"idac_round_info",
	metadata,
    Column("id", Integer, primary_key=True, nullable=False),
	Column("name", String(64)),
	Column("season", Integer),
	Column("start_dt", TIMESTAMP, server_default=func.now()),
	Column("end_dt", TIMESTAMP, server_default=func.now()),
    mysql_charset="utf8mb4",
)

round_info = Table(
	"idac_user_round_info",
	metadata,
    Column("id", Integer, primary_key=True, nullable=False),
    Column("user", ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade")),
	Column("round_id", Integer),
	Column("count", Integer),
	Column("win", Integer),
	Column("points", Integer),
    UniqueConstraint("user", "round_id", name="idac_user_round_info_uk"),
    mysql_charset="utf8mb4",
)

stamp = Table(
    "idac_user_stamp",
    metadata,
    Column("id", Integer, primary_key=True, nullable=False),
    Column(
        "user",
        ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"),
        nullable=False,
    ),
    Column("m_stamp_event_id", Integer),
    Column("select_flag", Integer),
    Column("stamp_masu", Integer),
    Column("daily_bonus", Integer),
    Column("weekly_bonus", Integer),
    Column("weekday_bonus", Integer),
    Column("weekend_bonus", Integer),
    Column("total_bonus", Integer),
    Column("day_total_bonus", Integer),
    Column("store_battle_bonus", Integer),
    Column("story_bonus", Integer),
    Column("online_battle_bonus", Integer),
    Column("timetrial_bonus", Integer),
    Column("fasteststreetlegaltheory_bonus", Integer),
    Column("collaboration_bonus", Integer),
    Column("add_bonus_daily_flag_1", Integer),
    Column("add_bonus_daily_flag_2", Integer),
    Column("add_bonus_daily_flag_3", Integer),
    Column("create_date_daily", TIMESTAMP, server_default=func.now()),
    Column("create_date_weekly", TIMESTAMP, server_default=func.now()),
    UniqueConstraint("user", "m_stamp_event_id", name="idac_user_stamp_uk"),
    mysql_charset="utf8mb4",
)

timetrial_event = Table(
    "idac_user_timetrial_event",
    metadata,
    Column("id", Integer, primary_key=True, nullable=False),
    Column(
        "user",
        ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"),
        nullable=False,
    ),
    Column("timetrial_event_id", Integer),
    Column("point", Integer),
    UniqueConstraint("user", "timetrial_event_id", name="idac_user_timetrial_event_uk"),
    mysql_charset="utf8mb4",
)


class IDACItemData(BaseData):
    def get_random_user_car(self, aime_id: int, version: int) -> Optional[List[Row]]:
        sql = (
            select(car)
            .where(and_(car.c.user == aime_id, car.c.version == version))
            .order_by(func.rand())
            .limit(1)
        )

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

    def get_random_car(self, version: int) -> Optional[List[Row]]:
        sql = select(car).where(car.c.version == version).order_by(func.rand()).limit(1)

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

    def get_car(
        self, aime_id: int, version: int, style_car_id: int
    ) -> Optional[List[Row]]:
        sql = select(car).where(
            and_(
                car.c.user == aime_id,
                car.c.version == version,
                car.c.style_car_id == style_car_id,
            )
        )

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

    def get_cars(
        self, version: int, aime_id: int, only_pickup: bool = False
    ) -> Optional[List[Row]]:
        if only_pickup:
            sql = select(car).where(
                and_(
                    car.c.user == aime_id,
                    car.c.version == version,
                    car.c.pickup_seq != 0,
                )
            )
        else:
            sql = select(car).where(
                and_(car.c.user == aime_id, car.c.version == version)
            )

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

    def get_ticket(self, aime_id: int, ticket_id: int) -> Optional[Row]:
        sql = select(ticket).where(
            ticket.c.user == aime_id, ticket.c.ticket_id == ticket_id
        )

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

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

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

    def get_story(self, aime_id: int, chapter_id: int) -> Optional[Row]:
        sql = select(story).where(
            and_(story.c.user == aime_id, story.c.chapter == chapter_id)
        )

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

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

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

    def get_story_episodes(self, aime_id: int, chapter_id: int) -> Optional[List[Row]]:
        sql = select(episode).where(
            and_(episode.c.user == aime_id, episode.c.chapter == chapter_id)
        )

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

    def get_story_episode(self, aime_id: int, episode_id: int) -> Optional[Row]:
        sql = select(episode).where(
            and_(episode.c.user == aime_id, episode.c.episode == episode_id)
        )

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

    def get_story_episode_difficulties(
        self, aime_id: int, episode_id: int
    ) -> Optional[List[Row]]:
        sql = select(difficulty).where(
            and_(difficulty.c.user == aime_id, difficulty.c.episode == episode_id)
        )

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

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

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

    def get_course(self, aime_id: int, course_id: int) -> Optional[Row]:
        sql = select(course).where(
            and_(course.c.user == aime_id, course.c.course_id == course_id)
        )

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

    def get_time_trial_courses(self, version: int) -> Optional[List[Row]]:
        sql = select(trial.c.course_id).where(trial.c.version == version).distinct()

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

    def get_time_trial_user_best_time_by_course_car(
        self, version: int, aime_id: int, course_id: int, style_car_id: int
    ) -> Optional[Row]:
        sql = select(trial).where(
            and_(
                trial.c.user == aime_id,
                trial.c.version == version,
                trial.c.course_id == course_id,
                trial.c.style_car_id == style_car_id,
            )
        )

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

    def get_time_trial_user_best_courses(
        self, version: int, aime_id: int
    ) -> Optional[List[Row]]:
        # get for a given aime_id the best time for each course
        subquery = (
            select(
                trial.c.version,
                func.min(trial.c.goal_time).label("min_goal_time"),
                trial.c.course_id,
            )
            .where(and_(trial.c.version == version, trial.c.user == aime_id))
            .group_by(trial.c.course_id)
            .subquery()
        )

        # now get the full row for each best time
        sql = select(trial).where(
            and_(
                trial.c.version == subquery.c.version,
                trial.c.goal_time == subquery.c.min_goal_time,
                trial.c.course_id == subquery.c.course_id,
                trial.c.user == aime_id,
            )
        )

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

    def get_time_trial_best_cars_by_course(
        self, version: int, course_id: int, aime_id: Optional[int] = None
    ) -> Optional[List[Row]]:
        subquery = select(
            trial.c.version,
            func.min(trial.c.goal_time).label("min_goal_time"),
            trial.c.style_car_id,
        ).where(
            and_(
                trial.c.version == version,
                trial.c.course_id == course_id,
            )
        )

        if aime_id is not None:
            subquery = subquery.where(trial.c.user == aime_id)

        subquery = subquery.group_by(trial.c.style_car_id).subquery()

        sql = select(trial).where(
            and_(
                trial.c.version == subquery.c.version,
                trial.c.goal_time == subquery.c.min_goal_time,
                trial.c.style_car_id == subquery.c.style_car_id,
                trial.c.course_id == course_id,
            )
        )

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

    def get_time_trial_ranking_by_course_total(
        self,
        version: int,
        course_id: int,
    ) -> Optional[List[Row]]:
        # count the number of rows returned by the query
        subquery = (
            select(
                trial.c.version,
                trial.c.user,
                func.min(trial.c.goal_time).label("min_goal_time"),
            )
            .where(and_(trial.c.version == version, trial.c.course_id == course_id))
            .group_by(trial.c.user)
        ).subquery()

        sql = (
            select(func.count().label("count"))
            .where(
                and_(
                    trial.c.version == subquery.c.version,
                    trial.c.user == subquery.c.user,
                    trial.c.goal_time == subquery.c.min_goal_time,
                ),
            )
        )

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

    def get_time_trial_ranking_by_course(
        self,
        version: int,
        course_id: int,
        style_car_id: Optional[int] = None,
        limit: Optional[int] = 10,
        offset: Optional[int] = 0,
    ) -> Optional[List[Row]]:
        # get the top 10 ranking by goal_time for a given course which is grouped by user
        subquery = select(
            trial.c.version,
            trial.c.user,
            func.min(trial.c.goal_time).label("min_goal_time"),
        ).where(and_(trial.c.version == version, trial.c.course_id == course_id))

        # if wanted filter only by style_car_id
        if style_car_id is not None:
            subquery = subquery.where(trial.c.style_car_id == style_car_id)

        subquery = subquery.group_by(trial.c.user).subquery()

        sql = (
            select(trial)
            .where(
                and_(
                    trial.c.version == subquery.c.version,
                    trial.c.user == subquery.c.user,
                    trial.c.goal_time == subquery.c.min_goal_time,
                ),
            )
            .order_by(trial.c.goal_time)
        )

        # limit the result if needed
        if limit is not None:
            sql = sql.limit(limit)

        # offset the result if needed
        if offset is not None:
            sql = sql.offset(offset)

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

    def get_time_trial_best_ranking_by_course(
        self, version: int, aime_id: int, course_id: int
    ) -> Optional[Row]:
        sql = (
            select(trial)
            .where(
                and_(
                    trial.c.version == version,
                    trial.c.user == aime_id,
                    trial.c.course_id == course_id,
                ),
            )
            .order_by(trial.c.goal_time)
            .limit(1)
        )

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

    def get_challenge(
        self, aime_id: int, vs_type: int, play_difficulty: int
    ) -> Optional[Row]:
        sql = select(challenge).where(
            and_(
                challenge.c.user == aime_id,
                challenge.c.vs_type == vs_type,
                challenge.c.play_difficulty == play_difficulty,
            )
        )

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

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

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

    def get_best_challenges_by_vs_type(
        self, aime_id: int, story_type: int = 4
    ) -> Optional[List[Row]]:
        subquery = (
            select(
                challenge.c.story_type,
                challenge.c.user,
                challenge.c.vs_type,
                func.max(challenge.c.play_difficulty).label("last_play_lv"),
            )
            .where(
                and_(challenge.c.user == aime_id, challenge.c.story_type == story_type)
            )
            .group_by(challenge.c.vs_type)
        )

        sql = (
            select(
                challenge.c.story_type,
                challenge.c.vs_type,
                challenge.c.cleared_difficulty.label("max_clear_lv"),
                challenge.c.play_difficulty.label("last_play_lv"),
                challenge.c.course_id,
                challenge.c.play_count,
            )
            .where(
                and_(
                    challenge.c.user == subquery.c.user,
                    challenge.c.vs_type == subquery.c.vs_type,
                    challenge.c.play_difficulty == subquery.c.last_play_lv,
                ),
            )
            .order_by(challenge.c.vs_type)
        )

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

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

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

    def get_theory_course_by_powerhouse_lv(
        self, aime_id: int, course_id: int, powerhouse_lv: int, count: int = 3
    ) -> Optional[List[Row]]:
        sql = (
            select(theory_course)
            .where(
                and_(
                    theory_course.c.user != aime_id,
                    theory_course.c.course_id == course_id,
                    theory_course.c.powerhouse_lv == powerhouse_lv,
                )
            )
            .order_by(func.rand())
            .limit(count)
        )

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

    def get_theory_course(self, aime_id: int, course_id: int) -> Optional[List[Row]]:
        sql = select(theory_course).where(
            and_(
                theory_course.c.user == aime_id, theory_course.c.course_id == course_id
            )
        )

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

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

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

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

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

    def get_theory_running_by_course(
        self, aime_id: int, course_id: int
    ) -> Optional[Row]:
        sql = select(theory_running).where(
            and_(
                theory_running.c.user == aime_id,
                theory_running.c.course_id == course_id,
            )
        )

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

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

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

    def get_vs_info_by_mode(self, aime_id: int, battle_mode: int) -> Optional[List[Row]]:
        sql = select(vs_info).where(
            and_(vs_info.c.user == aime_id, vs_info.c.battle_mode == battle_mode)
        )

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

    #This method returns a list of course_info
    def get_vs_course_infos_by_mode(self, aime_id: int, battle_mode: int) -> Optional[List[Row]]:
        sql = select(vs_course_info).where(vs_info.c.user == aime_id, vs_info.c.battle_mode == battle_mode)

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

    def get_stamps(self, aime_id: int) -> Optional[List[Row]]:
        sql = select(stamp).where(
            and_(
                stamp.c.user == aime_id,
            )
        )

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

    def get_timetrial_event(
        self, aime_id: int, timetrial_event_id: int
    ) -> Optional[Row]:
        sql = select(timetrial_event).where(
            and_(
                timetrial_event.c.user == aime_id,
                timetrial_event.c.timetrial_event_id == timetrial_event_id,
            )
        )

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

    def put_car(self, aime_id: int, version: int, car_data: Dict) -> Optional[int]:
        car_data["user"] = aime_id
        car_data["version"] = version

        sql = insert(car).values(**car_data)
        conflict = sql.on_duplicate_key_update(**car_data)
        result = self.execute(conflict)

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

    def put_ticket(self, aime_id: int, ticket_data: Dict) -> Optional[int]:
        ticket_data["user"] = aime_id

        sql = insert(ticket).values(**ticket_data)
        conflict = sql.on_duplicate_key_update(**ticket_data)
        result = self.execute(conflict)

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

    def put_story(self, aime_id: int, story_data: Dict) -> Optional[int]:
        story_data["user"] = aime_id

        sql = insert(story).values(**story_data)
        conflict = sql.on_duplicate_key_update(**story_data)
        result = self.execute(conflict)

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

    def put_story_episode_play_status(
        self, aime_id: int, chapter_id: int, play_status: int = 1
    ) -> Optional[int]:
        sql = (
            update(episode)
            .where(and_(episode.c.user == aime_id, episode.c.chapter == chapter_id))
            .values(play_status=play_status)
        )

        result = self.execute(sql)
        if result is None:
            self.logger.warn(
                f"put_story_episode_play_status: Failed to update! aime_id: {aime_id}"
            )
            return None
        return result.lastrowid

    def put_story_episode(
        self, aime_id: int, chapter_id: int, episode_data: Dict
    ) -> Optional[int]:
        episode_data["user"] = aime_id
        episode_data["chapter"] = chapter_id

        sql = insert(episode).values(**episode_data)
        conflict = sql.on_duplicate_key_update(**episode_data)
        result = self.execute(conflict)

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

    def put_story_episode_difficulty(
        self, aime_id: int, episode_id: int, difficulty_data: Dict
    ) -> Optional[int]:
        difficulty_data["user"] = aime_id
        difficulty_data["episode"] = episode_id

        sql = insert(difficulty).values(**difficulty_data)
        conflict = sql.on_duplicate_key_update(**difficulty_data)
        result = self.execute(conflict)

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

    def put_course(self, aime_id: int, course_data: Dict) -> Optional[int]:
        course_data["user"] = aime_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.warn(f"put_course: Failed to update! aime_id: {aime_id}")
            return None
        return result.lastrowid

    def put_time_trial(
        self, version: int, aime_id: int, time_trial_data: Dict
    ) -> Optional[int]:
        time_trial_data["user"] = aime_id
        time_trial_data["version"] = version

        sql = insert(trial).values(**time_trial_data)
        conflict = sql.on_duplicate_key_update(**time_trial_data)
        result = self.execute(conflict)

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

    def put_challenge(self, aime_id: int, challenge_data: Dict) -> Optional[int]:
        challenge_data["user"] = aime_id

        sql = insert(challenge).values(**challenge_data)
        conflict = sql.on_duplicate_key_update(**challenge_data)
        result = self.execute(conflict)

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

    def put_theory_course(
        self, aime_id: int, theory_course_data: Dict
    ) -> Optional[int]:
        theory_course_data["user"] = aime_id

        sql = insert(theory_course).values(**theory_course_data)
        conflict = sql.on_duplicate_key_update(**theory_course_data)
        result = self.execute(conflict)

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

    def put_theory_partner(
        self, aime_id: int, theory_partner_data: Dict
    ) -> Optional[int]:
        theory_partner_data["user"] = aime_id

        sql = insert(theory_partner).values(**theory_partner_data)
        conflict = sql.on_duplicate_key_update(**theory_partner_data)
        result = self.execute(conflict)

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

    def put_theory_running(
        self, aime_id: int, theory_running_data: Dict
    ) -> Optional[int]:
        theory_running_data["user"] = aime_id

        sql = insert(theory_running).values(**theory_running_data)
        conflict = sql.on_duplicate_key_update(**theory_running_data)
        result = self.execute(conflict)

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

    def put_vs_info(self, aime_id: int, battle_mode: int, vs_info_data: Dict) -> Optional[int]:
        vs_info_data["user"] = aime_id
        vs_info_data["battle_mode"] = battle_mode

        sql = insert(vs_info).values(**vs_info_data)
        conflict = sql.on_duplicate_key_update(**vs_info_data)
        result = self.execute(conflict)

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

    def put_vs_course_info(self, aime_id: int, battle_mode: int, course_info_data: Dict) -> Optional[int]:
        course_info_data["user"] = aime_id
        course_info_data["battle_mode"] = battle_mode

        sql = insert(vs_course_info).values(**course_info_data)
        conflict = sql.on_duplicate_key_update(**course_info_data)
        result = self.execute(conflict)

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

    def put_stamp(
        self, aime_id: int, stamp_data: Dict
    ) -> Optional[int]:
        stamp_data["user"] = aime_id

        sql = insert(stamp).values(**stamp_data)
        conflict = sql.on_duplicate_key_update(**stamp_data)
        result = self.execute(conflict)

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

    def put_timetrial_event(
        self, aime_id: int, time_trial_event_id: int, point: int
    ) -> Optional[int]:
        timetrial_event_data = {
            "user": aime_id,
            "timetrial_event_id": time_trial_event_id,
            "point": point,
        }

        sql = insert(timetrial_event).values(**timetrial_event_data)
        conflict = sql.on_duplicate_key_update(**timetrial_event_data)
        result = self.execute(conflict)

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