Files
GaCy-GO/tools/update_prydwen_endgame_teams.py
T

122 lines
4.7 KiB
Python

#!/usr/bin/env python3
"""Refresh the MoC recommendation snapshot from the credited Prydwen dataset."""
from __future__ import annotations
import argparse
import datetime as dt
import json
import pathlib
import urllib.request
DATASET_ROOT = "https://huggingface.co/datasets/LvlUrArti/MocDataProcessed/resolve/main"
REPOSITORY_ROOT = pathlib.Path(__file__).resolve().parents[1]
RESOURCE_PATH = REPOSITORY_ROOT / "resources" / "PrydwenEndgameTeams.json"
AVATAR_CONFIG_PATH = REPOSITORY_ROOT / "resources" / "AvatarConfig.json"
def fetch_json(path: str) -> object:
request = urllib.request.Request(
f"{DATASET_ROOT}/{path}", headers={"User-Agent": "GaC-hkrpg-go-recommendations/1.0"}
)
with urllib.request.urlopen(request, timeout=30) as response:
return json.load(response)
def normalized(value: str) -> str:
return "".join(char.lower() for char in value if char.isalnum())
def latest_dataset_version(config: dict[str, dict]) -> tuple[str, dict]:
available = []
for version, metadata in config.items():
mode = metadata.get("moc")
date_text = metadata.get("collect_date")
if not version[0].isdigit() or not isinstance(mode, dict) or not date_text:
continue
try:
collected = dt.datetime.strptime(date_text, "%d/%m/%Y").date()
except ValueError:
continue
available.append((collected, version, mode))
if not available:
raise RuntimeError("No dated MoC snapshot found in the source dataset config")
_, version, mode = max(available)
return version, mode
def avatar_tags() -> set[str]:
with AVATAR_CONFIG_PATH.open(encoding="utf-8") as source:
config = json.load(source)
return {
normalized(entry["AvatarVOTag"])
for entry in config.get("avatar_config", [])
if entry.get("AvatarVOTag")
}
def stage_team(version: str, floor: int, stage: int, tags: set[str], aliases: dict[str, str]) -> dict:
path = f"{version}/moc/comps/{floor}-{stage}_combined.json"
rows = fetch_json(path)
if not isinstance(rows, list):
raise RuntimeError(f"Unexpected dataset shape in {path}")
top = next((row for row in rows if row.get("rank") == 1), None)
if top is None:
raise RuntimeError(f"No rank-1 lineup found in {path}")
characters = [top.get(f"char_{slot}") for slot in ("one", "two", "three", "four")]
if any(not isinstance(slug, str) or not slug for slug in characters):
raise RuntimeError(f"Incomplete rank-1 lineup in {path}")
unresolved = [slug for slug in characters if normalized(aliases.get(slug, slug)) not in tags]
if unresolved:
raise RuntimeError(
f"Source slugs do not map to this Client's AvatarConfig: {', '.join(unresolved)}; "
"add their source-slug to AvatarVOTag aliases before refreshing"
)
return {"stage": stage, "rank": 1, "characters": characters}
def main() -> None:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--floor", type=int, required=True, help="Prydwen MoC floor shown by the Client")
parser.add_argument("--stage-count", type=int, required=True, help="number of stage teams to fetch")
args = parser.parse_args()
if args.floor <= 0 or args.stage_count <= 0:
parser.error("--floor and --stage-count must be positive")
config = fetch_json("config.json")
if not isinstance(config, dict):
raise RuntimeError("Unexpected source config shape")
version, cycle = latest_dataset_version(config)
aliases = {}
if RESOURCE_PATH.exists():
with RESOURCE_PATH.open(encoding="utf-8") as source:
aliases = json.load(source).get("avatar_aliases", {})
tags = avatar_tags()
stages = [
stage_team(version, args.floor, stage, tags, aliases)
for stage in range(1, args.stage_count + 1)
]
output = {
"source": {
"name": "LvlUrArti/MocDataProcessed",
"url": "https://huggingface.co/datasets/LvlUrArti/MocDataProcessed",
"license": "MIT",
"attribution": "Processed data shown on Prydwen; credit LvlUrArti",
"dataset_version": version,
"collect_date": dt.datetime.strptime(config[version]["collect_date"], "%d/%m/%Y").date().isoformat(),
"challenge_version": cycle.get("ver"),
"challenge_name": cycle.get("name"),
},
"avatar_aliases": aliases,
"mode": "moc",
"floor": args.floor,
"stages": stages,
}
RESOURCE_PATH.write_text(json.dumps(output, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
print(f"Updated {RESOURCE_PATH.relative_to(REPOSITORY_ROOT)} from dataset {version}: {len(stages)} stages")
if __name__ == "__main__":
main()