forked from Hay1tsme/artemis
1122 lines
37 KiB
Python
1122 lines
37 KiB
Python
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("vs_history", 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",
|
|
)
|
|
|
|
battle_gift = Table(
|
|
"idac_user_battle_gift",
|
|
metadata,
|
|
Column("id", Integer, primary_key=True, nullable=False),
|
|
Column(
|
|
"user",
|
|
ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"),
|
|
nullable=False,
|
|
),
|
|
Column("battle_gift_event_id", Integer),
|
|
Column("gift_id", Integer),
|
|
Column("gift_status", Integer),
|
|
Column("received_date", TIMESTAMP, server_default=func.now()),
|
|
UniqueConstraint("user", "battle_gift_event_id", "gift_id", name="idac_user_battle_gift_uk"),
|
|
mysql_charset="utf8mb4",
|
|
)
|
|
|
|
|
|
class IDACItemData(BaseData):
|
|
async 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 = await self.execute(sql)
|
|
if result is None:
|
|
return None
|
|
return result.fetchone()
|
|
|
|
async 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 = await self.execute(sql)
|
|
if result is None:
|
|
return None
|
|
return result.fetchone()
|
|
|
|
async 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 = await self.execute(sql)
|
|
if result is None:
|
|
return None
|
|
return result.fetchone()
|
|
|
|
async 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 = await self.execute(sql)
|
|
if result is None:
|
|
return None
|
|
return result.fetchall()
|
|
|
|
async 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 = await self.execute(sql)
|
|
if result is None:
|
|
return None
|
|
return result.fetchone()
|
|
|
|
async def get_tickets(self, aime_id: int) -> Optional[List[Row]]:
|
|
sql = select(ticket).where(ticket.c.user == aime_id)
|
|
|
|
result = await self.execute(sql)
|
|
if result is None:
|
|
return None
|
|
return result.fetchall()
|
|
|
|
async 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 = await self.execute(sql)
|
|
if result is None:
|
|
return None
|
|
return result.fetchone()
|
|
|
|
async def get_stories(self, aime_id: int) -> Optional[List[Row]]:
|
|
sql = select(story).where(story.c.user == aime_id)
|
|
|
|
result = await self.execute(sql)
|
|
if result is None:
|
|
return None
|
|
return result.fetchall()
|
|
|
|
async 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 = await self.execute(sql)
|
|
if result is None:
|
|
return None
|
|
return result.fetchall()
|
|
|
|
async 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 = await self.execute(sql)
|
|
if result is None:
|
|
return None
|
|
return result.fetchone()
|
|
|
|
async 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 = await self.execute(sql)
|
|
if result is None:
|
|
return None
|
|
return result.fetchall()
|
|
|
|
async def get_courses(self, aime_id: int) -> Optional[List[Row]]:
|
|
sql = select(course).where(course.c.user == aime_id)
|
|
|
|
result = await self.execute(sql)
|
|
if result is None:
|
|
return None
|
|
return result.fetchall()
|
|
|
|
async 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 = await self.execute(sql)
|
|
if result is None:
|
|
return None
|
|
return result.fetchone()
|
|
|
|
async def get_time_trial_courses(self, version: int) -> Optional[List[Row]]:
|
|
sql = select(trial.c.course_id).where(trial.c.version == version).distinct()
|
|
|
|
result = await self.execute(sql)
|
|
if result is None:
|
|
return None
|
|
return result.fetchall()
|
|
|
|
async 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 = await self.execute(sql)
|
|
if result is None:
|
|
return None
|
|
return result.fetchone()
|
|
|
|
async 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 = await self.execute(sql)
|
|
if result is None:
|
|
return None
|
|
return result.fetchall()
|
|
|
|
async 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 = await self.execute(sql)
|
|
if result is None:
|
|
return None
|
|
return result.fetchall()
|
|
|
|
async 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 = await self.execute(sql)
|
|
if result is None:
|
|
return None
|
|
return result.fetchone()
|
|
|
|
async 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 = await self.execute(sql)
|
|
if result is None:
|
|
return None
|
|
return result.fetchall()
|
|
|
|
async 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 = await self.execute(sql)
|
|
if result is None:
|
|
return None
|
|
return result.fetchone()
|
|
|
|
async 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 = await self.execute(sql)
|
|
if result is None:
|
|
return None
|
|
return result.fetchone()
|
|
|
|
async def get_challenges(self, aime_id: int) -> Optional[List[Row]]:
|
|
sql = select(challenge).where(challenge.c.user == aime_id)
|
|
|
|
result = await self.execute(sql)
|
|
if result is None:
|
|
return None
|
|
return result.fetchall()
|
|
|
|
async 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 = await self.execute(sql)
|
|
if result is None:
|
|
return None
|
|
return result.fetchall()
|
|
|
|
async def get_theory_courses(self, aime_id: int) -> Optional[List[Row]]:
|
|
sql = select(theory_course).where(theory_course.c.user == aime_id)
|
|
|
|
result = await self.execute(sql)
|
|
if result is None:
|
|
return None
|
|
return result.fetchall()
|
|
|
|
async 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 = await self.execute(sql)
|
|
if result is None:
|
|
return None
|
|
return result.fetchall()
|
|
|
|
async 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 = await self.execute(sql)
|
|
if result is None:
|
|
return None
|
|
return result.fetchone()
|
|
|
|
async def get_theory_partners(self, aime_id: int) -> Optional[List[Row]]:
|
|
sql = select(theory_partner).where(theory_partner.c.user == aime_id)
|
|
|
|
result = await self.execute(sql)
|
|
if result is None:
|
|
return None
|
|
return result.fetchall()
|
|
|
|
async def get_theory_running(self, aime_id: int) -> Optional[List[Row]]:
|
|
sql = select(theory_running).where(theory_running.c.user == aime_id)
|
|
|
|
result = await self.execute(sql)
|
|
if result is None:
|
|
return None
|
|
return result.fetchall()
|
|
|
|
async 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 = await self.execute(sql)
|
|
if result is None:
|
|
return None
|
|
return result.fetchone()
|
|
|
|
async def get_vs_infos(self, aime_id: int) -> Optional[List[Row]]:
|
|
sql = select(vs_info).where(vs_info.c.user == aime_id)
|
|
|
|
result = await self.execute(sql)
|
|
if result is None:
|
|
return None
|
|
return result.fetchall()
|
|
|
|
async 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 = await self.execute(sql)
|
|
if result is None:
|
|
return None
|
|
return result.fetchone()
|
|
|
|
# This method returns a list of course_info
|
|
async def get_vs_course_infos_by_mode(self, aime_id: int, battle_mode: int) -> Optional[List[Row]]:
|
|
sql = select(vs_course_info).where(
|
|
and_(vs_course_info.c.user == aime_id, vs_course_info.c.battle_mode == battle_mode)
|
|
)
|
|
|
|
result = await self.execute(sql)
|
|
if result is None:
|
|
return None
|
|
return result.fetchall()
|
|
|
|
async def get_stamps(self, aime_id: int) -> Optional[List[Row]]:
|
|
sql = select(stamp).where(
|
|
and_(
|
|
stamp.c.user == aime_id,
|
|
)
|
|
)
|
|
|
|
result = await self.execute(sql)
|
|
if result is None:
|
|
return None
|
|
return result.fetchall()
|
|
|
|
async 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 = await self.execute(sql)
|
|
if result is None:
|
|
return None
|
|
return result.fetchone()
|
|
|
|
async def get_battle_gifts(self, aime_id: int, battle_gift_event_id: int) -> Optional[Row]:
|
|
sql = select(battle_gift).where(
|
|
and_(
|
|
battle_gift.c.user == aime_id,
|
|
battle_gift.c.battle_gift_event_id == battle_gift_event_id,
|
|
)
|
|
)
|
|
|
|
result = await self.execute(sql)
|
|
if result is None:
|
|
return None
|
|
return result.fetchall()
|
|
|
|
async 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 = await 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
|
|
|
|
async 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 = await 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
|
|
|
|
async 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 = await 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
|
|
|
|
async 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 = await 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
|
|
|
|
async 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 = await 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
|
|
|
|
async 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 = await 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
|
|
|
|
async 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 = await 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
|
|
|
|
async 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 = await 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
|
|
|
|
async 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 = await 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
|
|
|
|
async 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 = await 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
|
|
|
|
async 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 = await 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
|
|
|
|
async 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 = await 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
|
|
|
|
async 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 = await 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
|
|
|
|
async 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 = await 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
|
|
|
|
async 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 = await self.execute(conflict)
|
|
|
|
if result is None:
|
|
self.logger.warn(f"putstamp: Failed to update! aime_id: {aime_id}")
|
|
return None
|
|
return result.lastrowid
|
|
|
|
async 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 = await 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
|
|
|
|
async def put_battle_gift(
|
|
self, aime_id: int, battle_gift_data: Dict
|
|
) -> Optional[int]:
|
|
battle_gift_data["user"] = aime_id
|
|
|
|
sql = insert(battle_gift).values(**battle_gift_data)
|
|
conflict = sql.on_duplicate_key_update(**battle_gift_data)
|
|
result = await self.execute(conflict)
|
|
|
|
if result is None:
|
|
self.logger.warn(
|
|
f"put_battle_gift: Failed to update! aime_id: {aime_id}"
|
|
)
|
|
return None
|
|
return result.lastrowid |