207 lines
6.9 KiB
Python
207 lines
6.9 KiB
Python
|
from typing import Dict, List, Optional
|
||
|
|
from sqlalchemy import Table, Column, UniqueConstraint, PrimaryKeyConstraint, and_
|
||
|
|
from sqlalchemy.types import Integer, String, TIMESTAMP, Boolean, JSON
|
||
|
|
from sqlalchemy.engine.base import Connection
|
||
|
|
from sqlalchemy.schema import ForeignKey
|
||
|
|
from sqlalchemy.sql import func, select
|
||
|
|
from sqlalchemy.dialects.mysql import insert
|
||
|
|
from sqlalchemy.engine import Row
|
||
|
|
|
||
|
|
from core.data.schema import BaseData, metadata
|
||
|
|
|
||
|
|
character = Table(
|
||
|
|
"chuni_item_character",
|
||
|
|
metadata,
|
||
|
|
Column("id", Integer, primary_key=True, nullable=False),
|
||
|
|
Column("user", ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"), nullable=False),
|
||
|
|
Column("characterId", Integer),
|
||
|
|
Column("level", Integer),
|
||
|
|
Column("param1", Integer),
|
||
|
|
Column("param2", Integer),
|
||
|
|
Column("isValid", Boolean),
|
||
|
|
Column("skillId", Integer),
|
||
|
|
Column("isNewMark", Boolean),
|
||
|
|
Column("playCount", Integer),
|
||
|
|
Column("friendshipExp", Integer),
|
||
|
|
Column("assignIllust", Integer),
|
||
|
|
Column("exMaxLv", Integer),
|
||
|
|
UniqueConstraint("user", "characterId", name="chuni_item_character_uk"),
|
||
|
|
mysql_charset='utf8mb4'
|
||
|
|
)
|
||
|
|
|
||
|
|
item = Table(
|
||
|
|
"chuni_item_item",
|
||
|
|
metadata,
|
||
|
|
Column("id", Integer, primary_key=True, nullable=False),
|
||
|
|
Column("user", ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"), nullable=False),
|
||
|
|
Column("itemId", Integer),
|
||
|
|
Column("itemKind", Integer),
|
||
|
|
Column("stock", Integer),
|
||
|
|
Column("isValid", Boolean),
|
||
|
|
UniqueConstraint("user", "itemId", "itemKind", name="chuni_item_item_uk"),
|
||
|
|
mysql_charset='utf8mb4'
|
||
|
|
)
|
||
|
|
|
||
|
|
duel = Table(
|
||
|
|
"chuni_item_duel",
|
||
|
|
metadata,
|
||
|
|
Column("id", Integer, primary_key=True, nullable=False),
|
||
|
|
Column("user", ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"), nullable=False),
|
||
|
|
Column("duelId", Integer),
|
||
|
|
Column("progress", Integer),
|
||
|
|
Column("point", Integer),
|
||
|
|
Column("isClear", Boolean),
|
||
|
|
Column("lastPlayDate", String(25)),
|
||
|
|
Column("param1", Integer),
|
||
|
|
Column("param2", Integer),
|
||
|
|
Column("param3", Integer),
|
||
|
|
Column("param4", Integer),
|
||
|
|
UniqueConstraint("user", "duelId", name="chuni_item_duel_uk"),
|
||
|
|
mysql_charset='utf8mb4'
|
||
|
|
)
|
||
|
|
|
||
|
|
map = Table(
|
||
|
|
"chuni_item_map",
|
||
|
|
metadata,
|
||
|
|
Column("id", Integer, primary_key=True, nullable=False),
|
||
|
|
Column("user", ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"), nullable=False),
|
||
|
|
Column("mapId", Integer),
|
||
|
|
Column("position", Integer),
|
||
|
|
Column("isClear", Boolean),
|
||
|
|
Column("areaId", Integer),
|
||
|
|
Column("routeNumber", Integer),
|
||
|
|
Column("eventId", Integer),
|
||
|
|
Column("rate", Integer),
|
||
|
|
Column("statusCount", Integer),
|
||
|
|
Column("isValid", Boolean),
|
||
|
|
UniqueConstraint("user", "mapId", name="chuni_item_map_uk"),
|
||
|
|
mysql_charset='utf8mb4'
|
||
|
|
)
|
||
|
|
|
||
|
|
map_area = Table(
|
||
|
|
"chuni_item_map_area",
|
||
|
|
metadata,
|
||
|
|
Column("id", Integer, primary_key=True, nullable=False),
|
||
|
|
Column("user", ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"), nullable=False),
|
||
|
|
Column("mapAreaId", Integer),
|
||
|
|
Column("rate", Integer),
|
||
|
|
Column("isClear", Boolean),
|
||
|
|
Column("isLocked", Boolean),
|
||
|
|
Column("position", Integer),
|
||
|
|
Column("statusCount", Integer),
|
||
|
|
Column("remainGridCount", Integer),
|
||
|
|
UniqueConstraint("user", "mapAreaId", name="chuni_item_map_area_uk"),
|
||
|
|
mysql_charset='utf8mb4'
|
||
|
|
)
|
||
|
|
|
||
|
|
class ChuniItemData(BaseData):
|
||
|
|
def put_character(self, user_id: int, character_data: Dict) -> Optional[int]:
|
||
|
|
character_data["user"] = user_id
|
||
|
|
|
||
|
|
character_data = self.fix_bools(character_data)
|
||
|
|
|
||
|
|
sql = insert(character).values(**character_data)
|
||
|
|
conflict = sql.on_duplicate_key_update(**character_data)
|
||
|
|
|
||
|
|
result = self.execute(conflict)
|
||
|
|
if result is None: return None
|
||
|
|
return result.lastrowid
|
||
|
|
|
||
|
|
def get_character(self, user_id: int, character_id: int) -> Optional[Dict]:
|
||
|
|
sql = select(character).where(and_(
|
||
|
|
character.c.user == user_id,
|
||
|
|
character.c.characterId == character_id
|
||
|
|
))
|
||
|
|
|
||
|
|
result = self.execute(sql)
|
||
|
|
if result is None: return None
|
||
|
|
return result.fetchone()
|
||
|
|
|
||
|
|
def get_characters(self, user_id: int) -> Optional[List[Row]]:
|
||
|
|
sql = select(character).where(character.c.user == user_id)
|
||
|
|
|
||
|
|
result = self.execute(sql)
|
||
|
|
if result is None: return None
|
||
|
|
return result.fetchall()
|
||
|
|
|
||
|
|
def put_item(self, user_id: int, item_data: Dict) -> Optional[int]:
|
||
|
|
item_data["user"] = user_id
|
||
|
|
|
||
|
|
item_data = self.fix_bools(item_data)
|
||
|
|
|
||
|
|
sql = insert(item).values(**item_data)
|
||
|
|
conflict = sql.on_duplicate_key_update(**item_data)
|
||
|
|
|
||
|
|
result = self.execute(conflict)
|
||
|
|
if result is None: return None
|
||
|
|
return result.lastrowid
|
||
|
|
|
||
|
|
def get_items(self, user_id: int, kind: int = None) -> Optional[List[Row]]:
|
||
|
|
if kind is None:
|
||
|
|
sql = select(item).where(item.c.user == user_id)
|
||
|
|
else:
|
||
|
|
sql = select(item).where(and_(
|
||
|
|
item.c.user == user_id,
|
||
|
|
item.c.itemKind == kind
|
||
|
|
))
|
||
|
|
|
||
|
|
result = self.execute(sql)
|
||
|
|
if result is None: return None
|
||
|
|
return result.fetchall()
|
||
|
|
|
||
|
|
def put_duel(self, user_id: int, duel_data: Dict) -> Optional[int]:
|
||
|
|
duel_data["user"] = user_id
|
||
|
|
|
||
|
|
duel_data = self.fix_bools(duel_data)
|
||
|
|
|
||
|
|
sql = insert(duel).values(**duel_data)
|
||
|
|
conflict = sql.on_duplicate_key_update(**duel_data)
|
||
|
|
|
||
|
|
result = self.execute(conflict)
|
||
|
|
if result is None: return None
|
||
|
|
return result.lastrowid
|
||
|
|
|
||
|
|
def get_duels(self, user_id: int) -> Optional[List[Row]]:
|
||
|
|
sql = select(duel).where(duel.c.user == user_id)
|
||
|
|
|
||
|
|
result = self.execute(sql)
|
||
|
|
if result is None: return None
|
||
|
|
return result.fetchall()
|
||
|
|
|
||
|
|
def put_map(self, user_id: int, map_data: Dict) -> Optional[int]:
|
||
|
|
map_data["user"] = user_id
|
||
|
|
|
||
|
|
map_data = self.fix_bools(map_data)
|
||
|
|
|
||
|
|
sql = insert(map).values(**map_data)
|
||
|
|
conflict = sql.on_duplicate_key_update(**map_data)
|
||
|
|
|
||
|
|
result = self.execute(conflict)
|
||
|
|
if result is None: return None
|
||
|
|
return result.lastrowid
|
||
|
|
|
||
|
|
def get_maps(self, user_id: int) -> Optional[List[Row]]:
|
||
|
|
sql = select(map).where(map.c.user == user_id)
|
||
|
|
|
||
|
|
result = self.execute(sql)
|
||
|
|
if result is None: return None
|
||
|
|
return result.fetchall()
|
||
|
|
|
||
|
|
def put_map_area(self, user_id: int, map_area_data: Dict) -> Optional[int]:
|
||
|
|
map_area_data["user"] = user_id
|
||
|
|
|
||
|
|
map_area_data = self.fix_bools(map_area_data)
|
||
|
|
|
||
|
|
sql = insert(map_area).values(**map_area_data)
|
||
|
|
conflict = sql.on_duplicate_key_update(**map_area_data)
|
||
|
|
|
||
|
|
result = self.execute(conflict)
|
||
|
|
if result is None: return None
|
||
|
|
return result.lastrowid
|
||
|
|
|
||
|
|
def get_map_areas(self, user_id: int) -> Optional[List[Row]]:
|
||
|
|
sql = select(map_area).where(map_area.c.user == user_id)
|
||
|
|
|
||
|
|
result = self.execute(sql)
|
||
|
|
if result is None: return None
|
||
|
|
return result.fetchall()
|