sorter: deepest-matching-dir re-homing + weak 0.50 confidence tier
Inner directory names now compete with the pack/file name: a subtree under a dir whose name matches the catalog better than the pack-level resolution is re-homed there (fixes multi-unit sculptor packs like 'Retro Elvez' dumping everything into one keyword class). Junk dir names (files/images/HEADS/2/...) never match. Hits between 0.50 and the threshold are accepted as low-confidence: [w] in the plan, confidence in the audit, listed in sorter_low_confidence.txt. Works in both provenance and manual-tree mode.
This commit is contained in:
+224
-3
@@ -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)
|
||||
|
||||
@@ -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()
|
||||
Reference in New Issue
Block a user