forked from Dniel97/artemis
457 lines
14 KiB
Python
457 lines
14 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",
|
|
)
|
|
|
|
gacha = Table(
|
|
"chuni_item_gacha",
|
|
metadata,
|
|
Column("id", Integer, primary_key=True, nullable=False),
|
|
Column(
|
|
"user",
|
|
ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"),
|
|
nullable=False,
|
|
),
|
|
Column("gachaId", Integer, nullable=False),
|
|
Column("totalGachaCnt", Integer, server_default="0"),
|
|
Column("ceilingGachaCnt", Integer, server_default="0"),
|
|
Column("dailyGachaCnt", Integer, server_default="0"),
|
|
Column("fiveGachaCnt", Integer, server_default="0"),
|
|
Column("elevenGachaCnt", Integer, server_default="0"),
|
|
Column("dailyGachaDate", TIMESTAMP, nullable=False, server_default=func.now()),
|
|
UniqueConstraint("user", "gachaId", name="chuni_item_gacha_uk"),
|
|
mysql_charset="utf8mb4",
|
|
)
|
|
|
|
print_state = Table(
|
|
"chuni_item_print_state",
|
|
metadata,
|
|
Column("id", Integer, primary_key=True, nullable=False),
|
|
Column(
|
|
"user",
|
|
ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"),
|
|
nullable=False,
|
|
),
|
|
Column("hasCompleted", Boolean, nullable=False, server_default="0"),
|
|
Column(
|
|
"limitDate", TIMESTAMP, nullable=False, server_default="2038-01-01 00:00:00.0"
|
|
),
|
|
Column("placeId", Integer),
|
|
Column("cardId", Integer),
|
|
Column("gachaId", Integer),
|
|
UniqueConstraint("id", "user", name="chuni_item_print_state_uk"),
|
|
mysql_charset="utf8mb4",
|
|
)
|
|
|
|
print_detail = Table(
|
|
"chuni_item_print_detail",
|
|
metadata,
|
|
Column("id", Integer, primary_key=True, nullable=False),
|
|
Column(
|
|
"user",
|
|
ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"),
|
|
nullable=False,
|
|
),
|
|
Column("cardId", Integer, nullable=False),
|
|
Column("printDate", TIMESTAMP, nullable=False),
|
|
Column("serialId", String(20), nullable=False),
|
|
Column("placeId", Integer, nullable=False),
|
|
Column("clientId", String(11), nullable=False),
|
|
Column("printerSerialId", String(20), nullable=False),
|
|
Column("printOption1", Boolean, server_default="0"),
|
|
Column("printOption2", Boolean, server_default="0"),
|
|
Column("printOption3", Boolean, server_default="0"),
|
|
Column("printOption4", Boolean, server_default="0"),
|
|
Column("printOption5", Boolean, server_default="0"),
|
|
Column("printOption6", Boolean, server_default="0"),
|
|
Column("printOption7", Boolean, server_default="0"),
|
|
Column("printOption8", Boolean, server_default="0"),
|
|
Column("printOption9", Boolean, server_default="0"),
|
|
Column("printOption10", Boolean, server_default="0"),
|
|
Column("created", String(255), server_default=""),
|
|
UniqueConstraint("serialId", name="chuni_item_print_detail_uk"),
|
|
mysql_charset="utf8mb4",
|
|
)
|
|
|
|
login_bonus = Table(
|
|
"chuni_item_login_bonus",
|
|
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("presetId", Integer, nullable=False),
|
|
Column("bonusCount", Integer, nullable=False, server_default="0"),
|
|
Column("lastUpdateDate", TIMESTAMP, server_default="2018-01-01 00:00:00.0"),
|
|
Column("isWatched", Boolean, server_default="0"),
|
|
Column("isFinished", Boolean, server_default="0"),
|
|
UniqueConstraint("version", "user", "presetId", name="chuni_item_login_bonus_uk"),
|
|
mysql_charset="utf8mb4",
|
|
)
|
|
|
|
|
|
class ChuniItemData(BaseData):
|
|
def put_login_bonus(
|
|
self, user_id: int, version: int, preset_id: int, **login_bonus_data
|
|
) -> Optional[int]:
|
|
sql = insert(login_bonus).values(
|
|
version=version, user=user_id, presetId=preset_id, **login_bonus_data
|
|
)
|
|
|
|
conflict = sql.on_duplicate_key_update(presetId=preset_id, **login_bonus_data)
|
|
|
|
result = self.execute(conflict)
|
|
if result is None:
|
|
return None
|
|
return result.lastrowid
|
|
|
|
def get_all_login_bonus(
|
|
self, user_id: int, version: int, is_finished: bool = False
|
|
) -> Optional[List[Row]]:
|
|
sql = login_bonus.select(
|
|
and_(
|
|
login_bonus.c.version == version,
|
|
login_bonus.c.user == user_id,
|
|
login_bonus.c.isFinished == is_finished,
|
|
)
|
|
)
|
|
|
|
result = self.execute(sql)
|
|
if result is None:
|
|
return None
|
|
return result.fetchall()
|
|
|
|
def get_login_bonus(
|
|
self, user_id: int, version: int, preset_id: int
|
|
) -> Optional[Row]:
|
|
sql = login_bonus.select(
|
|
and_(
|
|
login_bonus.c.version == version,
|
|
login_bonus.c.user == user_id,
|
|
login_bonus.c.presetId == preset_id,
|
|
)
|
|
)
|
|
|
|
result = self.execute(sql)
|
|
if result is None:
|
|
return None
|
|
return result.fetchone()
|
|
|
|
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()
|
|
|
|
def get_user_gachas(self, aime_id: int) -> Optional[List[Row]]:
|
|
sql = gacha.select(gacha.c.user == aime_id)
|
|
|
|
result = self.execute(sql)
|
|
if result is None:
|
|
return None
|
|
return result.fetchall()
|
|
|
|
def put_user_gacha(
|
|
self, aime_id: int, gacha_id: int, gacha_data: Dict
|
|
) -> Optional[int]:
|
|
sql = insert(gacha).values(user=aime_id, gachaId=gacha_id, **gacha_data)
|
|
|
|
conflict = sql.on_duplicate_key_update(
|
|
user=aime_id, gachaId=gacha_id, **gacha_data
|
|
)
|
|
result = self.execute(conflict)
|
|
|
|
if result is None:
|
|
self.logger.warn(f"put_user_gacha: Failed to insert! aime_id: {aime_id}")
|
|
return None
|
|
return result.lastrowid
|
|
|
|
def get_user_print_states(
|
|
self, aime_id: int, has_completed: bool = False
|
|
) -> Optional[List[Row]]:
|
|
sql = print_state.select(
|
|
and_(
|
|
print_state.c.user == aime_id,
|
|
print_state.c.hasCompleted == has_completed,
|
|
)
|
|
)
|
|
|
|
result = self.execute(sql)
|
|
if result is None:
|
|
return None
|
|
return result.fetchall()
|
|
|
|
def get_user_print_states_by_gacha(
|
|
self, aime_id: int, gacha_id: int, has_completed: bool = False
|
|
) -> Optional[List[Row]]:
|
|
sql = print_state.select(
|
|
and_(
|
|
print_state.c.user == aime_id,
|
|
print_state.c.gachaId == gacha_id,
|
|
print_state.c.hasCompleted == has_completed,
|
|
)
|
|
)
|
|
|
|
result = self.execute(sql)
|
|
if result is None:
|
|
return None
|
|
return result.fetchall()
|
|
|
|
def put_user_print_state(self, aime_id: int, **print_data) -> Optional[int]:
|
|
sql = insert(print_state).values(user=aime_id, **print_data)
|
|
|
|
conflict = sql.on_duplicate_key_update(user=aime_id, **print_data)
|
|
result = self.execute(conflict)
|
|
|
|
if result is None:
|
|
self.logger.warn(
|
|
f"put_user_print_state: Failed to insert! aime_id: {aime_id}"
|
|
)
|
|
return None
|
|
return result.lastrowid
|
|
|
|
def put_user_print_detail(
|
|
self, aime_id: int, serial_id: str, user_print_data: Dict
|
|
) -> Optional[int]:
|
|
sql = insert(print_detail).values(
|
|
user=aime_id, serialId=serial_id, **user_print_data
|
|
)
|
|
|
|
conflict = sql.on_duplicate_key_update(user=aime_id, **user_print_data)
|
|
result = self.execute(conflict)
|
|
|
|
if result is None:
|
|
self.logger.warn(
|
|
f"put_user_print_detail: Failed to insert! aime_id: {aime_id}"
|
|
)
|
|
return None
|
|
return result.lastrowid
|