"""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" 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", }) # 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)