forked from Hay1tsme/artemis
sao: backport changes from diana
This commit is contained in:
@ -1,24 +1,16 @@
|
||||
from typing import Optional, Dict, List
|
||||
from os import walk, path
|
||||
import urllib
|
||||
import csv
|
||||
from os import path
|
||||
from typing import Optional, Dict, List
|
||||
|
||||
from read import BaseReader
|
||||
from core.config import CoreConfig
|
||||
from titles.sao.database import SaoData
|
||||
from titles.sao.const import SaoConstants
|
||||
from read import BaseReader
|
||||
|
||||
from .database import SaoData
|
||||
from .const import SaoConstants
|
||||
|
||||
class SaoReader(BaseReader):
|
||||
def __init__(
|
||||
self,
|
||||
config: CoreConfig,
|
||||
version: int,
|
||||
bin_arg: Optional[str],
|
||||
opt_arg: Optional[str],
|
||||
extra: Optional[str],
|
||||
) -> None:
|
||||
super().__init__(config, version, bin_arg, opt_arg, extra)
|
||||
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 = SaoData(config)
|
||||
|
||||
try:
|
||||
@ -28,225 +20,207 @@ class SaoReader(BaseReader):
|
||||
except IndexError:
|
||||
self.logger.error(f"Invalid project SAO version {version}")
|
||||
exit(1)
|
||||
|
||||
|
||||
async def read(self) -> None:
|
||||
if path.exists(self.bin_dir):
|
||||
await self.read_csv(f"{self.bin_dir}")
|
||||
|
||||
else:
|
||||
self.logger.warn("Directory not found, nothing to import")
|
||||
|
||||
|
||||
def load_csv_file(self, file: str) -> List[Dict]:
|
||||
ret = []
|
||||
try:
|
||||
fullPath = self.bin_dir + "/" if not self.bin_dir.endswith("/") else ""
|
||||
fullPath += file
|
||||
with open(fullPath, encoding="UTF-8") as fp:
|
||||
reader = csv.DictReader(fp)
|
||||
for row in reader:
|
||||
tmp = {}
|
||||
|
||||
fkey = list(row.keys())[0]
|
||||
new_fkey = fkey.replace("// ", "")
|
||||
fval = row[fkey]
|
||||
row.pop(fkey)
|
||||
row[new_fkey] = fval
|
||||
|
||||
for k,v in row.items():
|
||||
if v == "-1":
|
||||
row[k] = None
|
||||
elif v.isdigit():
|
||||
row[k] = int(v)
|
||||
elif v.isdecimal():
|
||||
row[k] = float(v)
|
||||
elif v == "True":
|
||||
row[k] = True
|
||||
elif v == "False":
|
||||
row[k] = False
|
||||
ret.append(row)
|
||||
|
||||
except Exception as e:
|
||||
self.logger.warning(f"Couldn't read csv file {fullPath}, skipping - {e}")
|
||||
|
||||
return ret
|
||||
|
||||
async def read_csv(self, bin_dir: str) -> None:
|
||||
self.logger.info(f"Read csv from {bin_dir}")
|
||||
|
||||
self.logger.info("Now reading QuestScene.csv")
|
||||
try:
|
||||
fullPath = bin_dir + "/QuestScene.csv"
|
||||
with open(fullPath, encoding="UTF-8") as fp:
|
||||
reader = csv.DictReader(fp)
|
||||
for row in reader:
|
||||
questSceneId = row["QuestSceneId"]
|
||||
sortNo = row["SortNo"]
|
||||
name = row["Name"]
|
||||
enabled = True
|
||||
reader = self.load_csv_file("QuestScene.csv")
|
||||
if reader:
|
||||
for row in reader:
|
||||
self.logger.info(f"Adding quest {row['QuestSceneId']}")
|
||||
await self.data.static.put_quest(row)
|
||||
|
||||
self.logger.info(f"Added quest {questSceneId} | Name: {name}")
|
||||
|
||||
try:
|
||||
await self.data.static.put_quest(
|
||||
questSceneId,
|
||||
0,
|
||||
sortNo,
|
||||
name,
|
||||
enabled
|
||||
)
|
||||
except Exception as err:
|
||||
self.logger.error(err)
|
||||
except Exception:
|
||||
self.logger.warning(f"Couldn't read csv file in {self.bin_dir}, skipping")
|
||||
|
||||
self.logger.info("Now reading HeroLog.csv")
|
||||
try:
|
||||
fullPath = bin_dir + "/HeroLog.csv"
|
||||
with open(fullPath, encoding="UTF-8") as fp:
|
||||
reader = csv.DictReader(fp)
|
||||
for row in reader:
|
||||
heroLogId = row["HeroLogId"]
|
||||
name = row["Name"]
|
||||
nickname = row["Nickname"]
|
||||
rarity = row["Rarity"]
|
||||
skillTableSubId = row["SkillTableSubId"]
|
||||
awakeningExp = row["AwakeningExp"]
|
||||
flavorText = row["FlavorText"]
|
||||
enabled = True
|
||||
self.logger.info("Now reading Property.csv")
|
||||
reader = self.load_csv_file("Property.csv")
|
||||
if reader:
|
||||
for row in reader:
|
||||
self.logger.info(f"Adding property {row['PropertyId']}")
|
||||
await self.data.static.put_property(row)
|
||||
|
||||
self.logger.info(f"Added hero {heroLogId} | Name: {name}")
|
||||
|
||||
try:
|
||||
await self.data.static.put_hero(
|
||||
0,
|
||||
heroLogId,
|
||||
name,
|
||||
nickname,
|
||||
rarity,
|
||||
skillTableSubId,
|
||||
awakeningExp,
|
||||
flavorText,
|
||||
enabled
|
||||
)
|
||||
except Exception as err:
|
||||
self.logger.error(err)
|
||||
except Exception:
|
||||
self.logger.warning(f"Couldn't read csv file in {self.bin_dir}, skipping")
|
||||
|
||||
self.logger.info("Now reading Equipment.csv")
|
||||
try:
|
||||
fullPath = bin_dir + "/Equipment.csv"
|
||||
with open(fullPath, encoding="UTF-8") as fp:
|
||||
reader = csv.DictReader(fp)
|
||||
for row in reader:
|
||||
equipmentId = row["EquipmentId"]
|
||||
equipmentType = row["EquipmentType"]
|
||||
weaponTypeId = row["WeaponTypeId"]
|
||||
name = row["Name"]
|
||||
rarity = row["Rarity"]
|
||||
flavorText = row["FlavorText"]
|
||||
enabled = True
|
||||
reader = self.load_csv_file("Equipment.csv")
|
||||
if reader:
|
||||
for row in reader:
|
||||
self.logger.info(f"Adding equipment {row['EquipmentId']}")
|
||||
await self.data.static.put_equipment(row)
|
||||
|
||||
self.logger.info(f"Added equipment {equipmentId} | Name: {name}")
|
||||
|
||||
try:
|
||||
await self.data.static.put_equipment(
|
||||
0,
|
||||
equipmentId,
|
||||
name,
|
||||
equipmentType,
|
||||
weaponTypeId,
|
||||
rarity,
|
||||
flavorText,
|
||||
enabled
|
||||
)
|
||||
except Exception as err:
|
||||
self.logger.error(err)
|
||||
except Exception:
|
||||
self.logger.warning(f"Couldn't read csv file in {self.bin_dir}, skipping")
|
||||
self.logger.info("Now reading Skill.csv")
|
||||
reader = self.load_csv_file("Skill.csv")
|
||||
if reader:
|
||||
for row in reader:
|
||||
self.logger.info(f"Adding skill {row['SkillId']}")
|
||||
await self.data.static.put_skill(row)
|
||||
|
||||
self.logger.info("Now reading SkillTable.csv")
|
||||
reader = self.load_csv_file("SkillTable.csv")
|
||||
if reader:
|
||||
for row in reader:
|
||||
self.logger.info(f"Adding skill table {row['SkillId']} | SubId: {row['SkillTableSubId']}")
|
||||
await self.data.static.put_skill_table(row['SkillId'], row['SkillTableSubId'], row['Level'], row['AwakeningId'], row['SkillTableId'])
|
||||
|
||||
self.logger.info("Now reading HeroLog.csv")
|
||||
reader = self.load_csv_file("HeroLog.csv")
|
||||
if reader:
|
||||
for row in reader:
|
||||
self.logger.info(f"Adding hero {row['HeroLogId']}")
|
||||
await self.data.static.put_hero(row)
|
||||
|
||||
self.logger.info("Now reading Item.csv")
|
||||
try:
|
||||
fullPath = bin_dir + "/Item.csv"
|
||||
with open(fullPath, encoding="UTF-8") as fp:
|
||||
reader = csv.DictReader(fp)
|
||||
for row in reader:
|
||||
itemId = row["ItemId"]
|
||||
itemTypeId = row["ItemTypeId"]
|
||||
name = row["Name"]
|
||||
rarity = row["Rarity"]
|
||||
flavorText = row["FlavorText"]
|
||||
enabled = True
|
||||
reader = self.load_csv_file("Item.csv")
|
||||
if reader:
|
||||
for row in reader:
|
||||
self.logger.info(f"Adding item {row['ItemId']}")
|
||||
await self.data.static.put_item(row)
|
||||
|
||||
self.logger.info(f"Added item {itemId} | Name: {name}")
|
||||
|
||||
try:
|
||||
await self.data.static.put_item(
|
||||
0,
|
||||
itemId,
|
||||
name,
|
||||
itemTypeId,
|
||||
rarity,
|
||||
flavorText,
|
||||
enabled
|
||||
)
|
||||
except Exception as err:
|
||||
self.logger.error(err)
|
||||
except Exception:
|
||||
self.logger.warning(f"Couldn't read csv file in {self.bin_dir}, skipping")
|
||||
|
||||
self.logger.info("Now reading SupportLog.csv")
|
||||
try:
|
||||
fullPath = bin_dir + "/SupportLog.csv"
|
||||
with open(fullPath, encoding="UTF-8") as fp:
|
||||
reader = csv.DictReader(fp)
|
||||
for row in reader:
|
||||
supportLogId = row["SupportLogId"]
|
||||
charaId = row["CharaId"]
|
||||
name = row["Name"]
|
||||
rarity = row["Rarity"]
|
||||
salePrice = row["SalePrice"]
|
||||
skillName = row["SkillName"]
|
||||
enabled = True
|
||||
reader = self.load_csv_file("SupportLog.csv")
|
||||
if reader:
|
||||
for row in reader:
|
||||
supportLogId = row["SupportLogId"]
|
||||
charaId = row["CharaId"]
|
||||
name = row["Name"]
|
||||
rarity = row["Rarity"]
|
||||
salePrice = row["SalePrice"]
|
||||
skillName = row["SkillName"]
|
||||
enabled = True
|
||||
|
||||
self.logger.info(f"Adding support log {supportLogId}")
|
||||
await self.data.static.put_support_log(
|
||||
0,
|
||||
supportLogId,
|
||||
charaId,
|
||||
name,
|
||||
rarity,
|
||||
salePrice,
|
||||
skillName,
|
||||
enabled
|
||||
)
|
||||
|
||||
self.logger.info(f"Added support log {supportLogId} | Name: {name}")
|
||||
|
||||
try:
|
||||
await self.data.static.put_support_log(
|
||||
0,
|
||||
supportLogId,
|
||||
charaId,
|
||||
name,
|
||||
rarity,
|
||||
salePrice,
|
||||
skillName,
|
||||
enabled
|
||||
)
|
||||
except Exception as err:
|
||||
self.logger.error(err)
|
||||
except Exception:
|
||||
self.logger.warning(f"Couldn't read csv file in {self.bin_dir}, skipping")
|
||||
|
||||
self.logger.info("Now reading Title.csv")
|
||||
try:
|
||||
fullPath = bin_dir + "/Title.csv"
|
||||
with open(fullPath, encoding="UTF-8") as fp:
|
||||
reader = csv.DictReader(fp)
|
||||
for row in reader:
|
||||
titleId = row["TitleId"]
|
||||
displayName = row["DisplayName"]
|
||||
requirement = row["Requirement"]
|
||||
rank = row["Rank"]
|
||||
imageFilePath = row["ImageFilePath"]
|
||||
enabled = True
|
||||
reader = self.load_csv_file("Title.csv")
|
||||
if reader:
|
||||
for row in reader:
|
||||
titleId = row["TitleId"]
|
||||
displayName = row["DisplayName"]
|
||||
requirement = row["Requirement"]
|
||||
rank = row["Rank"]
|
||||
imageFilePath = row["ImageFilePath"]
|
||||
enabled = True
|
||||
|
||||
self.logger.info(f"Added title {titleId} | Name: {displayName}")
|
||||
|
||||
if len(titleId) > 5:
|
||||
try:
|
||||
await self.data.static.put_title(
|
||||
0,
|
||||
titleId,
|
||||
displayName,
|
||||
requirement,
|
||||
rank,
|
||||
imageFilePath,
|
||||
enabled
|
||||
)
|
||||
except Exception as err:
|
||||
self.logger.error(err)
|
||||
elif len(titleId) < 6: # current server code cannot have multiple lengths for the id
|
||||
continue
|
||||
except Exception:
|
||||
self.logger.warning(f"Couldn't read csv file in {self.bin_dir}, skipping")
|
||||
self.logger.info(f"Adding title {titleId}")
|
||||
await self.data.static.put_title(
|
||||
0,
|
||||
titleId,
|
||||
displayName,
|
||||
requirement,
|
||||
rank,
|
||||
imageFilePath,
|
||||
enabled
|
||||
)
|
||||
|
||||
self.logger.info("Now reading RareDropTable.csv")
|
||||
try:
|
||||
fullPath = bin_dir + "/RareDropTable.csv"
|
||||
with open(fullPath, encoding="UTF-8") as fp:
|
||||
reader = csv.DictReader(fp)
|
||||
for row in reader:
|
||||
questRareDropId = row["QuestRareDropId"]
|
||||
commonRewardId = row["CommonRewardId"]
|
||||
enabled = True
|
||||
self.logger.info("Now reading QuestRareDrop.csv")
|
||||
reader = self.load_csv_file("QuestRareDrop.csv")
|
||||
if reader:
|
||||
for row in reader:
|
||||
questRareDropId = row["QuestRareDropId"]
|
||||
commonRewardId = row["CommonRewardId"]
|
||||
enabled = True
|
||||
|
||||
self.logger.info(f"Added rare drop {questRareDropId} | Reward: {commonRewardId}")
|
||||
|
||||
try:
|
||||
await self.data.static.put_rare_drop(
|
||||
0,
|
||||
questRareDropId,
|
||||
commonRewardId,
|
||||
enabled
|
||||
)
|
||||
except Exception as err:
|
||||
self.logger.error(err)
|
||||
except Exception:
|
||||
self.logger.warning(f"Couldn't read csv file in {self.bin_dir}, skipping")
|
||||
self.logger.info(f"Adding rare drop {questRareDropId} | Reward: {commonRewardId}")
|
||||
await self.data.static.put_rare_drop(
|
||||
0,
|
||||
questRareDropId,
|
||||
commonRewardId,
|
||||
enabled
|
||||
)
|
||||
|
||||
self.logger.info("Now reading RewardTable.csv")
|
||||
reader = self.load_csv_file("RewardTable.csv")
|
||||
if reader:
|
||||
for row in reader:
|
||||
self.logger.info(f"Adding reward table {row['RewardTableId']} | Sub-ID: {row['RewardTableSubId']} | Reward {row['CommonRewardId']}")
|
||||
await self.data.static.put_reward_table(row)
|
||||
|
||||
self.logger.info("Now reading ExBonusTable.csv")
|
||||
reader = self.load_csv_file("ExBonusTable.csv")
|
||||
if reader:
|
||||
for row in reader:
|
||||
self.logger.info(f"Adding ex bonus {row['ExBonusTableId']} | Sub-ID: {row['ExBonusTableSubId']} | Reward {row['CommonRewardId']}")
|
||||
await self.data.static.put_ex_bonus(row)
|
||||
|
||||
self.logger.info("Now reading PlayerTraceTable.csv")
|
||||
reader = self.load_csv_file("PlayerTraceTable.csv")
|
||||
if reader:
|
||||
for row in reader:
|
||||
self.logger.info(f"Adding trace table {row['PlayerTraceTableId']} | Sub-ID: {row['PlayerTraceTableSubId']} | Reward {row['CommonRewardId']}")
|
||||
await self.data.static.put_player_trace(row)
|
||||
|
||||
self.logger.info("Now reading Episode.csv")
|
||||
reader = self.load_csv_file("Episode.csv")
|
||||
if reader:
|
||||
for row in reader:
|
||||
self.logger.info(f"Adding episode {row['EpisodeId']}")
|
||||
await self.data.static.put_episode(row)
|
||||
|
||||
self.logger.info("Now reading TrialTower.csv")
|
||||
reader = self.load_csv_file("TrialTower.csv")
|
||||
if reader:
|
||||
for row in reader:
|
||||
self.logger.info(f"Adding tower {row['TrialTowerId']}")
|
||||
await self.data.static.put_tower(row)
|
||||
|
||||
self.logger.info("Now reading ExTowerQuests.csv")
|
||||
reader = self.load_csv_file("ExTowerQuests.csv")
|
||||
if reader:
|
||||
for row in reader:
|
||||
self.logger.info(f"Adding ex tower {row['ExTowerQuestId']}")
|
||||
await self.data.static.put_ex_tower(row)
|
||||
|
||||
self.logger.info("Now reading SideQuest.csv")
|
||||
reader = self.load_csv_file("SideQuest.csv")
|
||||
if reader:
|
||||
for row in reader:
|
||||
self.logger.info(f"Adding side quest {row['SideQuestId']}")
|
||||
await self.data.static.put_side_quest(row)
|
||||
|
||||
Reference in New Issue
Block a user