diff --git a/dlphelper/sorter.py b/dlphelper/sorter.py index 3f8db05..9e1ca6d 100644 --- a/dlphelper/sorter.py +++ b/dlphelper/sorter.py @@ -41,6 +41,11 @@ CROSS_SYSTEM_MIN = 0.75 # sorter_low_confidence.txt). WEAK_MIN = 0.50 +# A hit whose name shares no word with the entry needs this much more +# score to be trusted (char-ratio flukes like "aeldari rangers" -> "Dark +# Reapers" sit at 0.60-0.67; real unit names usually share a token). +TOKENLESS_BONUS = 0.08 + # Run-artifact files that must never be sorted into a library RUN_ARTIFACTS = { "export.json", "export.html", "state.json", "manifest.jsonl", @@ -293,6 +298,18 @@ def _token_contained(a_clean: str, b_clean: str) -> bool: return True +def _shares_rare_token(a_clean: str, b_clean: str) -> bool: + """True when two cleaned names share at least one non-generic token + (>= 3 chars). Gates the weak tier: a sub-threshold hit with zero + shared words is a char-ratio fluke ("LICENSE.txt" -> "Veletaris Storm + Section"), not evidence. + """ + ta = set(_stem_tokens(a_clean)) + tb = set(_stem_tokens(b_clean)) + shared = (ta & tb) - GENERIC_TOKENS + return any(len(t) >= 3 for t in shared) + + def _apply_entry( item: SortItem, entry: dict, score: float, kind: str, confidence: str ) -> None: @@ -322,6 +339,47 @@ def _apply_entry( item.faction = entry_faction +def _accept_name_entry( + clean: str, entry: dict, score: float, threshold: float +): + """(score, entry, weak) when `entry` may adopt a name scoring `score`. + + Names sharing a rare token are trusted at the normal threshold (and + down to WEAK_MIN); tokenless hits need +TOKENLESS_BONUS and are never + weak. None when rejected. + """ + shared = _shares_rare_token(clean, entry["_clean"]) or any( + _shares_rare_token(clean, a) for a in entry["_clean_aliases"]) + if shared: + if score >= threshold: + return (score, entry, False) + if score >= WEAK_MIN: + return (score, entry, True) + return None + if score >= threshold + TOKENLESS_BONUS: + return (score, entry, False) + return None + + +def _pool_best( + clean: str, pool: list[dict], threshold: float +): + """Best ACCEPTED entry over a pool - normal beats weak, then higher + score; deeper-side ties are irrelevant here (single pool).""" + best_n = best_w = None + for entry in pool: + score = _best_ratio(clean, [entry["_clean"], *entry["_clean_aliases"]]) + a = _accept_name_entry(clean, entry, score, threshold) + if not a: + continue + if a[2]: + if best_w is None or a[0] > best_w[0]: + best_w = a + elif best_n is None or a[0] > best_n[0]: + best_n = a + return best_n or best_w + + def _match_name( clean: str, catalog: list[dict], @@ -352,26 +410,21 @@ def _match_name( 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: + same = _pool_best( + clean, [e for e in pool if e.get("faction") == faction], + threshold) + other = _pool_best( + clean, [e for e in pool if e.get("faction") != faction], + threshold) + if other and (other[2] or other[0] < CROSS_SYSTEM_MIN): + # a different faction must win strong, never weak 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) + if cands: + normals = [c for c in cands if not c[2]] + result = max(normals or cands, key=lambda c: c[0]) 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) + result = _pool_best(clean, pool, threshold) cache[key] = result or 0 return result @@ -561,9 +614,13 @@ 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. + elif best and best[0] >= WEAK_MIN and ( + _shares_rare_token(item.clean, best[1]["_clean"]) + or any(_shares_rare_token(_clean_name(n), best[1]["_clean"]) + for n in item.inner_names[:10])): + # Weak tier: a same-system hit below the threshold with at least + # one shared word, accepted only when nothing stronger exists. + # Flagged low confidence. _apply_entry(item, best[1], best[0], "weak", "low") else: if item.faction == UNSORTED: diff --git a/tests/test_sorter_units.py b/tests/test_sorter_units.py index dad2a20..3b085f5 100644 --- a/tests/test_sorter_units.py +++ b/tests/test_sorter_units.py @@ -24,6 +24,12 @@ CATALOG = { {"name": "Guardian Squad", "system": "Warhammer 40,000", "group": "Xenos Armies", "faction": "Aeldari", "class": "Battleline", "aliases": []}, + {"name": "Rangers", "system": "Warhammer 40,000", + "group": "Xenos Armies", "faction": "Aeldari", + "class": "Battleline", "aliases": []}, + {"name": "Dark Reapers", "system": "Warhammer 40,000", + "group": "Xenos Armies", "faction": "Aeldari", + "class": "Aspect Warriors", "aliases": []}, ] } @@ -121,6 +127,57 @@ def test_cross_system_needs_token_containment(): shutil.rmtree(root); shutil.rmtree(library) +def test_weak_needs_shared_token(): + """Sub-threshold hit with zero shared words is a fluke: reject. + + 'frontline squad' vs 'Guardian Squad' scores 0.552 but shares no + token - it must NOT re-home, while 'guardian weapon team' (0.588, + shares 'guardian') is accepted weak. + """ + root = Path(tempfile.mkdtemp(prefix="sorter_test_src_")) + pack = root / "Aeldari Pack" + write(pack / "frontline squad" / "f.stl") + write(pack / "guardian weapon team" / "g.stl") + library = make_lib() + (root / "extract_manifest.jsonl").write_text(json.dumps({ + "archive": str(root / "Aeldari Pack.zip"), + "dest": "Aeldari Pack", "unit": "Aeldari Pack.zip", + }), encoding="utf-8") + files = sorted((root / "Aeldari Pack").rglob("*")) + items = sort_plan(files, library, get_catalog(), provenance={ + "Aeldari Pack": "Aeldari Pack.zip"}, root=root) + by_name = {i.name: i for i in items} + f = by_name["f.stl"] + assert f.match_kind == "implicit", (f.match_kind, f.target) + assert "frontline squad" in str(f.target) + g = by_name["g.stl"] + assert g.match_kind == "weak" and g.confidence == "low" + shutil.rmtree(root); shutil.rmtree(library) + + +def test_tokenless_needs_bonus(): + """Same-faction adoption without a shared word needs +0.08. + + 'Aeldari Rangers' (0.667 vs 'Dark Reapers', tokenless) must not adopt + Dark Reapers; it instead adopts 'Rangers' (0.636, shares 'ranger'). + """ + root = Path(tempfile.mkdtemp(prefix="sorter_test_src_")) + pack = root / "Aeldari Pack" + write(pack / "Aeldari Rangers" / "r.stl") + library = make_lib() + (root / "extract_manifest.jsonl").write_text(json.dumps({ + "archive": str(root / "Aeldari Pack.zip"), + "dest": "Aeldari Pack", "unit": "Aeldari Pack.zip", + }), encoding="utf-8") + files = sorted((root / "Aeldari Pack").rglob("*")) + items = sort_plan(files, library, get_catalog(), provenance={ + "Aeldari Pack": "Aeldari Pack.zip"}, root=root) + it = next(i for i in items if i.name == "r.stl") + assert it.match_kind == "catalog" and it.unit == "Rangers", \ + (it.match_kind, it.unit, it.match_score) + 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_")) @@ -164,7 +221,8 @@ def load_catalog_path(): def main(): catalog = load_catalog_path() for fn in (test_prov_dir_rehome, test_prov_weak_tier, test_nonprov_rehome, - test_cross_system_needs_token_containment, test_apply_weak_doc): + test_cross_system_needs_token_containment, test_apply_weak_doc, + test_weak_needs_shared_token, test_tokenless_needs_bonus): fn() print("OK", fn.__name__)