forked from Hay1tsme/artemis
518 lines
18 KiB
Python
518 lines
18 KiB
Python
from typing import Optional, Dict, List
|
|
from sqlalchemy import Table, Column, UniqueConstraint, PrimaryKeyConstraint, and_
|
|
from sqlalchemy.types import Integer, String, TIMESTAMP, Boolean, JSON
|
|
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
|
|
|
|
profile = Table(
|
|
"wacca_profile",
|
|
metadata,
|
|
Column("id", Integer, primary_key=True, nullable=False),
|
|
Column(
|
|
"user",
|
|
ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"),
|
|
nullable=False,
|
|
),
|
|
Column("version", Integer),
|
|
Column("username", String(8), nullable=False),
|
|
Column("xp", Integer, server_default="0"),
|
|
Column("wp", Integer, server_default="0"),
|
|
Column("wp_total", Integer, server_default="0"),
|
|
Column("wp_spent", Integer, server_default="0"),
|
|
Column("dan_type", Integer, server_default="0"),
|
|
Column("dan_level", Integer, server_default="0"),
|
|
Column("title_0", Integer, server_default="0"),
|
|
Column("title_1", Integer, server_default="0"),
|
|
Column("title_2", Integer, server_default="0"),
|
|
Column("rating", Integer, server_default="0"),
|
|
Column("vip_expire_time", TIMESTAMP),
|
|
Column("always_vip", Boolean, server_default="0"),
|
|
Column("login_count", Integer, server_default="0"),
|
|
Column("login_count_consec", Integer, server_default="0"),
|
|
Column("login_count_days", Integer, server_default="0"),
|
|
Column("login_count_days_consec", Integer, server_default="0"),
|
|
Column("login_count_today", Integer, server_default="0"),
|
|
Column("playcount_single", Integer, server_default="0"),
|
|
Column("playcount_multi_vs", Integer, server_default="0"),
|
|
Column("playcount_multi_coop", Integer, server_default="0"),
|
|
Column("playcount_stageup", Integer, server_default="0"),
|
|
Column("friend_view_1", Integer),
|
|
Column("friend_view_2", Integer),
|
|
Column("friend_view_3", Integer),
|
|
Column("last_game_ver", String(50)),
|
|
Column("last_song_id", Integer, server_default="0"),
|
|
Column("last_song_difficulty", Integer, server_default="0"),
|
|
Column("last_folder_order", Integer, server_default="0"),
|
|
Column("last_folder_id", Integer, server_default="0"),
|
|
Column("last_song_order", Integer, server_default="0"),
|
|
Column("last_login_date", TIMESTAMP, server_default=func.now()),
|
|
Column("gate_tutorial_flags", JSON),
|
|
UniqueConstraint("user", "version", name="wacca_profile_uk"),
|
|
mysql_charset="utf8mb4",
|
|
)
|
|
|
|
option = Table(
|
|
"wacca_option",
|
|
metadata,
|
|
Column("id", Integer, primary_key=True, nullable=False),
|
|
Column(
|
|
"user",
|
|
ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"),
|
|
nullable=False,
|
|
),
|
|
Column("opt_id", Integer, nullable=False),
|
|
Column("value", Integer, nullable=False),
|
|
UniqueConstraint("user", "opt_id", name="wacca_option_uk"),
|
|
)
|
|
|
|
bingo = Table(
|
|
"wacca_bingo",
|
|
metadata,
|
|
Column(
|
|
"user",
|
|
ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"),
|
|
primary_key=True,
|
|
nullable=False,
|
|
),
|
|
Column("page_number", Integer, nullable=False),
|
|
Column("page_progress", JSON, nullable=False),
|
|
UniqueConstraint("user", "page_number", name="wacca_bingo_uk"),
|
|
mysql_charset="utf8mb4",
|
|
)
|
|
|
|
friend = Table(
|
|
"wacca_friend",
|
|
metadata,
|
|
Column(
|
|
"profile_sender",
|
|
ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"),
|
|
nullable=False,
|
|
),
|
|
Column(
|
|
"profile_reciever",
|
|
ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"),
|
|
nullable=False,
|
|
),
|
|
Column("is_accepted", Boolean, server_default="0"),
|
|
PrimaryKeyConstraint("profile_sender", "profile_reciever", name="arcade_owner_pk"),
|
|
mysql_charset="utf8mb4",
|
|
)
|
|
|
|
favorite = Table(
|
|
"wacca_favorite_song",
|
|
metadata,
|
|
Column("id", Integer, primary_key=True, nullable=False),
|
|
Column(
|
|
"user",
|
|
ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"),
|
|
nullable=False,
|
|
),
|
|
Column("song_id", Integer, nullable=False),
|
|
UniqueConstraint("user", "song_id", name="wacca_favorite_song_uk"),
|
|
mysql_charset="utf8mb4",
|
|
)
|
|
|
|
gate = Table(
|
|
"wacca_gate",
|
|
metadata,
|
|
Column("id", Integer, primary_key=True, nullable=False),
|
|
Column(
|
|
"user",
|
|
ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"),
|
|
nullable=False,
|
|
),
|
|
Column("gate_id", Integer, nullable=False),
|
|
Column("page", Integer, nullable=False, server_default="0"),
|
|
Column("progress", Integer, nullable=False, server_default="0"),
|
|
Column("loops", Integer, nullable=False, server_default="0"),
|
|
Column("last_used", TIMESTAMP, nullable=False, server_default=func.now()),
|
|
Column("mission_flag", Integer, nullable=False, server_default="0"),
|
|
Column("total_points", Integer, nullable=False, server_default="0"),
|
|
UniqueConstraint("user", "gate_id", name="wacca_gate_uk"),
|
|
)
|
|
|
|
|
|
class WaccaProfileData(BaseData):
|
|
def create_profile(
|
|
self, aime_id: int, username: str, version: int
|
|
) -> Optional[int]:
|
|
"""
|
|
Given a game version, aime id, and username, create a profile and return it's ID
|
|
"""
|
|
sql = insert(profile).values(user=aime_id, username=username, version=version)
|
|
|
|
conflict = sql.on_duplicate_key_update(username=sql.inserted.username)
|
|
|
|
result = self.execute(conflict)
|
|
if result is None:
|
|
self.logger.error(
|
|
f"{__name__} Failed to insert wacca profile! aime id: {aime_id} username: {username}"
|
|
)
|
|
return None
|
|
return result.lastrowid
|
|
|
|
def update_profile_playtype(
|
|
self, profile_id: int, play_type: int, game_version: str
|
|
) -> None:
|
|
sql = profile.update(profile.c.id == profile_id).values(
|
|
playcount_single=profile.c.playcount_single + 1
|
|
if play_type == 1
|
|
else profile.c.playcount_single,
|
|
playcount_multi_vs=profile.c.playcount_multi_vs + 1
|
|
if play_type == 2
|
|
else profile.c.playcount_multi_vs,
|
|
playcount_multi_coop=profile.c.playcount_multi_coop + 1
|
|
if play_type == 3
|
|
else profile.c.playcount_multi_coop,
|
|
playcount_stageup=profile.c.playcount_stageup + 1
|
|
if play_type == 4
|
|
else profile.c.playcount_stageup,
|
|
last_game_ver=game_version,
|
|
)
|
|
|
|
result = self.execute(sql)
|
|
if result is None:
|
|
self.logger.error(
|
|
f"update_profile: failed to update profile! profile: {profile_id}"
|
|
)
|
|
return None
|
|
|
|
def update_profile_lastplayed(
|
|
self,
|
|
profile_id: int,
|
|
last_song_id: int,
|
|
last_song_difficulty: int,
|
|
last_folder_order: int,
|
|
last_folder_id: int,
|
|
last_song_order: int,
|
|
) -> None:
|
|
sql = profile.update(profile.c.id == profile_id).values(
|
|
last_song_id=last_song_id,
|
|
last_song_difficulty=last_song_difficulty,
|
|
last_folder_order=last_folder_order,
|
|
last_folder_id=last_folder_id,
|
|
last_song_order=last_song_order,
|
|
)
|
|
result = self.execute(sql)
|
|
if result is None:
|
|
self.logger.error(
|
|
f"update_profile_lastplayed: failed to update profile! profile: {profile_id}"
|
|
)
|
|
return None
|
|
|
|
def update_profile_dan(
|
|
self, profile_id: int, dan_level: int, dan_type: int
|
|
) -> Optional[int]:
|
|
sql = profile.update(profile.c.id == profile_id).values(
|
|
dan_level=dan_level, dan_type=dan_type
|
|
)
|
|
|
|
result = self.execute(sql)
|
|
if result is None:
|
|
self.logger.warn(
|
|
f"update_profile_dan: Failed to update! profile {profile_id}"
|
|
)
|
|
return None
|
|
return result.lastrowid
|
|
|
|
def get_profile(self, profile_id: int = 0, aime_id: int = None) -> Optional[Row]:
|
|
"""
|
|
Given a game version and either a profile or aime id, return the profile
|
|
"""
|
|
if aime_id is not None:
|
|
sql = profile.select(profile.c.user == aime_id)
|
|
elif profile_id > 0:
|
|
sql = profile.select(profile.c.id == profile_id)
|
|
else:
|
|
self.logger.error(
|
|
f"get_profile: Bad arguments!! profile_id {profile_id} aime_id {aime_id}"
|
|
)
|
|
return None
|
|
|
|
result = self.execute(sql)
|
|
if result is None:
|
|
return None
|
|
return result.fetchone()
|
|
|
|
def get_options(self, user_id: int, option_id: int = None) -> Optional[List[Row]]:
|
|
"""
|
|
Get a specific user option for a profile, or all of them if none specified
|
|
"""
|
|
sql = option.select(
|
|
and_(
|
|
option.c.user == user_id,
|
|
option.c.opt_id == option_id if option_id is not None else True,
|
|
)
|
|
)
|
|
|
|
result = self.execute(sql)
|
|
if result is None:
|
|
return None
|
|
if option_id is not None:
|
|
return result.fetchone()
|
|
else:
|
|
return result.fetchall()
|
|
|
|
def update_option(self, user_id: int, option_id: int, value: int) -> Optional[int]:
|
|
sql = insert(option).values(user=user_id, opt_id=option_id, value=value)
|
|
|
|
conflict = sql.on_duplicate_key_update(value=value)
|
|
|
|
result = self.execute(conflict)
|
|
if result is None:
|
|
self.logger.error(
|
|
f"{__name__} failed to insert option! profile: {user_id}, option: {option_id}, value: {value}"
|
|
)
|
|
return None
|
|
|
|
return result.lastrowid
|
|
|
|
def add_favorite_song(self, user_id: int, song_id: int) -> Optional[int]:
|
|
sql = favorite.insert().values(user=user_id, song_id=song_id)
|
|
|
|
result = self.execute(sql)
|
|
if result is None:
|
|
self.logger.error(
|
|
f"{__name__} failed to insert favorite! profile: {user_id}, song_id: {song_id}"
|
|
)
|
|
return None
|
|
return result.lastrowid
|
|
|
|
def remove_favorite_song(self, user_id: int, song_id: int) -> None:
|
|
sql = favorite.delete(
|
|
and_(favorite.c.user == user_id, favorite.c.song_id == song_id)
|
|
)
|
|
|
|
result = self.execute(sql)
|
|
if result is None:
|
|
self.logger.error(
|
|
f"{__name__} failed to remove favorite! profile: {user_id}, song_id: {song_id}"
|
|
)
|
|
return None
|
|
|
|
def get_favorite_songs(self, user_id: int) -> Optional[List[Row]]:
|
|
sql = favorite.select(favorite.c.user == user_id)
|
|
|
|
result = self.execute(sql)
|
|
if result is None:
|
|
return None
|
|
return result.fetchall()
|
|
|
|
def get_gates(self, user_id: int) -> Optional[List[Row]]:
|
|
sql = select(gate).where(gate.c.user == user_id)
|
|
|
|
result = self.execute(sql)
|
|
if result is None:
|
|
return None
|
|
return result.fetchall()
|
|
|
|
def update_gate(
|
|
self,
|
|
user_id: int,
|
|
gate_id: int,
|
|
page: int,
|
|
progress: int,
|
|
loop: int,
|
|
mission_flag: int,
|
|
total_points: int,
|
|
) -> Optional[int]:
|
|
sql = insert(gate).values(
|
|
user=user_id,
|
|
gate_id=gate_id,
|
|
page=page,
|
|
progress=progress,
|
|
loops=loop,
|
|
mission_flag=mission_flag,
|
|
total_points=total_points,
|
|
)
|
|
|
|
conflict = sql.on_duplicate_key_update(
|
|
page=sql.inserted.page,
|
|
progress=sql.inserted.progress,
|
|
loops=sql.inserted.loops,
|
|
mission_flag=sql.inserted.mission_flag,
|
|
total_points=sql.inserted.total_points,
|
|
)
|
|
|
|
result = self.execute(conflict)
|
|
if result is None:
|
|
self.logger.error(
|
|
f"{__name__} failed to update gate! user: {user_id}, gate_id: {gate_id}"
|
|
)
|
|
return None
|
|
return result.lastrowid
|
|
|
|
def get_friends(self, user_id: int) -> Optional[List[Row]]:
|
|
sql = friend.select(friend.c.profile_sender == user_id)
|
|
|
|
result = self.execute(sql)
|
|
if result is None:
|
|
return None
|
|
return result.fetchall()
|
|
|
|
def profile_to_aime_user(self, profile_id: int) -> Optional[int]:
|
|
sql = select(profile.c.user).where(profile.c.id == profile_id)
|
|
|
|
result = self.execute(sql)
|
|
if result is None:
|
|
self.logger.info(
|
|
f"profile_to_aime_user: No user found for profile {profile_id}"
|
|
)
|
|
return None
|
|
|
|
this_profile = result.fetchone()
|
|
if this_profile is None:
|
|
self.logger.info(
|
|
f"profile_to_aime_user: No user found for profile {profile_id}"
|
|
)
|
|
return None
|
|
|
|
return this_profile["user"]
|
|
|
|
def session_login(
|
|
self, profile_id: int, is_new_day: bool, is_consec_day: bool
|
|
) -> None:
|
|
# TODO: Reset consec days counter
|
|
sql = profile.update(profile.c.id == profile_id).values(
|
|
login_count=profile.c.login_count + 1,
|
|
login_count_consec=profile.c.login_count_consec + 1,
|
|
login_count_days=profile.c.login_count_days + 1
|
|
if is_new_day
|
|
else profile.c.login_count_days,
|
|
login_count_days_consec=profile.c.login_count_days_consec + 1
|
|
if is_new_day and is_consec_day
|
|
else profile.c.login_count_days_consec,
|
|
login_count_today=1 if is_new_day else profile.c.login_count_today + 1,
|
|
last_login_date=func.now(),
|
|
)
|
|
|
|
result = self.execute(sql)
|
|
if result is None:
|
|
self.logger.error(
|
|
f"session_login: failed to update profile! profile: {profile_id}"
|
|
)
|
|
return None
|
|
|
|
def session_logout(self, profile_id: int) -> None:
|
|
sql = profile.update(profile.c.id == id).values(login_count_consec=0)
|
|
|
|
result = self.execute(sql)
|
|
if result is None:
|
|
self.logger.error(
|
|
f"{__name__} failed to update profile! profile: {profile_id}"
|
|
)
|
|
return None
|
|
|
|
def add_xp(self, profile_id: int, xp: int) -> None:
|
|
sql = profile.update(profile.c.id == profile_id).values(xp=profile.c.xp + xp)
|
|
|
|
result = self.execute(sql)
|
|
if result is None:
|
|
self.logger.error(
|
|
f"add_xp: Failed to update profile! profile_id {profile_id} xp {xp}"
|
|
)
|
|
return None
|
|
|
|
def add_wp(self, profile_id: int, wp: int) -> None:
|
|
sql = profile.update(profile.c.id == profile_id).values(
|
|
wp=profile.c.wp + wp,
|
|
wp_total=profile.c.wp_total + wp,
|
|
)
|
|
|
|
result = self.execute(sql)
|
|
if result is None:
|
|
self.logger.error(
|
|
f"add_wp: Failed to update profile! profile_id {profile_id} wp {wp}"
|
|
)
|
|
return None
|
|
|
|
def spend_wp(self, profile_id: int, wp: int) -> None:
|
|
sql = profile.update(profile.c.id == profile_id).values(
|
|
wp=profile.c.wp - wp,
|
|
wp_spent=profile.c.wp_spent + wp,
|
|
)
|
|
|
|
result = self.execute(sql)
|
|
if result is None:
|
|
self.logger.error(
|
|
f"spend_wp: Failed to update profile! profile_id {profile_id} wp {wp}"
|
|
)
|
|
return None
|
|
|
|
def activate_vip(self, profile_id: int, expire_time) -> None:
|
|
sql = profile.update(profile.c.id == profile_id).values(
|
|
vip_expire_time=expire_time
|
|
)
|
|
|
|
result = self.execute(sql)
|
|
if result is None:
|
|
self.logger.error(
|
|
f"activate_vip: Failed to update profile! profile_id {profile_id} expire_time {expire_time}"
|
|
)
|
|
return None
|
|
|
|
def update_user_rating(self, profile_id: int, new_rating: int) -> None:
|
|
sql = profile.update(profile.c.id == profile_id).values(rating=new_rating)
|
|
|
|
result = self.execute(sql)
|
|
if result is None:
|
|
self.logger.error(
|
|
f"update_user_rating: Failed to update profile! profile_id {profile_id} new_rating {new_rating}"
|
|
)
|
|
return None
|
|
|
|
def update_bingo(self, aime_id: int, page: int, progress: int) -> Optional[int]:
|
|
sql = insert(bingo).values(
|
|
user=aime_id, page_number=page, page_progress=progress
|
|
)
|
|
|
|
conflict = sql.on_duplicate_key_update(page_number=page, page_progress=progress)
|
|
|
|
result = self.execute(conflict)
|
|
if result is None:
|
|
self.logger.error(f"put_bingo: failed to update! aime_id: {aime_id}")
|
|
return None
|
|
return result.lastrowid
|
|
|
|
def get_bingo(self, aime_id: int) -> Optional[List[Row]]:
|
|
sql = select(bingo).where(bingo.c.user == aime_id)
|
|
|
|
result = self.execute(sql)
|
|
if result is None:
|
|
return None
|
|
return result.fetchone()
|
|
|
|
def get_bingo_page(self, aime_id: int, page: Dict) -> Optional[List[Row]]:
|
|
sql = select(bingo).where(
|
|
and_(bingo.c.user == aime_id, bingo.c.page_number == page)
|
|
)
|
|
|
|
result = self.execute(sql)
|
|
if result is None:
|
|
return None
|
|
return result.fetchone()
|
|
|
|
def update_vip_time(self, profile_id: int, time_left) -> None:
|
|
sql = profile.update(profile.c.id == profile_id).values(
|
|
vip_expire_time=time_left
|
|
)
|
|
|
|
result = self.execute(sql)
|
|
if result is None:
|
|
self.logger.error(f"Failed to update VIP time for profile {profile_id}")
|
|
|
|
def update_tutorial_flags(self, profile_id: int, flags: Dict) -> None:
|
|
sql = profile.update(profile.c.id == profile_id).values(
|
|
gate_tutorial_flags=flags
|
|
)
|
|
|
|
result = self.execute(sql)
|
|
if result is None:
|
|
self.logger.error(
|
|
f"Failed to update tutorial flags for profile {profile_id}"
|
|
)
|