refactor: all script refactor

.github/workflows/data_update.yml

@@ -28,13 +28,9 @@ jobs:
 
       - run: poetry install --no-interaction
 
-      - name: Run akashi_schedule
+      - name: Run script
         run: |
-          poetry run python -m script.akashi_schedule
-
-      - name: Run l10n
-        run: |
-          poetry run python -m script.l10n
+          poetry run python -m script.run
 
       - name: Commit changes
         id: commit

data/version.json

@@ -0,0 +1 @@
+{"version": "2024072712"}

script/__init__.py

@@ -1,9 +1,4 @@
 import os
-from datetime import datetime
 
 ROOT_PATH = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
 
-# generate a DATA_VERSION (YYYYMMDDHH)
-DATA_VERSION = datetime.strftime(datetime.utcnow(), '%Y%m%d%H')
-
-print(DATA_VERSION)

script/akashi_schedule.py

@@ -1,47 +1,59 @@
+import json
 import os.path
 
 from script import *
-import json
 
 
-def data_reformat(id, data, use_item_map):
-    res = {"id": int(id)}
-    for k, v in data.items():
-        if k == 'improvement':
-            res[k] = []
-            for i in v:
-                upgrade = i['upgrade']
-                req = i['req']
-                resource = i['resource']
-                req_item = i.get('require_item', None)
-
-                improve = {
-                    "req": [{
-                        "ship": item[1] if isinstance(item[1], list) else [],
-                        "day": day_stringify(item[0])
-                    } for item in req],
-                    "resource": resource_reformat(resource, req_item, use_item_map),
-                }
-                if upgrade:
-                    improve["upgrade"] = {"id": upgrade[0], "lv": upgrade[1]}
-
-                res[k].append(improve)
-    return res
-
-
-def resource_reformat(data_list, req_item, use_item_map):
-    res = {"base": data_list[0], "extra": []}
-    for i in data_list[1:]:
-        item = {
-            "dm": i[0:2],
-            "im": i[2:4],
+class Item:
+    def __init__(self, id, improve_data, use_item_map):
+        self.improvement = None
+        self.id = int(id)
+        self.improve_data = improve_data
+        self.use_item_map = use_item_map
+
+    def __dict__(self):
+        self.reformat()
+        res = {
+            "id": self.id,
+            "improvement": self.improvement
         }
-        if i[4] > 0:
-            item["slot"] = {str(i[4]): i[5]}
-        res["extra"].append(item)
-    if req_item:
-        res["extra"][2]["use"] = {str(use_item_map[item[0]]): item[1] for item in req_item}
-    return res
+        assert len(self.improvement) > 0
+        return res
+
+    def reformat(self):
+        self.improvement = []
+
+        for i in self.improve_data['improvement']:
+            upgrade = i['upgrade']
+            req = i['req']
+            resource = i['resource']
+            req_item = i.get('require_item', None)
+
+            improve = {
+                "req": [{
+                    "ship": item[1] if isinstance(item[1], list) else [],
+                    "day": day_stringify(item[0])
+                } for item in req],
+                "resource": self.resource_reformat(resource, req_item),
+            }
+            if upgrade:
+                improve["upgrade"] = {"id": upgrade[0], "lv": upgrade[1]}
+
+            self.improvement.append(improve)
+
+    def resource_reformat(self, data_list, req_item):
+        res = {"base": data_list[0], "extra": []}
+        for i in data_list[1:]:
+            item = {
+                "dm": i[0:2],
+                "im": i[2:4],
+            }
+            if i[4] > 0:
+                item["slot"] = {str(i[4]): i[5]}
+            res["extra"].append(item)
+        if req_item:
+            res["extra"][2]["use"] = {str(self.use_item_map[item[0]]): item[1] for item in req_item}
+        return res
 
 
 def day_stringify(day):
@@ -51,7 +63,7 @@ def day_stringify(day):
     return "".join([str(int(i)) for i in day])
 
 
-if __name__ == '__main__':
+def update_schedule(data_version):
     kcanotify_path = os.path.join(ROOT_PATH, 'repo', 'kcanotify-gamedata')
     kcanotify_data_path = os.path.join(kcanotify_path, 'files')
     data_path = os.path.join(ROOT_PATH, 'data')
@@ -63,10 +75,13 @@ def day_stringify(day):
         data_map = json.load(f)
     improve_list = []
     for k, v in data_map.items():
-        improve_list.append(data_reformat(k, v, item_map))
+        improve_list.append(Item(k, v, item_map).__dict__())
+
+    print(f'{len(improve_list)} items')
+
     with open(os.path.join(data_path, 'akashi_schedule.json'), 'w') as f:
         data = {
-            "data_version": str(DATA_VERSION),
+            "data_version": str(data_version),
             "items": improve_list
         }
-        json.dump(data, f, indent=2)
+        json.dump(data, f, indent=2)

script/l10n.py

@@ -1,72 +1,66 @@
 import json
 import os
 
-from kancolle.models import slot_item
 import requests
+from kancolle.models import slot_item
 
-from script import ROOT_PATH, DATA_VERSION
+from script import ROOT_PATH
 
 
-def req_kcdata_json():
-    from kancolle.data import KC_DATA_URL
-    ship_class_url = KC_DATA_URL + "slotitem/all.json"
-    return requests.get(ship_class_url).json()
+class Localization:
+    def __init__(self, version):
+        self.version = version
+        self.kcanotify_data_path = os.path.join(ROOT_PATH, 'repo', 'kcanotify-gamedata', 'files')
 
+    def update_item_l10n(self):
+        data_path = "slotitem/all.json"
+        items = slot_item.load_slot_item_list(self.req_kcdata_json(data_path))
 
-def load_kcanotify_l10n():
-    kcanotify_path = os.path.join(ROOT_PATH, 'repo', 'kcanotify-gamedata')
-    kcanotify_data_path = os.path.join(kcanotify_path, 'files')
-    item_en_path = os.path.join(kcanotify_data_path, 'items-en.json')
-    item_ko_path = os.path.join(kcanotify_data_path, 'items-ko.json')
-    item_sc_path = os.path.join(kcanotify_data_path, 'items-scn.json')
-    item_tc_path = os.path.join(kcanotify_data_path, 'items-tcn.json')
-    with open(item_en_path, 'r') as f:
-        item_en = json.load(f)
-    with open(item_ko_path, 'r') as f:
-        item_ko = json.load(f)
-    with open(item_sc_path, 'r') as f:
-        item_sc = json.load(f)
-    with open(item_tc_path, 'r') as f:
-        item_tc = json.load(f)
-    return item_en, item_ko, item_sc, item_tc
+        result = {item.id: {"ja": item.name} for item in items}
 
+        translations = self.load_kcanotify_l10n()
 
-def check_l10n(data):
-    for item in data.values():
-        try:
-            assert item.get("ja") is not None
-            assert item.get("en") is not None
-            assert item.get("ko") is not None
-            assert item.get("sc") is not None
-            assert item.get("tc") is not None
-        except AssertionError:
-            print(f'item {item["ja"]} is missing translation')
-            raise
+        for k, v in result.items():
+            for lang, translation in translations.items():
+                if v["ja"] in translation:
+                    result[k][lang] = translation[v["ja"]]
 
+        self.check_l10n(result)
 
-if __name__ == '__main__':
-    items = slot_item.load_slot_item_list(req_kcdata_json())
+        print(f'{len(result)} translations, {len(items)} items')
 
-    l10n_dict = {}
-    for item in items:
-        l10n_dict[item.id] = {"ja": item.name}
+        with open(os.path.join(ROOT_PATH, 'data', 'item_l10n.json'), 'w') as f:
+            data = {
+                "data_version": str(self.version),
+                "data": result
+            }
+            json.dump(data, f, indent=2, ensure_ascii=False)
 
-    item_en, item_ko, item_sc, item_tc = load_kcanotify_l10n()
-    for k, v in l10n_dict.items():
-        if v["ja"] in item_en:
-            l10n_dict[k]["en"] = item_en[v["ja"]]
-        if v["ja"] in item_ko:
-            l10n_dict[k]["ko"] = item_ko[v["ja"]]
-        if v["ja"] in item_sc:
-            l10n_dict[k]["sc"] = item_sc[v["ja"]]
-        if v["ja"] in item_tc:
-            l10n_dict[k]["tc"] = item_tc[v["ja"]]
+    def load_kcanotify_l10n(self):
+        kcanotify_data_path = self.kcanotify_data_path
+        item_paths = {
+            'en': os.path.join(kcanotify_data_path, 'items-en.json'),
+            'ko': os.path.join(kcanotify_data_path, 'items-ko.json'),
+            'sc': os.path.join(kcanotify_data_path, 'items-scn.json'),
+            'tc': os.path.join(kcanotify_data_path, 'items-tcn.json')
+        }
+        translations = {}
+        for lang, path in item_paths.items():
+            with open(path, 'r') as f:
+                translations[lang] = json.load(f)
+        return translations
 
-    check_l10n(l10n_dict)
+    @staticmethod
+    def check_l10n(data):
+        for item in data.values():
+            try:
+                assert all(k in item for k in ['ja', 'en', 'ko', 'sc', 'tc'])
+            except AssertionError:
+                print(item)
+                raise
 
-    with open(os.path.join(ROOT_PATH, 'data', 'item_l10n.json'), 'w') as f:
-        data = {
-            "data_version": str(DATA_VERSION),
-            "data": l10n_dict
-        }
-        json.dump(data, f, indent=2, ensure_ascii=False)
+    @staticmethod
+    def req_kcdata_json(path):
+        from kancolle.data import KC_DATA_URL
+        ship_class_url = KC_DATA_URL + path
+        return requests.get(ship_class_url).json()

script/run.py

@@ -0,0 +1,27 @@
+import json
+import os
+from datetime import datetime
+
+from script import ROOT_PATH
+from script import akashi_schedule
+from script.l10n import Localization
+
+if __name__ == '__main__':
+    data_version = datetime.strftime(datetime.utcnow(), '%Y%m%d%H')
+    print(data_version)
+
+    l10n = Localization(data_version)
+    l10n.update_item_l10n()
+
+    akashi_schedule.update_schedule(data_version)
+
+    data_version_path = os.path.join(ROOT_PATH, 'data', 'version.json')
+
+    with open(data_version_path, 'w') as f:
+        version_json = {
+            "version": data_version
+        }
+
+        f.write(json.dumps(version_json))
+
+    print('finish')