from logging import Logger from typing import Optional from os import walk, path, remove import xml.etree.ElementTree as ET from read import BaseReader from PIL import Image import configparser import glob from core.config import CoreConfig from titles.chuni.database import ChuniData from titles.chuni.const import ChuniConstants from titles.chuni.schema.static import music as MusicTable class ChuniReader(BaseReader): def __init__( self, config: CoreConfig, version: int, bin_dir: Optional[str], opt_dir: Optional[str], extra: Optional[str], ) -> None: super().__init__(config, version, bin_dir, opt_dir, extra) self.data = ChuniData(config) try: self.logger.info( f"Start importer for {ChuniConstants.game_ver_to_string(version)}" ) except IndexError: self.logger.error(f"Invalid chunithm version {version}") exit(1) async def read(self) -> None: data_dirs = [] if self.bin_dir is not None: data_dirs += self.get_data_directories(self.bin_dir) if self.opt_dir is not None: data_dirs += self.get_data_directories(self.opt_dir) we_diff = "4" if self.version >= ChuniConstants.VER_CHUNITHM_NEW: we_diff = "5" # Convert any old assets created with a previous version of the importer ChuniReader.ConvertOldAssets(self.logger) # character images could be stored anywhere across all the data dirs. Map them first self.logger.info(f"Mapping DDS image files...") dds_images = dict() for dir in data_dirs: self.map_dds_images(dds_images, f"{dir}/ddsImage") for dir in data_dirs: self.logger.info(f"Read from {dir}") this_opt_id = await self.read_opt_info(dir) # this also treats A000 as an opt, which is intended await self.read_events(f"{dir}/event", this_opt_id) await self.read_music(f"{dir}/music", we_diff, this_opt_id) await self.read_charges(f"{dir}/chargeItem", this_opt_id) await self.read_avatar(f"{dir}/avatarAccessory", this_opt_id) await self.read_login_bonus(f"{dir}/", this_opt_id) await self.read_nameplate(f"{dir}/namePlate", this_opt_id) await self.read_trophy(f"{dir}/trophy", this_opt_id) await self.read_character(f"{dir}/chara", dds_images, this_opt_id) await self.read_map_icon(f"{dir}/mapIcon", this_opt_id) await self.read_system_voice(f"{dir}/systemVoice", this_opt_id) await self.read_unlock_challenge(f"{dir}/unlockChallenge") async def read_login_bonus(self, root_dir: str, opt_id: Optional[int] = None) -> None: for root, dirs, files in walk(f"{root_dir}loginBonusPreset"): for dir in dirs: if path.exists(f"{root}/{dir}/LoginBonusPreset.xml"): with open(f"{root}/{dir}/LoginBonusPreset.xml", "r", encoding="utf-8") as fp: strdata = fp.read() xml_root = ET.fromstring(strdata) for name in xml_root.findall("name"): id = name.find("id").text name = name.find("str").text disableFlag = xml_root.find("disableFlag") # may not exist in older data is_enabled = True if (disableFlag is None or disableFlag.text == "false") else False result = await self.data.static.put_login_bonus_preset( self.version, id, name, is_enabled, opt_id ) if result is not None: self.logger.info(f"Inserted login bonus preset {id}") else: self.logger.warning(f"Failed to insert login bonus preset {id}") for bonus in xml_root.find("infos").findall("LoginBonusDataInfo"): for name in bonus.findall("loginBonusName"): bonus_id = name.find("id").text bonus_name = name.find("str").text if path.exists( f"{root_dir}/loginBonus/loginBonus{bonus_id}/LoginBonus.xml" ): with open( f"{root_dir}/loginBonus/loginBonus{bonus_id}/LoginBonus.xml", "rb", ) as fp: bytedata = fp.read() strdata = bytedata.decode("UTF-8") bonus_root = ET.fromstring(strdata) for present in bonus_root.findall("present"): present_id = present.find("id").text present_name = present.find("str").text item_num = int(bonus_root.find("itemNum").text) need_login_day_count = int( bonus_root.find("needLoginDayCount").text ) login_bonus_category_type = int( bonus_root.find("loginBonusCategoryType").text ) result = await self.data.static.put_login_bonus( self.version, id, bonus_id, bonus_name, present_id, present_name, item_num, need_login_day_count, login_bonus_category_type, opt_id ) if result is not None: self.logger.info(f"Inserted login bonus {bonus_id}") else: self.logger.warning( f"Failed to insert login bonus {bonus_id}" ) async def read_events(self, evt_dir: str, opt_id: Optional[int] = None) -> None: for root, dirs, files in walk(evt_dir): for dir in dirs: if path.exists(f"{root}/{dir}/Event.xml"): with open(f"{root}/{dir}/Event.xml", "r", encoding="utf-8") as fp: strdata = fp.read() xml_root = ET.fromstring(strdata) for name in xml_root.findall("name"): id = name.find("id").text name = name.find("str").text for substances in xml_root.findall("substances"): event_type = substances.find("type").text result = await self.data.static.put_event( self.version, id, event_type, name, opt_id ) if result is not None: self.logger.info(f"Inserted event {id}") else: self.logger.warning(f"Failed to insert event {id}") async def read_music(self, music_dir: str, we_diff: str = "4", opt_id: Optional[int] = None) -> None: max_title_len = MusicTable.columns["title"].type.length max_artist_len = MusicTable.columns["artist"].type.length for root, dirs, files in walk(music_dir): for dir in dirs: if path.exists(f"{root}/{dir}/Music.xml"): with open(f"{root}/{dir}/Music.xml", "r", encoding='utf-8') as fp: strdata = fp.read() xml_root = ET.fromstring(strdata) for name in xml_root.findall("name"): song_id = name.find("id").text title = name.find("str").text if len(title) > max_title_len: self.logger.warning(f"Truncating music {song_id} song title") title = title[:max_title_len] for artistName in xml_root.findall("artistName"): artist = artistName.find("str").text if len(artist) > max_artist_len: self.logger.warning(f"Truncating music {song_id} artist name") artist = artist[:max_artist_len] for genreNames in xml_root.findall("genreNames"): for list_ in genreNames.findall("list"): for StringID in list_.findall("StringID"): genre = StringID.find("str").text for jaketFile in xml_root.findall("jaketFile"): # nice typo, SEGA jacket_path = jaketFile.find("path").text # Save off image for use in frontend self.copy_image(jacket_path, f"{root}/{dir}", "titles/chuni/img/jacket/") for fumens in xml_root.findall("fumens"): for MusicFumenData in fumens.findall("MusicFumenData"): fumen_path = MusicFumenData.find("file").find("path") if fumen_path is not None: chart_type = MusicFumenData.find("type") chart_id = chart_type.find("id").text chart_diff = chart_type.find("str").text if chart_diff == "WorldsEnd" and chart_id == we_diff: # 4 in SDBT, 5 in SDHD level = float(xml_root.find("starDifType").text) we_chara = ( xml_root.find("worldsEndTagName") .find("str") .text ) else: level = float( f"{MusicFumenData.find('level').text}.{MusicFumenData.find('levelDecimal').text}" ) we_chara = None result = await self.data.static.put_music( self.version, song_id, chart_id, title, artist, level, genre, jacket_path, we_chara, opt_id ) if result is not None: self.logger.info( f"Inserted music {song_id} chart {chart_id}" ) else: self.logger.warning( f"Failed to insert music {song_id} chart {chart_id}" ) async def read_charges(self, charge_dir: str, opt_id: Optional[int] = None) -> None: for root, dirs, files in walk(charge_dir): for dir in dirs: if path.exists(f"{root}/{dir}/ChargeItem.xml"): with open(f"{root}/{dir}/ChargeItem.xml", "r", encoding='utf-8') as fp: strdata = fp.read() xml_root = ET.fromstring(strdata) for name in xml_root.findall("name"): id = name.find("id").text name = name.find("str").text expirationDays = xml_root.find("expirationDays").text consumeType = xml_root.find("consumeType").text sellingAppeal = bool(xml_root.find("sellingAppeal").text) result = await self.data.static.put_charge( self.version, id, name, expirationDays, consumeType, sellingAppeal, opt_id ) if result is not None: self.logger.info(f"Inserted charge {id}") else: self.logger.warning(f"Failed to insert charge {id}") async def read_avatar(self, avatar_dir: str, opt_id: Optional[int] = None) -> None: for root, dirs, files in walk(avatar_dir): for dir in dirs: if path.exists(f"{root}/{dir}/AvatarAccessory.xml"): with open(f"{root}/{dir}/AvatarAccessory.xml", "r", encoding='utf-8') as fp: strdata = fp.read() xml_root = ET.fromstring(strdata) for name in xml_root.findall("name"): id = name.find("id").text name = name.find("str").text sortName = xml_root.find("sortName").text category = xml_root.find("category").text defaultHave = xml_root.find("defaultHave").text == 'true' disableFlag = xml_root.find("disableFlag") # may not exist in older data is_enabled = True if (disableFlag is None or disableFlag.text == "false") else False for image in xml_root.findall("image"): iconPath = image.find("path").text self.copy_image(iconPath, f"{root}/{dir}", "titles/chuni/img/avatar/") for texture in xml_root.findall("texture"): texturePath = texture.find("path").text self.copy_image(texturePath, f"{root}/{dir}", "titles/chuni/img/avatar/") result = await self.data.static.put_avatar( self.version, id, name, category, iconPath, texturePath, is_enabled, defaultHave, sortName, opt_id ) if result is not None: self.logger.info(f"Inserted avatarAccessory {id}") else: self.logger.warning(f"Failed to insert avatarAccessory {id}") async def read_nameplate(self, nameplate_dir: str, opt_id: Optional[int] = None) -> None: for root, dirs, files in walk(nameplate_dir): for dir in dirs: if path.exists(f"{root}/{dir}/NamePlate.xml"): with open(f"{root}/{dir}/NamePlate.xml", "r", encoding='utf-8') as fp: strdata = fp.read() xml_root = ET.fromstring(strdata) for name in xml_root.findall("name"): id = name.find("id").text name = name.find("str").text sortName = name if xml_root.find("sortName") is None else xml_root.find("sortName").text defaultHave = xml_root.find("defaultHave").text == 'true' disableFlag = xml_root.find("disableFlag") # may not exist in older data is_enabled = True if (disableFlag is None or disableFlag.text == "false") else False for image in xml_root.findall("image"): texturePath = image.find("path").text self.copy_image(texturePath, f"{root}/{dir}", "titles/chuni/img/nameplate/") result = await self.data.static.put_nameplate( self.version, id, name, texturePath, is_enabled, defaultHave, sortName, opt_id ) if result is not None: self.logger.info(f"Inserted nameplate {id}") else: self.logger.warning(f"Failed to insert nameplate {id}") async def read_trophy(self, trophy_dir: str, opt_id: Optional[int] = None) -> None: for root, dirs, files in walk(trophy_dir): for dir in dirs: if path.exists(f"{root}/{dir}/Trophy.xml"): with open(f"{root}/{dir}/Trophy.xml", "r", encoding='utf-8') as fp: strdata = fp.read() xml_root = ET.fromstring(strdata) for name in xml_root.findall("name"): id = name.find("id").text name = name.find("str").text rareType = xml_root.find("rareType").text disableFlag = xml_root.find("disableFlag") # may not exist in older data is_enabled = True if (disableFlag is None or disableFlag.text == "false") else False defaultHave = xml_root.find("defaultHave").text == 'true' result = await self.data.static.put_trophy( self.version, id, name, rareType, is_enabled, defaultHave, opt_id ) if result is not None: self.logger.info(f"Inserted trophy {id}") else: self.logger.warning(f"Failed to insert trophy {id}") async def read_character(self, chara_dir: str, dds_images: dict, opt_id: Optional[int] = None) -> None: for root, dirs, files in walk(chara_dir): for dir in dirs: if path.exists(f"{root}/{dir}/Chara.xml"): with open(f"{root}/{dir}/Chara.xml", "r", encoding='utf-8') as fp: strdata = fp.read() # ET may choke if there is a & symbol (which is present in some character xml) if "&" in strdata: strdata = strdata.replace("&", "&") xml_root = ET.fromstring(strdata) for name in xml_root.findall("name"): id = name.find("id").text name = name.find("str").text sortName = name if xml_root.find("sortName") is None else xml_root.find("sortName").text for work in xml_root.findall("works"): worksName = work.find("str").text rareType = xml_root.find("rareType").text defaultHave = xml_root.find("defaultHave").text == 'true' disableFlag = xml_root.find("disableFlag") # may not exist in older data is_enabled = True if (disableFlag is None or disableFlag.text == "false") else False # character images are not stored alongside for image in xml_root.findall("defaultImages"): imageKey = image.find("str").text if imageKey in dds_images.keys(): (imageDir, imagePaths) = dds_images[imageKey] imagePath1 = imagePaths[0] if len(imagePaths) > 0 else "" imagePath2 = imagePaths[1] if len(imagePaths) > 1 else "" imagePath3 = imagePaths[2] if len(imagePaths) > 2 else "" # @note the third image is the image needed for the user box ui if imagePath3: self.copy_image(imagePath3, imageDir, "titles/chuni/img/character/") else: self.logger.warning(f"Character {id} only has {len(imagePaths)} images. Expected 3") else: self.logger.warning(f"Unable to location character {id} images") result = await self.data.static.put_character( self.version, id, name, sortName, worksName, rareType, imagePath1, imagePath2, imagePath3, is_enabled, defaultHave, opt_id ) if result is not None: self.logger.info(f"Inserted character {id}") else: self.logger.warning(f"Failed to insert character {id}") async def read_map_icon(self, mapicon_dir: str, opt_id: Optional[int] = None) -> None: for root, dirs, files in walk(mapicon_dir): for dir in dirs: if path.exists(f"{root}/{dir}/MapIcon.xml"): with open(f"{root}/{dir}/MapIcon.xml", "r", encoding='utf-8') as fp: strdata = fp.read() xml_root = ET.fromstring(strdata) for name in xml_root.findall("name"): id = name.find("id").text name = name.find("str").text sortName = name if xml_root.find("sortName") is None else xml_root.find("sortName").text for image in xml_root.findall("image"): iconPath = image.find("path").text self.copy_image(iconPath, f"{root}/{dir}", "titles/chuni/img/mapIcon/") defaultHave = xml_root.find("defaultHave").text == 'true' disableFlag = xml_root.find("disableFlag") # may not exist in older data is_enabled = True if (disableFlag is None or disableFlag.text == "false") else False result = await self.data.static.put_map_icon( self.version, id, name, sortName, iconPath, is_enabled, defaultHave, opt_id ) if result is not None: self.logger.info(f"Inserted map icon {id}") else: self.logger.warning(f"Failed to map icon {id}") async def read_system_voice(self, voice_dir: str, opt_id: Optional[int] = None) -> None: for root, dirs, files in walk(voice_dir): for dir in dirs: if path.exists(f"{root}/{dir}/SystemVoice.xml"): with open(f"{root}/{dir}/SystemVoice.xml", "r", encoding='utf-8') as fp: strdata = fp.read() xml_root = ET.fromstring(strdata) for name in xml_root.findall("name"): id = name.find("id").text name = name.find("str").text sortName = name if xml_root.find("sortName") is None else xml_root.find("sortName").text for image in xml_root.findall("image"): imagePath = image.find("path").text self.copy_image(imagePath, f"{root}/{dir}", "titles/chuni/img/systemVoice/") defaultHave = xml_root.find("defaultHave").text == 'true' disableFlag = xml_root.find("disableFlag") # may not exist in older data is_enabled = True if (disableFlag is None or disableFlag.text == "false") else False result = await self.data.static.put_system_voice( self.version, id, name, sortName, imagePath, is_enabled, defaultHave, opt_id ) if result is not None: self.logger.info(f"Inserted system voice {id}") else: self.logger.warning(f"Failed to system voice {id}") async def read_opt_info(self, directory: str) -> Optional[int]: if not path.exists(f"{directory}/data.conf"): self.logger.warning(f"{directory} does not contain data.conf, opt info will not be read") return None data_config = configparser.ConfigParser() if not data_config.read(f"{directory}/data.conf", 'utf-8'): self.logger.warning(f"{directory}/data.conf failed to read or parse, opt info will not be read") return None if 'Version' not in data_config: self.logger.warning(f"{directory}/data.conf contains no Version section, opt info will not be read") return None if 'Name' not in data_config['Version']: # Probably not worth checking that the other sections exist self.logger.warning(f"{directory}/data.conf contains no Name item in the Version section, opt info will not be read") return None if 'VerMajor' not in data_config['Version']: # Probably not worth checking that the other sections exist self.logger.warning(f"{directory}/data.conf contains no VerMajor item in the Version section, opt info will not be read") return None if 'VerMinor' not in data_config['Version']: # Probably not worth checking that the other sections exist self.logger.warning(f"{directory}/data.conf contains no VerMinor item in the Version section, opt info will not be read") return None if 'VerRelease' not in data_config['Version']: # Probably not worth checking that the other sections exist self.logger.warning(f"{directory}/data.conf contains no VerRelease item in the Version section, opt info will not be read") return None opt_seq = data_config['Version']['VerRelease'] opt_folder = path.basename(path.normpath(directory)) opt_id = await self.data.static.get_opt_by_version_folder(self.version, opt_folder) if not opt_id: opt_id = await self.data.static.put_opt(self.version, opt_folder, opt_seq) if not opt_id: self.logger.error(f"Failed to put opt folder info for {opt_folder}") return None else: opt_id = opt_id['id'] self.logger.info(f"Opt folder {opt_folder} (Database ID {opt_id}) contains {data_config['Version']['Name']} v{data_config['Version']['VerMajor']}.{data_config['Version']['VerMinor']}.{opt_seq}") return opt_id async def read_unlock_challenge(self, uc_dir: str) -> None: for root, dirs, files in walk(uc_dir): for dir in dirs: if path.exists(f"{root}/{dir}/UnlockChallenge.xml"): with open(f"{root}/{dir}/UnlockChallenge.xml", "r", encoding="utf-8") as fp: strdata = fp.read() xml_root = ET.fromstring(strdata) for name in xml_root.findall("name"): id = name.find("id").text name = name.find("str").text course_ids = [] for course in xml_root.find("musicList/list/UnlockChallengeMusicListSubData/unlockChallengeMusicData/courseList/list").findall("UnlockChallengeCourseListSubData"): course_id = course.find("unlockChallengeCourseData/courseName").find("id").text course_ids.append(course_id) # Build keyword arguments dynamically for up to 5 course IDs course_kwargs = { f"course_id{i+1}": course_ids[i] for i in range(min(5, len(course_ids))) } result = await self.data.static.put_unlock_challenge( self.version, id, name, **course_kwargs ) if result is not None: self.logger.info(f"Inserted unlock challenge {id}") else: self.logger.warning(f"Failed to unlock challenge {id}") def copy_image(self, filename: str, src_dir: str, dst_dir: str) -> None: # Convert the image to webp so we can easily display it in the frontend file_src = path.join(src_dir, filename) (basename, ext) = path.splitext(filename) file_dst = path.join(dst_dir, basename) + ".webp" if path.exists(file_src) and not path.exists(file_dst): try: im = Image.open(file_src) im.save(file_dst) except Exception: self.logger.warning(f"Failed to convert {filename} to webp") def ConvertOldAssets(logger: Logger): """ Converts any previously-imported png files to webp. In the initial version of the userbox/avatar frontend support, png images were used, scraped via read.py. The amount of data pushed once a lot of stuff was unlocked was noticeable so the frontend now uses webp format for these assets. If any png files are present, convert them to webp now. """ # Find all pngs under the /img directory png_files = glob.glob(f'titles/chuni/img/**/*.png', recursive=True) if len(png_files) > 0: logger.info(f'Found {len(png_files)} old assets. Converting to webp... (may take a few minutes)') for img_png in png_files: img_webp = path.splitext(img_png)[0] + '.webp' try: # convert to webp im = Image.open(img_png) im.save(img_webp) # delete the original file remove(img_png) except Exception as e: logger.warning(f'Failed to convert {img_png} to webp') logger.info(f'Conversion complete') def map_dds_images(self, image_dict: dict, dds_dir: str) -> None: for root, dirs, files in walk(dds_dir): for dir in dirs: directory = f"{root}/{dir}" if path.exists(f"{directory}/DDSImage.xml"): with open(f"{directory}/DDSImage.xml", "r", encoding='utf-8') as fp: strdata = fp.read() xml_root = ET.fromstring(strdata) for name in xml_root.findall("name"): name = name.find("str").text images = [] i = 0 while xml_root.findall(f"ddsFile{i}"): for ddsFile in xml_root.findall(f"ddsFile{i}"): images += [ddsFile.find("path").text] i += 1 image_dict[name] = (directory, images)