"""Cross-check the catalog against the installed game (`ackit verify --catalog`). For every archive the catalog in use (shipped CSV or the user-cache scan) is compared with a fresh scan: missing a sound of the game that is not in the catalog extra a catalog row that is not in the game duplicate a target or a name used twice size/sha1 the catalog's original size or checksum differs from the game (game modified or other build) category the row's group/category is not what the classification rules give for its name character the verified speaker differs from data/catalog/characters.csv language the archive code does not match the WAD's language folder copies (BANK) stored copies or prefetch heads differ from the location index Run it on unmodified game files (`ackit status`), otherwise modified sounds show up as size/sha1 issues. """ import collections import hashlib import os import random from .. import banks, classify, subtitles from ..catalog import (Catalog, apply_lines, available_archives, bank_names, bank_sounds, build_lines, character_of, is_shipped, line_names, load_archive, load_banks, load_characters, mark_heads, read_csv, scan_archive, sfx_names, whole_recordings) from ..inject import read_target from ..paths import catalog_dir from ..wad import Wad FOLDER_LANG = {"ENGLISH(US)": "INT", "ITALIAN": "ITA", "FRENCH(FRANCE)": "FRA", "GERMAN": "DEU", "SPANISH(SPAIN)": "ESN"} class Report: def __init__(self): self.checked = collections.Counter() self.issues = [] # (archive, kind, key, detail) self.facts = [] # what was established about the game's relations (full check) def add(self, archive, kind, key, detail=""): self.issues.append((archive, kind, key, detail)) def counts(self): return collections.Counter((a, k) for a, k, _, _ in self.issues) @property def ok(self): return not self.issues def _expected_category(s, names): if s.archive == banks.ARCHIVE: return classify.classify(s.name, bank=s.bank) pack = None if s.is_pack_media: pack = names.get(int(s.target.split("/")[0], 16), s.target.split("/")[0]) return classify.classify(s.name, pack=pack) def _check_rows(rep, archive, rows, characters, names): seen_t, seen_n = set(), {} for s in rows: rep.checked[archive] += 1 if s.archive != archive: rep.add(archive, "language", s.key, "row says archive %s" % s.archive) if s.target in seen_t: rep.add(archive, "duplicate", s.key, "target") seen_t.add(s.target) n = (s.language, s.name.lower()) # BANK: names are unique per language if n in seen_n or (s.language and ("", n[1]) in seen_n): rep.add(archive, "duplicate", s.key, "name %s (also %s)" % (s.name, seen_n.get(n) or seen_n[("", n[1])])) seen_n[n] = s.key g, c = _expected_category(s, names) if (s.group, s.category) != (g, c): rep.add(archive, "category", s.key, "%s/%s, rules give %s/%s" % (s.group, s.category, g, c)) want = character_of(s, characters) if s.character != want: rep.add(archive, "character", s.key, "%r, curated list says %r" % (s.character, want)) def check_wad(rep, game, archive, rows, characters, lines=None): """Rows vs a fresh scan of the WAD. With `lines` (full check) the fresh sounds get the game's line identities and names, so names, aliases and lines are compared too.""" path = game.wad(archive) if archive != "SFX": folder = os.path.basename(os.path.dirname(path)).upper() if FOLDER_LANG.get(folder, archive) != archive: rep.add(archive, "language", archive, "WAD in folder %s" % folder) _check_rows(rep, archive, rows, characters, bank_names(game.cooked)) fresh_list = scan_archive(archive, path, characters=characters) if lines is not None and archive in lines.langs: apply_lines(fresh_list, lines, lines.names) fresh = {s.target: s for s in fresh_list} if lines is None and archive in game.languages: # quick check: line names need the banks, not rescanned byk = {s.target: s for s in rows} for t, s in fresh.items(): if t in byk: s.name, s.aliases, s.line = byk[t].name, byk[t].aliases, byk[t].line byk = {s.target: s for s in rows} for t in sorted(set(fresh) & set(byk)): f, s = fresh[t], byk[t] if (f.name, tuple(f.aliases), f.line) != (s.name, tuple(s.aliases), s.line): rep.add(archive, "name", s.key, "catalog %s %s line %s, game %s %s line %s" % (s.name, list(s.aliases), s.line or "-", f.name, list(f.aliases), f.line or "-")) cat = {s.target: s for s in rows} for t in sorted(set(fresh) - set(cat)): rep.add(archive, "missing", "%s:%s" % (archive, t), fresh[t].name) for t in sorted(set(cat) - set(fresh)): rep.add(archive, "extra", "%s:%s" % (archive, t), cat[t].name) wad = Wad(path) for t in sorted(set(cat) & set(fresh)): s = cat[t] if s.size and s.size != fresh[t].size and not s.is_pack_media: rep.add(archive, "size", s.key, "catalog %d, game %d" % (s.size, fresh[t].size)) if s.sha1: h = hashlib.sha1(read_target(path, t, wad)).hexdigest()[:16] if h != s.sha1: rep.add(archive, "sha1", s.key, "game file differs from the catalog's original") def check_banks(rep, game, rows, locations, characters, quick=False, sample=40, progress=None): """BANK rows, copies, prefetch heads and subtitles vs a scan of the packages. The full check (not quick) also derives the game's lines from that scan and returns them (else None).""" _check_rows(rep, banks.ARCHIVE, rows, characters, bank_names(game.cooked)) files = banks.bank_files(game.cooked) if quick: pk = [f for f in files if f.lower().endswith(".upk")] rnd = random.Random(1) files = [f for f in files if f.startswith("SFX/")] + sorted(rnd.sample(pk, min(sample, len(pk)))) col = subtitles.Collector() sc = banks.scan(game.cooked, progress, files=files, visit=col) known = {n.split("#")[0] for s in rows for n in s.names()} # BANK names for a in game.languages: # + speech line names of the catalogs p = os.path.join(catalog_dir(), "%s.csv" % a) if os.path.exists(p): known.update(n.split("#")[0] for s in read_csv(p) for n in s.names()) col.resolve({n for n in sc.names | known if n.lower().startswith("speech_")}) check_subtitles(rep, game, col, set(files), quick) lines = None index = collections.defaultdict(set) for m, ls in locations.items(): for l in ls: if l.file in set(files): index[(l.media, l.kind)].add((l.file, l.offset, l.size, l.sound if l.kind == banks.PREFETCH_HEAD else "")) if not quick: sc.names.update(col.dialogues) # as the build does: subtitle objects name speech events too lines = build_lines(game, sc) mark_heads(sc, whole_recordings(sc, lines)) # WAD recordings also stored whole in banks else: # a sample cannot derive them: the index says which they are mark_heads(sc, {m for m, k in index if k == banks.PREFETCH_HEAD} - sc.prefetch) found = collections.defaultdict(set) for l in sc.locations: head = l.kind == banks.PREFETCH_HEAD found[(l.media, l.kind)].add((l.file, l.offset, l.size, lines.head_key(l) if head and lines else "")) if lines is None: # quick: a subset of the packages cannot resolve the heads' lines index = {k: {(f, o, s, "") for f, o, s, _ in v} for k, v in index.items()} for k in sorted(set(found) | set(index)): if found.get(k, set()) != index.get(k, set()): rep.add(banks.ARCHIVE, "copies", "%08X" % k[0], "%s: index %d, game %d" % ("head" if k[1] == banks.PREFETCH_HEAD else "copies", len(index.get(k, ())), len(found.get(k, ())))) cat = {s.target: s for s in rows} fresh_list = bank_sounds(sc, characters, sfx_names(game)) if lines is not None: lines.names = line_names(game, sc, lines, {s.key: s.name for s in fresh_list}) apply_lines(fresh_list, lines, lines.names) for s in fresh_list: s.character = character_of(s, characters) fresh = {s.target: s for s in fresh_list} for t in sorted(set(fresh) - set(cat)): rep.add(banks.ARCHIVE, "missing", "BANK:" + t, fresh[t].bank) if not quick: for t in sorted(set(cat) - set(fresh)): rep.add(banks.ARCHIVE, "extra", "BANK:" + t, cat[t].name) for t in sorted(set(cat) & set(fresh)): s, f = cat[t], fresh[t] if s.sha1 and f.sha1 and s.sha1 != f.sha1: rep.add(banks.ARCHIVE, "sha1", s.key, "game file differs from the catalog's original") if f.size < s.size: rep.add(banks.ARCHIVE, "size", s.key, "catalog %d, game %d" % (s.size, f.size)) if lines is not None and (f.name, f.language, f.line) != (s.name, s.language, s.line): rep.add(banks.ARCHIVE, "name", s.key, "catalog %s [%s] line %s, game %s [%s] line %s" % (s.name, s.language, s.line or "-", f.name, f.language, f.line or "-")) for m, ls in locations.items(): for l in ls: if l.kind == banks.PREFETCH_HEAD and not l.sound: rep.add(banks.ARCHIVE, "copies", "%08X" % m, "prefetch head without a speech line") return sc, lines, col def check_subtitles(rep, game, col, files, quick): """Subtitle copies found in the scanned packages vs SUB_locations + the SUB_.csv texts.""" from ..catalog import _shipped_index, catalog_dir loc_path = os.path.join(catalog_dir(), subtitles.LOCATIONS) if not os.path.exists(loc_path) or not is_shipped(game, banks.ARCHIVE): return index = collections.defaultdict(set) for k, ls in subtitles.read_locations(loc_path).items(): for l in ls: if l.file in files: index[k].add((l.file, l.pos, l.size)) found, texts = collections.defaultdict(set), {} for speech, copies in col.dialogues.items(): for rel, d in copies: for f in d.fields: k = subtitles.key(f.lang, speech, f.index) found[k].add((rel, f.pos, f.size)) texts[k] = f.text for k in sorted(set(found) | set(index)): if found.get(k, set()) != index.get(k, set()): rep.add("SUB", "copies", k, "index %d, game %d" % (len(index.get(k, ())), len(found.get(k, ())))) langs = sorted(_shipped_index().get("SUB", {}).get("lines", {})) bank_lang = {s.key: s.language for s in read_csv(os.path.join(catalog_dir(), "BANK.csv"))} for lang in langs: voice = lang if lang in game.languages else "INT" # languages without own voices: INT voices p = os.path.join(catalog_dir(), "SUB_%s.csv" % lang) rows = subtitles.read_texts(p, lang) rep.checked["SUB_" + lang] += len(rows) seen = set() for s in rows: k = subtitles.key(lang, s.speech, s.line) if k in seen: rep.add("SUB", "duplicate", k) seen.add(k) if k in texts and texts[k] != s.text: rep.add("SUB", "text", k, "catalog %r, game %r" % (s.text[:40], texts[k][:40])) for snd in s.sounds: if snd.startswith(banks.ARCHIVE + ":"): b = bank_lang.get(snd) if b and voice not in b.split(";"): # '' = stored for every language rep.add("SUB", "language", k, "linked to %s (%s)" % (snd, b)) elif not snd.startswith(voice + ":"): rep.add("SUB", "language", k, "linked to %s, not a %s voice line" % (snd, voice)) def check_lines(rep, game, sc, lines, locations): """The game's relations, re-derived from the game files (ackit/lines.py) and compared with the catalogs: links a broken one-to-one relation in the game data (a WEM without its table row, a Sound object with two recordings in one language, a line without a recording in some language, ...) line a catalog row whose line identity / name differs from the game's subtitle a subtitle linked to a different line in one language than in INT Facts (rep.facts) give the counts, prefetch heads compared byte by byte with the start of their WAD recording, and the length agreement of a line's recordings across languages (independent evidence).""" for kind, detail in lines.issues: rep.add("LINES", "links", kind, detail) st = lines.stats for lid, miss in sorted(st["lines_missing_a_language"].items()): rep.facts.append("line %s (%s) has no recording in %s" % (lid, lines.names.get(lid, "?"), ", ".join(miss))) for k in st["without_line"]: rep.add("LINES", "links", k, "a recording of no line") # independent of the BKPK tables: each WAD recording is the source of a Sound object in a bank of the # language the tables give it (localized banks: one per declared language) named = total = 0 for L in lines.langs: for m, k in lines.owners.get(L, {}).items(): if not k.startswith(banks.ARCHIVE + ":"): total += 1 if m in sc.lang_media.get(L, ()): named += 1 else: rep.add("LINES", "links", k, "media %08X is not a source in any %s bank" % (m, L)) rep.facts.append("WAD recordings named as a source by a Sound object of their own language's bank: %d of %d" % (named, total)) rep.facts.append("voice languages %s: %d recordings (WAD WEMs %s, %d of them second places of a recording " "stored twice, + localized in-memory), %d lines, %d of them with exactly one recording in every " "language" % ("/".join(lines.langs), st["recordings"], "/".join(str(st["wad_wems"].get(L, 0)) for L in lines.langs), len(st["copies"]), st["lines"], st["lines_in_every_language"])) # catalog rows vs the game's lines rows = {} for L in lines.langs + [banks.ARCHIVE]: p = os.path.join(catalog_dir(), "%s.csv" % L) if os.path.exists(p): for s in read_csv(p): rows[s.key] = s for lid, per in lines.members.items(): for L, k in per.items(): s = rows.get(k) if s is None: rep.add("LINES", "line", k, "recording of line %s is not in the catalog" % lid) elif s.line != lid or s.name != lines.names[lid]: rep.add("LINES", "line", k, "catalog %s/%s, game %s/%s" % (s.line, s.name, lid, lines.names[lid])) for k, s in rows.items(): if s.line and lines.line_of.get(k) != s.line: rep.add("LINES", "line", k, "catalog says line %s, the game data %s" % (s.line, lines.line_of.get(k))) # prefetch heads: every stored head is the start of the WAD recording it is mapped to wads = {L: (game.wad(L), Wad(game.wad(L))) for L in lines.langs} heads = [l for ls in locations.values() for l in ls if l.kind == banks.PREFETCH_HEAD] by_file = collections.defaultdict(list) for l in heads: by_file[l.file].append(l) same = 0 for rel, ls in by_file.items(): r = banks.open_file(banks.file_path(game.cooked, rel)) for l in ls: L = l.sound.split(":")[0] if L not in wads: rep.add("LINES", "head", "%08X" % l.media, "head in %s maps to %r" % (rel, l.sound)) continue w = read_target(wads[L][0], l.sound.split(":", 1)[1], wads[L][1]) if r.read(l.offset, l.size) == (w + bytes(l.size))[:l.size]: same += 1 else: rep.add("LINES", "head", "%08X" % l.media, "head in %s is not the start of %s" % (rel, l.sound)) same_copies = 0 for k, primary in sorted(lines.copy_of.items()): # one recording stored in two WAD entries L = k.split(":")[0] a = read_target(wads[L][0], k.split(":", 1)[1], wads[L][1]) b = read_target(wads[L][0], primary.split(":", 1)[1], wads[L][1]) if a == b: same_copies += 1 else: rep.add("LINES", "copies", k, "differs from %s (same media ID)" % primary) rep.facts.append("recordings stored twice in a WAD (same media ID, written together): %d, byte-identical: %d" % (len(lines.copy_of), same_copies)) rep.facts.append("prefetch heads and whole bank copies: %d of %d are byte-identical to the start of their WAD " "recording" % (same, len(heads))) # independent evidence: the recordings of a line have similar lengths in every language ratios = [] for lid, per in lines.members.items(): secs = [rows[k].seconds for k in per.values() if k in rows and rows[k].seconds > 0] if len(secs) > 1: ratios.append((max(secs) / min(secs), lid)) if ratios: ratios.sort() within = sum(1 for r, _ in ratios if r <= 1.5) rep.facts.append("lengths across languages: %d of %d lines within a factor 1.5, median %.2f, longest ratio " "%.1f (%s)" % (within, len(ratios), ratios[len(ratios) // 2][0], ratios[-1][0], lines.names.get(ratios[-1][1], ""))) return rows def check_subtitle_lines(rep, rows): """A subtitle line (speech event + line index) must point to the same line in every language.""" from ..catalog import _shipped_index langs = sorted(_shipped_index().get("SUB", {}).get("lines", {})) ref = {} p = os.path.join(catalog_dir(), "SUB_INT.csv") if os.path.exists(p): for s in subtitles.read_texts(p, "INT"): if s.sounds and all(k in rows for k in s.sounds): ref[(s.speech, s.line)] = [rows[k].line for k in s.sounds] linked = total = 0 for lang in langs: for s in subtitles.read_texts(os.path.join(catalog_dir(), "SUB_%s.csv" % lang), lang): total += 1 if not s.sound: continue linked += 1 unknown = [k for k in s.sounds if k not in rows] want = ref.get((s.speech, s.line)) if unknown: rep.add("SUB", "link", subtitles.key(lang, s.speech, s.line), "linked to unknown %s" % unknown) elif want and [rows[k].line for k in s.sounds] != want: rep.add("SUB", "link", subtitles.key(lang, s.speech, s.line), "lines %s, INT subtitle is on lines %s" % ([rows[k].line for k in s.sounds], want)) rep.facts.append("subtitles: %d strings in %d languages, %d linked to a voice line, each to the same line as in INT" % (total, len(langs), linked)) SAME_PERSON = {("mr freeze", "mister freeze"), ("batman", "bruce wayne"), ("two face", "twoface"), ("azrael", "mysterious stranger")} def _person(s): return " ".join(s.lower().replace("'", "").replace(".", "").replace("_", " ").replace("-", " ").split()) def check_speakers(rep, rows, col): """Independent evidence for the curated speaker labels: the dialogue object of a subtitled line names its speaker. Facts only (the lists are curated by listening; differences are for a person to judge).""" speaker = {} for speech, copies in col.dialogues.items(): for _, d in copies: if d.speaker: speaker.setdefault(speech, d.speaker) p = os.path.join(catalog_dir(), "SUB_INT.csv") if not os.path.exists(p): return agree, unknown, differ = 0, 0, [] for s in subtitles.read_texts(p, "INT"): spk = speaker.get(s.speech, "") if s.line != 0 or not spk: continue for k in s.sounds: r = rows.get(k) if not r or not r.character: continue who = [_person(c) for c in r.character.split("; ")] g = _person(spk) if any(g in w or w in g or (w, g) in SAME_PERSON for w in who): agree += 1 elif who == ["unknown"]: unknown += 1 else: differ.append("%s (list: %s, game: %s)" % (r.name, r.character, spk)) rep.facts.append("speakers: %d labelled voice lines agree with the speaker field of their dialogue object, " "%d listed as Unknown are named there, %d differ%s" % (agree, unknown, len(differ), (": " + "; ".join(differ)) if differ else "")) def check_menu(rep, game): """Menu texts: the same keys in every language (facts; differences are the game's own).""" from .. import loctext root = os.path.join(game.root, "BmGame", "Localization") if not os.path.isdir(root): return keys = {} for lang in loctext.languages(root): try: keys[lang] = {(os.path.splitext(e.file)[0].upper(), e.section, e.name, e.occurrence) for e in loctext.load(root, lang)} except Exception as e: # unreadable folder: report, do not stop rep.add("MENU", "read", lang, str(e)) if "INT" not in keys: return for lang, ks in sorted(keys.items()): extra, miss = ks - keys["INT"], keys["INT"] - ks rep.facts.append("menu text %s: %d keys%s" % (lang, len(ks), "" if not (extra or miss) else " (%d not in INT, %d of INT missing)" % (len(extra), len(miss)))) def run(game, archives=None, quick=False, log=print, progress=None): """Check the given archives (default: all). Returns a Report.""" rep = Report() archives = archives or available_archives(game) cat = Catalog(game, archives, progress) by_arch = collections.defaultdict(list) for s in cat.sounds: by_arch[s.archive].append(s) lines = None if banks.ARCHIVE in archives: # first: the full check derives the game's lines from this scan log("checking %s ..." % banks.ARCHIVE) rows = load_banks(game, progress)[0] # every language, not only the loaded one sc, lines, col = check_banks(rep, game, rows, cat.locations, load_characters(lang=None), quick=quick, progress=progress) for a in archives: if a == banks.ARCHIVE: continue log("checking %s ..." % a) rows_a = by_arch[a] if a in cat.archives else load_archive(game, a, progress) check_wad(rep, game, a, rows_a, load_characters(lang=a), lines) if lines is not None and is_shipped(game, banks.ARCHIVE): log("checking the relations ...") rows_all = check_lines(rep, game, sc, lines, cat.locations) check_subtitle_lines(rep, rows_all) check_speakers(rep, rows_all, col) check_menu(rep, game) characters = load_characters(lang=game.language()) known = {n.lower() for s in cat.sounds for n in s.names()} # speaker names are the same in every language; only meaningful for the build the lists were made for if any(a in game.languages and is_shipped(game, a) for a in archives): for n in sorted(characters): if n not in known: rep.add("characters.csv", "extra", n, "no sound with this name") return rep