-
Notifications
You must be signed in to change notification settings - Fork 10
/
gf_utils.py
118 lines (104 loc) · 3.33 KB
/
gf_utils.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
# %%
import json
import logging
import os
import socket
from collections.abc import MutableMapping
from pathlib import Path
from socket import timeout
from urllib import request
from urllib.error import URLError
logger = logging.getLogger(__name__)
# %%
special_keys = {
"achievement": "identity",
"attendance_info": None,
"auto_mission": "mission_id",
"bingo_task_info": "task_id",
"chess_creation_logic": None,
"daily_info": "identity",
"daily": "identity",
"enemy_standard_attribute": "level",
"equip_category": "category",
"equip_exp_info": "level",
"equip_type": "type",
"furniture_establish_info": "establish_id",
"game_config_info": "parameter_name",
"guild_level": "lv",
"gun_exp_info": "lv",
"gun_obtain_info": "obtain_id",
"kalina_favor_info": "level",
"main_quest_info": "identity",
"mission_draw_bonus": None,
"mission_event_prize_info": "mission_id",
"mission_targettrain_battlesetting": "difficult_level",
"sangvis_advance": "lv",
"sangvis_exp": "lv",
"seven_attendance_info": None,
"seven_spendpoint_info": None,
"squad_chip_exp": "lv",
"squad_exp": "lv",
"squad_rank": "star_id",
"squad_type": "type_id",
"weekly_info": "identity",
"weekly": "identity",
}
class GameData(MutableMapping):
def __init__(self, stc_dir, to_dict=True) -> None:
self.stc_dir = Path(stc_dir)
self.to_dict = to_dict
self.__keys = [p.name[:-5] for p in self.stc_dir.glob("*.json")]
self.__data = {}
def __get_stc_dict(self, name):
logger.debug(f"Reading {name}.json")
with (self.stc_dir / f"{name}.json").open("r", encoding="utf-8") as f:
data = json.load(f)
if self.to_dict and len(data) > 0:
k = (
"id"
if "id" in data[0].keys()
else (special_keys[name] if name in special_keys.keys() else None)
)
if k is not None:
data = {d[k]: d for d in data}
return data
def __getitem__(self, key):
if key not in self.__keys:
raise KeyError(key)
if key not in self.__data:
self.__data[key] = self.__get_stc_dict(key)
return self.__data[key]
def __getattr__(self, k):
return self[k]
def __call__(self, k):
return self[k]
def __setitem__(self, key, value):
pass
def __delitem__(self, key):
pass
def __iter__(self):
return iter(self.__keys)
def __len__(self):
return len(self.__keys)
def download(url, path, max_retry=10, timeout_sec=5):
socket.setdefaulttimeout(timeout_sec)
fname = os.path.split(path)[-1]
logger.info(f"Start downloading {fname}")
for i in range(max_retry):
try:
if not os.path.exists(path):
request.urlretrieve(url, path + ".tmp")
os.rename(path + ".tmp", path)
except Exception as e:
if i + 1 < max_retry:
logger.warning(
f"Failed to download {fname} for {i+1}/{max_retry} tries"
)
continue
else:
logger.exception(repr(e))
raise
else:
logger.info(f"Successfully downloaded {fname}")
break
return path