from typing import Dict, List, Optional from sqlalchemy import Table, Column, UniqueConstraint, PrimaryKeyConstraint, and_ from sqlalchemy.types import Integer, String, TIMESTAMP, Boolean, JSON, Float from sqlalchemy.engine.base import Connection from sqlalchemy.engine import Row from sqlalchemy.schema import ForeignKey from sqlalchemy.sql import func, select from sqlalchemy.dialects.mysql import insert from core.data.schema import BaseData, metadata quest = Table( "sao_static_quest", metadata, Column("id", Integer, primary_key=True, nullable=False), Column("version", Integer), Column("questSceneId", Integer), Column("sortNo", Integer), Column("name", String(255)), Column("enabled", Boolean), UniqueConstraint( "version", "questSceneId", name="sao_static_quest_uk" ), mysql_charset="utf8mb4", ) hero = Table( "sao_static_hero_list", metadata, Column("id", Integer, primary_key=True, nullable=False), Column("version", Integer), Column("heroLogId", Integer), Column("name", String(255)), Column("nickname", String(255)), Column("rarity", Integer), Column("skillTableSubId", Integer), Column("awakeningExp", Integer), Column("flavorText", String(255)), Column("enabled", Boolean), UniqueConstraint( "version", "heroLogId", name="sao_static_hero_list_uk" ), mysql_charset="utf8mb4", ) equipment = Table( "sao_static_equipment_list", metadata, Column("id", Integer, primary_key=True, nullable=False), Column("version", Integer), Column("equipmentId", Integer), Column("equipmentType", Integer), Column("weaponTypeId", Integer), Column("name", String(255)), Column("rarity", Integer), Column("flavorText", String(255)), Column("enabled", Boolean), UniqueConstraint( "version", "equipmentId", name="sao_static_equipment_list_uk" ), mysql_charset="utf8mb4", ) item = Table( "sao_static_item_list", metadata, Column("id", Integer, primary_key=True, nullable=False), Column("version", Integer), Column("itemId", Integer), Column("itemTypeId", Integer), Column("name", String(255)), Column("rarity", Integer), Column("flavorText", String(255)), Column("enabled", Boolean), UniqueConstraint( "version", "itemId", name="sao_static_item_list_uk" ), mysql_charset="utf8mb4", ) support = Table( "sao_static_support_log_list", metadata, Column("id", Integer, primary_key=True, nullable=False), Column("version", Integer), Column("supportLogId", Integer), Column("charaId", Integer), Column("name", String(255)), Column("rarity", Integer), Column("salePrice", Integer), Column("skillName", String(255)), Column("enabled", Boolean), UniqueConstraint( "version", "supportLogId", name="sao_static_support_log_list_uk" ), mysql_charset="utf8mb4", ) title = Table( "sao_static_title_list", metadata, Column("id", Integer, primary_key=True, nullable=False), Column("version", Integer), Column("titleId", Integer), Column("displayName", String(255)), Column("requirement", Integer), Column("rank", Integer), Column("imageFilePath", String(255)), Column("enabled", Boolean), UniqueConstraint( "version", "titleId", name="sao_static_title_list_uk" ), mysql_charset="utf8mb4", ) class SaoStaticData(BaseData): def put_quest( self, questSceneId: int, version: int, sortNo: int, name: str, enabled: bool ) -> Optional[int]: sql = insert(quest).values( questSceneId=questSceneId, version=version, sortNo=sortNo, name=name, tutorial=tutorial, ) conflict = sql.on_duplicate_key_update( name=name, questSceneId=questSceneId, version=version ) result = self.execute(conflict) if result is None: return None return result.lastrowid def put_hero( self, version: int, heroLogId: int, name: str, nickname: str, rarity: int, skillTableSubId: int, awakeningExp: int, flavorText: str, enabled: bool ) -> Optional[int]: sql = insert(hero).values( version=version, heroLogId=heroLogId, name=name, nickname=nickname, rarity=rarity, skillTableSubId=skillTableSubId, awakeningExp=awakeningExp, flavorText=flavorText, enabled=enabled ) conflict = sql.on_duplicate_key_update( name=name, heroLogId=heroLogId ) result = self.execute(conflict) if result is None: return None return result.lastrowid def put_equipment( self, version: int, equipmentId: int, name: str, equipmentType: int, weaponTypeId:int, rarity: int, flavorText: str, enabled: bool ) -> Optional[int]: sql = insert(equipment).values( version=version, equipmentId=equipmentId, name=name, equipmentType=equipmentType, weaponTypeId=weaponTypeId, rarity=rarity, flavorText=flavorText, enabled=enabled ) conflict = sql.on_duplicate_key_update( name=name, equipmentId=equipmentId ) result = self.execute(conflict) if result is None: return None return result.lastrowid def put_item( self, version: int, itemId: int, name: str, itemTypeId: int, rarity: int, flavorText: str, enabled: bool ) -> Optional[int]: sql = insert(item).values( version=version, itemId=itemId, name=name, itemTypeId=itemTypeId, rarity=rarity, flavorText=flavorText, enabled=enabled ) conflict = sql.on_duplicate_key_update( name=name, itemId=itemId ) result = self.execute(conflict) if result is None: return None return result.lastrowid def put_support_log( self, version: int, supportLogId: int, charaId: int, name: str, rarity: int, salePrice: int, skillName: str, enabled: bool ) -> Optional[int]: sql = insert(support).values( version=version, supportLogId=supportLogId, charaId=charaId, name=name, rarity=rarity, salePrice=salePrice, skillName=skillName, enabled=enabled ) conflict = sql.on_duplicate_key_update( name=name, supportLogId=supportLogId ) result = self.execute(conflict) if result is None: return None return result.lastrowid def put_title( self, version: int, titleId: int, displayName: str, requirement: int, rank: int, imageFilePath: str, enabled: bool ) -> Optional[int]: sql = insert(title).values( version=version, titleId=titleId, displayName=displayName, requirement=requirement, rank=rank, imageFilePath=imageFilePath, enabled=enabled ) conflict = sql.on_duplicate_key_update( displayName=displayName, titleId=titleId ) result = self.execute(conflict) if result is None: return None return result.lastrowid def get_quests_ids(self, version: int, enabled: bool) -> Optional[List[Dict]]: sql = quest.select(quest.c.version == version and quest.c.enabled == enabled).order_by( quest.c.questSceneId.asc() ) result = self.execute(sql) if result is None: return None return [list[2] for list in result.fetchall()] def get_hero_id(self, heroLogId: int) -> Optional[Dict]: sql = hero.select(hero.c.heroLogId == heroLogId) result = self.execute(sql) if result is None: return None return result.fetchone() def get_hero_ids(self, version: int, enabled: bool) -> Optional[List[Dict]]: sql = hero.select(hero.c.version == version and hero.c.enabled == enabled).order_by( hero.c.heroLogId.asc() ) result = self.execute(sql) if result is None: return None return [list[2] for list in result.fetchall()] def get_equipment_id(self, equipmentId: int) -> Optional[Dict]: sql = equipment.select(equipment.c.equipmentId == equipmentId) result = self.execute(sql) if result is None: return None return result.fetchone() def get_equipment_ids(self, version: int, enabled: bool) -> Optional[List[Dict]]: sql = equipment.select(equipment.c.version == version and equipment.c.enabled == enabled).order_by( equipment.c.equipmentId.asc() ) result = self.execute(sql) if result is None: return None return [list[2] for list in result.fetchall()] def get_item_ids(self, version: int, enabled: bool) -> Optional[List[Dict]]: sql = item.select(item.c.version == version and item.c.enabled == enabled).order_by( item.c.itemId.asc() ) result = self.execute(sql) if result is None: return None return [list[2] for list in result.fetchall()] def get_support_log_ids(self, version: int, enabled: bool) -> Optional[List[Dict]]: sql = support.select(support.c.version == version and support.c.enabled == enabled).order_by( support.c.supportLogId.asc() ) result = self.execute(sql) if result is None: return None return [list[2] for list in result.fetchall()] def get_title_ids(self, version: int, enabled: bool) -> Optional[List[Dict]]: sql = title.select(title.c.version == version and title.c.enabled == enabled).order_by( title.c.titleId.asc() ) result = self.execute(sql) if result is None: return None return [list[2] for list in result.fetchall()]