artemis/titles/sao/schema/static.py

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()]