Bennett/artemis
1
0
Fork 0
mirror of https://gitea.tendokyu.moe/Hay1tsme/artemis.git synced 2026-02-13 03:07:29 +08:00
Code Issues Packages Projects Releases Wiki Activity

chuni: added login bonus (+importer), fixed config strings

Browse Source
This commit is contained in:
Dniel97
2023-03-28 18:28:57 +02:00
parent 541fe76a7c
commit 1aa92458f4
8 changed files with 424 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files

75
titles/chuni/schema/item.py
View File

@@ -184,8 +184,73 @@ print_detail = Table(
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
@@ -335,7 +400,7 @@ class ChuniItemData(BaseData):
sql = print_state.select(
and_(
print_state.c.user == aime_id,
print_state.c.hasCompleted == has_completed
print_state.c.hasCompleted == has_completed,
)
)
@@ -351,7 +416,7 @@ class ChuniItemData(BaseData):
and_(
print_state.c.user == aime_id,
print_state.c.gachaId == gacha_id,
print_state.c.hasCompleted == has_completed
print_state.c.hasCompleted == has_completed,
)
)
@@ -380,9 +445,7 @@ class ChuniItemData(BaseData):
user=aime_id, serialId=serial_id, **user_print_data
)
conflict = sql.on_duplicate_key_update(
user=aime_id, **user_print_data
)
conflict = sql.on_duplicate_key_update(user=aime_id, **user_print_data)
result = self.execute(conflict)
if result is None:
@@ -390,4 +453,4 @@ class ChuniItemData(BaseData):
f"put_user_print_detail: Failed to insert! aime_id: {aime_id}"
)
return None
return result.lastrowid
return result.lastrowid

140
titles/chuni/schema/static.py
View File

@@ -122,8 +122,148 @@ gacha_cards = Table(
mysql_charset="utf8mb4",
)
login_bonus_preset = Table(
"chuni_static_login_bonus_preset",
metadata,
Column("id", Integer, primary_key=True, nullable=False),
Column("version", Integer, nullable=False),
Column("presetName", String(255), nullable=False),
Column("isEnabled", Boolean, server_default="1"),
UniqueConstraint("version", "id", name="chuni_static_login_bonus_preset_uk"),
mysql_charset="utf8mb4",
)
login_bonus = Table(
"chuni_static_login_bonus",
metadata,
Column("id", Integer, primary_key=True, nullable=False),
Column("version", Integer, nullable=False),
Column(
"presetId",
ForeignKey(
"chuni_static_login_bonus_preset.id",
ondelete="cascade",
onupdate="cascade",
),
nullable=False,
),
Column("loginBonusId", Integer, nullable=False),
Column("loginBonusName", String(255), nullable=False),
Column("presentId", Integer, nullable=False),
Column("presentName", String(255), nullable=False),
Column("itemNum", Integer, nullable=False),
Column("needLoginDayCount", Integer, nullable=False),
Column("loginBonusCategoryType", Integer, nullable=False),
UniqueConstraint(
"version", "presetId", "loginBonusId", name="chuni_static_login_bonus_uk"
),
mysql_charset="utf8mb4",
)
class ChuniStaticData(BaseData):
def put_login_bonus(
self,
version: int,
preset_id: int,
login_bonus_id: int,
login_bonus_name: str,
present_id: int,
present_ame: str,
item_num: int,
need_login_day_count: int,
login_bonus_category_type: int,
) -> Optional[int]:
sql = insert(login_bonus).values(
version=version,
presetId=preset_id,
loginBonusId=login_bonus_id,
loginBonusName=login_bonus_name,
presentId=present_id,
presentName=present_ame,
itemNum=item_num,
needLoginDayCount=need_login_day_count,
loginBonusCategoryType=login_bonus_category_type,
)
conflict = sql.on_duplicate_key_update(
loginBonusName=login_bonus_name,
presentName=present_ame,
itemNum=item_num,
needLoginDayCount=need_login_day_count,
loginBonusCategoryType=login_bonus_category_type,
)
result = self.execute(conflict)
if result is None:
return None
return result.lastrowid
def get_login_bonus(
self, version: int, preset_id: int,
) -> Optional[List[Row]]:
sql = login_bonus.select(
and_(
login_bonus.c.version == version,
login_bonus.c.presetId == preset_id,
)
).order_by(login_bonus.c.needLoginDayCount.desc())
result = self.execute(sql)
if result is None:
return None
return result.fetchall()
def get_login_bonus_by_required_days(
self, version: int, preset_id: int, need_login_day_count: int
) -> Optional[Row]:
sql = login_bonus.select(
and_(
login_bonus.c.version == version,
login_bonus.c.presetId == preset_id,
login_bonus.c.needLoginDayCount == need_login_day_count,
)
)
result = self.execute(sql)
if result is None:
return None
return result.fetchone()
def put_login_bonus_preset(
self, version: int, preset_id: int, preset_name: str, is_enabled: bool
) -> Optional[int]:
sql = insert(login_bonus_preset).values(
id=preset_id,
version=version,
presetName=preset_name,
isEnabled=is_enabled,
)
conflict = sql.on_duplicate_key_update(
presetName=preset_name, isEnabled=is_enabled
)
result = self.execute(conflict)
if result is None:
return None
return result.lastrowid
def get_login_bonus_presets(
self, version: int, is_enabled: bool = True
) -> Optional[List[Row]]:
sql = login_bonus_preset.select(
and_(
login_bonus_preset.c.version == version,
login_bonus_preset.c.isEnabled == is_enabled,
)
)
result = self.execute(sql)
if result is None:
return None
return result.fetchall()
def put_event(
self, version: int, event_id: int, type: int, name: str
) -> Optional[int]: