283 lines
9.7 KiB
Python
283 lines
9.7 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""把 SDGB 官方更新包（A0xx）里的曲目 / 称号合并进 maipulse 的 master 数据。
|
||
|
||
用法：
|
||
python tools/apply_update.py "C:\\Users\\20558\\Downloads\\SDGB_A041_20261020134418_0"
|
||
python tools/apply_update.py <包目录> --dry 只看差异，不写盘
|
||
|
||
包里有什么就用什么：
|
||
music/music*/Music.xml -> data/master/music.json（定数、bpm、版本、谱师全在这）
|
||
title/title*/Title.xml -> data/master/title.json
|
||
其它目录（event / map / information / MovieData / SoundData / AssetBundleImages）是机台资源，
|
||
maipulse 用不到，直接跳过。
|
||
|
||
★ 只增量合并：已存在的 id 用包里更完整的字段覆盖空值，不删任何旧条目。
|
||
★ 写盘前自动备份到 data/_backup/<时间戳>/。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import io
|
||
import json
|
||
import os
|
||
import re
|
||
import shutil
|
||
import sys
|
||
import time
|
||
import xml.etree.ElementTree as ET
|
||
|
||
BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
DATA_DIR = os.path.join(BASE_DIR, "data")
|
||
MASTER_DIR = os.path.join(DATA_DIR, "master")
|
||
MASTER_INDEX = os.path.join(DATA_DIR, "master.json")
|
||
BACKUP_ROOT = os.path.join(DATA_DIR, "_backup")
|
||
|
||
|
||
def txt(node, path, default=""):
|
||
e = node.find(path)
|
||
if e is None:
|
||
return default
|
||
return (e.text or default) or default
|
||
|
||
|
||
def num(node, path, default=0):
|
||
e = node.find(path)
|
||
if e is None:
|
||
return default
|
||
try:
|
||
return int((e.text or "").strip())
|
||
except (ValueError, AttributeError):
|
||
return default
|
||
|
||
|
||
def flag(node, path, default=False):
|
||
return (txt(node, path, "true" if default else "false") or "").strip().lower() == "true"
|
||
|
||
|
||
# ---------------------------------------------------------------- 解析
|
||
|
||
def parse_music(path: str) -> dict:
|
||
root = ET.parse(path).getroot()
|
||
notes = []
|
||
for n in root.findall("./notesData/Notes"):
|
||
if not flag(n, "./isEnable", True):
|
||
notes.append({"level": 0, "levelDecimal": 0, "designer": ""})
|
||
continue
|
||
notes.append({
|
||
"level": num(n, "./level"),
|
||
"levelDecimal": num(n, "./levelDecimal"),
|
||
"designer": txt(n, "./notesDesigner/str"),
|
||
})
|
||
# 谱面固定补到 6 档（BAS/ADV/EXP/MAS/Re:MAS + 宴），和现有 master 对齐
|
||
while len(notes) < 6:
|
||
notes.append({"level": 0, "levelDecimal": 0, "designer": ""})
|
||
return {
|
||
"id": num(root, "./name/id"),
|
||
"name": txt(root, "./name/str"),
|
||
"sortName": txt(root, "./sortName"),
|
||
"artist": txt(root, "./artistName/str"),
|
||
"genre": txt(root, "./genreName/str"),
|
||
"bpm": num(root, "./bpm"),
|
||
"version": num(root, "./version"),
|
||
"addVersion": txt(root, "./AddVersion/str"),
|
||
"lockType": num(root, "./lockType"),
|
||
"disable": flag(root, "./disable"),
|
||
"notes": notes,
|
||
}
|
||
|
||
|
||
def parse_title(path: str) -> dict:
|
||
root = ET.parse(path).getroot()
|
||
return {
|
||
"id": num(root, "./name/id"),
|
||
"name": txt(root, "./name/str"),
|
||
# Title.xml 里叫 <genre>，Music.xml 里叫 <genreName>，两个都兼容
|
||
"genre": txt(root, "./genre/str") or txt(root, "./genreName/str"),
|
||
"release": txt(root, "./releaseTagName/str"),
|
||
"lock": txt(root, "./dispCond"),
|
||
"norm": txt(root, "./normText"),
|
||
"isDefault": flag(root, "./isDefault"),
|
||
"disable": flag(root, "./disable"),
|
||
}
|
||
|
||
|
||
def pack_of(pkg: str) -> str:
|
||
"""从包目录名取出包号：SDGB_A041_20261020134418_0 -> A041。
|
||
|
||
目录名里没有 A0xx 时回退取全部数字段里第一个 3 位纯数字，
|
||
再不行返回空串（这条就归到「无归属」）。
|
||
"""
|
||
for part in re.split(r"[\\/_\-\s]+", os.path.basename(os.path.normpath(pkg))):
|
||
if re.fullmatch(r"A\d{3}", part):
|
||
return part
|
||
return ""
|
||
|
||
|
||
def collect(pkg: str, sub: str, filename: str, parser, pack: str = "") -> list:
|
||
out = []
|
||
base = os.path.join(pkg, sub)
|
||
if not os.path.isdir(base):
|
||
return out
|
||
for name in sorted(os.listdir(base)):
|
||
p = os.path.join(base, name, filename)
|
||
if os.path.isfile(p):
|
||
try:
|
||
item = parser(p)
|
||
except ET.ParseError as exc:
|
||
print(" 跳过 %s：%s" % (p, exc))
|
||
continue
|
||
if item.get("id"):
|
||
if pack:
|
||
item["pack"] = pack
|
||
out.append(item)
|
||
return out
|
||
|
||
|
||
# ---------------------------------------------------------------- 合并
|
||
|
||
def merge(old: list, new: list):
|
||
"""按 id 增量合并。返回 (合并后列表, 新增, 更新)。"""
|
||
index = {x["id"]: i for i, x in enumerate(old)}
|
||
added, updated = [], []
|
||
for item in new:
|
||
i = index.get(item["id"])
|
||
if i is None:
|
||
old.append(item)
|
||
index[item["id"]] = len(old) - 1
|
||
added.append(item)
|
||
continue
|
||
cur = old[i]
|
||
dirty = False
|
||
for k, v in item.items():
|
||
if k == "pack":
|
||
# 包归属记首次出现的那个包：已有值不覆盖，
|
||
# 空值（早先水鱼补的曲）才由包补上
|
||
if v and not cur.get("pack"):
|
||
cur["pack"] = v
|
||
dirty = True
|
||
continue
|
||
if k == "notes":
|
||
# 谱面：包里启用的档位比旧的更权威
|
||
if any(n["level"] or n["levelDecimal"] for n in v):
|
||
if v != cur.get("notes"):
|
||
cur["notes"] = v
|
||
dirty = True
|
||
continue
|
||
if v in ("", 0, False, None) and cur.get(k) not in ("", 0, False, None):
|
||
continue # 包里是空值就不覆盖已有值
|
||
if cur.get(k) != v:
|
||
cur[k] = v
|
||
dirty = True
|
||
if dirty:
|
||
updated.append(item["id"])
|
||
old.sort(key=lambda x: x["id"])
|
||
return old, added, updated
|
||
|
||
|
||
def backup(paths: list) -> str:
|
||
stamp = time.strftime("%Y%m%d-%H%M%S")
|
||
dst = os.path.join(BACKUP_ROOT, stamp)
|
||
os.makedirs(dst, exist_ok=True)
|
||
for p in paths:
|
||
if os.path.isfile(p):
|
||
shutil.copy2(p, os.path.join(dst, os.path.basename(p)))
|
||
return dst
|
||
|
||
|
||
def write_json(path: str, obj):
|
||
with io.open(path, "w", encoding="utf-8", newline="\n") as fp:
|
||
json.dump(obj, fp, ensure_ascii=False, separators=(",", ":"))
|
||
|
||
|
||
def rebuild_index():
|
||
"""master.json 是总索引：{kind: {kind, count, items}}。"""
|
||
index = {}
|
||
if os.path.isfile(MASTER_INDEX):
|
||
try:
|
||
index = json.load(io.open(MASTER_INDEX, encoding="utf-8"))
|
||
except (OSError, ValueError):
|
||
index = {}
|
||
for name in os.listdir(MASTER_DIR):
|
||
if not name.endswith(".json"):
|
||
continue
|
||
kind = name[:-5]
|
||
p = os.path.join(MASTER_DIR, name)
|
||
try:
|
||
items = json.load(io.open(p, encoding="utf-8"))
|
||
except (OSError, ValueError):
|
||
continue
|
||
index[kind] = {"kind": kind, "count": len(items), "items": items}
|
||
write_json(MASTER_INDEX, index)
|
||
return {k: v["count"] for k, v in index.items()}
|
||
|
||
|
||
def main():
|
||
args = [a for a in sys.argv[1:] if not a.startswith("--")]
|
||
if not args:
|
||
print(__doc__)
|
||
return 1
|
||
pkg = os.path.abspath(args[0])
|
||
dry = "--dry" in sys.argv
|
||
if not os.path.isdir(pkg):
|
||
print("包目录不存在：%s" % pkg)
|
||
return 1
|
||
|
||
print("包：%s" % pkg)
|
||
ver = os.path.join(pkg, "DataConfig.xml")
|
||
if os.path.isfile(ver):
|
||
try:
|
||
r = ET.parse(ver).getroot()
|
||
print("版本：%s.%s.%s" % (num(r, "./version/major"), num(r, "./version/minor"),
|
||
num(r, "./version/release")))
|
||
except ET.ParseError:
|
||
pass
|
||
|
||
music_path = os.path.join(MASTER_DIR, "music.json")
|
||
title_path = os.path.join(MASTER_DIR, "title.json")
|
||
music_old = json.load(io.open(music_path, encoding="utf-8"))
|
||
title_old = json.load(io.open(title_path, encoding="utf-8"))
|
||
|
||
pkg_pack = pack_of(pkg)
|
||
pkg_music = collect(pkg, "music", "Music.xml", parse_music, pkg_pack)
|
||
pkg_title = collect(pkg, "title", "Title.xml", parse_title, pkg_pack)
|
||
print("包号：%s%s" % (pkg_pack or "(未识别)",
|
||
"" if pkg_pack else "，曲目会归到「无归属」"))
|
||
print("包里：曲目 %d 首 / 称号 %d 个" % (len(pkg_music), len(pkg_title)))
|
||
|
||
music_new, m_add, m_upd = merge(music_old, pkg_music)
|
||
title_new, t_add, t_upd = merge(title_old, pkg_title)
|
||
|
||
print("\n曲目新增 %d 首：" % len(m_add))
|
||
for it in m_add:
|
||
lv = "/".join("%d%s" % (n["level"], ("." + str(n["levelDecimal"])) if n["levelDecimal"] else "")
|
||
for n in it["notes"][:4])
|
||
print(" %-7d %-30s v%-6d %s" % (it["id"], it["name"][:28], it["version"], lv))
|
||
if m_upd:
|
||
print("曲目更新 %d 首：%s" % (len(m_upd), ", ".join(map(str, m_upd))))
|
||
|
||
print("\n称号新增 %d 个：" % len(t_add))
|
||
for it in t_add:
|
||
print(" %-7d %s" % (it["id"], it["name"]))
|
||
if t_upd:
|
||
print("称号更新 %d 个：%s" % (len(t_upd), ", ".join(map(str, t_upd))))
|
||
|
||
print("\n总计：曲目 %d -> %d，称号 %d -> %d"
|
||
% (len(music_old) - len(m_add), len(music_new),
|
||
len(title_old) - len(t_add), len(title_new)))
|
||
|
||
if dry:
|
||
print("\n--dry：没有写盘")
|
||
return 0
|
||
|
||
dst = backup([music_path, title_path, MASTER_INDEX])
|
||
print("\n已备份到 %s" % dst)
|
||
write_json(music_path, music_new)
|
||
write_json(title_path, title_new)
|
||
counts = rebuild_index()
|
||
print("已写入。master.json 计数：")
|
||
for k, v in sorted(counts.items()):
|
||
print(" %-8s %d" % (k, v))
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main()) |