Files
Warhammer_DLP_Helper/sorter.py
T
Ben_Kosytorz 86fcb3aecc extract stage + system-locked matching + provenance unit mode
- tools/extract_all.py: recursive extraction of all archives (incl.
  nested) into a plain-file tree; nested archives are deleted after
  successful extraction, plain files are hardlinked, failures go to
  .broken/, extract_manifest.jsonl records archive->dest->unit.
- sorter --provenance: classify each extraction directory once under
  the archive's name (files of one archive are one unit; inner
  structure kept), so loose STLs sort with their model set.
- matching discipline: a fuzzy match never leaves the file's system;
  system-less files need CROSS_SYSTEM_MIN (0.75) for a catalog unit,
  and a same-system faction jump below 0.75 is rejected. Fixes 40k
  files landing in Middle-earth/Blood Bowl/AoS buckets.
- speed for big trees: lru_cache on keyword matchers and pair scores,
  quick_ratio pruning; keywords: night lord, csm.
2026-10-05 08:09:22 +02:00

116 lines
4.2 KiB
Python

#!/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" / "gwshop_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("--provenance", type=Path, default=None, metavar="FILE",
help="extract_manifest.jsonl from tools/extract_all.py: "
"sort whole extraction directories as units")
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")
provenance = None
if args.provenance:
provenance = {}
with args.provenance.open(encoding="utf-8") as fh:
for line in fh:
rec = json.loads(line)
provenance[rec["dest"]] = rec["unit"]
log.info("Provenance: %d extraction directories", len(provenance))
items = sort_plan(files, args.library, catalog, threshold=args.threshold,
provenance=provenance, root=source)
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)