From cccf10a30895dca7b9107076b7c59a10aae4b874 Mon Sep 17 00:00:00 2001 From: Ben_Kosytorz Date: Sun, 4 Oct 2026 21:17:52 +0200 Subject: [PATCH] Sorter: fuzzysort files into the miniset-aligned library via hardlinks - docs/SORTING.md: target hierarchy (game > faction group > faction > subfaction > class > unit, miniset.net naming), canonical-name policy, hardlink policy, unzip outlook - data/library_catalog.json: canonical catalog (name/aliases/subfaction/class) - dlphelper/sorter.py: inspect (zip via zipfile, 7z via 7z l), keyword classification incl. inner-name voting for hash-named archives, fuzzy matching (difflib), plan/apply split, hardlink preferred with copy fallback, append-only audit, existing targets skipped, run artifacts filtered, HTML-page detection for bad downloads - sorter.py CLI (plan by default, --apply to act) --- data/library_catalog.json | 140 ++++++++++++++++ dlphelper/sorter.py | 331 ++++++++++++++++++++++++++++++++++++++ docs/SORTING.md | 111 +++++++++++++ sorter.py | 102 ++++++++++++ 4 files changed, 684 insertions(+) create mode 100644 data/library_catalog.json create mode 100644 dlphelper/sorter.py create mode 100644 docs/SORTING.md create mode 100644 sorter.py diff --git a/data/library_catalog.json b/data/library_catalog.json new file mode 100644 index 0000000..173cda0 --- /dev/null +++ b/data/library_catalog.json @@ -0,0 +1,140 @@ +{ + "comment": "Canonical library catalog. Fields: system/group/faction/subfaction/class/name/aliases. Names and aliases are matched case-insensitively; subfaction and class may be null. Aligned with miniset.net faction names.", + "entries": [ + { + "system": "Warhammer 40,000", + "group": "Armies of the Imperium", + "faction": "Adeptus Astartes", + "subfaction": "Space Wolves", + "class": "Fast Attack", + "name": "Thunderwolf Cavalry", + "aliases": ["thunder wolf", "thunderwolves", "thunder wolf cavalry"] + }, + { + "system": "Warhammer 40,000", + "group": "Armies of the Imperium", + "faction": "Adeptus Astartes", + "subfaction": "Space Wolves", + "class": "Elites", + "name": "Wulfen", + "aliases": ["wulfen dreadnought"] + }, + { + "system": "Warhammer 40,000", + "group": "Armies of Chaos", + "faction": "Chaos Daemons", + "subfaction": null, + "class": "HQ", + "name": "Keeper of Secrets", + "aliases": ["keeper of secrets and shalaxi helbane", "shalaxi helbane", "keepers of secrets"] + }, + { + "system": "Warhammer 40,000", + "group": "Xenos Armies", + "faction": "Drukhari", + "subfaction": null, + "class": null, + "name": "The Twin Lance", + "aliases": ["twin lance"] + }, + { + "system": "Warhammer 40,000", + "group": "Armies of the Imperium", + "faction": "Adeptus Astartes", + "subfaction": null, + "class": "Elites", + "name": "Terminator Squad", + "aliases": ["terminators", "terminator squad"] + }, + { + "system": "Warhammer 40,000", + "group": "Armies of the Imperium", + "faction": "Adeptus Astartes", + "subfaction": null, + "class": "Elites", + "name": "Dreadnought", + "aliases": ["dread"] + }, + { + "system": "Warhammer 40,000", + "group": "Armies of the Imperium", + "faction": "Adeptus Astartes", + "subfaction": null, + "class": "HQ", + "name": "Captain", + "aliases": ["captain in gravis armour", "primaris captain"] + }, + { + "system": "Warhammer 40,000", + "group": "Armies of the Imperium", + "faction": "Adeptus Astartes", + "subfaction": null, + "class": "Heavy Support", + "name": "Hellblaster Squad", + "aliases": ["hellblasters"] + }, + { + "system": "Warhammer 40,000", + "group": "Armies of the Imperium", + "faction": "Adeptus Astartes", + "subfaction": null, + "class": "Troops", + "name": "Intercessor Squad", + "aliases": ["intercessors"] + }, + { + "system": "Warhammer 40,000", + "group": "Xenos Armies", + "faction": "T'au Empire", + "subfaction": null, + "class": "HQ", + "name": "Commander Farsight", + "aliases": ["farsight"] + }, + { + "system": "Warhammer 40,000", + "group": "Xenos Armies", + "faction": "T'au Empire", + "subfaction": null, + "class": "HQ", + "name": "Commander Shadowsun", + "aliases": ["shadowsun"] + }, + { + "system": "Warhammer 40,000", + "group": "Xenos Armies", + "faction": "T'au Empire", + "subfaction": null, + "class": "Fast Attack", + "name": "Pathfinder Team", + "aliases": ["pathfinders"] + }, + { + "system": "Warhammer 40,000", + "group": "Xenos Armies", + "faction": "T'au Empire", + "subfaction": null, + "class": "Dedicated Transport", + "name": "Devilfish", + "aliases": ["devil fish"] + }, + { + "system": "Warhammer 40,000", + "group": "Xenos Armies", + "faction": "T'au Empire", + "subfaction": null, + "class": null, + "name": "Assorted Drones", + "aliases": ["drones"] + }, + { + "system": "Warhammer 40,000", + "group": "Armies of the Imperium", + "faction": "Adepta Sororitas", + "subfaction": null, + "class": null, + "name": "Battle Sisters", + "aliases": ["battle sister squad", "battle sisters squad"] + } + ] +} diff --git a/dlphelper/sorter.py b/dlphelper/sorter.py new file mode 100644 index 0000000..3483ae3 --- /dev/null +++ b/dlphelper/sorter.py @@ -0,0 +1,331 @@ +"""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 pathlib import Path + +from dlphelper.catalogue import _clean_name, _match_class, _match_faction + +log = logging.getLogger(__name__) + +ARCHIVE_SUFFIXES = {".zip", ".7z", ".rar", ".tar", ".gz"} +UNSORTED = "_unsorted" +UNCLASSIFIED_CLASS = "Unclassified" + +# 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", +} + + +@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 | implicit | unclassified + 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, text=True, 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_name(" ".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_name(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 (miniset naming), extended as the taxonomy grows +GROUP_OF_FACTION: dict[str, str] = { + faction: group + for group, factions in { + "Armies of the Imperium": [ + "Adeptus Astartes", "Astra Militarum", "Adeptus Mechanicus", + "Adepta Sororitas", "Adeptus Custodes", "Imperial Knights", + "Inquisition & Officio Assassinorum", + ], + "Armies of Chaos": [ + "Chaos Space Marines", "Death Guard", "Thousand Sons", + "Chaos Daemons", "Chaos Knights", + ], + "Xenos Armies": [ + "Orks", "T'au Empire", "Tyranids", "Genestealer Cults", "Necrons", + "Aeldari", "Drukhari", "Harlequins", "Ynnari", + ], + }.items() + for faction in factions +} + + +def load_catalog(path: Path) -> list[dict]: + data = json.loads(path.read_text(encoding="utf-8")) + entries = data["entries"] if isinstance(data, dict) else data + 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]) -> float: + return max( + (difflib.SequenceMatcher(None, clean, c).ratio() for c in candidates), + default=0.0, + ) + + +def match_catalog(item: SortItem, catalog: list[dict], threshold: float) -> None: + """Fuzzy-match against catalog entries in the same faction (or all).""" + if item.faction == UNSORTED: + return + same_faction = [e for e in catalog if e["faction"] == item.faction] + pool = same_faction or catalog # fall back to whole catalog if faction unknown + best: tuple[float, dict] | None = None + for entry in pool: + cands = [entry["_clean"], *entry["_clean_aliases"]] + score = _best_ratio(item.clean, cands) + if item.inner_names: + inner_clean = [_clean_name(n) for n in item.inner_names[:10]] + for c in inner_clean: + score = max(score, _best_ratio(c, cands) * 0.95) # inner slightly discounted + if best is None or score > best[0]: + best = (score, entry) + if best and best[0] >= threshold: + entry = best[1] + item.unit = entry["name"] + item.subfaction = entry.get("subfaction") + item.klass = item.klass or entry.get("class") + item.match_score = best[0] + item.match_kind = "catalog" + if item.faction == UNSORTED: + item.system = entry["system"] + item.faction = entry["faction"] + item.group = entry.get("group") or GROUP_OF_FACTION.get(entry["faction"], UNSORTED) + 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" + + +def build_target(item: SortItem, library: Path) -> Path: + parts: list[str] = [] + if item.system == "Unknown": + parts = [UNSORTED] + 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) + 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 sort_plan( + files: list[Path], + library: Path, + catalog: list[dict], + threshold: float = 0.62, +) -> list[SortItem]: + 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 + item = inspect(f) + classify_item(item) + match_catalog(item, catalog, threshold) + item.target = build_target(item, library) + 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 = 0 + 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 + audit.write(json.dumps({ + "from": str(src), "to": str(dest), "action": item.action, + "unit": item.unit, "faction": item.faction, "class": item.klass, + }) + "\n") + 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": "+", "implicit": "~", "none": "?"}[item.match_kind] + lines.append(f"[{flag}] {src}\n -> {item.target}") + return "\n".join(lines) diff --git a/docs/SORTING.md b/docs/SORTING.md new file mode 100644 index 0000000..d662e49 --- /dev/null +++ b/docs/SORTING.md @@ -0,0 +1,111 @@ +# Library structure & sorting pipeline + +This document defines the target structure of the collection ("the library") +and how the sorter places files into it. It is the working agreement for all +sorting-related code; change it here first, then in code. + +## Target hierarchy (miniset.net-aligned) + +The hierarchy mirrors miniset.net ("Miniatures Collectors Guide"), extended +with the 40k force-organization class, which miniset itself uses as a +"specification" tag on set pages: + +``` +/ + Warhammer 40,000/ # game + Armies of the Imperium/ # faction group (miniset naming) + Adeptus Astartes/ # faction + Space Wolves/ # sub-faction (optional) + Fast Attack/ # force-org class + Thunderwolf Cavalry/ # canonical unit name + Thunderwolf Cavalry.7z # the file(s), or unpacked contents +``` + +Layer rules: + +| Layer | Source of truth | Example | May be omitted? | +|-------|-----------------|---------|-----------------| +| Game | `warhammer_data.SYSTEMS` | `Warhammer 40,000` | no (else: `_unsorted/`) | +| Faction group | Imperium/Chaos/Xenos/etc. | `Armies of Chaos` | no (derived from faction) | +| Faction | miniset.net faction names | `Death Guard` | no (else `_unsorted/` in game) | +| Sub-faction | miniset sub-factions | `Space Wolves` | **yes** — only when known | +| Class | force-org slots | `Fast Attack` | **yes** — only when known | +| Unit | canonical model name (lexicanum spelling) | `Thunderwolf Cavalry` | yes (files drop directly into the class dir) | + +- Files without a faction but with a game land in `/_unsorted/`. +- Fully unclassifiable files land in `/_unsorted/` — never guessed. +- Everything in between is optional and filled in as knowledge grows. +- Sub-faction folders are only created when the file name states it + (`Thunderwolf` → Space Wolves); otherwise files stay in the faction dir so + we don't invent hierarchy. + +## Canonical names + +Canonical unit names follow Lexicanum/GW spelling (`Keeper of Secrets`, +`Death Korps of Krieg`), because that is what cross-seeding, wikis and +miniset.net all agree on. Release names stay untouched inside the unit +folder. + +## Sorting pipeline + +```mermaid +graph LR + A[archive/dir] --> B[read name + zip/7z listing if possible] + B --> C[classify: system + faction + class
keyword tables] + C --> D[fuzzy match unit name
against library catalog] + D -->|score >= threshold| E[target: catalog entry] + D -->|no match| F[implicit entry: own cleaned name
flagged for review] + E & F --> G[plan: dry-run report] + G -->|apply| H[hardlink into library
fallback: copy] +``` + +1. **Inspect** — read the file name; for archives, also read the *inner* + file list (zip via stdlib `zipfile`, 7z via the installed `7z l` binary — + no Python dependency). Inner names improve classification for hash-named + archives and are recorded in the report. +2. **Classify** — game/faction/class from the keyword tables + (`warhammer_data.py`), same as `catalogue.py`. +3. **Match** — fuzzy-match the (inner) name against `data/library_catalog.json` + entries within the same faction; best sequence-match above + `--threshold` (default 0.62) wins. The catalog provides canonical names, + sub-factions and classes. +4. **Plan** — default output: a dry-run table `source → target` plus + `sorter_plan.jsonl`. Nothing is touched. +5. **Apply** — create the target and **hardlink** the file there + (`os.link`); if the filesystems differ, fall back to a real copy with a + warning. Originals are never modified or deleted — that is what keeps the + *arr-style workflow possible (remux/seed/sort independently). Every + applied action is appended to `sorter_applied.jsonl` and re-runs skip + targets that already exist. + +## Why hardlinks, and the unzip question + +- Hardlinks make "same file, two positions" free (0 extra bytes) — the + original stays for seeding/history, the library copy is what torrenting + sees. Limitation: hardlinks need source and library on the same + filesystem; the sorter detects this and copies instead (warned). +- Zip contents **can** be read without extracting (that is how the sorter + peeks). Modifying a member in place is technically possible with + `zipfile` but is not useful here. +- Endgame (later stage): unpack archives into the unit folder so the + library is plain files, easier to sort per-model and to torrent. The + unpack step will live in the sorter (`--unpack`, 7z binary) but stays off + until the link-based flow is proven on a small batch. + +## Catalog file (`data/library_catalog.json`) + +```json +{ + "system": "Warhammer 40,000", + "group": "Armies of the Imperium", + "faction": "Adeptus Astartes", + "subfaction": "Space Wolves", + "class": "Fast Attack", + "name": "Thunderwolf Cavalry", + "aliases": ["thunder wolf", "thunderwolf cavalry squad"] +} +``` + +Entries are matched case-insensitively against cleaned names. Growing the +catalog is the main lever for match quality; unknown files surface in the +plan report with their best score so gaps are obvious. diff --git a/sorter.py b/sorter.py new file mode 100644 index 0000000..83c424f --- /dev/null +++ b/sorter.py @@ -0,0 +1,102 @@ +#!/usr/bin/env python3 +"""Sorter CLI: fuzzysort collection files into the library (plan/apply). + + # plan (default, touches nothing) + python3 sorter.py --library /path/to/library + + # apply: hardlink files into the library (originals stay!) + python3 sorter.py --library /path/to/library --apply + + # widen/narrow fuzzy matching + python3 sorter.py --library ... --threshold 0.5 + + # sort a single archive + python3 sorter.py "Thunderwolf Cavalry.7z" --library ... + +Flags: + --catalog FILE default data/library_catalog.json + --audit FILE default /sorter_applied.jsonl +""" + +from __future__ import annotations + +import argparse +import json +import logging +import sys +from pathlib import Path + +from dlphelper.sorter import ( + format_plan, + load_catalog, + sort_plan, + apply_plan, +) + +log = logging.getLogger("sorter") + +DEFAULT_CATALOG = Path(__file__).resolve().parent / "data" / "library_catalog.json" + + +def main() -> None: + parser = argparse.ArgumentParser( + description="Sort files into the library structure (dry-run by default)." + ) + parser.add_argument("source", type=Path, + help="a directory to sort or a single file") + parser.add_argument("--library", type=Path, required=True, metavar="DIR", + help="library root the files will be linked into") + parser.add_argument("--catalog", type=Path, default=DEFAULT_CATALOG, metavar="FILE") + parser.add_argument("--threshold", type=float, default=0.62, + help="fuzzy match threshold (default 0.62)") + parser.add_argument("--apply", action="store_true", + help="create the targets (hardlink, fallback copy)") + parser.add_argument("--audit", type=Path, default=None, metavar="FILE") + parser.add_argument("--json", action="store_true", + help="print the plan as JSON instead of text") + args = parser.parse_args() + + logging.basicConfig(level=logging.INFO, + format="%(asctime)s %(levelname)-7s %(message)s", + datefmt="%H:%M:%S") + + source = args.source + files = [source] if source.is_file() else sorted(source.rglob("*")) + catalog = load_catalog(args.catalog) + audit = args.audit or (args.library / "sorter_applied.jsonl") + + items = sort_plan(files, args.library, catalog, threshold=args.threshold) + + if args.json: + print(json.dumps([ + { + "source": str(i.path), "name": i.name, "size": i.size, + "system": i.system, "group": i.group, "faction": i.faction, + "subfaction": i.subfaction, "class": i.klass, "unit": i.unit, + "score": round(i.match_score, 3), "kind": i.match_kind, + "inner_count": len(i.inner_names), + "target": str(i.target), + } + for i in items + ], indent=2, ensure_ascii=False)) + else: + print(format_plan(items)) + + n_catalog = sum(1 for i in items if i.match_kind == "catalog") + n_implicit = sum(1 for i in items if i.match_kind == "implicit") + n_none = sum(1 for i in items if i.match_kind == "none") + log.info("Plan: %d files -> %d catalog matches [+] / %d implicit [~] / " + "%d unclassifiable [?]", + len(items), n_catalog, n_implicit, n_none) + + if args.apply: + linked, copied, skipped = apply_plan(items, audit) + log.info("Applied: %d hardlinked, %d copied, %d skipped (audit: %s)", + linked, copied, skipped, audit) + + +if __name__ == "__main__": + try: + main() + except KeyboardInterrupt: + sys.exit(130)