148 lines
4.5 KiB
Python
148 lines
4.5 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""backfill_pack.py —— 给曲目回填「差分包归属」字段。
|
||
|
||
扫描 StreamingAssets 下的 A000 / A005 / A011 ... 各包的 music 目录，
|
||
得到「曲目 id -> 首次出现的包号」，写进：
|
||
|
||
data/master/music.json 每条记录加 pack 字段
|
||
data/master.json music.items 同步（/api/master 读的是这个）
|
||
|
||
归属规则：A000 之外的包优先（大包号覆盖小包号），剩下的归 A000，
|
||
哪个包目录都没有的记空串。
|
||
|
||
用法：
|
||
python tools/backfill_pack.py
|
||
python tools/backfill_pack.py --src "D:\\bs\\Package\\Sinmai_Data\\StreamingAssets" --dry
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import io
|
||
import json
|
||
import os
|
||
import re
|
||
import sys
|
||
|
||
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")
|
||
|
||
DEFAULT_SRC = r"D:\bs\Package\Sinmai_Data\StreamingAssets"
|
||
|
||
|
||
def read_json(path):
|
||
with io.open(path, encoding="utf-8") as fp:
|
||
return json.load(fp)
|
||
|
||
|
||
def write_json(path, obj):
|
||
"""无 BOM UTF-8，缩进 1 空格保持体积。"""
|
||
tmp = path + ".tmp"
|
||
with io.open(tmp, "w", encoding="utf-8", newline="\n") as fp:
|
||
json.dump(obj, fp, ensure_ascii=False, separators=(",", ":"))
|
||
os.replace(tmp, path)
|
||
|
||
|
||
def scan_packs(src):
|
||
"""返回 (pack 列表, {pack: set(id)})，A000 排第一，其余按包号升序。"""
|
||
if not os.path.isdir(src):
|
||
raise SystemExit("找不到 StreamingAssets 目录：" + src)
|
||
packs = sorted([d for d in os.listdir(src) if re.fullmatch(r"A\d{3}", d)],
|
||
key=lambda x: (x != "A000", x))
|
||
out = {}
|
||
for p in packs:
|
||
d = os.path.join(src, p, "music")
|
||
if not os.path.isdir(d):
|
||
continue
|
||
ids = set()
|
||
for name in os.listdir(d):
|
||
if not os.path.isdir(os.path.join(d, name)):
|
||
continue
|
||
m = re.search(r"(\d+)$", name)
|
||
if m:
|
||
ids.add(int(m.group(1)))
|
||
if ids:
|
||
out[p] = ids
|
||
return packs, out
|
||
|
||
|
||
def owner_map(packs, pack_ids):
|
||
"""id -> 包号。A000 之外的包按顺序覆盖，没有目录的留空。"""
|
||
owner = {}
|
||
for p in packs:
|
||
if p == "A000":
|
||
continue
|
||
for i in pack_ids.get(p, ()):
|
||
owner[i] = p
|
||
for i in pack_ids.get("A000", ()):
|
||
owner.setdefault(i, "A000")
|
||
return owner
|
||
|
||
|
||
def main(argv=None):
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("--src", default=DEFAULT_SRC)
|
||
ap.add_argument("--dry", action="store_true", help="只打印统计，不写盘")
|
||
args = ap.parse_args(argv)
|
||
|
||
packs, pack_ids = scan_packs(args.src)
|
||
print("扫到包:", ", ".join(packs))
|
||
for p in packs:
|
||
if p in pack_ids:
|
||
print(" %-6s %5d 首" % (p, len(pack_ids[p])))
|
||
|
||
owner = owner_map(packs, pack_ids)
|
||
|
||
# ---- master/music.json
|
||
mpath = os.path.join(MASTER_DIR, "music.json")
|
||
music = read_json(mpath)
|
||
from collections import Counter
|
||
stat = Counter()
|
||
changed = 0
|
||
for row in music:
|
||
p = owner.get(int(row["id"]), "")
|
||
if row.get("pack") != p:
|
||
changed += 1
|
||
row["pack"] = p
|
||
stat[p or "(无归属)"] += 1
|
||
|
||
order = [p for p in packs if p in pack_ids] + [""]
|
||
print("\n=== 回填结果：%d 首，变更 %d 条 ===" % (len(music), changed))
|
||
for p in order:
|
||
key = p or "(无归属)"
|
||
if stat.get(key):
|
||
print(" %-8s %5d" % (key, stat[key]))
|
||
total = sum(stat.values())
|
||
print(" 合计 %5d" % total)
|
||
|
||
if args.dry:
|
||
print("\n--dry，未写盘")
|
||
return 0
|
||
|
||
write_json(mpath, music)
|
||
print("\n已写", mpath, os.path.getsize(mpath), "字节")
|
||
|
||
# ---- master.json 里的 music 镜像
|
||
idx = read_json(MASTER_INDEX)
|
||
node = idx.get("music")
|
||
if isinstance(node, dict) and isinstance(node.get("items"), list):
|
||
pmap = {int(r["id"]): r.get("pack", "") for r in music}
|
||
n = 0
|
||
for row in node["items"]:
|
||
p = pmap.get(int(row["id"]), "")
|
||
if row.get("pack") != p:
|
||
n += 1
|
||
row["pack"] = p
|
||
node["count"] = len(node["items"])
|
||
write_json(MASTER_INDEX, idx)
|
||
print("已写 %s %d 字节（music %d 条，变更 %d）"
|
||
% (MASTER_INDEX, os.path.getsize(MASTER_INDEX), len(node["items"]), n))
|
||
else:
|
||
print("master.json 里没有 music.items，跳过")
|
||
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main()) |