pgw00g/artemis
1
0
Fork 0
forked from Dniel97/artemis
Code Pull Requests Activity
d024b2eeb8
artemis/titles/idac/schema/profile.py

441 lines
17 KiB
Python
Raw Blame History

from typing import Dict, List, Optional
from sqlalchemy import Table, Column, UniqueConstraint, PrimaryKeyConstraint, and_
from sqlalchemy.types import Integer, String, TIMESTAMP, Boolean, JSON, BigInteger
from sqlalchemy.engine.base import Connection
from sqlalchemy.schema import ForeignKey
from sqlalchemy.sql import func, select
from sqlalchemy.engine import Row
from sqlalchemy.dialects.mysql import insert
from core.data.schema import BaseData, metadata
from core.config import CoreConfig
profile = Table(
"idac_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, nullable=False),
Column("username", String(8)),
Column("country", Integer),
Column("store", Integer),
Column("team_id", Integer, server_default="0"),
Column("total_play", Integer, server_default="0"),
Column("daily_play", Integer, server_default="0"),
Column("day_play", Integer, server_default="0"),
Column("mileage", Integer, server_default="0"),
Column("asset_version", Integer, server_default="1"),
Column("last_play_date", TIMESTAMP, server_default=func.now()),
Column("mytitle_id", Integer, server_default="0"),
Column("mytitle_efffect_id", Integer, server_default="0"),
Column("sticker_id", Integer, server_default="0"),
Column("sticker_effect_id", Integer, server_default="0"),
Column("papercup_id", Integer, server_default="0"),
Column("tachometer_id", Integer, server_default="0"),
Column("aura_id", Integer, server_default="0"),
Column("aura_color_id", Integer, server_default="0"),
Column("aura_line_id", Integer, server_default="0"),
Column("bgm_id", Integer, server_default="0"),
Column("keyholder_id", Integer, server_default="0"),
Column("start_menu_bg_id", Integer, server_default="0"),
Column("use_car_id", Integer, server_default="1"),
Column("use_style_car_id", Integer, server_default="1"),
Column("bothwin_count", Integer, server_default="0"),
Column("bothwin_score", Integer, server_default="0"),
Column("subcard_count", Integer, server_default="0"),
Column("vs_history", Integer, server_default="0"),
Column("stamp_key_assign_0", Integer),
Column("stamp_key_assign_1", Integer),
Column("stamp_key_assign_2", Integer),
Column("stamp_key_assign_3", Integer),
Column("name_change_category", Integer, server_default="0"),
Column("factory_disp", Integer, server_default="0"),
Column("create_date", TIMESTAMP, server_default=func.now()),
Column("cash", Integer, server_default="0"),
Column("dressup_point", Integer, server_default="0"),
Column("avatar_point", Integer, server_default="0"),
Column("total_cash", Integer, server_default="0"),
UniqueConstraint("user", "version", name="idac_profile_uk"),
mysql_charset="utf8mb4",
)
# No point setting defaults since the game sends everything on profile creation anyway
config = Table(
"idac_profile_config",
metadata,
Column("id", Integer, primary_key=True, nullable=False),
Column(
"user",
ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"),
nullable=False,
),
Column("config_id", Integer),
Column("steering_intensity", Integer),
Column("transmission_type", Integer),
Column("default_viewpoint", Integer),
Column("favorite_bgm", Integer),
Column("bgm_volume", Integer),
Column("se_volume", Integer),
Column("master_volume", Integer),
Column("store_battle_policy", Integer),
Column("battle_onomatope_display", Integer),
Column("cornering_guide", Integer),
Column("minimap", Integer),
Column("line_guide", Integer),
Column("ghost", Integer),
Column("race_exit", Integer),
Column("result_skip", Integer),
Column("stamp_select_skip", Integer),
UniqueConstraint("user", name="idac_profile_config_uk"),
mysql_charset="utf8mb4",
)
# No point setting defaults since the game sends everything on profile creation anyway
avatar = Table(
"idac_profile_avatar",
metadata,
Column("id", Integer, primary_key=True, nullable=False),
Column(
"user",
ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"),
nullable=False,
),
Column("sex", Integer),
Column("face", Integer),
Column("eye", Integer),
Column("mouth", Integer),
Column("hair", Integer),
Column("glasses", Integer),
Column("face_accessory", Integer),
Column("body", Integer),
Column("body_accessory", Integer),
Column("behind", Integer),
Column("bg", Integer),
Column("effect", Integer),
Column("special", Integer),
UniqueConstraint("user", name="idac_profile_avatar_uk"),
mysql_charset="utf8mb4",
)
rank = Table(
"idac_profile_rank",
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("story_rank_exp", Integer, server_default="0"),
Column("story_rank", Integer, server_default="1"),
Column("time_trial_rank_exp", Integer, server_default="0"),
Column("time_trial_rank", Integer, server_default="1"),
Column("online_battle_rank_exp", Integer, server_default="0"),
Column("online_battle_rank", Integer, server_default="1"),
Column("store_battle_rank_exp", Integer, server_default="0"),
Column("store_battle_rank", Integer, server_default="1"),
Column("theory_exp", Integer, server_default="0"),
Column("theory_rank", Integer, server_default="1"),
Column("pride_group_id", Integer, server_default="0"),
Column("pride_point", Integer, server_default="0"),
Column("grade_exp", Integer, server_default="0"),
Column("grade", Integer, server_default="1"),
Column("grade_reward_dist", Integer, server_default="0"),
Column("story_rank_reward_dist", Integer, server_default="0"),
Column("time_trial_rank_reward_dist", Integer, server_default="0"),
Column("online_battle_rank_reward_dist", Integer, server_default="0"),
Column("store_battle_rank_reward_dist", Integer, server_default="0"),
Column("theory_rank_reward_dist", Integer, server_default="0"),
Column("max_attained_online_battle_rank", Integer, server_default="1"),
Column("max_attained_pride_point", Integer, server_default="0"),
Column("is_last_max", Integer, server_default="0"),
UniqueConstraint("user", "version", name="idac_profile_rank_uk"),
mysql_charset="utf8mb4",
)
stock = Table(
"idac_profile_stock",
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("mytitle_list", String(1024), server_default=""),
Column("mytitle_new_list", String(1024), server_default=""),
Column("avatar_face_list", String(255), server_default=""),
Column("avatar_face_new_list", String(255), server_default=""),
Column("avatar_eye_list", String(255), server_default=""),
Column("avatar_eye_new_list", String(255), server_default=""),
Column("avatar_hair_list", String(255), server_default=""),
Column("avatar_hair_new_list", String(255), server_default=""),
Column("avatar_body_list", String(255), server_default=""),
Column("avatar_body_new_list", String(255), server_default=""),
Column("avatar_mouth_list", String(255), server_default=""),
Column("avatar_mouth_new_list", String(255), server_default=""),
Column("avatar_glasses_list", String(255), server_default=""),
Column("avatar_glasses_new_list", String(255), server_default=""),
Column("avatar_face_accessory_list", String(255), server_default=""),
Column("avatar_face_accessory_new_list", String(255), server_default=""),
Column("avatar_body_accessory_list", String(255), server_default=""),
Column("avatar_body_accessory_new_list", String(255), server_default=""),
Column("avatar_behind_list", String(255), server_default=""),
Column("avatar_behind_new_list", String(255), server_default=""),
Column("avatar_bg_list", String(255), server_default=""),
Column("avatar_bg_new_list", String(255), server_default=""),
Column("avatar_effect_list", String(255), server_default=""),
Column("avatar_effect_new_list", String(255), server_default=""),
Column("avatar_special_list", String(255), server_default=""),
Column("avatar_special_new_list", String(255), server_default=""),
Column("stamp_list", String(255), server_default=""),
Column("stamp_new_list", String(255), server_default=""),
Column("keyholder_list", String(256), server_default=""),
Column("keyholder_new_list", String(256), server_default=""),
Column("papercup_list", String(255), server_default=""),
Column("papercup_new_list", String(255), server_default=""),
Column("tachometer_list", String(255), server_default=""),
Column("tachometer_new_list", String(255), server_default=""),
Column("aura_list", String(255), server_default=""),
Column("aura_new_list", String(255), server_default=""),
Column("aura_color_list", String(255), server_default=""),
Column("aura_color_new_list", String(255), server_default=""),
Column("aura_line_list", String(255), server_default=""),
Column("aura_line_new_list", String(255), server_default=""),
Column("bgm_list", String(255), server_default=""),
Column("bgm_new_list", String(255), server_default=""),
Column("dx_color_list", String(255), server_default=""),
Column("dx_color_new_list", String(255), server_default=""),
Column("start_menu_bg_list", String(255), server_default=""),
Column("start_menu_bg_new_list", String(255), server_default=""),
Column("under_neon_list", String(255), server_default=""),
UniqueConstraint("user", "version", name="idac_profile_stock_uk"),
mysql_charset="utf8mb4",
)
theory = Table(
"idac_profile_theory",
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("play_count", Integer, server_default="0"),
Column("play_count_multi", Integer, server_default="0"),
Column("partner_id", Integer),
Column("partner_progress", Integer),
Column("partner_progress_score", Integer),
Column("practice_start_rank", Integer, server_default="0"),
Column("general_flag", Integer, server_default="0"),
Column("vs_history", Integer, server_default="0"),
Column("vs_history_multi", Integer, server_default="0"),
Column("win_count", Integer, server_default="0"),
Column("win_count_multi", Integer, server_default="0"),
UniqueConstraint("user", "version", name="idac_profile_theory_uk"),
mysql_charset="utf8mb4",
)
class IDACProfileData(BaseData):
def __init__(self, cfg: CoreConfig, conn: Connection) -> None:
super().__init__(cfg, conn)
self.date_time_format_ext = (
"%Y-%m-%d %H:%M:%S.%f" # needs to be lopped off at [:-5]
)
self.date_time_format_short = "%Y-%m-%d"
def get_profile(self, aime_id: int, version: int) -> Optional[Row]:
sql = select(profile).where(
and_(
profile.c.user == aime_id,
profile.c.version == version,
)
)
result = self.execute(sql)
if result is None:
return None
return result.fetchone()
def get_different_random_profiles(
self, aime_id: int, version: int, count: int = 9
) -> Optional[Row]:
sql = (
select(profile)
.where(
and_(
profile.c.user != aime_id,
profile.c.version == version,
)
)
.order_by(func.rand())
.limit(count)
)
result = self.execute(sql)
if result is None:
return None
return result.fetchall()
def get_profile_config(self, aime_id: int) -> Optional[Row]:
sql = select(config).where(
and_(
config.c.user == aime_id,
)
)
result = self.execute(sql)
if result is None:
return None
return result.fetchone()
def get_profile_avatar(self, aime_id: int) -> Optional[Row]:
sql = select(avatar).where(
and_(
avatar.c.user == aime_id,
)
)
result = self.execute(sql)
if result is None:
return None
return result.fetchone()
def get_profile_rank(self, aime_id: int, version: int) -> Optional[Row]:
sql = select(rank).where(
and_(
rank.c.user == aime_id,
rank.c.version == version,
)
)
result = self.execute(sql)
if result is None:
return None
return result.fetchone()
def get_profile_stock(self, aime_id: int, version: int) -> Optional[Row]:
sql = select(stock).where(
and_(
stock.c.user == aime_id,
stock.c.version == version,
)
)
result = self.execute(sql)
if result is None:
return None
return result.fetchone()
def get_profile_theory(self, aime_id: int, version: int) -> Optional[Row]:
sql = select(theory).where(
and_(
theory.c.user == aime_id,
theory.c.version == version,
)
)
result = self.execute(sql)
if result is None:
return None
return result.fetchone()
def put_profile(
self, aime_id: int, version: int, profile_data: Dict
) -> Optional[int]:
profile_data["user"] = aime_id
profile_data["version"] = version
sql = insert(profile).values(**profile_data)
conflict = sql.on_duplicate_key_update(**profile_data)
result = self.execute(conflict)
if result is None:
self.logger.warn(f"put_profile: Failed to update! aime_id: {aime_id}")
return None
return result.lastrowid
def put_profile_config(self, aime_id: int, config_data: Dict) -> Optional[int]:
config_data["user"] = aime_id
sql = insert(config).values(**config_data)
conflict = sql.on_duplicate_key_update(**config_data)
result = self.execute(conflict)
if result is None:
self.logger.warn(
f"put_profile_config: Failed to update! aime_id: {aime_id}"
)
return None
return result.lastrowid
def put_profile_avatar(self, aime_id: int, avatar_data: Dict) -> Optional[int]:
avatar_data["user"] = aime_id
sql = insert(avatar).values(**avatar_data)
conflict = sql.on_duplicate_key_update(**avatar_data)
result = self.execute(conflict)
if result is None:
self.logger.warn(
f"put_profile_avatar: Failed to update! aime_id: {aime_id}"
)
return None
return result.lastrowid
def put_profile_rank(
self, aime_id: int, version: int, rank_data: Dict
) -> Optional[int]:
rank_data["user"] = aime_id
rank_data["version"] = version
sql = insert(rank).values(**rank_data)
conflict = sql.on_duplicate_key_update(**rank_data)
result = self.execute(conflict)
if result is None:
self.logger.warn(f"put_profile_rank: Failed to update! aime_id: {aime_id}")
return None
return result.lastrowid
def put_profile_stock(
self, aime_id: int, version: int, stock_data: Dict
) -> Optional[int]:
stock_data["user"] = aime_id
stock_data["version"] = version
sql = insert(stock).values(**stock_data)
conflict = sql.on_duplicate_key_update(**stock_data)
result = self.execute(conflict)
if result is None:
self.logger.warn(f"put_profile_stock: Failed to update! aime_id: {aime_id}")
return None
return result.lastrowid
def put_profile_theory(
self, aime_id: int, version: int, theory_data: Dict
) -> Optional[int]:
theory_data["user"] = aime_id
theory_data["version"] = version
sql = insert(theory).values(**theory_data)
conflict = sql.on_duplicate_key_update(**theory_data)
result = self.execute(conflict)
if result is None:
self.logger.warn(
f"put_profile_theory: Failed to update! aime_id: {aime_id}"
)
return None
return result.lastrowid
View Git Blame Copy Permalink