refactor: drop players.json support

utils/export_b50.py

@@ -0,0 +1,180 @@
+import orjson as json
+from typing import Callable
+from datetime import datetime
+from decimal import Decimal, getcontext
+
+from hasher import salted_hash_userid
+
+
+with open("musicDB.json", "r", encoding="utf-8") as f:
+    music_db = json.loads(f.read())
+
+music_db = {entry["id"]: entry for entry in music_db}
+
+# Set Decimal precision
+getcontext().prec = 28
+
+# Constants
+SSS_PLUS_THRESHOLD = Decimal("100.5")
+SSS_PLUS_FACTOR = Decimal("0.224")
+SSS_PRO_THRESHOLD = Decimal("100.4999")
+SSS_PRO_FACTOR = Decimal("0.222")
+SSS_THRESHOLD = Decimal("100.0")
+SSS_FACTOR = Decimal("0.216")
+SS_PLUS_PRO_THRESHOLD = Decimal("99.9999")
+SS_PLUS_PRO_FACTOR = Decimal("0.214")
+SS_PLUS_THRESHOLD = Decimal("99.5")
+SS_PLUS_FACTOR = Decimal("0.211")
+SS_THRESHOLD = Decimal("99.0")
+SS_FACTOR = Decimal("0.208")
+S_PLUS_PRO_THRESHOLD = Decimal("98.9999")
+S_PLUS_PRO_FACTOR = Decimal("0.206")
+S_PLUS_THRESHOLD = Decimal("98.0")
+S_PLUS_FACTOR = Decimal("0.203")
+S_THRESHOLD = Decimal("97.0")
+S_FACTOR = Decimal("0.2")
+AAA_PRO_THRESHOLD = Decimal("96.9999")
+AAA_PRO_FACTOR = Decimal("0.176")
+AAA_THRESHOLD = Decimal("94.0")
+AAA_FACTOR = Decimal("0.168")
+AA_THRESHOLD = Decimal("90.0")
+AA_FACTOR = Decimal("0.152")
+A_THRESHOLD = Decimal("80.0")
+A_FACTOR = Decimal("0.136")
+
+
+def dx_rating(difficulty: Decimal, achievement: int) -> int:
+    ach = Decimal(achievement) / Decimal("10000")
+    if ach > Decimal("101.0") or ach < A_THRESHOLD:
+        return 0
+    if ach >= SSS_PLUS_THRESHOLD:
+        factor = SSS_PLUS_FACTOR
+        ach = Decimal("100.5")
+    elif ach >= SSS_PRO_THRESHOLD:
+        factor = SSS_PRO_FACTOR
+    elif ach >= SSS_THRESHOLD:
+        factor = SSS_FACTOR
+    elif ach >= SS_PLUS_PRO_THRESHOLD:
+        factor = SS_PLUS_PRO_FACTOR
+    elif ach >= SS_PLUS_THRESHOLD:
+        factor = SS_PLUS_FACTOR
+    elif ach >= SS_THRESHOLD:
+        factor = SS_FACTOR
+    elif ach >= S_PLUS_PRO_THRESHOLD:
+        factor = S_PLUS_PRO_FACTOR
+    elif ach >= S_PLUS_THRESHOLD:
+        factor = S_PLUS_FACTOR
+    elif ach >= S_THRESHOLD:
+        factor = S_FACTOR
+    elif ach >= AAA_PRO_THRESHOLD:
+        factor = AAA_PRO_FACTOR
+    elif ach >= AAA_THRESHOLD:
+        factor = AAA_FACTOR
+    elif ach >= AA_THRESHOLD:
+        factor = AA_FACTOR
+    elif ach >= A_THRESHOLD:
+        factor = A_FACTOR
+    else:
+        return 0
+    result = (factor * difficulty * ach).quantize(Decimal("1."), rounding="ROUND_FLOOR")
+    return int(result)
+
+
+def clean_b50(b50: dict[str, str | dict]):
+    urating: dict[str, list[dict[str, int]]] = b50["userRating"]
+
+    def add_rating(entry: dict[str, int]):
+        """
+        ```
+        {
+            "musicId": 11638,
+            "level": 2,
+            "romVersion": 24005,
+            "achievement": 988145
+        }
+        ```
+        - level: EXPERT
+        - ver: DX, 1.40.05
+        - ach: 98.8145%
+        """
+
+        entry["musicTitle"] = None
+        entry["difficulty"] = None
+        entry["dxRating"] = 0
+
+        music_info = music_db.get(entry["musicId"])
+        if music_info is None:
+            return
+
+        entry["musicTitle"] = music_info["name"]
+
+        levels = [
+            level for level in music_info["levels"] if level["level"] == entry["level"]
+        ]
+
+        if levels:
+            level: dict[str, str | int] = levels.pop()
+            difficulty = level["difficulty"]
+
+            entry["difficulty"] = difficulty
+            entry["dxRating"] = dx_rating(
+                difficulty=Decimal(difficulty),
+                achievement=entry["achievement"],
+            )
+
+    for b35 in urating["ratingList"]:
+        add_rating(b35)
+    for b15 in urating["newRatingList"]:
+        add_rating(b15)
+
+    urating["rating"] = sum(
+        map(
+            lambda lst: sum(map(lambda entry: entry["dxRating"], urating[lst])),
+            ["ratingList", "newRatingList"],
+        )
+    )
+
+
+def record_time(*, _: list[datetime] = []):
+    last_time = _
+    if not last_time:
+        last_time.append(datetime.now())
+    else:
+        new = datetime.now()
+        diff = (new - last_time.pop()).total_seconds()
+        last_time.append(new)
+        return diff
+
+
+def process(
+    clean_fields: Callable[[dict], None],
+    input_file: str,
+    output_file: str,
+):
+    record_time()
+    with open(input_file, "rb") as f:
+        data = json.loads(f.read())
+    print(f"loaded, cost {record_time():.2f}s")
+
+    for entry in data:
+        entry["userId"] = salted_hash_userid(entry["userId"])
+        clean_fields(entry)
+    print(f"processed, cost {record_time():.2f}s")
+
+    with open(output_file, "wb") as f:
+        f.write(json.dumps(data))
+    print(f"written out, cost {record_time():.2f}s")
+
+    return data
+
+
+def main():
+    process(
+        clean_b50,
+        "b50.json",
+        "b50_pub.json",
+    )
+
+
+if __name__ == "__main__":
+    main()