Ben_Kosytorz/Warhammer_DLP_Helper
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

extract stage + system-locked matching + provenance unit mode

Browse Source 
- tools/extract_all.py: recursive extraction of all archives (incl.
  nested) into a plain-file tree; nested archives are deleted after
  successful extraction, plain files are hardlinked, failures go to
  .broken/, extract_manifest.jsonl records archive->dest->unit.
- sorter --provenance: classify each extraction directory once under
  the archive's name (files of one archive are one unit; inner
  structure kept), so loose STLs sort with their model set.
- matching discipline: a fuzzy match never leaves the file's system;
  system-less files need CROSS_SYSTEM_MIN (0.75) for a catalog unit,
  and a same-system faction jump below 0.75 is rejected. Fixes 40k
  files landing in Middle-earth/Blood Bowl/AoS buckets.
- speed for big trees: lru_cache on keyword matchers and pair scores,
  quick_ratio pruning; keywords: night lord, csm.
This commit is contained in:
Ben_Kosytorz
2026-10-05 08:09:22 +02:00
parent 7e28518a52
commit 86fcb3aecc
7 changed files with 455 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dlphelper/sorter.py
+164 -22
View File
@@ -19,6 +19,7 @@ import shutil
import subprocess
import zipfile
from dataclasses import dataclass, field
from functools import lru_cache
from pathlib import Path
from dlphelper.catalogue import _clean_name, _match_class, _match_faction
@@ -29,6 +30,10 @@ log = logging.getLogger(__name__)
ARCHIVE_SUFFIXES = {".zip", ".7z", ".rar", ".tar", ".gz"}
UNSORTED = "_unsorted"
UNCLASSIFIED_CLASS = "Unclassified"
# A file whose system the keyword tables could not identify only gets a
# catalog unit when the fuzzy hit is strong - otherwise it would wander
# into other games' sections (Middle-earth, Blood Bowl, ...) on a guess.
CROSS_SYSTEM_MIN = 0.75
# Run-artifact files that must never be sorted into a library
RUN_ARTIFACTS = {
@@ -169,10 +174,70 @@ def load_catalog(path: Path) -> list[dict]:
def _best_ratio(clean: str, candidates: list[str]) -> float:
return max(
(difflib.SequenceMatcher(None, clean, c).ratio() for c in candidates),
default=0.0,
)
"""Best sequence ratio, pruned with cheap upper-bound checks.
lru_cached because the same (name, entry) pair is scored over and
over across thousands of files.
"""
return _best_ratio_cached(clean, tuple(candidates))
@lru_cache(maxsize=None)
def _best_ratio_cached(clean: str, candidates: tuple[str, ...]) -> float:
best = 0.0
toks = set(_stem_tokens(clean))
best = _containment_boost(best, toks, candidates)
for c in candidates:
sm = difflib.SequenceMatcher(None, clean, c)
if sm.real_quick_ratio() < best or sm.quick_ratio() < best:
continue
r = sm.ratio()
if r > best:
best = r
if best == 1.0:
break
return best
def _stem_tokens(clean: str) -> list[str]:
"""Tokens with a crude plural 's' stripped, for containment checks."""
return [t[:-1] if len(t) > 3 and t.endswith("s") else t
for t in clean.split()]
def _containment_boost(best: float, toks: set[str], candidates: tuple[str, ...]) -> float:
"""Token-containment bonus: a name whose tokens all appear in the
other (or vice versa) matches strongly even when string ratio is low
("telemon" vs "Telemon Heavy Dreadnought"). Guarded: needs >= 2
tokens, or a single rare one (>= 7 chars), and the smaller side must
not consist of generic army words ("chaos lord" is contained in half
the catalog) so "head" or "guard" never trigger it."""
if not toks:
return best
for c in candidates:
ctoks = set(_stem_tokens(c))
if not ctoks:
continue
small = toks if toks <= ctoks else (ctoks if ctoks <= toks else None)
if small is None or any(t in GENERIC_TOKENS for t in small):
continue
if len(small) >= 2 or len(next(iter(small))) >= 7:
if best < 0.9:
best = 0.9
break
return best
# words too generic for token-containment matching: entries built from
# them alone would claim any file whose name happens to mention them
GENERIC_TOKENS = frozenset({
"chaos", "lord", "space", "marine", "guard", "knight", "dark",
"death", "blood", "war", "warrior", "daemon", "hero", "god",
"free", "black", "white", "grey", "green", "gold", "iron",
"steel", "storm", "soul", "shadow", "bone", "ghost", "king",
"clan", "tribe", "legion", "regiment", "squad", "warhound",
"orc", "ogre", "elf", "dwarf", "giant", "company", "chapter",
}) # noqa: C405
def _best_in_pool(
@@ -207,21 +272,45 @@ def match_catalog(item: SortItem, catalog: list[dict], threshold: float) -> None
return
if item.match_kind == "invalid":
return # HTML-page masquerading as an archive; never fuzzy-match it
system_known = item.system != "Unknown"
same_faction = [e for e in catalog if e.get("faction") == item.faction]
pool = same_faction or catalog # fall back to whole catalog if faction unknown
best = _best_in_pool(item, pool, use_inner=False)
if best is None or best[0] < threshold:
# Outer name not confident: let the archive's inner file names vote
# (this is the only signal for hash-named gofile archives).
best = _best_in_pool(item, pool, use_inner=True)
if best is None or best[0] < threshold:
# The keyword tables can lag behind GW's current categorization
# (e.g. units moved between factions); if the faction pool has no
# confident match, try the whole catalog before giving up.
if same_faction:
best = _best_in_pool(item, catalog, use_inner=False)
if system_known:
# a fuzzy match must never jump into another game's section
same_faction = [e for e in same_faction if e.get("system") == item.system]
if same_faction:
best = _best_in_pool(item, same_faction, use_inner=False)
if best is None or best[0] < threshold:
# Outer name not confident: let the archive's inner file names
# vote (the only signal for hash-named gofile archives).
best = _best_in_pool(item, same_faction, use_inner=True)
if best is None or best[0] < threshold:
# The keyword tables can lag behind GW's current categorization
# (e.g. units moved between factions); retry within the same
# system before giving up - never across systems.
sys_pool = [e for e in catalog if e.get("system") == item.system]
best = _best_in_pool(item, sys_pool, use_inner=False)
if best is None or best[0] < threshold:
best = _best_in_pool(item, catalog, use_inner=True)
best = _best_in_pool(item, sys_pool, use_inner=True)
# a retry hit from a *different* faction is only trusted when
# it is strong; weak inner-name votes must not re-home the file
if best is not None and best[0] >= threshold \
and best[1].get("faction") != item.faction \
and best[0] < CROSS_SYSTEM_MIN:
best = None
elif system_known:
sys_pool = [e for e in catalog if e.get("system") == item.system]
best = _best_in_pool(item, sys_pool, use_inner=False)
if best is None or best[0] < threshold:
best = _best_in_pool(item, sys_pool, use_inner=True)
else:
# Neither system nor faction known: whole catalog, but only a
# strong hit (CROSS_SYSTEM_MIN) may adopt another game's section;
# weaker guesses stay in _unsorted.
best = _best_in_pool(item, catalog, use_inner=False)
if best is None or best[0] < threshold:
best = _best_in_pool(item, catalog, use_inner=True)
if best is not None and best[0] < CROSS_SYSTEM_MIN:
best = None
if best and best[0] >= threshold:
entry = best[1]
item.unit = entry["name"]
@@ -278,10 +367,13 @@ def match_catalog(item: SortItem, catalog: list[dict], threshold: float) -> None
item.match_kind = "implicit"
def build_target(item: SortItem, library: Path) -> Path:
def build_target(item: SortItem, library: Path, inner_rel: Path | None = None) -> Path:
parts: list[str] = []
if item.system == "Unknown":
parts = [UNSORTED]
if inner_rel is not None and item.unit:
# extracted archives keep their unit folder even when unplaced
parts.append(_display_name(item.unit))
else:
parts = [item.system]
if item.faction != UNSORTED:
@@ -296,6 +388,8 @@ def build_target(item: SortItem, library: Path) -> Path:
else:
parts.append(UNSORTED)
target_dir = Path(library).joinpath(*parts)
if inner_rel is not None:
target_dir = target_dir.joinpath(*inner_rel.parts)
return target_dir / item.name
@@ -314,7 +408,20 @@ def sort_plan(
library: Path,
catalog: list[dict],
threshold: float = 0.62,
provenance: dict[str, str] | None = None,
root: Path | None = None,
) -> list[SortItem]:
"""Build sort items for `files`.
`provenance` (with `root`) activates extracted-archive mode: it maps
extraction directories (relative to `root`) to the archive they came
from. All files inside such a directory are classified and matched
once under the archive's name and then placed together, keeping the
archive's inner structure - a loose "Backpack.stl" alone is noise,
under "The Twin Lance/" it belongs to its unit.
"""
prov = provenance or {}
unit_cache: dict[str, dict] = {}
items = []
for f in sorted(files):
if not f.is_file():
@@ -322,10 +429,45 @@ def sort_plan(
if (f.name.endswith(".part") or f.name.startswith(".")
or f.name in RUN_ARTIFACTS):
continue
item = inspect(f)
classify_item(item)
match_catalog(item, catalog, threshold)
item.target = build_target(item, library)
unit_name = dest_key = None
inner_rel: Path | None = None
if root is not None and prov:
rel = f.relative_to(root)
p = rel.parent
while p != Path("."):
if str(p) in prov:
dest_key, unit_name = str(p), prov[str(p)]
inner_rel = rel.relative_to(p)
break
p = p.parent
if unit_name is None:
item = inspect(f)
classify_item(item)
match_catalog(item, catalog, threshold)
else:
# one classification per extraction directory, reused for
# every file inside it
fields = unit_cache.get(dest_key)
if fields is None:
listing = sorted(
str(g.relative_to(root / dest_key))
for g in (root / dest_key).rglob("*") if g.is_file()
)[:40]
probe = SortItem(
path=f, size=0, name=unit_name,
clean=_clean_name(unit_name), inner_names=listing,
)
classify_item(probe)
match_catalog(probe, catalog, threshold)
fields = {k: getattr(probe, k) for k in (
"system", "group", "faction", "subfaction", "klass",
"unit", "match_score", "match_kind",
)}
unit_cache[dest_key] = fields
item = inspect(f)
for k, v in fields.items():
setattr(item, k, v)
item.target = build_target(item, library, inner_rel=inner_rel)
item.action = "link"
items.append(item)
return items