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

Sorter: fuzzysort files into the miniset-aligned library via hardlinks

Browse Source 
- 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)
This commit is contained in:
Ben_Kosytorz
2026-10-04 21:17:52 +02:00
parent acb3415878
commit cccf10a308
4 changed files with 684 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
data/library_catalog.json
+140
View File
@@ -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"]
}
]
}
dlphelper/sorter.py
+331
View File
@@ -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"<!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_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)
docs/SORTING.md
+111
View File
@@ -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:
```
<library>/
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 `<game>/_unsorted/`.
- Fully unclassifiable files land in `<library>/_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<br/>keyword tables]
C --> D[fuzzy match unit name<br/>against library catalog]
D -->|score >= threshold| E[target: catalog entry]
D -->|no match| F[implicit entry: own cleaned name<br/>flagged for review]
E & F --> G[plan: dry-run report]
G -->|apply| H[hardlink into library<br/>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.
sorter.py
+102
View File
@@ -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 <dir> --library /path/to/library
# apply: hardlink files into the library (originals stay!)
python3 sorter.py <dir> --library /path/to/library --apply
# widen/narrow fuzzy matching
python3 sorter.py <dir> --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 <library>/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)