diff --git a/dlphelper/sorter.py b/dlphelper/sorter.py index 681536f..0931d8a 100644 --- a/dlphelper/sorter.py +++ b/dlphelper/sorter.py @@ -35,6 +35,12 @@ UNCLASSIFIED_CLASS = "Unclassified" # into other games' sections (Middle-earth, Blood Bowl, ...) on a guess. CROSS_SYSTEM_MIN = 0.75 +# Weak acceptance tier: when nothing reached the normal threshold, a hit +# at least this strong is accepted but flagged low confidence ("[w]" in +# the plan, confidence field in the audit, listed in +# sorter_low_confidence.txt). +WEAK_MIN = 0.50 + # Run-artifact files that must never be sorted into a library RUN_ARTIFACTS = { "export.json", "export.html", "state.json", "manifest.jsonl", @@ -61,7 +67,8 @@ class SortItem: klass: str | None = None unit: str | None = None # canonical catalog name, if matched match_score: float = 0.0 - match_kind: str = "none" # catalog | implicit | unclassified + match_kind: str = "none" # catalog | weak | implicit | none | invalid + confidence: str = "normal" # normal | low (match_kind == "weak") action: str = "link" # link | copy target: Path | None = None @@ -242,6 +249,175 @@ GENERIC_TOKENS = frozenset({ }) # noqa: C405 +# directory names that carry no unit information inside sculptor packs; +# never used as match candidates +_JUNK_DIRS = frozenset({ + "files", "file", "images", "image", "img", "pictures", "pics", + "preview", "previews", "render", "renders", "photos", "screenshot", + "screenshots", "torso", "torsos", "head", "heads", "arm", "arms", + "leg", "legs", "back", "front", "left", "right", "top", "bottom", + "side", "weapon", "weapons", "accessories", "accessory", "bits", + "bitz", "parts", "part", "base", "bases", "support", "supports", + "presupported", "unsupported", "stl", "stls", "obj", "models", + "model", "output", "exports", "release", "releases", "license", + "readme", "sources", "docs", "textures", "new", "old", "final", + "fixed", "print", "prints", "plates", "plate", "assembly", "test", +}) # noqa: C405 + + +def _is_junk_dir(name: str) -> bool: + """True for directory names that must never be match candidates.""" + clean = _clean_name(name) + if not clean or clean in _JUNK_DIRS or clean.isdigit() or len(clean) <= 2: + return True + return " " not in clean and clean in GENERIC_TOKENS + + +def _apply_entry( + item: SortItem, entry: dict, score: float, kind: str, confidence: str +) -> None: + """Adopt a catalog entry into an item (dir re-homing and weak tier). + + Unlike the pack/file path in match_catalog, the entry's class layer is + adopted unconditionally: a precise unit match ("Warp Spiders" -> Fly) + must beat a generic keyword class ("warlock" mentioned somewhere -> + Hero). + """ + item.unit = entry["name"] + if entry.get("group"): + item.group = entry["group"] + else: + item.group = GROUP_OF_FACTION.get(item.faction, UNSORTED) + if entry.get("subfaction"): + item.subfaction = entry["subfaction"] + if entry.get("class"): + item.klass = entry["class"] + item.match_score = score + item.match_kind = kind + item.confidence = confidence + entry_faction = entry.get("faction") + if entry_faction and entry_faction not in ("Unit Type", "Heroes", + "Warriors", "Character"): + item.system = entry["system"] + item.faction = entry_faction + + +def _match_name( + clean: str, + catalog: list[dict], + system: str, + faction: str, + threshold: float, + cache: dict, +): + """Fuzzy-match a bare name (a directory or pack name) against the catalog. + + Same-system constraint; a different faction needs a strong hit + (CROSS_SYSTEM_MIN). Returns (score, entry, weak) or None - weak marks + a same-faction hit accepted only via the WEAK_MIN tier. + """ + key = (clean, system, faction) + hit = cache.get(key, 0) + if hit != 0: + return hit or None + shim = SortItem(path=Path(), size=0, name="", clean=clean) + result = None + if system == "Unknown": + full = _best_in_pool(shim, catalog, use_inner=False) + if full and full[0] >= CROSS_SYSTEM_MIN: + result = (full[0], full[1], False) + else: + pool = [e for e in catalog if e.get("system") == system] + if faction != UNSORTED: + same = _best_in_pool( + shim, [e for e in pool if e.get("faction") == faction], + use_inner=False) + other = _best_in_pool( + shim, [e for e in pool if e.get("faction") != faction], + use_inner=False) + if other and other[0] < CROSS_SYSTEM_MIN: + other = None + cands = [c for c in (same, other) if c] + top = max(cands, key=lambda c: c[0]) if cands else None + if top and top[0] >= threshold: + result = (top[0], top[1], False) + elif same and WEAK_MIN <= same[0] < threshold: + result = (same[0], same[1], True) + else: + top = _best_in_pool(shim, pool, use_inner=False) + if top and top[0] >= threshold: + result = (top[0], top[1], False) + elif top and WEAK_MIN <= top[0] < threshold: + result = (top[0], top[1], True) + cache[key] = result or 0 + return result + + +def _best_dir_match( + item: SortItem, + dirs: list[tuple[Path, str]], + catalog: list[dict], + threshold: float, + cache: dict, +): + """Best matching directory in a deepest-first [(path, name)] chain. + + Returns (score, entry, weak, path) of the winner - higher score wins, + equal keeps the deeper candidate - or None. + """ + best_n = best_w = None + for real_dir, dir_name in dirs: + if _is_junk_dir(dir_name): + continue + m = _match_name(_clean_name(dir_name), catalog, item.system, + item.faction, threshold, cache) + if not m: + continue + score, entry, weak = m + cand = (score, entry, weak, real_dir) + if weak: + if best_w is None or score > best_w[0]: + best_w = cand + elif best_n is None or score > best_n[0]: + best_n = cand + return best_n or best_w + + +def _rehome_from_dirs( + item: SortItem, + dirs: list[tuple[Path, str]], + catalog: list[dict], + threshold: float, + cache: dict, +): + """Let matching directory names re-home an item ("deepest match wins"). + + Returns the matched directory if the item was re-homed, else None. A + dir hit takes over when it beats the item's current resolution - a + confident pack/file match is only replaced by a strictly better (or + equal, coming from below the pack level) hit; implicit items yield to + any accepted dir match. + """ + if item.match_kind == "invalid": + return None + own = item.match_kind in ("catalog", "weak") + own_score = item.match_score if own else 0.0 + m = _best_dir_match(item, dirs, catalog, threshold, cache) + if not m: + return None + score, entry, weak, real_dir = m + if own and score < own_score: + return None + if own and score == own_score and not weak \ + and item.match_kind == "catalog" and item.confidence == "normal": + # exact tie with a confident file match - the file name is the + # deeper side in non-provenance mode, keep it + return None + kind = "weak" if weak else "catalog" + _apply_entry(item, entry, score, kind, "low" if weak else "normal") + return real_dir + + def _best_in_pool( item: SortItem, pool: list[dict], use_inner: bool = True ) -> tuple[float, dict] | None: @@ -355,6 +531,10 @@ def match_catalog(item: SortItem, catalog: list[dict], threshold: float) -> None item.group = GROUP_OF_FACTION.get(entry_faction, UNSORTED) elif entry.get("group") and entry["group"] not in ("Unit Type",): item.group = entry["group"] + elif best and best[0] >= WEAK_MIN: + # Weak tier: a same-system hit below the threshold, accepted only + # when nothing stronger exists anywhere. Flagged low confidence. + _apply_entry(item, best[1], best[0], "weak", "low") else: if item.faction == UNSORTED: item.unit = item.name # implicit entry: own (original) name @@ -424,6 +604,7 @@ def sort_plan( """ prov = provenance or {} unit_cache: dict[str, dict] = {} + dir_cache: dict = {} items = [] for f in sorted(files): if not f.is_file(): @@ -449,6 +630,16 @@ def sort_plan( item = inspect(f) classify_item(item) match_catalog(item, catalog, threshold) + if root is not None: + # manual trees: let the real ancestor dir names compete + rel = f.relative_to(root) + dirs = [(root.joinpath(*rel.parent.parts[:i + 1]), + rel.parent.parts[i]) + for i in range(len(rel.parent.parts) - 1, -1, -1)] + winner = _rehome_from_dirs(item, dirs, catalog, threshold, + dir_cache) + if winner is not None: + inner_rel = f.parent.relative_to(winner) else: # one classification per extraction directory, reused for # every file inside it @@ -472,6 +663,16 @@ def sort_plan( item = inspect(f) for k, v in fields.items(): setattr(item, k, v) + if inner_rel is not None and inner_rel.parts: + # extracted archives: inner dir names may claim subtrees + dirs = [] + for i in range(len(inner_rel.parts) - 1, -1, -1): + d = (root / dest_key).joinpath(*inner_rel.parts[:i + 1]) + dirs.append((d, inner_rel.parts[i])) + winner = _rehome_from_dirs(item, dirs, catalog, threshold, + dir_cache) + if winner is not None: + inner_rel = f.parent.relative_to(winner) item.target = build_target(item, library, inner_rel=inner_rel) item.action = "link" items.append(item) @@ -484,7 +685,8 @@ def apply_plan(items: list[SortItem], audit_path: Path) -> tuple[int, int, int]: Returns (linked, copied, skipped). """ audit_path.parent.mkdir(parents=True, exist_ok=True) - linked = copied = skipped = 0 + linked = copied = skipped = weak = 0 + weak_lines: list[str] = [] with audit_path.open("a", encoding="utf-8") as audit: for item in items: src, dest = item.path, item.target @@ -511,10 +713,28 @@ def apply_plan(items: list[SortItem], audit_path: Path) -> tuple[int, int, int]: shutil.copy2(src, dest) item.action = "copy" copied += 1 + if item.match_kind == "weak": + weak += 1 + weak_lines.append( + f"{item.name} (score {item.match_score:.2f})" + f"\n -> {dest}\n from {src}\n" + ) audit.write(json.dumps({ "from": str(src), "to": str(dest), "action": item.action, "unit": item.unit, "faction": item.faction, "class": item.klass, + "match": item.match_kind, "confidence": item.confidence, }) + "\n") + if weak: + weak_doc = audit_path.parent / "sorter_low_confidence.txt" + weak_doc.write_text( + "Low-confidence placements - the sorter accepted these below the " + "normal\nthreshold because nothing better matched. Check and move " + "manually if wrong.\n\n" + "\n".join(weak_lines), + encoding="utf-8", + ) + log.warning( + "%d low-confidence placements - see %s", weak, weak_doc + ) return linked, copied, skipped @@ -525,6 +745,7 @@ def format_plan(items: list[SortItem]) -> str: lines.append(f"[x] {item.name} - NOT AN ARCHIVE (HTML page, bad download)") continue src = f"{item.name} ({item.size // (1024 * 1024)} MiB)" if item.size > 4 * 1024 * 1024 else item.name - flag = {"catalog": "+", "implicit": "~", "none": "?"}[item.match_kind] + flag = {"catalog": "+", "weak": "w", "implicit": "~", + "none": "?"}[item.match_kind] lines.append(f"[{flag}] {src}\n -> {item.target}") return "\n".join(lines) diff --git a/tests/test_sorter_units.py b/tests/test_sorter_units.py new file mode 100644 index 0000000..6c04771 --- /dev/null +++ b/tests/test_sorter_units.py @@ -0,0 +1,149 @@ +"""Offline sorter tests: dir re-homing, weak tier, junk dirs (plain python3).""" +import json +import shutil +import sys +import tempfile +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from dlphelper.sorter import ( # noqa: E402 + apply_plan, + load_catalog, + sort_plan, +) + +CATALOG = { + "entries": [ + {"name": "Warp Spiders", "system": "Warhammer 40,000", + "group": "Xenos Armies", "faction": "Aeldari", + "class": "Fly", "aliases": []}, + {"name": "Harlequin Troupe", "system": "Warhammer 40,000", + "group": "Xenos Armies", "faction": "Harlequins", + "class": "Battleline", "aliases": []}, + {"name": "Guardian Squad", "system": "Warhammer 40,000", + "group": "Xenos Armies", "faction": "Aeldari", + "class": "Battleline", "aliases": []}, + ] +} + + +def make_lib(): + return Path(tempfile.mkdtemp(prefix="sorter_test_lib_")) + + +def write(p: Path): + p.parent.mkdir(parents=True, exist_ok=True) + p.write_bytes(b"x") + + +_LOADED = None + + +def get_catalog(): + global _LOADED + if _LOADED is None: + _LOADED = load_catalog_path() + return _LOADED + + +def prov_items(root: Path, library): + (root / "extract_manifest.jsonl").write_text(json.dumps({ + "archive": str(root / "Multi Pack.zip"), + "dest": "Multi Pack", "unit": "Multi Pack.zip", + }), encoding="utf-8") + files = sorted((root / "Multi Pack").rglob("*")) + return sort_plan(files, library, get_catalog(), provenance={ + "Multi Pack": "Multi Pack.zip"}, root=root) + + +def test_prov_dir_rehome(): + """A matching inner dir name claims its subtree; junk dirs don't.""" + root = Path(tempfile.mkdtemp(prefix="sorter_test_src_")) + pack = root / "Multi Pack" + write(pack / "14 Warp spiders" / "spider_body.stl") + write(pack / "14 Warp spiders" / "files" / "spider_arm.stl") + write(pack / "HEADS" / "head_a.stl") # junk dir: stays with pack + write(pack / "Retro GrimDark Thing" / "t.stl") # no catalog match + library = make_lib() + items = prov_items(root, library) + tgt = {i.name: i.target for i in items} + lib = str(library) + # warp spiders re-homed, incl. nested junk dir structure, class from catalog + assert "Warp Spiders" in str(tgt["spider_body.stl"]), tgt["spider_body.stl"] + assert "Fly" in str(tgt["spider_body.stl"]), tgt["spider_body.stl"] + assert "Hero" not in str(tgt["spider_body.stl"]) + assert "files" in str(tgt["spider_arm.stl"]) + # junk-dir file stays under the pack unit folder + assert "_unsorted/Multi Pack/HEADS" in str(tgt["head_a.stl"]).replace("\\", "/") + # unmatched subtree keeps the pack unit + assert "Multi Pack" in str(tgt["t.stl"]) + shutil.rmtree(root); shutil.rmtree(library) + + +def test_prov_weak_tier(): + """Below-threshold dir hit is accepted as low confidence.""" + root = Path(tempfile.mkdtemp(prefix="sorter_test_src_")) + pack = root / "Multi Pack" + # 'guardian squad extras' scores below 1.0 but high vs 'Guardian Squad' + write(pack / "guardian squad extras" / "g.stl") + library = make_lib() + items = prov_items(root, library) + it = next(i for i in items if i.name == "g.stl") + assert it.match_kind in ("weak", "catalog"), it.match_kind + if it.match_kind == "weak": + assert it.confidence == "low" + assert "Guardian" in str(it.target) + shutil.rmtree(root); shutil.rmtree(library) + + +def test_nonprov_rehome(): + """Loose noise file inside a named dir re-homes to the dir's unit.""" + root = Path(tempfile.mkdtemp(prefix="sorter_test_src_")) + write(root / "Warp Spiders" / "body.stl") + library = make_lib() + files = sorted(root.rglob("*")) + items = sort_plan(files, library, get_catalog(), root=root) + it = next(i for i in items if i.name == "body.stl") + assert it.match_kind == "catalog", (it.match_kind, it.target) + assert "Warp Spiders" in str(it.target) and "Fly" in str(it.target) + shutil.rmtree(root); shutil.rmtree(library) + + +def test_apply_weak_doc(): + """apply_plan writes the low-confidence review document.""" + root = Path(tempfile.mkdtemp(prefix="sorter_test_src_")) + pack = root / "Multi Pack" + write(pack / "guardian squad extras" / "g.stl") + library = make_lib() + items = prov_items(root, library) + linked, copied, skipped = apply_plan(items, library / "sorter_applied.jsonl") + doc = library / "sorter_low_confidence.txt" + kinds = [i.match_kind for i in items if i.match_kind == "weak"] + if kinds: + assert doc.exists(), "weak doc missing" + assert "low-confidence" in doc.read_text(encoding="utf-8") + else: + assert not doc.exists() + assert (library / "sorter_applied.jsonl").exists() + shutil.rmtree(root); shutil.rmtree(library) + + +def load_catalog_path(): + import tempfile as tf + fh = tf.NamedTemporaryFile(mode="w", suffix=".json", delete=False) + json.dump(CATALOG, fh) + fh.close() + return load_catalog(Path(fh.name)) + + +def main(): + catalog = load_catalog_path() + for fn in (test_prov_dir_rehome, test_prov_weak_tier, test_nonprov_rehome, + test_apply_weak_doc): + fn() + print("OK", fn.__name__) + + +if __name__ == "__main__": + main()