Ben_Kosytorz
8f3a91bcd4
The library may be a seeded torrent whose file set is frozen; the sorter_applied.jsonl / sorter_low_confidence.txt bookkeeping now goes to <library>/../_logs/ by default. Explicit --audit still wins.
141 lines
5.5 KiB
Python
141 lines
5.5 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>/../_logs/sorter_applied.jsonl
|
|
(bookkeeping stays OUT of the library, which may be a
|
|
seeded torrent; --audit still overrides)
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import logging
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
from dlphelper.sorter import (
|
|
DROPPED_SYSTEMS,
|
|
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")
|
|
parser.add_argument("--drop-system", action="append", default=[],
|
|
metavar="NAME",
|
|
help="never place into this game system "
|
|
"(default-dropped: Middle-Earth)")
|
|
parser.add_argument("--keep-system", action="append", default=[],
|
|
metavar="NAME",
|
|
help="re-enable a default-dropped game system")
|
|
args = parser.parse_args()
|
|
|
|
dropped = frozenset(DROPPED_SYSTEMS - set(args.keep_system)
|
|
| set(args.drop_system))
|
|
if args.keep_system or args.drop_system:
|
|
log.info("System filter adjusted: dropping %s", ", ".join(sorted(dropped)) or "nothing")
|
|
|
|
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(
|
|
p for p in source.rglob("*")
|
|
# hidden trees like .broken/ must never reach the plan
|
|
if not any(part.startswith(".") for part in p.relative_to(source).parts)
|
|
)
|
|
catalog = load_catalog(args.catalog, dropped=dropped)
|
|
if dropped:
|
|
log.info("Dropped systems: %s", ", ".join(sorted(dropped)))
|
|
# Bookkeeping must never land inside the library itself (it may be a
|
|
# seeded torrent whose file set is frozen): default to a _logs folder
|
|
# beside it. An explicit --audit overrides this.
|
|
audit = args.audit or (args.library.parent / "_logs" / "sorter_applied.jsonl")
|
|
log.info("Audit file: %s", audit)
|
|
|
|
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)
|