cw cmd and unstuck and more
This commit is contained in:
1 parent
ca5831c928
commit
ffd6b2e853
11 files changed
+1370
-2
No files matched your search
@@ -0,0 +1,437 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Build the read-only item encyclopedia from pinned game and image data."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
import tempfile
|
||||
import time
|
||||
import urllib.error
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
GAME_PROJECT = "Dimbreath/turnbasedgamedata"
|
||||
ASSET_PROJECT = "Mar-7th/StarRailRes"
|
||||
ASSET_LANGUAGES = {
|
||||
"th": "th", "en": "en", "zh-CN": "cn", "zh-TW": "cht", "ja": "jp",
|
||||
"ko": "kr", "vi": "vi", "id": "id", "es": "es", "fr": "fr",
|
||||
"de": "de", "ru": "ru", "pt": "pt",
|
||||
}
|
||||
TEXTMAP_TOKENS = {
|
||||
"th": ("TH",), "en": ("EN",), "zh-CN": ("CHS",), "zh-TW": ("CHT",),
|
||||
"ja": ("JP",), "ko": ("KR",), "vi": ("VI",), "id": ("ID",),
|
||||
"es": ("ES",), "fr": ("FR",), "de": ("DE",), "ru": ("RU",), "pt": ("PT",),
|
||||
}
|
||||
TEXT_FIELDS = ("ItemName", "ItemDesc", "ItemBGDesc")
|
||||
ICON_FIELDS = ("ItemIconPath", "ItemCurrencyIconPath", "ItemAvatarIconPath", "ItemFigureIconPath")
|
||||
|
||||
|
||||
def request_bytes(url: str, attempts: int = 4) -> bytes:
|
||||
request = urllib.request.Request(url, headers={"User-Agent": "GaC-Srtool-item-catalog/1.0"})
|
||||
last_error = None
|
||||
for attempt in range(attempts):
|
||||
try:
|
||||
with urllib.request.urlopen(request, timeout=90) as response:
|
||||
return response.read()
|
||||
except (OSError, urllib.error.URLError) as error:
|
||||
last_error = error
|
||||
if attempt + 1 < attempts:
|
||||
time.sleep(1.5 * (attempt + 1))
|
||||
raise RuntimeError(f"Could not download {url}: {last_error}")
|
||||
|
||||
|
||||
def request_json(url: str):
|
||||
return json.loads(request_bytes(url))
|
||||
|
||||
|
||||
def cache_path(cache_dir: Path, key: str) -> Path:
|
||||
return cache_dir / (re.sub(r"[^A-Za-z0-9_.-]+", "_", key) + ".json")
|
||||
|
||||
|
||||
def cached_json(url: str, cache_dir: Path, key: str):
|
||||
path = cache_path(cache_dir, key)
|
||||
if not path.exists():
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
payload = request_bytes(url)
|
||||
temporary = path.with_suffix(".tmp")
|
||||
temporary.write_bytes(payload)
|
||||
temporary.replace(path)
|
||||
return json.loads(path.read_bytes())
|
||||
|
||||
|
||||
def gitlab_api(project: str, endpoint: str) -> str:
|
||||
return f"https://gitlab.com/api/v4/projects/{urllib.parse.quote(project, safe='')}/{endpoint}"
|
||||
|
||||
|
||||
def gitlab_raw(project: str, revision: str, file_path: str) -> str:
|
||||
encoded = urllib.parse.quote(file_path, safe="")
|
||||
return gitlab_api(project, f"repository/files/{encoded}/raw?ref={revision}")
|
||||
|
||||
|
||||
def discover_files(project: str, revision: str, directory: str) -> list[str]:
|
||||
names: list[str] = []
|
||||
page = 1
|
||||
while True:
|
||||
query = urllib.parse.urlencode({
|
||||
"ref": revision, "path": directory, "recursive": "true", "per_page": 100, "page": page,
|
||||
})
|
||||
response = request_json(gitlab_api(project, f"repository/tree?{query}"))
|
||||
names.extend(entry["path"] for entry in response if entry.get("type") == "blob")
|
||||
if len(response) < 100:
|
||||
break
|
||||
page += 1
|
||||
if page > 100:
|
||||
raise RuntimeError(f"Unexpectedly large file listing under {directory}")
|
||||
return sorted(names)
|
||||
|
||||
|
||||
def fetch_game_json_files(paths: list[str], revision: str, cache_dir: Path) -> dict[str, object]:
|
||||
result: dict[str, object] = {}
|
||||
|
||||
def load_one(path: str):
|
||||
return path, cached_json(gitlab_raw(GAME_PROJECT, revision, path), cache_dir, f"game_{path}")
|
||||
|
||||
with ThreadPoolExecutor(max_workers=8) as pool:
|
||||
futures = [pool.submit(load_one, path) for path in paths]
|
||||
for future in as_completed(futures):
|
||||
path, value = future.result()
|
||||
result[path] = value
|
||||
return result
|
||||
|
||||
|
||||
def parse_rarity(value, index_row=None) -> int | None:
|
||||
if isinstance(value, int):
|
||||
return value if 1 <= value <= 5 else None
|
||||
if index_row and isinstance(index_row.get("rarity"), int):
|
||||
rarity = index_row["rarity"]
|
||||
return rarity if 1 <= rarity <= 5 else None
|
||||
rarity = str(value or "").lower()
|
||||
for label, number in (("superrare", 5), ("veryrare", 5), ("rare", 4), ("notnormal", 3), ("normal", 2), ("low", 1)):
|
||||
if label in rarity:
|
||||
return number
|
||||
return None
|
||||
|
||||
|
||||
def item_hash(row: dict, field: str):
|
||||
value = row.get(field)
|
||||
if isinstance(value, dict):
|
||||
value = value.get("Hash")
|
||||
if value is None or value == "":
|
||||
return None
|
||||
return str(value)
|
||||
|
||||
|
||||
def row_image_paths(row: dict) -> list[str]:
|
||||
values = []
|
||||
for field in ICON_FIELDS:
|
||||
value = row.get(field)
|
||||
if isinstance(value, str) and value and value not in values:
|
||||
values.append(value.replace("\\", "/").lstrip("/"))
|
||||
return values
|
||||
|
||||
|
||||
def rows_preference(path: str) -> tuple[int, str]:
|
||||
base = Path(path).name
|
||||
if base == "ItemConfig.json":
|
||||
return 0, path
|
||||
if base in ("ItemConfigEquipment.json", "ItemConfigRelic.json"):
|
||||
return 1, path
|
||||
return 2, path
|
||||
|
||||
|
||||
def resolve_source_icon(game_paths: list[str], category: str, item_id: int, indexes: dict, asset_files: set[str]) -> str | None:
|
||||
index_group = {"item": "items", "lightcone": "light_cones", "relic": "relics"}[category]
|
||||
index_row = indexes.get("en", {}).get(index_group, {}).get(str(item_id))
|
||||
if isinstance(index_row, dict):
|
||||
icon = index_row.get("icon")
|
||||
if isinstance(icon, str) and icon in asset_files:
|
||||
return icon
|
||||
|
||||
for source_path in game_paths:
|
||||
candidates: list[str] = []
|
||||
if source_path.startswith("SpriteOutput/ItemIcon/LightConeIcons/"):
|
||||
candidates.append(f"icon/light_cone/{item_id}.png")
|
||||
if source_path.startswith("SpriteOutput/ItemIcon/RelicIcons/"):
|
||||
relic_row = indexes.get("en", {}).get("relics", {}).get(str(item_id))
|
||||
if isinstance(relic_row, dict):
|
||||
candidates.append(relic_row.get("icon", ""))
|
||||
if source_path.startswith("SpriteOutput/ItemIcon/"):
|
||||
suffix = source_path.removeprefix("SpriteOutput/ItemIcon/")
|
||||
candidates.extend((f"icon/item/{suffix}", f"icon/{suffix}"))
|
||||
if source_path.startswith("SpriteOutput/AvatarIcon/Avatar/"):
|
||||
suffix = source_path.removeprefix("SpriteOutput/AvatarIcon/Avatar/")
|
||||
candidates.extend((f"image/character_portrait/{suffix}", f"icon/avatar/{suffix}"))
|
||||
if source_path.startswith("SpriteOutput/AvatarRoundIcon/Avatar/"):
|
||||
suffix = source_path.removeprefix("SpriteOutput/AvatarRoundIcon/Avatar/")
|
||||
candidates.append(f"icon/avatar/{suffix}")
|
||||
if source_path.startswith("SpriteOutput/AvatarShopIcon/Avatar/"):
|
||||
suffix = source_path.removeprefix("SpriteOutput/AvatarShopIcon/Avatar/")
|
||||
candidates.extend((f"icon/avatar/{suffix}", f"image/character_portrait/{suffix}"))
|
||||
if source_path.startswith("SpriteOutput/AvatarShopIcon/AvatarSkin/"):
|
||||
suffix = source_path.removeprefix("SpriteOutput/AvatarShopIcon/AvatarSkin/")
|
||||
candidates.extend((f"image/character_portrait/{suffix}", f"icon/item/Skin/{suffix}"))
|
||||
if source_path.startswith("SpriteOutput/ItemIcon/Skin/"):
|
||||
suffix = source_path.removeprefix("SpriteOutput/ItemIcon/Skin/")
|
||||
candidates.extend((f"icon/item/Skin/{suffix}", f"icon/item/{suffix}"))
|
||||
if source_path.startswith("SpriteOutput/ItemFigures/"):
|
||||
suffix = source_path.removeprefix("SpriteOutput/ItemFigures/")
|
||||
candidates.extend((f"image/item/{suffix}", f"image/items/{suffix}"))
|
||||
for candidate in candidates:
|
||||
if candidate and candidate in asset_files:
|
||||
return candidate
|
||||
return None
|
||||
|
||||
|
||||
def read_asset_indexes(revision: str, cache_dir: Path, asset_files: set[str]) -> dict[str, dict[str, dict]]:
|
||||
base = f"https://raw.githubusercontent.com/{ASSET_PROJECT}/{revision}/index_new"
|
||||
kinds = ("items", "light_cones", "relics")
|
||||
indexes: dict[str, dict[str, dict]] = {}
|
||||
jobs = {}
|
||||
with ThreadPoolExecutor(max_workers=12) as pool:
|
||||
for language, asset_language in ASSET_LANGUAGES.items():
|
||||
indexes[language] = {}
|
||||
for kind in kinds:
|
||||
url = f"{base}/{asset_language}/{kind}.json"
|
||||
key = f"starrailres_{asset_language}_{kind}"
|
||||
jobs[pool.submit(cached_json, url, cache_dir, key)] = (language, kind)
|
||||
for future in as_completed(jobs):
|
||||
language, kind = jobs[future]
|
||||
try:
|
||||
data = future.result()
|
||||
except RuntimeError:
|
||||
# Some locales omit a data family; localized text maps fill its names.
|
||||
data = {}
|
||||
if not isinstance(data, dict):
|
||||
data = {}
|
||||
indexes[language][kind] = data
|
||||
# The pinned repository tree is the authority for whether an icon URL exists.
|
||||
for language in indexes:
|
||||
for kind, table in indexes[language].items():
|
||||
for row in table.values():
|
||||
icon = row.get("icon") if isinstance(row, dict) else None
|
||||
if icon and icon not in asset_files:
|
||||
row.pop("icon", None)
|
||||
return indexes
|
||||
|
||||
|
||||
def textmap_files_by_language(paths: list[str]) -> dict[str, list[str]]:
|
||||
result = {}
|
||||
for language, tokens in TEXTMAP_TOKENS.items():
|
||||
matched = []
|
||||
for path in paths:
|
||||
name = Path(path).name.upper()
|
||||
if not name.startswith("TEXTMAP") or not name.endswith(".JSON"):
|
||||
continue
|
||||
token = name[len("TEXTMAP"):-len(".JSON")].split("_")[0]
|
||||
if token in tokens:
|
||||
matched.append(path)
|
||||
result[language] = sorted(matched)
|
||||
return result
|
||||
|
||||
|
||||
def localized_textmaps(paths: list[str], revision: str, cache_dir: Path, wanted_hashes: set[str]) -> dict[str, dict[str, str]]:
|
||||
maps: dict[str, dict[str, str]] = {language: {} for language in TEXTMAP_TOKENS}
|
||||
by_language = textmap_files_by_language(paths)
|
||||
for language, files in by_language.items():
|
||||
if not files or not wanted_hashes:
|
||||
continue
|
||||
print(f"Reading {language} text maps ({len(files)} file(s))", flush=True)
|
||||
for path in files:
|
||||
data = cached_json(gitlab_raw(GAME_PROJECT, revision, path), cache_dir, f"game_{path}")
|
||||
if isinstance(data, dict):
|
||||
for key in wanted_hashes:
|
||||
value = data.get(key)
|
||||
if isinstance(value, str) and value:
|
||||
maps[language][key] = value
|
||||
del data
|
||||
print(f" localized strings: {len(maps[language])}", flush=True)
|
||||
return maps
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("--inventory-config", required=True, type=Path, help="Go resource InventoryItemConfig.json")
|
||||
parser.add_argument("--output", type=Path, default=Path("assets/item-catalog.json"))
|
||||
parser.add_argument("--cache-dir", type=Path, default=Path(tempfile.gettempdir()) / "gac-srtool-item-catalog-cache")
|
||||
args = parser.parse_args()
|
||||
|
||||
inventory = json.loads(args.inventory_config.read_text(encoding="utf-8"))
|
||||
game_revision = inventory["source"]["revision"]
|
||||
supported_ids = {int(value) for value in inventory["item_ids"]}
|
||||
supported_lcs = {int(row["id"]) for row in inventory.get("lightcones", [])}
|
||||
supported_relics = {int(row["relic_id"]) for row in inventory.get("relics", [])}
|
||||
supported_ids.update(supported_lcs)
|
||||
supported_ids.update(supported_relics)
|
||||
asset_revision = "dbe8cdfcfb0bf657b9fe1cf92d5537f118ccb487"
|
||||
|
||||
print("Discovering game item tables...", flush=True)
|
||||
excel_files = discover_files(GAME_PROJECT, game_revision, "ExcelOutput")
|
||||
candidates = [path for path in excel_files if "ItemConfig" in Path(path).name and path.lower().endswith(".json")]
|
||||
tables = fetch_game_json_files(candidates, game_revision, args.cache_dir)
|
||||
item_rows: dict[int, list[tuple[str, dict]]] = {}
|
||||
table_scan = []
|
||||
for path in sorted(candidates):
|
||||
data = tables[path]
|
||||
valid_rows = [row for row in data if isinstance(row, dict) and isinstance(row.get("ID"), int) and row.get("ID") > 0 and isinstance(row.get("ItemMainType"), str)] if isinstance(data, list) else []
|
||||
table_scan.append({"table": path, "rows": len(data) if isinstance(data, list) else 0, "itemRows": len(valid_rows)})
|
||||
for row in valid_rows:
|
||||
item_rows.setdefault(row["ID"], []).append((path, row))
|
||||
definition_tables = [row["table"] for row in table_scan if row["itemRows"]]
|
||||
if not item_rows:
|
||||
raise RuntimeError("No item definition rows were found in ItemConfig tables")
|
||||
print(f"Accepted {len(definition_tables)} tables; {sum(row['itemRows'] for row in table_scan)} rows; {len(item_rows)} unique IDs", flush=True)
|
||||
|
||||
print("Reading pinned asset index...", flush=True)
|
||||
tree_url = f"https://api.github.com/repos/{ASSET_PROJECT}/git/trees/{asset_revision}?recursive=1"
|
||||
tree = request_json(tree_url)
|
||||
if tree.get("truncated"):
|
||||
raise RuntimeError("Pinned asset tree was truncated; refusing incomplete image coverage")
|
||||
asset_files = {entry["path"] for entry in tree.get("tree", []) if entry.get("type") == "blob"}
|
||||
indexes = read_asset_indexes(asset_revision, args.cache_dir, asset_files)
|
||||
|
||||
hashes = set()
|
||||
for rows in item_rows.values():
|
||||
for _, row in rows:
|
||||
for field in TEXT_FIELDS:
|
||||
value = item_hash(row, field)
|
||||
if value:
|
||||
hashes.add(value)
|
||||
print(f"Resolving {len(hashes)} game text keys across 13 locales...", flush=True)
|
||||
textmap_paths = discover_files(GAME_PROJECT, game_revision, "TextMap")
|
||||
textmaps = localized_textmaps(textmap_paths, game_revision, args.cache_dir, hashes)
|
||||
|
||||
existing_item_names = indexes.get("en", {}).get("items", {})
|
||||
existing_lc_names = indexes.get("en", {}).get("light_cones", {})
|
||||
existing_relic_names = indexes.get("en", {}).get("relics", {})
|
||||
merged = []
|
||||
name_coverage = {language: 0 for language in ASSET_LANGUAGES}
|
||||
description_coverage = {language: 0 for language in ASSET_LANGUAGES}
|
||||
background_description_coverage = {language: 0 for language in ASSET_LANGUAGES}
|
||||
for item_id in sorted(item_rows):
|
||||
source_rows = sorted(item_rows[item_id], key=lambda pair: rows_preference(pair[0]))
|
||||
all_rows = [row for _, row in source_rows]
|
||||
all_paths = []
|
||||
for row in all_rows:
|
||||
for path in row_image_paths(row):
|
||||
if path not in all_paths:
|
||||
all_paths.append(path)
|
||||
preferred = all_rows[0]
|
||||
if item_id in supported_lcs:
|
||||
category, index_kind = "lightcone", "light_cones"
|
||||
elif item_id in supported_relics:
|
||||
category, index_kind = "relic", "relics"
|
||||
else:
|
||||
category, index_kind = "item", "items"
|
||||
index_rows = indexes.get("en", {}).get(index_kind, {})
|
||||
index_row = index_rows.get(str(item_id), {}) if isinstance(index_rows, dict) else {}
|
||||
name_hash = next((item_hash(row, "ItemName") for _, row in source_rows if item_hash(row, "ItemName")), None)
|
||||
desc_hash = next((item_hash(row, "ItemDesc") for _, row in source_rows if item_hash(row, "ItemDesc")), None)
|
||||
bg_desc_hash = next((item_hash(row, "ItemBGDesc") for _, row in source_rows if item_hash(row, "ItemBGDesc")), None)
|
||||
names = {}
|
||||
descriptions = {}
|
||||
background_descriptions = {}
|
||||
for language in ASSET_LANGUAGES:
|
||||
locale_index = indexes.get(language, {}).get(index_kind, {})
|
||||
locale_row = locale_index.get(str(item_id), {}) if isinstance(locale_index, dict) else {}
|
||||
textmap = textmaps.get(language, {})
|
||||
name = textmap.get(name_hash) if name_hash else None
|
||||
if not name and isinstance(locale_row, dict):
|
||||
name = locale_row.get("name")
|
||||
if not name and isinstance(index_row, dict):
|
||||
name = index_row.get("name")
|
||||
if name:
|
||||
names[language] = name
|
||||
name_coverage[language] += 1
|
||||
desc = textmap.get(desc_hash) if desc_hash else None
|
||||
if not desc and isinstance(locale_row, dict):
|
||||
desc = locale_row.get("desc")
|
||||
if desc and language:
|
||||
descriptions[language] = desc
|
||||
description_coverage[language] += 1
|
||||
bg_desc = textmap.get(bg_desc_hash) if bg_desc_hash else None
|
||||
if bg_desc:
|
||||
background_descriptions[language] = bg_desc
|
||||
background_description_coverage[language] += 1
|
||||
|
||||
# Fill sparse locales with the English name only if the source has no localized text.
|
||||
english_name = names.get("en") or (index_row.get("name") if isinstance(index_row, dict) else None) or f"Item {item_id}"
|
||||
for language in ASSET_LANGUAGES:
|
||||
names.setdefault(language, english_name)
|
||||
icon = resolve_source_icon(all_paths, category, item_id, indexes, asset_files)
|
||||
rarity = parse_rarity(preferred.get("Rarity"), index_row)
|
||||
output_row = {
|
||||
"id": item_id,
|
||||
"category": category,
|
||||
"type": preferred.get("ItemMainType", ""),
|
||||
"subType": preferred.get("ItemSubType", ""),
|
||||
"rarity": rarity,
|
||||
"names": names,
|
||||
"descriptions": descriptions,
|
||||
"backgroundDescriptions": background_descriptions,
|
||||
"icon": icon,
|
||||
"gameIconPaths": all_paths,
|
||||
"sourceTables": sorted({path for path, _ in source_rows}),
|
||||
"supported": item_id in supported_ids,
|
||||
}
|
||||
pile_limit = next((row.get("PileLimit") for _, row in source_rows if isinstance(row.get("PileLimit"), int)), None)
|
||||
if pile_limit:
|
||||
output_row["pileLimit"] = pile_limit
|
||||
if category == "lightcone":
|
||||
output_row["path"] = index_row.get("path") if isinstance(index_row, dict) else None
|
||||
elif category == "relic":
|
||||
if isinstance(index_row, dict):
|
||||
output_row["setId"] = int(index_row["set_id"]) if str(index_row.get("set_id", "")).isdigit() else None
|
||||
output_row["slot"] = index_row.get("type")
|
||||
merged.append(output_row)
|
||||
|
||||
counts = {
|
||||
"total": len(merged),
|
||||
"definitionRows": sum(row["itemRows"] for row in table_scan),
|
||||
"uniqueIds": len(merged),
|
||||
"supported": sum(row["supported"] for row in merged),
|
||||
"withImages": sum(bool(row["icon"]) for row in merged),
|
||||
"withoutConfiguredImage": sum(not row["icon"] for row in merged),
|
||||
"withDescriptions": sum(bool(row["descriptions"] or row["backgroundDescriptions"]) for row in merged),
|
||||
"nameCoverageByLanguage": name_coverage,
|
||||
"descriptionCoverageByLanguage": description_coverage,
|
||||
"backgroundDescriptionCoverageByLanguage": background_description_coverage,
|
||||
"candidateTables": len(candidates),
|
||||
"definitionTables": len(definition_tables),
|
||||
}
|
||||
output = {
|
||||
"source": {
|
||||
"gameRepository": f"https://gitlab.com/{GAME_PROJECT}",
|
||||
"gameRevision": game_revision,
|
||||
"candidateTables": len(candidates),
|
||||
"gameTables": definition_tables,
|
||||
"tableScan": table_scan,
|
||||
"assetRepository": ASSET_PROJECT,
|
||||
"assetRevision": asset_revision,
|
||||
"assetBaseUrl": f"https://cdn.jsdelivr.net/gh/{ASSET_PROJECT}@{asset_revision}/",
|
||||
"assetFallbackBaseUrl": f"https://raw.githubusercontent.com/{ASSET_PROJECT}/{asset_revision}/",
|
||||
"languages": list(ASSET_LANGUAGES),
|
||||
"notes": "Includes every ItemConfig table with item-definition rows at the pinned game-data revision. Items absent from InventoryItemConfig are browse-only; the catalog does not add or summon items. Icons use only files present in the pinned public StarRailRes tree. gameIconPaths retain the source paths for items without a mapped public icon.",
|
||||
},
|
||||
"counts": counts,
|
||||
"items": merged,
|
||||
}
|
||||
args.output.parent.mkdir(parents=True, exist_ok=True)
|
||||
args.output.write_text(json.dumps(output, ensure_ascii=False, separators=(",", ":")) + "\n", encoding="utf-8")
|
||||
print(json.dumps(counts, ensure_ascii=False, indent=2), flush=True)
|
||||
print(f"Wrote {len(merged)} item rows to {args.output}", flush=True)
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
raise SystemExit(main())
|
||||
except Exception as error: # give the CLI one actionable line on network/schema failures
|
||||
print(f"item catalog build failed: {error}", file=sys.stderr)
|
||||
raise
|
||||
Reference in new issue
Block a user