900 lines
36 KiB
Python
900 lines
36 KiB
Python
"""Sorter: place collection files into the library via plan/apply.
|
|
|
|
See docs/SORTING.md for the structure and the pipeline this implements:
|
|
inspect (name + archive listing) -> classify (keyword tables) -> fuzzy match
|
|
against the library catalog -> plan -> apply (hardlink, fallback copy).
|
|
|
|
The sorter never deletes or modifies source files. Applying is append-only:
|
|
targets that already exist are skipped, actions are audited in
|
|
sorter_applied.jsonl.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import difflib
|
|
import json
|
|
import logging
|
|
import os
|
|
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
|
|
from dlphelper.warhammer_data import GROUP_OF_FACTION
|
|
|
|
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 (Blood Bowl, ...) on a guess.
|
|
CROSS_SYSTEM_MIN = 0.75
|
|
|
|
# Game systems whose sections the sorter must never place into. The
|
|
# collection is warhammer-only: everything else (Middle-earth proxies,
|
|
# ...) stays in _unsorted even when a name matches the catalog. The
|
|
# catalog snapshot itself keeps all systems; the filter lives here.
|
|
DROPPED_SYSTEMS = frozenset({"Middle-Earth"})
|
|
|
|
# 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
|
|
|
|
# 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",
|
|
"failed_urls.txt", "catalogue.md", "catalogue.jsonl",
|
|
"sorter_applied.jsonl", "catalogue_moves.jsonl", "urls.txt",
|
|
"urls_0807.txt", "urls_gofile.txt", "failed_0807.txt",
|
|
"extract_manifest.jsonl",
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class SortItem:
|
|
"""One file to sort, plus everything learned about it."""
|
|
|
|
path: Path
|
|
size: int
|
|
name: str # original file name
|
|
clean: str # cleaned lower-case name
|
|
inner_names: list[str] = field(default_factory=list) # archive listing
|
|
system: str = "Unknown"
|
|
group: str = UNSORTED
|
|
faction: str = UNSORTED
|
|
subfaction: str | None = None
|
|
klass: str | None = None
|
|
unit: str | None = None # canonical catalog name, if matched
|
|
match_score: float = 0.0
|
|
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
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# inspection
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def list_zip(path: Path, limit: int = 40) -> list[str]:
|
|
"""Inner names of a .zip without extracting."""
|
|
try:
|
|
with zipfile.ZipFile(path) as zf:
|
|
return zf.namelist()[:limit]
|
|
except (zipfile.BadZipFile, OSError, RuntimeError) as exc:
|
|
log.debug("zip listing failed for %s: %s", path.name, exc)
|
|
return []
|
|
|
|
|
|
def list_7z(path: Path, limit: int = 40) -> list[str]:
|
|
"""Inner names of a .7z via the installed 7z binary, without extracting."""
|
|
if not shutil.which("7z"):
|
|
return []
|
|
try:
|
|
proc = subprocess.run(
|
|
["7z", "l", "-ba", "-slt", str(path)],
|
|
stdout=subprocess.PIPE, stderr=subprocess.DEVNULL,
|
|
encoding="utf-8", errors="replace", timeout=120,
|
|
)
|
|
except (OSError, subprocess.TimeoutExpired) as exc:
|
|
log.debug("7z listing failed for %s: %s", path.name, exc)
|
|
return []
|
|
names: list[str] = []
|
|
for block in proc.stdout.split("Path = ")[1:]:
|
|
name = block.splitlines()[0].strip() if block.splitlines() else ""
|
|
# -slt lists directories too; skip entries ending in /
|
|
if name and not name.endswith("/"):
|
|
names.append(name)
|
|
if len(names) >= limit:
|
|
break
|
|
return names
|
|
|
|
|
|
def inspect(path: Path) -> SortItem:
|
|
size = path.stat().st_size if path.exists() else 0
|
|
item = SortItem(path=path, size=size, name=path.name, clean=_clean_name(path.name))
|
|
if not path.is_file():
|
|
return item
|
|
if path.suffix.lower() in ARCHIVE_SUFFIXES:
|
|
# An archive that is not an archive is almost always a saved HTML
|
|
# error/login page that got downloaded instead of the file.
|
|
if path.suffix.lower() == ".zip":
|
|
item.inner_names = list_zip(path)
|
|
else:
|
|
item.inner_names = list_7z(path)
|
|
if item.inner_names == [] and path.stat().st_size > 0:
|
|
head = b""
|
|
with path.open("rb") as fh:
|
|
head = fh.read(512)
|
|
if b"<!doctype html" in head.lower() or b"<html" in head.lower():
|
|
log.warning("%s is an HTML page, not an archive (bad download)", path.name)
|
|
item.match_kind = "invalid"
|
|
return item
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# classification & matching
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def classify_item(item: SortItem) -> None:
|
|
"""Keyword classification from the file name, then refined by inner names."""
|
|
hit = _match_faction(item.clean)
|
|
if hit:
|
|
item.system, item.faction = hit
|
|
item.klass = _match_class(item.clean) or (
|
|
_match_class(_clean_inner(" ".join(item.inner_names)))
|
|
if item.inner_names else None
|
|
)
|
|
|
|
if item.faction == UNSORTED and item.inner_names:
|
|
# hash-named archives: vote on the inner names' classification
|
|
votes: dict[tuple[str, str], int] = {}
|
|
for inner in item.inner_names:
|
|
ihit = _match_faction(_clean_inner(inner))
|
|
if ihit:
|
|
votes[ihit] = votes.get(ihit, 0) + 1
|
|
if votes:
|
|
best = max(votes.items(), key=lambda kv: kv[1])[0]
|
|
item.system, item.faction = best
|
|
if item.klass is None:
|
|
inner_classes = [
|
|
_match_class(_clean_name(inner)) for inner in item.inner_names
|
|
]
|
|
cvotes: dict[str, int] = {}
|
|
for c in inner_classes:
|
|
if c:
|
|
cvotes[c] = cvotes.get(c, 0) + 1
|
|
if cvotes:
|
|
item.klass = max(cvotes.items(), key=lambda kv: kv[1])[0]
|
|
|
|
item.group = GROUP_OF_FACTION.get(item.faction, UNSORTED)
|
|
|
|
|
|
# faction group layer (GW naming, shared with warhammer_data)
|
|
|
|
|
|
def load_catalog(path: Path,
|
|
dropped: frozenset[str] | None = DROPPED_SYSTEMS) -> list[dict]:
|
|
"""Load catalog entries, dropping non-collected game systems by default
|
|
(pass dropped=frozenset() to keep everything)."""
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
entries = data["entries"] if isinstance(data, dict) else data
|
|
if dropped:
|
|
entries = [e for e in entries if e.get("system") not in dropped]
|
|
for e in entries:
|
|
e["_clean"] = _clean_name(e["name"])
|
|
e["_clean_aliases"] = [_clean_name(a) for a in e.get("aliases", [])]
|
|
return entries
|
|
|
|
|
|
def _best_ratio(
|
|
clean: str, candidates: list[str], allow_single: bool = True
|
|
) -> float:
|
|
"""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), allow_single)
|
|
|
|
|
|
@lru_cache(maxsize=None)
|
|
def _best_ratio_cached(
|
|
clean: str, candidates: tuple[str, ...], allow_single: bool = True
|
|
) -> float:
|
|
best = 0.0
|
|
toks = set(_stem_tokens(clean))
|
|
best = _containment_boost(best, toks, candidates, allow_single)
|
|
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, ...],
|
|
allow_single: bool = True,
|
|
) -> 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. Inner-name votes
|
|
pass allow_single=False: one short inner file name ("lieutenant")
|
|
must not claim a whole pack."""
|
|
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 (allow_single and 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",
|
|
"weapon", "gun", "gunner", "turret", "carriage", "upgrade",
|
|
}) # 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",
|
|
"characters", "character",
|
|
}) # noqa: C405
|
|
|
|
|
|
def _clean_inner(name: str) -> str:
|
|
"""Clean an archive-internal (relative) path: dir names join the vote
|
|
("2E Catachany fellaz/Catachan/Jungle Fighters/x.stl" -> "2e catachany
|
|
fellaz catachan jungle fighters x") - the file stem alone loses the
|
|
archive's own folder structure."""
|
|
return _clean_name(name.replace("/", " ").replace("\\", " "))
|
|
|
|
|
|
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 _token_contained(a_clean: str, b_clean: str) -> bool:
|
|
"""True when one name's (stemmed) tokens all appear in the other's.
|
|
|
|
Guarded like _containment_boost: the smaller side must not consist of
|
|
generic army words alone. Used to gate cross-system adoption, where
|
|
plain char-ratio lies ("aeldari rangers" vs "dwarf rangers").
|
|
"""
|
|
ta = set(_stem_tokens(a_clean))
|
|
tb = set(_stem_tokens(b_clean))
|
|
if not ta or not tb:
|
|
return False
|
|
small = ta if ta <= tb else (tb if tb <= ta else None)
|
|
if small is None:
|
|
return False
|
|
if len(small) == 1:
|
|
# a lone generic word ("guard", "orc") must not carry a match
|
|
return next(iter(small)) not in GENERIC_TOKENS
|
|
return True
|
|
|
|
|
|
def _contained_somewhere(item: SortItem, entry: dict) -> bool:
|
|
"""Token containment between the entry (or an alias) and the outer
|
|
name or any inner (relative-path) name."""
|
|
cands = [entry["_clean"], *entry["_clean_aliases"]]
|
|
probes = [item.clean, *[_clean_inner(n) for n in item.inner_names[:10]]]
|
|
return any(_token_contained(p, c) for p in probes for c in cands)
|
|
|
|
|
|
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:
|
|
"""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 _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],
|
|
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 and (
|
|
_token_contained(clean, full[1]["_clean"])
|
|
or any(_token_contained(clean, a)
|
|
for a in full[1]["_clean_aliases"])):
|
|
result = (full[0], full[1], False)
|
|
else:
|
|
pool = [e for e in catalog if e.get("system") == system]
|
|
if faction != UNSORTED:
|
|
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]
|
|
if cands:
|
|
normals = [c for c in cands if not c[2]]
|
|
result = max(normals or cands, key=lambda c: c[0])
|
|
else:
|
|
result = _pool_best(clean, pool, threshold)
|
|
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,
|
|
prefer_dirs_on_tie: bool = False,
|
|
):
|
|
"""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:
|
|
# exact tie: the deeper side wins, which is the dir only in
|
|
# provenance mode (pack name sits above the inner dirs); in manual
|
|
# trees the file name itself is the deepest side - keep it
|
|
if not (prefer_dirs_on_tie and not weak):
|
|
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:
|
|
"""Best fuzzy score of `item` against all entries in `pool`.
|
|
|
|
Inner archive names are only consulted on request; a confident outer
|
|
(file) name must win over votes from random inner STL filenames.
|
|
"""
|
|
best: tuple[float, dict] | None = None
|
|
for entry in pool:
|
|
cands = [entry["_clean"], *entry["_clean_aliases"]]
|
|
score = _best_ratio(item.clean, cands)
|
|
if use_inner and item.inner_names:
|
|
inner_clean = [_clean_inner(n) for n in item.inner_names[:10]]
|
|
for c in inner_clean:
|
|
# no single-token containment boost for inner votes
|
|
score = max(score, _best_ratio(c, cands, allow_single=False) * 0.95)
|
|
if best is None or score > best[0]:
|
|
best = (score, entry)
|
|
return best
|
|
|
|
|
|
def match_catalog(item: SortItem, catalog: list[dict], threshold: float) -> None:
|
|
"""Fuzzy-match against catalog entries in the same faction (or all).
|
|
|
|
The matched entry is hierarchy-authoritative: group, subfaction and
|
|
class come from the catalog (GW shop categories), and when the keyword
|
|
classification had no faction the entry's system/faction is adopted too.
|
|
"""
|
|
if item.faction == UNSORTED and not item.inner_names and not item.clean:
|
|
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]
|
|
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. Keep the
|
|
# higher of the two scores (a retry must not downgrade a
|
|
# weak-eligible same-faction hit).
|
|
sys_pool = [e for e in catalog if e.get("system") == item.system]
|
|
sys_best = _best_in_pool(item, sys_pool, use_inner=False)
|
|
if sys_best is not None and (best is None or sys_best[0] > best[0]):
|
|
best = sys_best
|
|
if best is None or best[0] < threshold:
|
|
sys_best = _best_in_pool(item, sys_pool, use_inner=True)
|
|
if sys_best is not None and (best is None or sys_best[0] > best[0]):
|
|
best = sys_best
|
|
# a retry hit from a *different* faction is only trusted when
|
|
# it is strong AND anchored by token containment; weak
|
|
# inner-name votes must not re-home the file
|
|
if best is not None and best[1].get("faction") != item.faction \
|
|
and (best[0] < CROSS_SYSTEM_MIN
|
|
or not _contained_somewhere(item, best[1])):
|
|
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) with token containment may adopt
|
|
# another game's section; weaker or unanchored 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
|
|
elif best is not None and not _contained_somewhere(item, best[1]):
|
|
best = None
|
|
if best and best[0] >= threshold:
|
|
entry = best[1]
|
|
item.unit = entry["name"]
|
|
if entry.get("group"):
|
|
item.group = entry["group"]
|
|
elif item.faction == UNSORTED:
|
|
item.group = GROUP_OF_FACTION.get(entry["faction"], UNSORTED)
|
|
if entry.get("subfaction"):
|
|
item.subfaction = entry["subfaction"]
|
|
# GW's class layer (shop category) beats the keyword guess, but do
|
|
# not *narrow* it: "Battleline" is GW's bucket for core infantry,
|
|
# a keyword-derived Hero/Elite is more specific for a proxy file.
|
|
catalog_klass = entry.get("class")
|
|
if catalog_klass:
|
|
# a matched catalog unit's class layer is authoritative for
|
|
# the whole subtree - a keyword-voted class ("warlock" seen
|
|
# somewhere -> Hero) must not survive next to a precise unit
|
|
item.klass = catalog_klass
|
|
item.match_score = best[0]
|
|
item.match_kind = "catalog"
|
|
# The catalog entry's faction is authoritative - but only adopt it
|
|
# when it is a real army faction. Generic GW branches like
|
|
# "Unit Type" or factionless entries would degrade a confident
|
|
# keyword faction to a junk folder.
|
|
entry_faction = entry.get("faction")
|
|
generic_factions = {"Unit Type", "Heroes", "Warriors", "Character"}
|
|
if item.faction == UNSORTED:
|
|
if entry_faction and entry_faction not in generic_factions:
|
|
item.system = entry["system"]
|
|
item.faction = entry_faction
|
|
if entry.get("group") and entry["group"] not in ("Unit Type",):
|
|
item.group = entry["group"]
|
|
else:
|
|
item.group = GROUP_OF_FACTION.get(item.faction, UNSORTED)
|
|
elif entry_faction and entry_faction not in generic_factions and entry_faction != item.faction:
|
|
# GW moved this unit to another faction (or the keyword tables
|
|
# are stale): trust the catalog.
|
|
item.system = entry["system"]
|
|
item.faction = entry_faction
|
|
if entry.get("group") and entry["group"] != "Unit Type":
|
|
item.group = entry["group"]
|
|
else:
|
|
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 and (
|
|
_shares_rare_token(item.clean, best[1]["_clean"])
|
|
or any(_shares_rare_token(_clean_inner(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:
|
|
item.unit = item.name # implicit entry: own (original) name
|
|
item.match_score = best[0] if best else 0.0
|
|
item.match_kind = "implicit"
|
|
else:
|
|
# Faction was known but nothing fuzzy-matched: keep the keyword
|
|
# faction and use the file's own name as the unit, but adopt the
|
|
# catalog's class layer when it adds information.
|
|
item.unit = item.name
|
|
item.match_score = best[0] if best else 0.0
|
|
item.match_kind = "implicit"
|
|
|
|
|
|
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:
|
|
if item.group:
|
|
parts.append(item.group)
|
|
parts.append(item.faction)
|
|
if item.subfaction:
|
|
parts.append(item.subfaction)
|
|
parts.append(item.klass or UNCLASSIFIED_CLASS)
|
|
if item.unit:
|
|
parts.append(_display_name(item.unit))
|
|
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
|
|
|
|
|
|
def _display_name(unit: str) -> str:
|
|
"""Implicit entries carry the original filename; strip the extension."""
|
|
p = Path(unit)
|
|
return p.stem if p.suffix else unit
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# plan / apply
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _drop_dropped_system(item: SortItem) -> None:
|
|
"""Dropped systems (see DROPPED_SYSTEMS) count as unclassified: the
|
|
collection does not cover them, so their names must not place files
|
|
into those sections. Called after keyword classification, before
|
|
matching - the item can still be re-homed by other signals."""
|
|
if item.system in DROPPED_SYSTEMS:
|
|
item.system = "Unknown"
|
|
item.group = item.faction = UNSORTED
|
|
item.subfaction = None
|
|
|
|
|
|
def sort_plan(
|
|
files: list[Path],
|
|
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] = {}
|
|
dir_cache: dict = {}
|
|
items = []
|
|
for f in sorted(files):
|
|
if not f.is_file():
|
|
continue
|
|
if (f.name.endswith(".part") or f.name.startswith(".")
|
|
or f.name in RUN_ARTIFACTS):
|
|
continue
|
|
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)]
|
|
# dirs-only relative path (the file name itself is
|
|
# appended by build_target; keeping it here would
|
|
# duplicate it in the target path)
|
|
inner_rel = rel.relative_to(p).parent
|
|
break
|
|
p = p.parent
|
|
if unit_name is None:
|
|
item = inspect(f)
|
|
classify_item(item)
|
|
_drop_dropped_system(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
|
|
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)
|
|
_drop_dropped_system(probe)
|
|
match_catalog(probe, catalog, threshold)
|
|
fields = {k: getattr(probe, k) for k in (
|
|
"system", "group", "faction", "subfaction", "klass",
|
|
"unit", "match_score", "match_kind", "confidence",
|
|
)}
|
|
unit_cache[dest_key] = fields
|
|
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, prefer_dirs_on_tie=True)
|
|
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)
|
|
return items
|
|
|
|
|
|
def apply_plan(items: list[SortItem], audit_path: Path) -> tuple[int, int, int]:
|
|
"""Create targets. Hardlink preferred; copy when filesystems differ.
|
|
|
|
Returns (linked, copied, skipped).
|
|
"""
|
|
audit_path.parent.mkdir(parents=True, exist_ok=True)
|
|
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
|
|
if item.match_kind == "invalid":
|
|
log.error("Not sorting %s: it is an HTML page, not an archive", src.name)
|
|
skipped += 1
|
|
continue
|
|
if dest is None or not src.is_file():
|
|
skipped += 1
|
|
continue
|
|
if dest.exists():
|
|
log.info("Skip, target exists: %s", dest)
|
|
skipped += 1
|
|
continue
|
|
dest.parent.mkdir(parents=True, exist_ok=True)
|
|
if os.stat(src).st_dev == os.stat(dest.parent).st_dev:
|
|
os.link(src, dest)
|
|
item.action = "link"
|
|
linked += 1
|
|
else:
|
|
log.warning(
|
|
"Different filesystem - COPYING instead of hardlinking %s", src.name
|
|
)
|
|
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
|
|
|
|
|
|
def format_plan(items: list[SortItem]) -> str:
|
|
lines = []
|
|
for item in items:
|
|
if item.match_kind == "invalid":
|
|
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": "+", "weak": "w", "implicit": "~",
|
|
"none": "?"}[item.match_kind]
|
|
lines.append(f"[{flag}] {src}\n -> {item.target}")
|
|
return "\n".join(lines)
|