akanyan/artemis
1
0
Fork 0
forked from Hay1tsme/artemis
Code Pull Requests Activity

add RewardList support

Browse Source 
add PresentList support
add reading rewards to read.py
add Ranking Music List support
This commit is contained in:
phantomlan 2023-11-07 02:14:15 +01:00 committed by phantomlan
parent 4da886a083
commit 1897e8002d
4 changed files with 181 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

44
titles/ongeki/base.py
View File

@ -157,7 +157,21 @@ class OngekiBase:
return {"type": data["type"], "length": 0, "gameIdlistList": []}
def handle_get_game_ranking_api_request(self, data: Dict) -> Dict:
game_ranking_list = self.data.static.get_ranking_list()
ranking_list = []
for music in game_ranking_list:
tmp = music._asdict()
ranking_list.append(tmp)
if ranking_list is None:
return {"length": 0, "gameRankingList": []}
return {
"type": data["type"],
#"length": len(ranking_list),
"gameRankingList": ranking_list,
}
def handle_get_game_point_api_request(self, data: Dict) -> Dict:
"""
@ -215,11 +229,39 @@ class OngekiBase:
return {"returnCode": 1, "apiName": "ExtendLockTimeApi"}
def handle_get_game_reward_api_request(self, data: Dict) -> Dict:
# TODO: reward list
get_game_rewards = self.data.static.get_reward_list(self.version)
reward_list = []
for reward in get_game_rewards:
tmp = reward._asdict()
tmp.pop("id")
tmp.pop("version")
tmp.pop("rewardname")
reward_list.append(tmp)
if reward_list is None:
return {"length": 0, "gameRewardList": []}
return {
"length": len(reward_list),
"gameRewardList": reward_list,
}
def handle_get_game_present_api_request(self, data: Dict) -> Dict:
get_present = self.data.static.get_present_list(self.version)
present_list = []
for present in get_present:
tmp = present._asdict()
tmp.pop("id")
tmp.pop("version")
present_list.append(tmp)
if present_list is None:
return {"length": 0, "gamePresentList": []}
return {
"length": len(present_list),
"gamePresentList": present_list,
}
def handle_get_game_message_api_request(self, data: Dict) -> Dict:
return {"length": 0, "gameMessageList": []}

28
titles/ongeki/const.py
View File

@ -43,6 +43,34 @@ class OngekiConstants:
],
)
REWARD_TYPES: Enum = Enum(
"REWARD_TYPES",
[
"Card",
"NamePlate",
"Trophy",
"LimitBreakItem",
"AlmightyJewel",
"Money",
"Music",
"ProfileVoice",
"Present",
"ChapterJewel",
"GachaTicket",
"KaikaItem",
"ExpUpItem",
"IntimateUpItem",
"BookItem",
"SystemVoice",
"Costume",
"Medal",
"Attachment",
"UnlockItem",
"Max",
"None",
],
)
class CM_GACHA_KINDS(Enum):
Normal = 0
Pickup = 1

26
titles/ongeki/read.py
View File

@ -44,6 +44,7 @@ class OngekiReader(BaseReader):
self.read_events(f"{dir}/event")
self.read_music(f"{dir}/music")
self.read_card(f"{dir}/card")
self.read_reward(f"{dir}/reward")
def read_card(self, base_dir: str) -> None:
self.logger.info(f"Reading cards from {base_dir}...")
@ -171,3 +172,28 @@ class OngekiReader(BaseReader):
self.version, song_id, chart_id, title, artist, genre, level
)
self.logger.info(f"Added song {song_id} chart {chart_id}")
def read_reward(self, base_dir: str) -> None:
self.logger.info(f"Reading rewards from {base_dir}...")
for root, dirs, files in os.walk(base_dir):
for dir in dirs:
if os.path.exists(f"{root}/{dir}/Reward.xml"):
strdata = ""
with open(f"{root}/{dir}/Reward.xml", "r", encoding="utf-8") as f:
strdata = f.read()
troot = ET.fromstring(strdata)
if root is None:
continue
name = troot.find("Name")
rewardId = name.find("id").text
rewardname = name.find("str").text
itemKind = OngekiConstants.REWARD_TYPES[troot.find("ItemType").text].value
itemId = troot.find("RewardItem").find("ItemName").find("id").text
self.data.static.put_reward(self.version, rewardId, rewardname, itemKind, itemId)
self.logger.info(f"Added reward {rewardId}")

81
titles/ongeki/schema/static.py
View File

@ -96,6 +96,45 @@ cards = Table(
mysql_charset="utf8mb4",
)
music_ranking = Table(
"ongeki_static_music_ranking_list",
metadata,
Column("id", Integer, primary_key=True, nullable=False),
Column("musicId", Integer, nullable=False),
Column("point", Integer, nullable=False),
Column("userName", String(255)),
UniqueConstraint("musicId", name="ongeki_static_music_ranking_uk"),
mysql_charset="utf8mb4",
)
rewards = Table(
"ongeki_static_rewards",
metadata,
Column("id", Integer, primary_key=True, nullable=False),
Column("version", Integer, nullable=False),
Column("rewardId", Integer, nullable=False),
Column("rewardname", String(255), nullable=False),
Column("itemKind", Integer, nullable=False),
Column("itemId", Integer, nullable=False),
UniqueConstraint("version","itemKind","rewardId", name="ongeki_static_rewards_uk"),
mysql_charset="utf8mb4",
)
present = Table(
"ongeki_static_present_list",
metadata,
Column("id", Integer, primary_key=True, nullable=False),
Column("version", Integer, nullable=False),
Column("presentId", Integer, nullable=False),
Column("presentName", String(255), nullable=False),
Column("rewardId", Integer, nullable=False),
Column("stock", Integer, nullable=False),
Column("message", String(255)),
Column("startDate", String(25), nullable=False),
Column("endDate", String(25), nullable=False),
UniqueConstraint("version","presentId", name="ongeki_static_present_list_uk"),
mysql_charset="utf8mb4",
)
class OngekiStaticData(BaseData):
def put_card(self, version: int, card_id: int, **card_data) -> Optional[int]:
@ -333,3 +372,45 @@ class OngekiStaticData(BaseData):
if result is None:
return None
return result.fetchone()
def get_ranking_list(self) -> Optional[List[Dict]]:
sql = select(music_ranking.c.musicId.label('id'), music_ranking.c.point, music_ranking.c.userName)
result = self.execute(sql)
if result is None:
return None
return result.fetchall()
def put_reward(self, version: int, rewardId: int, rewardname: str, itemKind: int, itemId: int) -> Optional[int]:
sql = insert(rewards).values(
version=version,
rewardId=rewardId,
rewardname=rewardname,
itemKind=itemKind,
itemId=itemId,
)
conflict = sql.on_duplicate_key_update(
rewardname=rewardname,
)
result = self.execute(conflict)
if result is None:
self.logger.warning(f"Failed to insert reward! reward_id: {rewardId}")
return None
return result.lastrowid
def get_reward_list(self, version: int) -> Optional[List[Dict]]:
sql = select(rewards).where(rewards.c.version == version)
result = self.execute(sql)
if result is None:
self.logger.warning(f"Failed to load reward list")
return None
return result.fetchall()
def get_present_list(self, version: int) -> Optional[List[Dict]]:
sql = select(present).where(present.c.version == version)
result = self.execute(sql)
if result is None:
self.logger.warning(f"Failed to load present list")
return None
return result.fetchall()