1
0
mirror of synced 2024-12-27 03:34:08 +01:00
artemis/titles/sao/schema/static.py

368 lines
12 KiB
Python
Raw Normal View History

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",
)
2023-06-03 17:42:50 +02:00
rare_drop = Table(
"sao_static_rare_drop_list",
metadata,
Column("id", Integer, primary_key=True, nullable=False),
Column("version", Integer),
Column("questRareDropId", Integer),
Column("commonRewardId", Integer),
Column("enabled", Boolean),
UniqueConstraint(
"version", "questRareDropId", "commonRewardId", name="sao_static_rare_drop_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):
2024-01-09 20:42:17 +01:00
async 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,
2024-01-09 20:42:17 +01:00
enabled=enabled,
)
conflict = sql.on_duplicate_key_update(
name=name, questSceneId=questSceneId, version=version
)
2024-01-09 20:42:17 +01:00
result = await self.execute(conflict)
if result is None:
return None
return result.lastrowid
2024-01-09 20:42:17 +01:00
async 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
)
2024-01-09 20:42:17 +01:00
result = await self.execute(conflict)
if result is None:
return None
return result.lastrowid
2024-01-09 20:42:17 +01:00
async 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
)
2024-01-09 20:42:17 +01:00
result = await self.execute(conflict)
if result is None:
return None
return result.lastrowid
2024-01-09 20:42:17 +01:00
async 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
)
2024-01-09 20:42:17 +01:00
result = await self.execute(conflict)
if result is None:
return None
return result.lastrowid
2024-01-09 20:42:17 +01:00
async 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
)
2024-01-09 20:42:17 +01:00
result = await self.execute(conflict)
if result is None:
return None
return result.lastrowid
2023-06-03 17:42:50 +02:00
2024-01-09 20:42:17 +01:00
async def put_rare_drop( self, version: int, questRareDropId: int, commonRewardId: int, enabled: bool ) -> Optional[int]:
2023-06-03 17:42:50 +02:00
sql = insert(rare_drop).values(
version=version,
questRareDropId=questRareDropId,
commonRewardId=commonRewardId,
enabled=enabled,
)
conflict = sql.on_duplicate_key_update(
questRareDropId=questRareDropId, commonRewardId=commonRewardId, version=version
)
2024-01-09 20:42:17 +01:00
result = await self.execute(conflict)
2023-06-03 17:42:50 +02:00
if result is None:
return None
return result.lastrowid
2024-01-09 20:42:17 +01:00
async 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
)
2024-01-09 20:42:17 +01:00
result = await self.execute(conflict)
if result is None:
return None
return result.lastrowid
2024-01-09 20:42:17 +01:00
async def get_quests_id(self, sortNo: int) -> Optional[Dict]:
sql = quest.select(quest.c.sortNo == sortNo)
2024-01-09 20:42:17 +01:00
result = await self.execute(sql)
if result is None:
return None
return result.fetchone()
2024-01-09 20:42:17 +01:00
async 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()
)
2024-01-09 20:42:17 +01:00
result = await self.execute(sql)
if result is None:
return None
return [list[2] for list in result.fetchall()]
2024-01-09 20:42:17 +01:00
async def get_hero_id(self, heroLogId: int) -> Optional[Dict]:
2023-05-30 20:29:50 +02:00
sql = hero.select(hero.c.heroLogId == heroLogId)
2024-01-09 20:42:17 +01:00
result = await self.execute(sql)
2023-05-30 20:29:50 +02:00
if result is None:
return None
return result.fetchone()
2024-01-09 20:42:17 +01:00
async 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()
)
2024-01-09 20:42:17 +01:00
result = await self.execute(sql)
if result is None:
return None
return [list[2] for list in result.fetchall()]
2024-01-09 20:42:17 +01:00
async def get_equipment_id(self, equipmentId: int) -> Optional[Dict]:
sql = equipment.select(equipment.c.equipmentId == equipmentId)
2024-01-09 20:42:17 +01:00
result = await self.execute(sql)
if result is None:
return None
return result.fetchone()
2024-01-09 20:42:17 +01:00
async 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()
)
2024-01-09 20:42:17 +01:00
result = await self.execute(sql)
if result is None:
return None
return [list[2] for list in result.fetchall()]
2024-01-09 20:42:17 +01:00
async def get_item_id(self, itemId: int) -> Optional[Dict]:
sql = item.select(item.c.itemId == itemId)
2024-01-09 20:42:17 +01:00
result = await self.execute(sql)
if result is None:
return None
return result.fetchone()
2023-06-03 17:42:50 +02:00
2024-01-09 20:42:17 +01:00
async def get_rare_drop_id(self, questRareDropId: int) -> Optional[Dict]:
2023-06-03 17:42:50 +02:00
sql = rare_drop.select(rare_drop.c.questRareDropId == questRareDropId)
2024-01-09 20:42:17 +01:00
result = await self.execute(sql)
2023-06-03 17:42:50 +02:00
if result is None:
return None
return result.fetchone()
2024-01-09 20:42:17 +01:00
async 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()
)
2024-01-09 20:42:17 +01:00
result = await self.execute(sql)
if result is None:
return None
return [list[2] for list in result.fetchall()]
2024-01-09 20:42:17 +01:00
async 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()
)
2024-01-09 20:42:17 +01:00
result = await self.execute(sql)
if result is None:
return None
return [list[2] for list in result.fetchall()]
2024-01-09 20:42:17 +01:00
async 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()
)
2024-01-09 20:42:17 +01:00
result = await self.execute(sql)
if result is None:
return None
return [list[2] for list in result.fetchall()]