chore: start transition layout — python into version-python/, scaffold version-rust/
- git mv of all existing files into version-python/ (history preserved) - version-rust/: minimal cargo scaffold (edition 2024), no deps yet - root AGENTS.md documents the transition rules and artifact/schema contract - main is untouched; the python layout there stays as-is
This commit is contained in:
@@ -0,0 +1,3 @@
|
||||
"""Warhammer DLP Helper - collect and organize files shared in Discord channels."""
|
||||
|
||||
__version__ = "0.1.0"
|
||||
@@ -0,0 +1,305 @@
|
||||
"""Catalogue: classify collected files into game system / faction / class.
|
||||
|
||||
The catalogue is the bridge between "a pile of downloads" and the later
|
||||
library stage: it reads manifest.jsonl (with full source attribution) and/or
|
||||
scans arbitrary directories for archive files, classifies each file against
|
||||
the taxonomy in warhammer_data.py, and writes machine- (catalogue.jsonl) and
|
||||
human-readable (catalogue.md) reports.
|
||||
|
||||
Classification is keyword-based and deliberately conservative: files that
|
||||
cannot be classified land in the "unclassified" bucket so gaps become visible
|
||||
instead of producing wrong assignments. A manual overrides file can pin
|
||||
individual files.
|
||||
|
||||
Nothing is moved by default - use --apply for that (it moves files into
|
||||
<library-root>/<System>/<Faction>/<Class>/ and records every move in
|
||||
catalogue_moves.jsonl so it can be audited/undone).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import re
|
||||
import shutil
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
|
||||
from dlphelper.warhammer_data import (
|
||||
FACTION_ALIASES,
|
||||
GROUP_OF_FACTION,
|
||||
SYSTEMS,
|
||||
UNIT_CLASSES,
|
||||
)
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
ARCHIVE_SUFFIXES = {".zip", ".7z", ".rar", ".tar", ".gz"}
|
||||
UNKNOWN = "Unknown"
|
||||
UNCLASSIFIED_FACTION = "Unclassified"
|
||||
UNCLASSIFIED_CLASS = "Unclassified"
|
||||
|
||||
|
||||
@dataclass
|
||||
class CatalogueEntry:
|
||||
"""One classified file."""
|
||||
|
||||
path: str
|
||||
name: str
|
||||
size: int
|
||||
system: str = UNKNOWN
|
||||
faction: str = UNCLASSIFIED_FACTION
|
||||
klass: str = UNCLASSIFIED_CLASS
|
||||
source: dict = field(default_factory=dict) # manifest attribution if known
|
||||
|
||||
|
||||
from functools import lru_cache
|
||||
|
||||
|
||||
@lru_cache(maxsize=None)
|
||||
def _clean_name(name: str) -> str:
|
||||
stem = Path(name).stem
|
||||
stem = re.sub(r"[_\.\+\[\]\(\)]+", " ", stem)
|
||||
stem = re.sub(r"\s+", " ", stem).strip().lower()
|
||||
return stem
|
||||
|
||||
|
||||
def _kw_regex(keyword: str) -> str:
|
||||
"""Word-boundary regex for a keyword, tolerating a plural 's'."""
|
||||
return rf"(?<![a-z0-9]){re.escape(keyword)}s?(?![a-z0-9])"
|
||||
|
||||
|
||||
@lru_cache(maxsize=None)
|
||||
def _match_faction(clean: str) -> tuple[str, str] | None:
|
||||
"""Return (system, faction) with the most specific keyword hit."""
|
||||
best: tuple[int, str, str] | None = None # (len, system, faction)
|
||||
for system, aliases in FACTION_ALIASES.items():
|
||||
for keyword, faction in aliases.items():
|
||||
if re.search(_kw_regex(keyword), clean):
|
||||
if best is None or len(keyword) > best[0]:
|
||||
best = (len(keyword), system, faction)
|
||||
return (best[1], best[2]) if best else None
|
||||
|
||||
|
||||
@lru_cache(maxsize=None)
|
||||
def _match_class(clean: str) -> str | None:
|
||||
"""Return the most specific force-organization class keyword hit."""
|
||||
best: tuple[int, str] | None = None
|
||||
for keyword, klass in UNIT_CLASSES.items():
|
||||
if re.search(_kw_regex(keyword), clean):
|
||||
if best is None or len(keyword) > best[0]:
|
||||
best = (len(keyword), klass)
|
||||
return best[1] if best else None
|
||||
|
||||
|
||||
def classify(name: str) -> tuple[str, str, str]:
|
||||
"""Classify a file name -> (system, faction, class)."""
|
||||
clean = _clean_name(name)
|
||||
faction_hit = _match_faction(clean)
|
||||
klass = _match_class(clean)
|
||||
|
||||
if faction_hit:
|
||||
system, faction = faction_hit
|
||||
# Non-40k systems: keep their class if we found one, else Unclassified
|
||||
if system != "Warhammer 40,000" and klass is None:
|
||||
return system, faction, UNCLASSIFIED_CLASS
|
||||
return system, faction, klass or UNCLASSIFIED_CLASS
|
||||
|
||||
# No faction match: a class keyword alone cannot attribute a system
|
||||
return UNKNOWN, UNCLASSIFIED_FACTION, UNCLASSIFIED_CLASS
|
||||
|
||||
|
||||
def scan_manifest(manifest_path: Path) -> list[CatalogueEntry]:
|
||||
"""Entries from a manifest.jsonl (preferred: has source attribution)."""
|
||||
entries: list[CatalogueEntry] = []
|
||||
seen: set[str] = set()
|
||||
if not manifest_path.exists():
|
||||
return entries
|
||||
for line in manifest_path.read_text(encoding="utf-8").splitlines():
|
||||
line = line.strip()
|
||||
if not line:
|
||||
continue
|
||||
try:
|
||||
row = json.loads(line)
|
||||
except ValueError:
|
||||
log.warning("Skipping malformed manifest line: %s", line[:80])
|
||||
continue
|
||||
path = row.get("path") or ""
|
||||
if path in seen:
|
||||
continue
|
||||
seen.add(path)
|
||||
entries.append(CatalogueEntry(
|
||||
path=path,
|
||||
name=row.get("name") or Path(path).name,
|
||||
size=int(row.get("size") or 0),
|
||||
source={
|
||||
"channel": row.get("channel"),
|
||||
"url": row.get("url"),
|
||||
"message_id": row.get("message_id"),
|
||||
"timestamp": row.get("timestamp"),
|
||||
"author": row.get("author"),
|
||||
},
|
||||
))
|
||||
return entries
|
||||
|
||||
|
||||
def scan_directory(directory: Path) -> list[CatalogueEntry]:
|
||||
"""Entries from walking a directory tree for archive files (manual era)."""
|
||||
entries: list[CatalogueEntry] = []
|
||||
for path in sorted(directory.rglob("*")):
|
||||
if not path.is_file() or path.suffix.lower() not in ARCHIVE_SUFFIXES:
|
||||
continue
|
||||
try:
|
||||
size = path.stat().st_size
|
||||
except OSError:
|
||||
size = 0
|
||||
entries.append(CatalogueEntry(path=str(path), name=path.name, size=size))
|
||||
return entries
|
||||
|
||||
|
||||
def load_overrides(path: Path | None) -> dict[str, dict]:
|
||||
"""overrides.json: {"/path/or/filename": {"faction": ..., "class": ..., "system": ...}}"""
|
||||
if path is None or not path.exists():
|
||||
return {}
|
||||
return json.loads(path.read_text(encoding="utf-8"))
|
||||
|
||||
|
||||
def apply_overrides(entries: list[CatalogueEntry], overrides: dict[str, dict]) -> int:
|
||||
count = 0
|
||||
by_name = {e.name.lower(): e for e in entries}
|
||||
for key, fix in overrides.items():
|
||||
entry = by_name.get(key.lower())
|
||||
if entry is None:
|
||||
log.warning("Override target not found: %s", key)
|
||||
continue
|
||||
if "system" in fix:
|
||||
entry.system = fix["system"]
|
||||
if "faction" in fix:
|
||||
entry.faction = fix["faction"]
|
||||
if "class" in fix:
|
||||
entry.klass = fix["class"]
|
||||
count += 1
|
||||
return count
|
||||
|
||||
|
||||
def format_size(n: int) -> str:
|
||||
size = float(n)
|
||||
for unit in ("B", "KiB", "MiB", "GiB", "TiB"):
|
||||
if size < 1024 or unit == "TiB":
|
||||
return f"{size:.1f} {unit}"
|
||||
size /= 1024
|
||||
return f"{n} B"
|
||||
|
||||
|
||||
def write_reports(
|
||||
entries: list[CatalogueEntry],
|
||||
out_md: Path,
|
||||
out_jsonl: Path,
|
||||
) -> None:
|
||||
# machine report
|
||||
with out_jsonl.open("w", encoding="utf-8") as fh:
|
||||
for e in entries:
|
||||
fh.write(json.dumps({
|
||||
"path": e.path,
|
||||
"name": e.name,
|
||||
"size": e.size,
|
||||
"system": e.system,
|
||||
"faction": e.faction,
|
||||
"class": e.klass,
|
||||
"source": {k: v for k, v in e.source.items() if v},
|
||||
}, ensure_ascii=False) + "\n")
|
||||
|
||||
# human report
|
||||
grouped: dict[str, dict[str, dict[str, list[CatalogueEntry]]]] = {}
|
||||
for e in entries:
|
||||
grouped.setdefault(e.system, {}).setdefault(e.faction, {}).setdefault(
|
||||
e.klass, []).append(e)
|
||||
|
||||
lines: list[str] = ["# Warhammer Collection Catalogue", ""]
|
||||
total_size = sum(e.size for e in entries)
|
||||
classified = sum(1 for e in entries if e.faction != UNCLASSIFIED_FACTION)
|
||||
lines.append(f"{len(entries)} files, {format_size(total_size)} total, "
|
||||
f"{classified} classified, "
|
||||
f"{len(entries) - classified} unclassified")
|
||||
lines.append("")
|
||||
|
||||
for system in sorted(grouped):
|
||||
factions = grouped[system]
|
||||
all_sys_files = [e for f in factions.values() for c in f.values() for e in c]
|
||||
sys_size = sum(e.size for e in all_sys_files)
|
||||
lines.append(f"## {system} ({len(all_sys_files)} files, {format_size(sys_size)})")
|
||||
for faction in sorted(factions):
|
||||
classes = factions[faction]
|
||||
fac_size = sum(e.size for c in classes.values() for e in c)
|
||||
lines.append(f"### {faction} ({format_size(fac_size)})")
|
||||
for klass in sorted(classes):
|
||||
files = classes[klass]
|
||||
lines.append(f"- **{klass}** ({len(files)} files, {format_size(sum(e.size for e in files))})")
|
||||
for e in sorted(files, key=lambda x: x.name.lower()):
|
||||
src = ""
|
||||
if e.source.get("url"):
|
||||
src = f" — from {e.source['url']}"
|
||||
lines.append(f" - {e.name} ({format_size(e.size)}){src}")
|
||||
lines.append("")
|
||||
|
||||
out_md.write_text("\n".join(lines), encoding="utf-8")
|
||||
|
||||
|
||||
def apply_moves(
|
||||
entries: list[CatalogueEntry],
|
||||
library_root: Path,
|
||||
moves_log: Path,
|
||||
) -> int:
|
||||
"""Move classified files into the library structure. Audited in moves_log."""
|
||||
moves_log.parent.mkdir(parents=True, exist_ok=True)
|
||||
moved = 0
|
||||
with moves_log.open("a", encoding="utf-8") as logfh:
|
||||
for e in entries:
|
||||
if e.faction == UNCLASSIFIED_FACTION:
|
||||
continue # never move what we cannot classify
|
||||
if e.system == UNKNOWN:
|
||||
continue
|
||||
src = Path(e.path)
|
||||
if not src.exists():
|
||||
log.warning("Cannot move, missing: %s", src)
|
||||
continue
|
||||
dest_dir = (
|
||||
library_root / e.system
|
||||
/ GROUP_OF_FACTION.get(e.faction, "Unclassified")
|
||||
/ e.faction / e.klass
|
||||
)
|
||||
dest = dest_dir / src.name
|
||||
if dest.exists():
|
||||
log.warning("Skip, destination exists: %s", dest)
|
||||
continue
|
||||
dest_dir.mkdir(parents=True, exist_ok=True)
|
||||
shutil.move(str(src), str(dest))
|
||||
logfh.write(json.dumps({"from": str(src), "to": str(dest)}) + "\n")
|
||||
e.path = str(dest)
|
||||
moved += 1
|
||||
return moved
|
||||
|
||||
|
||||
def build_catalogue(
|
||||
manifests: list[Path],
|
||||
scan_dirs: list[Path],
|
||||
overrides_path: Path | None,
|
||||
) -> list[CatalogueEntry]:
|
||||
entries: list[CatalogueEntry] = []
|
||||
seen: set[str] = set()
|
||||
for m in manifests:
|
||||
for e in scan_manifest(m):
|
||||
if e.path not in seen:
|
||||
seen.add(e.path)
|
||||
entries.append(e)
|
||||
for d in scan_dirs:
|
||||
for e in scan_directory(d):
|
||||
if e.path not in seen:
|
||||
seen.add(e.path)
|
||||
entries.append(e)
|
||||
|
||||
for e in entries:
|
||||
e.system, e.faction, e.klass = classify(e.name)
|
||||
|
||||
apply_overrides(entries, load_overrides(overrides_path))
|
||||
return entries
|
||||
@@ -0,0 +1,160 @@
|
||||
"""Export a Discord channel/thread with DiscordChatExporter and extract URLs.
|
||||
|
||||
The channels this tool targets encode their download links as base64 inside
|
||||
message content, and occasionally as plain links. We export the channel as
|
||||
JSON (richer metadata than the HTML export: message id, timestamp, author),
|
||||
decode every base64 candidate, and keep only URLs for supported hosts.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import re
|
||||
import subprocess
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from urllib.parse import urlparse
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
DCE_IMAGE = "tyrrrz/discordchatexporter:stable"
|
||||
|
||||
# The channels encode links as base64 that always starts with this prefix
|
||||
# (https://...); grab.sh established this regex, keep it compatible.
|
||||
B64_RE = re.compile(r"aHR0cHM6Ly[89][A-Za-z0-9+/_-]*")
|
||||
URL_RE = re.compile(r"https://[^\s<>\"'\)\]\}]+")
|
||||
|
||||
SUPPORTED_HOSTS = {
|
||||
"gofile": ("gofile.io", "www.gofile.io"),
|
||||
"0807": ("0807.st", "www.0807.st"),
|
||||
}
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class FoundUrl:
|
||||
"""One download URL found in the channel, with source attribution."""
|
||||
|
||||
url: str
|
||||
host: str # "gofile" | "0807" (set after classification)
|
||||
message_id: str | None = None
|
||||
timestamp: str | None = None
|
||||
author: str | None = None
|
||||
encoded: bool = False # found via base64 decode vs. plain link
|
||||
|
||||
|
||||
def export_channel(channel_id: str, out_dir: Path, token: str, force: bool = False) -> Path:
|
||||
"""Export the channel to out_dir/export.json using the DCE docker image.
|
||||
|
||||
The token is passed to the container via the DISCORD_TOKEN env var (DCE
|
||||
reads it natively) so it never appears on the docker command line. TOKEN
|
||||
is set as well for DCE versions that only know that name.
|
||||
"""
|
||||
dest = out_dir / "export.json"
|
||||
if dest.exists() and not force:
|
||||
log.info("Reusing existing export %s (pass --fresh to re-export)", dest)
|
||||
return dest
|
||||
|
||||
out_dir.mkdir(parents=True, exist_ok=True)
|
||||
env = dict(os.environ)
|
||||
env["DISCORD_TOKEN"] = token
|
||||
env["TOKEN"] = token
|
||||
cmd = [
|
||||
"docker", "run", "--rm",
|
||||
"--user", f"{os.getuid()}:{os.getgid()}",
|
||||
"-v", f"{out_dir.resolve()}:/out",
|
||||
"-e", "DISCORD_TOKEN",
|
||||
"-e", "TOKEN",
|
||||
DCE_IMAGE,
|
||||
"export", "-c", channel_id, "-f", "Json", "-o", "/out/export.json",
|
||||
]
|
||||
log.info("Exporting channel %s via DiscordChatExporter ...", channel_id)
|
||||
try:
|
||||
subprocess.run(cmd, env=env, check=True)
|
||||
except FileNotFoundError as exc:
|
||||
raise RuntimeError(
|
||||
"docker not found - the Discord export needs the docker image "
|
||||
f"{DCE_IMAGE}"
|
||||
) from exc
|
||||
except subprocess.CalledProcessError as exc:
|
||||
raise RuntimeError(
|
||||
f"DiscordChatExporter failed with exit code {exc.returncode} "
|
||||
"(check the token and the channel id)"
|
||||
) from exc
|
||||
if not dest.exists():
|
||||
raise RuntimeError("DiscordChatExporter produced no output file")
|
||||
return dest
|
||||
|
||||
|
||||
def classify(url: str) -> str | None:
|
||||
"""Return the host key ('gofile' | '0807') for a URL, else None."""
|
||||
try:
|
||||
hostname = (urlparse(url).hostname or "").lower()
|
||||
except ValueError:
|
||||
return None
|
||||
for key, hosts in SUPPORTED_HOSTS.items():
|
||||
if hostname in hosts:
|
||||
return key
|
||||
return None
|
||||
|
||||
|
||||
def _decode_b64(candidate: str) -> str | None:
|
||||
s = candidate.replace("_", "/").replace("-", "+")
|
||||
s = s.rstrip("=")
|
||||
s += "=" * ((4 - len(s) % 4) % 4)
|
||||
try:
|
||||
decoded = base64.b64decode(s, validate=False).decode("utf-8", "strict")
|
||||
except Exception: # noqa: BLE001 - malformed candidate, just skip it
|
||||
return None
|
||||
# One encoded blob can contain the URL plus padding/newlines
|
||||
stripped = decoded.strip()
|
||||
line = stripped.splitlines()[0] if stripped else ""
|
||||
return line or None
|
||||
|
||||
|
||||
def extract_urls(export_path: Path) -> list[FoundUrl]:
|
||||
"""Extract supported URLs from a DCE JSON export, deduplicated."""
|
||||
data = json.loads(export_path.read_text(encoding="utf-8"))
|
||||
messages = data if isinstance(data, list) else data.get("messages", [])
|
||||
|
||||
found: dict[str, FoundUrl] = {}
|
||||
stats = {"b64": 0, "plain": 0, "unsupported": 0}
|
||||
|
||||
for msg in messages:
|
||||
content = str(msg.get("content") or "")
|
||||
if not content:
|
||||
continue
|
||||
author = (msg.get("author") or {}).get("name")
|
||||
attrib = {
|
||||
"message_id": str(msg.get("id") or "") or None,
|
||||
"timestamp": msg.get("timestamp"),
|
||||
"author": author,
|
||||
}
|
||||
|
||||
for match in B64_RE.finditer(content):
|
||||
stats["b64"] += 1
|
||||
decoded = _decode_b64(match.group(0))
|
||||
if not decoded:
|
||||
continue
|
||||
host = classify(decoded)
|
||||
if host is None:
|
||||
stats["unsupported"] += 1
|
||||
continue
|
||||
found.setdefault(decoded, FoundUrl(url=decoded, host=host, encoded=True, **attrib))
|
||||
|
||||
for match in URL_RE.finditer(content):
|
||||
url = match.group(0).rstrip(".,;)")
|
||||
host = classify(url)
|
||||
if host is None:
|
||||
continue
|
||||
stats["plain"] += 1
|
||||
found.setdefault(url, FoundUrl(url=url, host=host, encoded=False, **attrib))
|
||||
|
||||
log.info(
|
||||
"Extraction: %d base64 candidates, %d plain links, %d unsupported/undecodable, "
|
||||
"%d unique supported URLs",
|
||||
stats["b64"], stats["plain"], stats["unsupported"], len(found),
|
||||
)
|
||||
return sorted(found.values(), key=lambda f: f.url)
|
||||
@@ -0,0 +1,70 @@
|
||||
"""Cross-run, cross-channel record of completed URLs.
|
||||
|
||||
This is what prevents re-downloading your collection: state.json only knows
|
||||
about one OUT_DIR, but the same gofile link posted in a second channel would
|
||||
otherwise be downloaded again into that channel's OUT_DIR. The global store
|
||||
(default: ~/.local/share/dlphelper/urls.json) records every successfully
|
||||
downloaded URL once, forever, regardless of OUT_DIR.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def default_global_path() -> Path:
|
||||
data_home = os.environ.get("XDG_DATA_HOME") or str(
|
||||
Path.home() / ".local" / "share"
|
||||
)
|
||||
return Path(data_home) / "dlphelper" / "urls.json"
|
||||
|
||||
|
||||
class GlobalStore:
|
||||
"""JSON file mapping url -> {status, channel, host, files, updated}."""
|
||||
|
||||
def __init__(self, path: Path | None = None):
|
||||
self.path = path
|
||||
self.records: dict[str, dict] = {}
|
||||
if self.path and self.path.exists():
|
||||
try:
|
||||
self.records = json.loads(self.path.read_text(encoding="utf-8"))
|
||||
if not isinstance(self.records, dict):
|
||||
raise ValueError("global store root must be an object")
|
||||
except (ValueError, OSError) as exc:
|
||||
backup = self.path.with_suffix(".json.bak")
|
||||
log.warning("Corrupt global store %s (%s); backed up to %s",
|
||||
self.path, exc, backup)
|
||||
self.path.replace(backup)
|
||||
self.records = {}
|
||||
|
||||
def is_done(self, url: str) -> bool:
|
||||
if not self.records:
|
||||
return False
|
||||
rec = self.records.get(url)
|
||||
return bool(rec) and rec.get("status") == "ok"
|
||||
|
||||
def mark(self, url: str, **fields) -> None:
|
||||
if self.path is None:
|
||||
return
|
||||
rec = self.records.setdefault(url, {"status": "ok"})
|
||||
for key, value in fields.items():
|
||||
if value is not None:
|
||||
rec[key] = value
|
||||
rec["status"] = "ok"
|
||||
rec["updated"] = time.strftime("%Y-%m-%dT%H:%M:%S")
|
||||
self.save()
|
||||
|
||||
def save(self) -> None:
|
||||
if self.path is None:
|
||||
return
|
||||
self.path.parent.mkdir(parents=True, exist_ok=True)
|
||||
tmp = self.path.with_suffix(".json.tmp")
|
||||
tmp.write_text(json.dumps(self.records, indent=2, sort_keys=True),
|
||||
encoding="utf-8")
|
||||
tmp.replace(self.path)
|
||||
@@ -0,0 +1,239 @@
|
||||
"""Minimal GoFile API client - one guest account per run.
|
||||
|
||||
The API interaction mirrors Lysagxra/GoFileDownloader (same headers, website
|
||||
token scheme and endpoints), but with two deliberate differences:
|
||||
|
||||
1. A single guest account is created per run (GoFileDownloader creates one per
|
||||
URL, which trips Gofile's rate limit after ~6 requests).
|
||||
2. Failures raise exceptions instead of sys.exit(), so the orchestrator can
|
||||
track them per file/URL and keep going.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import logging
|
||||
import os
|
||||
import time
|
||||
from pathlib import Path
|
||||
from urllib.parse import urlparse
|
||||
|
||||
import requests
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
API = "https://api.gofile.io"
|
||||
USER_AGENT = "Mozilla/5.0"
|
||||
LOCALE = "en-US"
|
||||
TOKEN_SECRET = "12af056dacea0b" # public website-token seed, see GoFileDownloader # noqa: S105
|
||||
TOKEN_WINDOW_SEC = 14400
|
||||
CHUNK_SIZE = 1024 * 1024
|
||||
|
||||
COMMON_HEADERS = {
|
||||
"User-Agent": USER_AGENT,
|
||||
"Accept-Encoding": "gzip",
|
||||
"Accept": "*/*",
|
||||
"Connection": "keep-alive",
|
||||
}
|
||||
EXTENDED_HEADERS = {
|
||||
**COMMON_HEADERS,
|
||||
"Sec-Fetch-Dest": "empty",
|
||||
"Sec-Fetch-Mode": "cors",
|
||||
"Sec-Fetch-Site": "same-site",
|
||||
"Pragma": "no-cache",
|
||||
"Cache-Control": "no-cache",
|
||||
}
|
||||
|
||||
|
||||
class GoFileError(Exception):
|
||||
"""Base error for GoFile operations."""
|
||||
|
||||
|
||||
class GoFilePasswordRequired(GoFileError):
|
||||
"""The content is password protected and none (or a wrong one) was given."""
|
||||
|
||||
|
||||
class GoFileNotFound(GoFileError):
|
||||
"""The content/file no longer exists (404 / status notFound).
|
||||
|
||||
Dead links never come back, so the orchestrator must not retry these.
|
||||
"""
|
||||
|
||||
|
||||
class GoFileEmpty(GoFileError):
|
||||
"""The content exists but contains no files (e.g. emptied folder).
|
||||
|
||||
Stable state like a dead link: no retry.
|
||||
"""
|
||||
|
||||
|
||||
class GoFileAuthError(GoFileError):
|
||||
"""The account token was rejected; the client can re-auth once."""
|
||||
|
||||
|
||||
def content_id_from_url(url: str) -> str:
|
||||
"""Extract the content id from a gofile.io/d/<id> URL."""
|
||||
parts = [p for p in urlparse(url).path.split("/") if p]
|
||||
if len(parts) >= 2 and parts[-2] == "d":
|
||||
return parts[-1]
|
||||
raise GoFileError(f"not a gofile content URL (expected gofile.io/d/<id>): {url}")
|
||||
|
||||
|
||||
class GoFileClient:
|
||||
"""GoFile API client that reuses one guest account token for the whole run."""
|
||||
|
||||
def __init__(self, password: str | None = None):
|
||||
self._password_hash = (
|
||||
hashlib.sha256(password.encode()).hexdigest() if password else None
|
||||
)
|
||||
self._token: str | None = None
|
||||
self.session = requests.Session()
|
||||
|
||||
# -- authentication -----------------------------------------------------
|
||||
|
||||
def ensure_token(self, max_attempts: int = 8, wait_seconds: int = 20) -> str:
|
||||
"""Create a guest account; backs off and retries on rate limiting."""
|
||||
if self._token:
|
||||
return self._token
|
||||
|
||||
last_status = "unknown"
|
||||
for attempt in range(1, max_attempts + 1):
|
||||
try:
|
||||
resp = self.session.post(
|
||||
f"{API}/accounts", headers=COMMON_HEADERS, timeout=15
|
||||
)
|
||||
data = resp.json()
|
||||
except (ValueError, requests.RequestException) as exc:
|
||||
last_status = f"request error: {exc}"
|
||||
log.warning(
|
||||
"GoFile account creation attempt %d/%d failed: %s",
|
||||
attempt, max_attempts, last_status,
|
||||
)
|
||||
else:
|
||||
if isinstance(data, dict) and data.get("status") == "ok":
|
||||
self._token = data["data"]["token"]
|
||||
log.info("GoFile guest account created (token reused for whole run)")
|
||||
return self._token
|
||||
last_status = str(data.get("status", data)[:200] if data else data)
|
||||
log.warning(
|
||||
"GoFile account creation attempt %d/%d failed: %s",
|
||||
attempt, max_attempts, last_status,
|
||||
)
|
||||
if attempt < max_attempts:
|
||||
time.sleep(wait_seconds * attempt)
|
||||
raise GoFileError(
|
||||
f"GoFile account creation failed after {max_attempts} attempts "
|
||||
f"(last: {last_status})"
|
||||
)
|
||||
|
||||
def reset(self) -> None:
|
||||
"""Drop the current token so the next ensure_token() creates a new one."""
|
||||
self._token = None
|
||||
|
||||
def _website_token(self) -> str:
|
||||
window = int(time.time()) // TOKEN_WINDOW_SEC
|
||||
seed = f"{USER_AGENT}::{LOCALE}::{self._token}::{window}::{TOKEN_SECRET}"
|
||||
return hashlib.sha256(seed.encode()).hexdigest()
|
||||
|
||||
def _auth_headers(self) -> dict:
|
||||
return {
|
||||
**EXTENDED_HEADERS,
|
||||
"Authorization": f"Bearer {self._token}",
|
||||
"X-Website-Token": self._website_token(),
|
||||
"X-BL": LOCALE,
|
||||
}
|
||||
|
||||
# -- listing ------------------------------------------------------------
|
||||
|
||||
def list_files(self, url: str) -> list[tuple[str, str, str]]:
|
||||
"""Return (relpath, filename, download_link) for every file under a URL."""
|
||||
cid = content_id_from_url(url)
|
||||
results: list[tuple[str, str, str]] = []
|
||||
self._walk(cid, [], results)
|
||||
# Direct files at the content root get their own folder named after the id
|
||||
normalized = [
|
||||
(relpath if relpath else cid, name, link) for relpath, name, link in results
|
||||
]
|
||||
return normalized
|
||||
|
||||
def _walk(self, cid: str, prefix: list[str], results: list[tuple[str, str, str]]) -> None:
|
||||
query = "?cache=true&sortField=createTime&sortDirection=1"
|
||||
if self._password_hash:
|
||||
query += f"&password={self._password_hash}"
|
||||
try:
|
||||
resp = self.session.get(
|
||||
f"{API}/contents/{cid}{query}", headers=self._auth_headers(), timeout=30
|
||||
)
|
||||
except requests.RequestException as exc:
|
||||
raise GoFileError(f"content request for {cid} failed: {exc}") from exc
|
||||
|
||||
if resp.status_code == 404:
|
||||
raise GoFileNotFound(f"content {cid} not found (dead link)")
|
||||
if resp.status_code in (401, 403):
|
||||
raise GoFileAuthError(f"content request denied ({resp.status_code}) for {cid}")
|
||||
try:
|
||||
data = resp.json()
|
||||
except ValueError as exc:
|
||||
raise GoFileError(f"non-JSON response for {cid}: {resp.text[:200]}") from exc
|
||||
|
||||
if data.get("status") == "notFound" or "notFound" in str(data.get("status", "")):
|
||||
raise GoFileNotFound(f"content {cid} does not exist anymore (dead link)")
|
||||
if data.get("status") != "ok":
|
||||
raise GoFileError(f"content request failed for {cid}: {data.get('status')}")
|
||||
|
||||
d = data["data"]
|
||||
if d.get("passwordStatus") not in (None, "passwordOk"):
|
||||
raise GoFilePasswordRequired(
|
||||
f"content {cid} is password protected (use --password)"
|
||||
)
|
||||
|
||||
if d.get("type") == "folder":
|
||||
here = prefix + [d["name"]]
|
||||
for child in (d.get("children") or {}).values():
|
||||
if child.get("type") == "folder":
|
||||
self._walk(child["id"], here, results)
|
||||
else:
|
||||
results.append(("/".join(here), child["name"], child["link"]))
|
||||
else:
|
||||
results.append(("/".join(prefix), d["name"], d["link"]))
|
||||
|
||||
# -- downloading ----------------------------------------------------------
|
||||
|
||||
def download_file(self, link: str, dest: Path) -> int:
|
||||
"""Download one file; returns the file size. Skips existing files."""
|
||||
dest.parent.mkdir(parents=True, exist_ok=True)
|
||||
if dest.exists() and dest.stat().st_size > 0:
|
||||
log.info("Skip existing: %s", dest)
|
||||
return dest.stat().st_size
|
||||
|
||||
tmp = dest.with_name(dest.name + ".part")
|
||||
headers = {
|
||||
**EXTENDED_HEADERS,
|
||||
"Cookie": f"accountToken={self._token}",
|
||||
"Referer": link,
|
||||
"Origin": link,
|
||||
}
|
||||
try:
|
||||
with self.session.get(
|
||||
link, headers=headers, stream=True, timeout=(10, 60), allow_redirects=True
|
||||
) as resp:
|
||||
if resp.status_code == 404:
|
||||
raise GoFileNotFound(f"file {dest.name} not found (dead link)")
|
||||
if resp.status_code != 200:
|
||||
raise GoFileError(f"HTTP {resp.status_code} while downloading {dest.name}")
|
||||
total = int(resp.headers.get("Content-Length") or 0)
|
||||
written = 0
|
||||
with tmp.open("wb") as fh:
|
||||
for chunk in resp.iter_content(chunk_size=CHUNK_SIZE):
|
||||
fh.write(chunk)
|
||||
written += len(chunk)
|
||||
if written == 0:
|
||||
raise GoFileError(f"empty body for {dest.name}")
|
||||
os.replace(tmp, dest)
|
||||
return written
|
||||
finally:
|
||||
if tmp.exists() and not dest.exists():
|
||||
try:
|
||||
tmp.unlink()
|
||||
except OSError:
|
||||
pass
|
||||
@@ -0,0 +1,353 @@
|
||||
"""Games Workshop webshop catalog via Algolia.
|
||||
|
||||
The warhammer.com shop is behind an AWS WAF challenge for non-browser
|
||||
clients, but its product search runs on Algolia and the app-id plus the
|
||||
public search-only key ship with every browser page. This module uses them
|
||||
to enumerate the whole product index and normalize it into the sorter's
|
||||
catalog schema (system/group/faction/subfaction/class/name/aliases).
|
||||
|
||||
Credentials are read from (first hit wins):
|
||||
1. env ALGOLIA_APP_ID + ALGOLIA_API_KEY
|
||||
2. algolia.json next to the repo root (gitignored):
|
||||
{"app_id": "...", "api_key": "..."}
|
||||
|
||||
The key is the *search-only* public key (never an admin key). Treat any
|
||||
key found in algolia.json as still secret: never log or commit it.
|
||||
|
||||
CLI: python3 -m dlphelper.gwshop OUT.json [--index NAME] [--raw RAW.json]
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import time
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import requests
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
DEFAULT_INDEX = "prod-lazarus-product-en-ww"
|
||||
# The read host (-dsn) serves searches; credentials also work as query
|
||||
# parameters, which keeps every request a plain GET/POST with no headers
|
||||
# beyond content-type.
|
||||
_DSN_HOST = "https://{app_id}-dsn.algolia.net"
|
||||
|
||||
_PAGE_SIZE = 1000 # Algolia max hitsPerPage per /query page
|
||||
_RETRY_DELAYS = (2, 5, 15)
|
||||
|
||||
|
||||
def load_credentials() -> tuple[str, str]:
|
||||
"""Resolve (app_id, api_key) from env or algolia.json."""
|
||||
app_id = os.environ.get("ALGOLIA_APP_ID", "")
|
||||
api_key = os.environ.get("ALGOLIA_API_KEY", "")
|
||||
if app_id and api_key:
|
||||
return app_id, api_key
|
||||
cfg = Path(__file__).resolve().parent.parent / "algolia.json"
|
||||
if cfg.is_file():
|
||||
data = json.loads(cfg.read_text(encoding="utf-8"))
|
||||
app_id = data.get("app_id", "") or app_id
|
||||
api_key = data.get("api_key", "") or api_key
|
||||
if not app_id or not api_key:
|
||||
raise SystemExit(
|
||||
"Algolia credentials missing: set ALGOLIA_APP_ID and "
|
||||
"ALGOLIA_API_KEY, or create algolia.json with "
|
||||
'{"app_id": "...", "api_key": "..."} (gitignored).'
|
||||
)
|
||||
return app_id, api_key
|
||||
|
||||
|
||||
class GwShopClient:
|
||||
"""Minimal Algolia search client for one application."""
|
||||
|
||||
def __init__(self, app_id: str, api_key: str) -> None:
|
||||
self.base = _DSN_HOST.format(app_id=app_id)
|
||||
self.params = {
|
||||
"x-algolia-application-id": app_id,
|
||||
"x-algolia-api-key": api_key,
|
||||
}
|
||||
self.session = requests.Session()
|
||||
|
||||
def _request(self, method: str, path: str, body: dict | None = None) -> Any:
|
||||
url = f"{self.base}{path}"
|
||||
last_err: Exception | None = None
|
||||
for attempt, delay in enumerate((0,) + _RETRY_DELAYS, start=1):
|
||||
if delay:
|
||||
log.info("Algolia retry %d in %ds", attempt, delay)
|
||||
time.sleep(delay)
|
||||
try:
|
||||
r = self.session.request(
|
||||
method,
|
||||
url,
|
||||
params=self.params,
|
||||
json=body,
|
||||
timeout=30,
|
||||
)
|
||||
if r.status_code == 200:
|
||||
return r.json()
|
||||
last_err = RuntimeError(
|
||||
f"Algolia {method} {path} -> {r.status_code}: {r.text[:200]}"
|
||||
)
|
||||
if r.status_code in (401, 403): # credentials: no point retrying
|
||||
break
|
||||
except requests.RequestException as exc:
|
||||
last_err = exc
|
||||
raise RuntimeError(f"Algolia request failed: {last_err}") from last_err
|
||||
|
||||
def list_indices(self) -> list[dict]:
|
||||
return self._request("GET", "/1/indexes").get("items", [])
|
||||
|
||||
# ---- full index retrieval -------------------------------------------
|
||||
# The search-only key cannot use /browse (403) and normal pagination is
|
||||
# capped at 1000 cumulative results. So we partition the index on the
|
||||
# GameSystemsRoot hierarchical facet: each branch is fetched with an
|
||||
# explicit filter, recursing deeper only when a branch exceeds one page.
|
||||
|
||||
_FACET = "GameSystemsRoot"
|
||||
|
||||
@staticmethod
|
||||
def _escape(value: str) -> str:
|
||||
return value.replace("\\", "\\\\").replace('"', '\\"')
|
||||
|
||||
def _filter_expr(self, path: list[tuple[int, str]]) -> str:
|
||||
return " AND ".join(
|
||||
f'{self._FACET}.lvl{lvl}:"{self._escape(value)}"'
|
||||
for lvl, value in path
|
||||
)
|
||||
|
||||
def _search_all(self, index: str, filters: str | None) -> tuple[list[dict], int]:
|
||||
"""All hits matching `filters` (must be <= pagination cap)."""
|
||||
out: list[dict] = []
|
||||
params = f"query=&hitsPerPage={_PAGE_SIZE}&page=0"
|
||||
if filters:
|
||||
params += f"&filters={requests.utils.quote(filters, safe='')}"
|
||||
page = self._request(
|
||||
"POST", f"/1/indexes/{index}/query", body={"params": params}
|
||||
)
|
||||
out.extend(page.get("hits", []))
|
||||
return out, int(page.get("nbHits") or 0)
|
||||
|
||||
def _facet_counts(
|
||||
self, index: str, level: int, filters: str | None
|
||||
) -> dict[str, int]:
|
||||
params = f"query=&hitsPerPage=0&facets={self._FACET}.lvl{level}"
|
||||
if filters:
|
||||
params += f"&filters={requests.utils.quote(filters, safe='')}"
|
||||
page = self._request(
|
||||
"POST", f"/1/indexes/{index}/query", body={"params": params}
|
||||
)
|
||||
facets = page.get("facets") or {}
|
||||
return facets.get(f"{self._FACET}.lvl{level}") or {}
|
||||
|
||||
def browse(self, index: str, hits_per_page: int = _PAGE_SIZE) -> list[dict]:
|
||||
"""Every record of the index, via facet-partitioned search."""
|
||||
hits, total = self._search_all(index, None)
|
||||
if total <= hits_per_page:
|
||||
return hits
|
||||
log.info("gwshop %s: %d hits, partitioning on %s", index, total, self._FACET)
|
||||
out = self._partition(index, [], total)
|
||||
covered = len({h.get("objectID") for h in out})
|
||||
if covered < total:
|
||||
log.warning(
|
||||
"gwshop %s: %d of %d index records left out of the dump "
|
||||
"(no fetchable %s branch path)",
|
||||
index, total - covered, total, self._FACET,
|
||||
)
|
||||
return out
|
||||
|
||||
def _partition(
|
||||
self, index: str, path: list[tuple[int, str]], total: int
|
||||
) -> list[dict]:
|
||||
level = len(path)
|
||||
filters = self._filter_expr(path) if path else None
|
||||
counts = self._facet_counts(index, level, filters)
|
||||
out: list[dict] = []
|
||||
covered: set[Any] = set()
|
||||
for value, count in sorted(counts.items()):
|
||||
sub_path = path + [(level, value)]
|
||||
sub_filters = self._filter_expr(sub_path)
|
||||
if count <= _PAGE_SIZE:
|
||||
hits, nb = self._search_all(index, sub_filters)
|
||||
out.extend(hits)
|
||||
covered.update(h.get("objectID") for h in hits)
|
||||
log.info(
|
||||
"gwshop %s: %s -> %d hits", index, sub_filters[:80], nb
|
||||
)
|
||||
else:
|
||||
sub = self._partition(index, sub_path, count)
|
||||
out.extend(sub)
|
||||
covered.update(h.get("objectID") for h in sub)
|
||||
# nbHits summed across branches double-counts products listed in
|
||||
# several branches, so coverage is counted per unique objectID.
|
||||
residual = total - len(covered)
|
||||
if residual > 0:
|
||||
log.warning(
|
||||
"gwshop %s: %d hits in branch carry no %s.lvl%d "
|
||||
"(shallow facet path, left out of the dump)",
|
||||
index, residual, self._FACET, level,
|
||||
)
|
||||
return out
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# normalization
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _clean(node: str) -> str:
|
||||
return node.strip()
|
||||
|
||||
|
||||
def hierarchy_paths(hit: dict) -> list[list[str]]:
|
||||
"""Every full path of the GameSystemsRoot hierarchical facet.
|
||||
|
||||
A hit may belong to several branches (e.g. an army box also being a
|
||||
"Gaming Rules" entry); each branch is returned as its own path
|
||||
[system, group, faction, subfaction_or_class, ...].
|
||||
"""
|
||||
root = hit.get("GameSystemsRoot") or {}
|
||||
paths: list[list[str]] = []
|
||||
for lvl_key in sorted(root, key=lambda k: int(k[3:])):
|
||||
for value in root.get(lvl_key) or []:
|
||||
paths.append([_clean(p) for p in value.split(" > ")])
|
||||
return paths
|
||||
|
||||
|
||||
def normalize(hit: dict) -> dict:
|
||||
"""One Algolia hit -> one or more catalog entries (usually one)."""
|
||||
name = _clean(hit.get("name") or hit.get("objectID") or "")
|
||||
entry: dict[str, Any] = {
|
||||
"name": name,
|
||||
"aliases": [],
|
||||
"sku": hit.get("sku"),
|
||||
"slug": hit.get("slug"),
|
||||
"productType": hit.get("productType"),
|
||||
"source": "gwshop",
|
||||
"gw_paths": hierarchy_paths(hit),
|
||||
"collections": hit.get("_collections") or [],
|
||||
}
|
||||
return entry
|
||||
|
||||
|
||||
def build_catalog(hits: list[dict]) -> dict:
|
||||
"""Index dump -> catalog document in the sorter's schema shape.
|
||||
|
||||
The hierarchy layer (system/group/faction/subfaction) is expanded from
|
||||
gw_paths; name/aliases stay per product so the fuzzy matcher works on
|
||||
the product name, exactly like data/library_catalog.json entries.
|
||||
"""
|
||||
entries: list[dict] = []
|
||||
seen: set[tuple] = set()
|
||||
for hit in hits:
|
||||
base = normalize(hit)
|
||||
paths = base["gw_paths"]
|
||||
if not paths:
|
||||
key = (hit.get("objectID"), ())
|
||||
if key in seen:
|
||||
continue
|
||||
seen.add(key)
|
||||
entries.append(
|
||||
{
|
||||
"system": None,
|
||||
"group": None,
|
||||
"faction": None,
|
||||
"subfaction": None,
|
||||
"class": None,
|
||||
"name": base["name"],
|
||||
"aliases": [],
|
||||
"sku": base["sku"],
|
||||
"slug": base["slug"],
|
||||
"productType": base["productType"],
|
||||
"source": "gwshop",
|
||||
"collections": base["collections"],
|
||||
}
|
||||
)
|
||||
continue
|
||||
# primary branch = longest path (most specific), e.g.
|
||||
# [40k, Space Marines, Raven Guard, Walker]
|
||||
primary = max(paths, key=len)
|
||||
system = primary[0] if len(primary) > 0 else None
|
||||
group = primary[1] if len(primary) > 1 else None
|
||||
faction = primary[2] if len(primary) > 2 else None
|
||||
# The LAST level is the class/category layer (GW uses shop
|
||||
# categories there: Battleline, Walker, Vehicle, "Rules &
|
||||
# Accessories", product lines like "Cadian"). Deeper sources
|
||||
# (miniset style) may carry a real subfaction before it.
|
||||
if len(primary) >= 5:
|
||||
subfaction = primary[3]
|
||||
klass = primary[4]
|
||||
elif len(primary) == 4:
|
||||
subfaction = None
|
||||
klass = primary[3]
|
||||
else:
|
||||
subfaction = None
|
||||
klass = None
|
||||
key = (hit.get("objectID"), tuple(primary))
|
||||
if key in seen:
|
||||
continue
|
||||
seen.add(key)
|
||||
entries.append(
|
||||
{
|
||||
"system": system,
|
||||
"group": group,
|
||||
"faction": faction,
|
||||
"subfaction": subfaction,
|
||||
"class": klass,
|
||||
"name": base["name"],
|
||||
"aliases": [],
|
||||
"sku": base["sku"],
|
||||
"slug": base["slug"],
|
||||
"productType": base["productType"],
|
||||
"source": "gwshop",
|
||||
"collections": base["collections"],
|
||||
}
|
||||
)
|
||||
return {
|
||||
"comment": (
|
||||
"Generated by dlphelper.gwshop from the Games Workshop webshop "
|
||||
"(Algolia index, deduplicated by product). Source-agnostic "
|
||||
"catalog schema: system/group/faction/subfaction/class/name/aliases."
|
||||
),
|
||||
"source": "gwshop",
|
||||
"generated_entries": len(entries),
|
||||
"entries": entries,
|
||||
}
|
||||
|
||||
|
||||
def main() -> None:
|
||||
logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Dump the GW webshop catalog via Algolia."
|
||||
)
|
||||
parser.add_argument("out", help="output catalog JSON path")
|
||||
parser.add_argument("--index", default=DEFAULT_INDEX)
|
||||
parser.add_argument("--raw", help="also write the raw Algolia hits here")
|
||||
args = parser.parse_args()
|
||||
|
||||
app_id, api_key = load_credentials()
|
||||
client = GwShopClient(app_id, api_key)
|
||||
hits = client.browse(args.index)
|
||||
log.info("%d raw hits from %s", len(hits), args.index)
|
||||
|
||||
if args.raw:
|
||||
Path(args.raw).write_text(
|
||||
json.dumps(hits, ensure_ascii=False, indent=1), encoding="utf-8"
|
||||
)
|
||||
log.info("raw hits written to %s", args.raw)
|
||||
|
||||
catalog = build_catalog(hits)
|
||||
Path(args.out).write_text(
|
||||
json.dumps(catalog, ensure_ascii=False, indent=1), encoding="utf-8"
|
||||
)
|
||||
log.info(
|
||||
"%d catalog entries written to %s",
|
||||
catalog["generated_entries"],
|
||||
args.out,
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,899 @@
|
||||
"""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"<!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_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",
|
||||
"weapon", "gun", "gunner", "turret", "carriage", "upgrade",
|
||||
}) # 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)
|
||||
@@ -0,0 +1,78 @@
|
||||
"""Persistent per-URL state: makes runs resumable and failures trackable.
|
||||
|
||||
state.json lives in the run's OUT_DIR and maps every URL seen to a record:
|
||||
|
||||
{
|
||||
"status": "ok" | "failed" | "error",
|
||||
"attempts": <int>,
|
||||
"error": "<last error message>",
|
||||
"files": [{"path": ..., "size": ...}, ...],
|
||||
"updated": "<iso timestamp>"
|
||||
}
|
||||
|
||||
Re-running skips URLs whose status is "ok"; failed URLs are retried and also
|
||||
written to failed_urls.txt for easy manual retry lists.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class State:
|
||||
"""JSON-backed state store for one run directory."""
|
||||
|
||||
def __init__(self, path: Path):
|
||||
self.path = path
|
||||
self.records: dict[str, dict] = {}
|
||||
if path.exists():
|
||||
try:
|
||||
self.records = json.loads(path.read_text(encoding="utf-8"))
|
||||
if not isinstance(self.records, dict):
|
||||
raise ValueError("state root must be an object")
|
||||
except (ValueError, OSError) as exc:
|
||||
backup = path.with_suffix(".json.bak")
|
||||
log.warning("Corrupt state file %s (%s); moving to %s and starting fresh",
|
||||
path, exc, backup)
|
||||
path.replace(backup)
|
||||
self.records = {}
|
||||
|
||||
def is_done(self, url: str) -> bool:
|
||||
rec = self.records.get(url)
|
||||
return bool(rec) and rec.get("status") == "ok"
|
||||
|
||||
def pending(self, urls: list[str]) -> list[str]:
|
||||
return [u for u in urls if not self.is_done(u)]
|
||||
|
||||
def mark(self, url: str, **fields) -> dict:
|
||||
rec = self.records.setdefault(url, {"status": "pending", "attempts": 0, "files": []})
|
||||
for key, value in fields.items():
|
||||
if value is not None:
|
||||
rec[key] = value
|
||||
rec["attempts"] = rec.get("attempts", 0) + 1
|
||||
rec["updated"] = time.strftime("%Y-%m-%dT%H:%M:%S")
|
||||
self.save()
|
||||
return rec
|
||||
|
||||
def failures(self) -> list[str]:
|
||||
return [u for u, r in self.records.items() if r.get("status") != "ok"]
|
||||
|
||||
def save(self) -> None:
|
||||
self.path.parent.mkdir(parents=True, exist_ok=True)
|
||||
tmp = self.path.with_suffix(".json.tmp")
|
||||
tmp.write_text(
|
||||
json.dumps(self.records, indent=2, sort_keys=True), encoding="utf-8"
|
||||
)
|
||||
tmp.replace(self.path)
|
||||
|
||||
|
||||
def append_manifest(manifest_path: Path, row: dict) -> None:
|
||||
"""Append one JSON line to manifest.jsonl (per downloaded file)."""
|
||||
manifest_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
with manifest_path.open("a", encoding="utf-8") as fh:
|
||||
fh.write(json.dumps(row, ensure_ascii=False) + "\n")
|
||||
@@ -0,0 +1,529 @@
|
||||
"""Warhammer taxonomy and unit keyword tables used by the catalogue.
|
||||
|
||||
Naming follows the Games Workshop webshop hierarchy (the same one
|
||||
dlphelper.gwshop dumps from Algolia):
|
||||
|
||||
<System>/<Group>/<Faction>/<Class> e.g.
|
||||
Warhammer 40,000/Space Marines/Ultramarines/Walker
|
||||
|
||||
Two layers:
|
||||
- SYSTEMS: game system -> faction -> keyword aliases (matched against file
|
||||
names). Faction detection is the primary signal; unit names double as
|
||||
faction hints since they are listed under their faction. Space Marine
|
||||
chapters are their own GW factions under the "Space Marines" group.
|
||||
- CLASSES: keyword -> class. With GW as the catalog source the class layer
|
||||
holds GW shop categories (unit types: Battleline, Infantry, Walker,
|
||||
Vehicle, Flyer, Monster, Mounted, Hero, Titanic).
|
||||
|
||||
The unit table grows with the collection; it does not attempt to enumerate
|
||||
every unit Games Workshop ever released - unknown files land in the
|
||||
"unclassified" bucket of the catalogue report so gaps become visible.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Game systems and factions (keyword aliases, lowercase)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
SYSTEMS: dict[str, dict[str, list[str]]] = {
|
||||
"Warhammer 40,000": {
|
||||
# -- Space Marines (chapters are their own GW factions) --------------
|
||||
"Space Marines": [
|
||||
"space marine", "astartes", "primaris", "intercessor", "reiver",
|
||||
"aggressor", "inceptor", "hellblaster", "eradicator", "bladeguard",
|
||||
"sternguard", "devastator", "tactical squad", "terminator squad",
|
||||
"assault marine", "assault intercessor", "outrider", "scout squad",
|
||||
"vanguard veteran", "invictor", "impulsor", "gladiator",
|
||||
"land speeder", "drop pod", "inner circle", "honour guard",
|
||||
],
|
||||
"Ultramarines": ["ultramarines", "guilliman", "calgar"],
|
||||
"Blood Angels": [
|
||||
"blood angel", "sanguinary guard", "sanguinor", "death company",
|
||||
"baal predator", "mephiston", "dante", "astorath", "flesh tearers",
|
||||
"lamberto", "furioso",
|
||||
],
|
||||
"Dark Angels": [
|
||||
"dark angel", "ravenwing", "deathwing", "inner circle companions",
|
||||
"azrael", "asmodai", "belial", "samuel", "lion el'jonson",
|
||||
"lion el jonson", "nephilim jetfighter",
|
||||
],
|
||||
"Space Wolves": [
|
||||
"space wolf", "space wolves", "wulfen", "thunderwolf", "wolf guard",
|
||||
"rune priest", "wolf priest", "wolf lord", "logan grimnar",
|
||||
"ragnar blackmane", "canis wolfborn", "fenrisian wolf",
|
||||
"stormfang", "murderfang", "lukas the trickster",
|
||||
],
|
||||
"Black Templars": [
|
||||
"black templar", "crusader squad", "high marshal helbrecht",
|
||||
"grimaldus", "the emperor's champion", "emperors champion",
|
||||
"castellan", "land raider crusader",
|
||||
],
|
||||
"Imperial Fists": [
|
||||
"imperial fist", "rogal dorn", "lupos", "tor garadon",
|
||||
"horatius hold", "dorn's arrow",
|
||||
],
|
||||
"Raven Guard": [
|
||||
"raven guard", "kayvaan shrike", "corvus", "sable brand",
|
||||
"shadow captains",
|
||||
],
|
||||
"Iron Hands": [
|
||||
"iron hand", "iron hands", "ferrus manus", "iron father",
|
||||
"medusan immortals",
|
||||
],
|
||||
"White Scars": ["white scar", "jaghatai khan", "stormseer", "korsos"],
|
||||
"Salamanders": [
|
||||
"salamander", "vulkan", "he'stan", "adrax agatone",
|
||||
"drake-scale", "flamecaster",
|
||||
],
|
||||
"Grey Knights": [
|
||||
"grey knight", "grand master", "brother-captain", "nemesis dreadknight",
|
||||
"draigo", "castellan crowe", "sternguard knight",
|
||||
],
|
||||
"Deathwatch": [
|
||||
"deathwatch", "watch captain", "watch master", "kill team cassius",
|
||||
"overkill", "corvus blackstar",
|
||||
],
|
||||
"Crimson Fists": ["crimson fist", "kantor", "rynn's world"],
|
||||
"Flesh Tearers": ["flesh tearer", "gabriel seth", "amissum"],
|
||||
# -- Imperium --------------------------------------------------------
|
||||
"Astra Militarum": [
|
||||
"astra militarum", "imperial guard", "cadian", "catachan", "krieg",
|
||||
"death korps", "tempestus", "scion", "valhallan", "steel legion",
|
||||
"tallarn", "vostroyan", "tanith", "gaunt's ghosts", "leman russ",
|
||||
"chimera", "basilisk", "hydra", "wyvern", "manticore", "deathstrike",
|
||||
"sentinel", "taurox", "commissar", "bullgryn", "ogryn", "ratling",
|
||||
"heavy weapon squad", "baneblade", "stormlord", "shadowsword",
|
||||
"hellhammer", "malcador", "macharius", "valkyrie", "vulture",
|
||||
"primaris psyker", "astropath", "priest",
|
||||
],
|
||||
"Adeptus Mechanicus": [
|
||||
"adeptus mechanicus", "mechanicum", "admech", "skitarii",
|
||||
"cult mechanicus", "sicarian", "ruststalker", "infiltrator",
|
||||
"kataphron", "kastelan", "ironstrider", "sydonian",
|
||||
"onager", "dunecrawler", "maniple", "archmagos", "magos", "dominus",
|
||||
"manipulus", "enginseer", "transuranic", "tech-priest", "cybernetica",
|
||||
"thallax", "castellax", "vorax", "ordinatus", "belisarius cawl",
|
||||
],
|
||||
"Adepta Sororitas": [
|
||||
"adepta sororitas", "sisters of battle", "sister of battle",
|
||||
"battle sister", "repentia", "retributor", "canoness", "dialogus",
|
||||
"hospitaller", "seraphim", "zephyrim", "paragon warsuit",
|
||||
"morvenn vahl", "junith", "immersa", "exorcist", "immolator",
|
||||
],
|
||||
"Adeptus Custodes": [
|
||||
"adeptus custodes", "custodes", "custodian", "allarus", "vertus praetor",
|
||||
"venatari", "aquilon", "sagittarum", "hetaeron", "trajann", "constantin valdor",
|
||||
"sisters of silence", "witchseeker", "prosecutor", "vigilator",
|
||||
"caladius", "coronus", "telemon", "achillus", "galatus", "pallas",
|
||||
],
|
||||
"Imperial Knights": [
|
||||
"imperial knight", "knight errant", "knight paladin", "knight crusader",
|
||||
"knight warden", "knight gallant", "knight lancer", "knight acheron",
|
||||
"knight castellan", "knight valiant", "knight preceptor", "questoris",
|
||||
"armiger", "warglaive", "helverin", "morialtis", "canis rex",
|
||||
"cerastus", "atropos", "seneschal",
|
||||
],
|
||||
"Imperial Agents": [
|
||||
"inquisition", "inquisitor", "astra telepathica", "witch hunter",
|
||||
"coteaz", "eisenhorn", "ravenor", "kyria draxus",
|
||||
"officio assassinorum", "culexus", "vindicare", "callidus", "eversor",
|
||||
"saint celestine", "living saint", "assassin",
|
||||
],
|
||||
# -- Chaos -----------------------------------------------------------
|
||||
"Chaos Space Marines": [
|
||||
"chaos space marine", "black legion", "iron warriors", "word bearers",
|
||||
"night lords", "night lord", "alpha legion", "chaos terminator", "chosen", "havoc",
|
||||
"raptor", "warp talon", "chaos cultist", "cultist", "daemon prince",
|
||||
"chaos lord", "chaos sorcerer", "abaddon", "kharn", "ahriman",
|
||||
"typhus", "fabius bile", "vashtorr", "defiler", "forgefiend",
|
||||
"maulerfiend", "helldrake", "helbrute", "chaos spawn",
|
||||
"obliterator", "chaos rhino", "chaos vindicator", "spartan assault",
|
||||
"mastodon", "typhon", "csm",
|
||||
],
|
||||
"Death Guard": [
|
||||
"death guard", "nurgle", "plague marine", "deathshroud", "mortarion",
|
||||
"plaguecaster", "poxwalker", "blightlord", "foetid bloat-drone",
|
||||
"plagueburst crawler", "myphitic blight-hauler", "blight hauler",
|
||||
"tallyman", "foul blightspawn", "biologus putrifier",
|
||||
"plague surgeon", "necrosius",
|
||||
],
|
||||
"Thousand Sons": [
|
||||
"thousand sons", "tzeentch", "rubric marine", "rubricae",
|
||||
"scarab occult", "magnus the red", "ahzek ahriman", "arhiman",
|
||||
"exalted sorcerer", "osiron", "mutalith vortex beast", "tzaangor",
|
||||
"prospero",
|
||||
],
|
||||
"Chaos Daemons": [
|
||||
"chaos daemon", "chaos daemons", "daemonette", "bloodletter",
|
||||
"plaguebearer", "horror", "screamer", "flesh hound", "juggernaut",
|
||||
"keeper of secrets", "skarbrand", "kairos fateweaver", "rotigus",
|
||||
"be'lakor", "great unclean one", "lord of change", "soul grinder",
|
||||
"skull cannon", "burning chariot", "exalted flamer",
|
||||
"fiend of slaanesh", "plague drone", "nurgling", "herald",
|
||||
"skulltaker", "khorne", "slaanesh", "skull lord",
|
||||
"shalaxi helbane",
|
||||
],
|
||||
"World Eaters": [
|
||||
"world eater", "angron", "eightbound", "exalted eightbound",
|
||||
"jakhals", "berserker", "skull lord",
|
||||
],
|
||||
"Emperor's Children": [
|
||||
"emperor's children", "emperors children", "fulgrim",
|
||||
"noise marine", "lucius the eternal", "flawless blades",
|
||||
],
|
||||
"Iron Warriors": ["iron warrior", "perturabo"],
|
||||
"Red Corsairs": ["red corsair", "huron blackheart"],
|
||||
"Chaos Knights": [
|
||||
"chaos knight", "knight desecrator", "knight despoiler",
|
||||
"knight rampager", "kharibdyss", "renegade knight",
|
||||
],
|
||||
# -- Xenos -----------------------------------------------------------
|
||||
"Orks": [
|
||||
"ork", "orks", "orkz", "nob", "boyz", "grot", "weirdboy", "painboy",
|
||||
"warboss", "ghazghkull", "meganob", "deff dread", "killa kan",
|
||||
"deffkopta", "stormboy", "kommando", "burna boy", "loota",
|
||||
"trukk", "battlewagon", "gorkanaut", "morkanaut", "stompa", "squig",
|
||||
"zagstruk", "snikrot", "speed freek", "boomdakka", "flash gitz",
|
||||
"wazbom blasta", "dakka jet", "burna-bomber", "blitza-bomber",
|
||||
"da troopz", "dad and sonz", "ass sassins", "abby boi",
|
||||
],
|
||||
"T'au Empire": [
|
||||
"tau", "t'au", "ta'u", "kroot", "vespid", "battlesuit", "crisis suit",
|
||||
"broadside", "r'varna", "ghostkeel", "stormsurge",
|
||||
"fire warrior", "pathfinder team", "devilfish", "hammerhead",
|
||||
"sky ray", "piranha", "sniper drone team", "ethereal",
|
||||
"commander farsight", "commander shadowsun", "aun'va", "darkstrider",
|
||||
"longstrike", "mont'ka", "assorted drones", "breacher",
|
||||
"bomber", "tetra", "barracuda", "sun shark", "razorshark",
|
||||
"kv128 stormsurge", "tx4 piranha", "kv46",
|
||||
],
|
||||
"Tyranids": [
|
||||
"tyranid", "hive tyrant", "swarmlord", "carnifex", "hormagaunt",
|
||||
"termagant", "termagaunt", "genestealer brood", "zoanthrope", "venomthrope",
|
||||
"exocrine", "haruspex", "tyranid prime", "tyranid warrior",
|
||||
"ravener", "gargoyle", "spore mine", "mawloc", "trygon",
|
||||
"harridan", "harpy", "crone", "tervigon", "barbed hierodule",
|
||||
"hierodule", "hierophant", "neurothrope", "deathleaper", "lictor", "pyrovore",
|
||||
"biovore", "broodlord", "bio-titan", "dominatrix",
|
||||
],
|
||||
"Genestealer Cults": [
|
||||
"genestealer cult", "genestealer", "aberrant", "acolyte", "neophyte hybrid",
|
||||
"primus", "magus", "patriarch", "purestrain genestealer",
|
||||
"sanctus", "kelermorph", "goliath truck", "rockgrinder",
|
||||
"jackal alcaeus", "cult limo", "hijacked",
|
||||
],
|
||||
"Necrons": [
|
||||
"necron", "c'tan", "canoptek", "wraith", "spyder", "scarab",
|
||||
"tesseract vault", "doomsday ark", "annihilation barge",
|
||||
"ghost ark", "monolith", "doom scythe", "night scythe",
|
||||
"imotekh", "szarekh", "nemesor zahndrekh", "vargard obyron",
|
||||
"lokhrust", "skorpekh", "flayed one", "deathmark", "lychguard",
|
||||
"necron warrior", "necron immortal", "ovoid", "dzaan",
|
||||
],
|
||||
"Aeldari": [
|
||||
"aeldari", "eldar", "craftworld", "ulthwe", "biel-tan", "iyanden",
|
||||
"alaitoc", "saim-hann", "farseer", "warlock",
|
||||
"spiritseer", "autarch", "asurmen", "guardian defender",
|
||||
"ranger", "wraithguard", "wraithblades", "wraithlord",
|
||||
"wraithknight", "dire avenger", "howling banshee",
|
||||
"striking scorpion", "swooping hawk", "warp spider",
|
||||
"fire dragon", "dark reaper", "shining spears", "windrider",
|
||||
"vyper", "wave serpent", "fire prism", "night spinner",
|
||||
"war walker", "hemlock wraithfighter", "crimson hunter",
|
||||
"shade runner", "skyrunner",
|
||||
],
|
||||
"Drukhari": [
|
||||
"drukhari", "dark eldar", "kabal", "wych", "wrack", "haemonculus",
|
||||
"coven", "reaver jetbike", "hellion", "scourge", "mandrake",
|
||||
"kabalite warrior", "raider", "ravager", "venom", "razorwing",
|
||||
"voidraven", "grotesque", "talos", "cronos",
|
||||
"beast pack", "khymera", "razorwing flock", "asdrubael vect",
|
||||
"lilith", "sathonyx", "luriel",
|
||||
],
|
||||
"Harlequins": [
|
||||
"harlequin", "death jester", "shadowseer", "solitaire",
|
||||
"troupe", "starweaver", "voidweaver",
|
||||
],
|
||||
"Leagues of Votann": [
|
||||
"leagues of votann", "votann", "hearthkyn", "ironkin",
|
||||
"brokhyr", "kahl", "cthonian", "transhyperic", "thegn",
|
||||
],
|
||||
},
|
||||
"Age of Sigmar": {
|
||||
"Stormcast Eternals": [
|
||||
"stormcast", "eternal", "liberator", "sequitor", "evocator",
|
||||
"vindictor", "judicators", "retributors", "sacrosanct",
|
||||
"dracothian", "lord-arcane", "knight-questor", "neave blacktalon",
|
||||
"stormdrake", "krondys", "karazai",
|
||||
],
|
||||
"Seraphon": ["seraphon", "saurus", "skink", "slann", "carnosaur",
|
||||
"bastiladon", "stegadon", "razordon", "salamander"],
|
||||
"Orruk Warclans": ["orruk", "warclan", "kruleboyz", "ironjawz",
|
||||
"brute", "gorger", "weirdnob", "warchanter"],
|
||||
"Gloomspite Gitz": ["gloomspite", "gitz", "moonclan", "spiderfang",
|
||||
"troggoth", "loontouched", "zarbag", "skragrot"],
|
||||
"Sylvaneth": ["sylvaneth", "treelord", "dryad", "branchwraith",
|
||||
"alarielle", "kurnothi"],
|
||||
"Ossiarch Bonereapers": ["ossiarch", "bonereaper", "mortek", "keldrek",
|
||||
"katakros"],
|
||||
"Nighthaunt": ["nighthaunt", "grimghast", "bladegheist", "chainghast",
|
||||
"kurdoss", "lady olynder"],
|
||||
"Lumineth Realm-lords": ["lumineth", "aelf", "vanari", "teclis",
|
||||
"light of eltharion"],
|
||||
"Daughters of Khaine": ["daughters of khaine", "khainite", "witch aelf",
|
||||
"morgwaeth", "morathi"],
|
||||
"Fyreslayers": ["fyreslayer", "vulkite", "hearthguard", "auric",
|
||||
"grimhold"],
|
||||
"Kharadron Overlords": ["kharadron", "skyfarer", "endofters",
|
||||
"grundstok", "arcanaut"],
|
||||
"Idoneth Deepkin": ["idoneth", "deepkin", "namarti", "isharann",
|
||||
"volturnos"],
|
||||
"Blades of Khorne": ["blades of khorne", "khorne", "skullgrinder",
|
||||
"blood warrior", "wrathmonger"],
|
||||
"Maggotkin of Nurgle": ["maggotkin", "nurgle", "blightking",
|
||||
"rotmire", "orghotts"],
|
||||
"Slaves to Darkness": ["slaves to darkness", "chaos warrior",
|
||||
"chaos knight", "archaon", "gaunt summoner",
|
||||
"everchosen"],
|
||||
"Skaven": ["skaven", "clanrat", "stormvermin", "verminlord",
|
||||
"grey seer", "warplock", "ratling gun"],
|
||||
"Soulblight Gravelords": ["soulblight", "gravelord", "vampire",
|
||||
"deathrattle", "koldhunter", "belladamma"],
|
||||
"Cities of Sigmar": ["cities of sigmar", "freeguild", "windspeaker",
|
||||
"callis and toll", "hammerers"],
|
||||
},
|
||||
"The Horus Heresy": {
|
||||
"Space Marine Legions": [
|
||||
"legiones astartes", "pre-heresy", "horus heresy", "justaerin",
|
||||
"seeker squad", "suzerain", "invictarus", "veteran tactical",
|
||||
"contemptor", "leviathan pattern", "sicaran", "fellblade",
|
||||
"kratos", "fire raptor", "eisenstein",
|
||||
],
|
||||
"Mechanicum": ["mechanicum", "fulgurite", "magos", "thallax"],
|
||||
},
|
||||
"Other Games": {
|
||||
"Kill Team": [
|
||||
"kill team", "killteam", "killzone", "operative",
|
||||
"hand of the archon", "void-dance troupe", "veteran guardsmen",
|
||||
"spector squad", "wolf scouts", "scout squad kill",
|
||||
],
|
||||
"Necromunda": [
|
||||
"necromunda", "goliath gang", "escher", "orlock", "cawdor",
|
||||
"van saar", "ash wastes", "underhive", "sumpkrocs",
|
||||
"forge boss", "sumpkroc", "matriarch", "phyrr cat",
|
||||
],
|
||||
"Adeptus Titanicus": [
|
||||
"adepticus titanicus", "titan legio", "reaver titan",
|
||||
"warhound titan", "warlord titan", "imperator titan",
|
||||
"warmaster titan", "titanicus",
|
||||
],
|
||||
"Warhammer Underworlds": ["warhammer underworlds", "underworlds",
|
||||
"warband", "shadeborn", "embergard"],
|
||||
"Aeronautica Imperialis": ["aeronautica", "t'au air caste",
|
||||
"lightning fighter", "thunderbolt"],
|
||||
},
|
||||
"The Old World": {
|
||||
"Kingdom of Bretonnia": ["bretonnia", "grail knight", "peasant",
|
||||
"questing knight", "perrin"],
|
||||
"Orc and Goblin Tribes": ["greenskin", "goblin", "night goblin",
|
||||
"forest goblin", "black orc"],
|
||||
"Tomb Kings of Khemri": ["tomb king", "khemri", "necrolith",
|
||||
"skeleton archer"],
|
||||
"Empire of Man": ["empire of man", "state troop", "greatsword",
|
||||
"empire knight"],
|
||||
"Warriors of Chaos": ["chaos dwarf", "dog of war", "chaos marauder"],
|
||||
"High Elf Realms": ["high elf", "eltharion", "phoenix guard"],
|
||||
"Wood Elf Realms": ["wood elf", "glade guard", "waywatcher"],
|
||||
"Beastmen Brayherds": ["beastmen", "brayherd", "ghorgon", "cygor"],
|
||||
"Dwarfen Mountain Holds": ["dwarf", "ironbreaker", "hammerer"],
|
||||
},
|
||||
"Middle-Earth": {
|
||||
"Rivendell": ["rivendell", "elrond", "elendil", "isildur", "gil-galad",
|
||||
"beorning", "grimbeorn", "rumil", "orophin"],
|
||||
"Minas Tirith": ["minas tirith", "gondor", "boromir", "faramir"],
|
||||
"Mordor": ["mordor", "orc", "orcs", "nazgul", "witch-king"],
|
||||
"Iron Hills": ["iron hills", "dwarf", "mattock", "dwalin"],
|
||||
"Moria": ["moria", "cave troll", "balrog"],
|
||||
},
|
||||
}
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Classes (keyword -> class). GW shop categories / unit types.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
CLASSES: list[tuple[str, list[str]]] = [
|
||||
("Titanic", [
|
||||
"titan", "stompa", "gargantuan", "hierophant", "harridan", "hierodule",
|
||||
"imperator", "warmaster", "tesseract vault", "plaguereaper",
|
||||
"warhound", "warlord titan", "ta'unar", "spermato", "titandeath",
|
||||
]),
|
||||
("Fortification", ["fortification", "bunker", "bastion", "wall of martyrs"]),
|
||||
("Flyer", [
|
||||
"flyer", "heldrake", "dakka jet", "burna-bomber", "blitza-bomber",
|
||||
"bomber", "sky ray", "stormraven", "storm talon", "stormhawk",
|
||||
"caestus", "valkyrie", "vulture", "thunderbolt", "lightning",
|
||||
"marauder", "razorwing", "voidraven", "crimson hunter",
|
||||
"hemlock", "doom scythe", "night scythe", "harpy", "crone",
|
||||
"fire raptor", "barracuda", "sun shark", "razorshark",
|
||||
"wazbom blasta", "nephilim jetfighter", "corvus blackstar",
|
||||
]),
|
||||
("Vehicle", [
|
||||
"transport", "rhino", "razorback", "chimera", "taurox", "trukk",
|
||||
"battlewagon", "devilfish", "wave serpent", "ghost ark",
|
||||
"land raider", "repulsor", "drop pod", "starweaver", "voidweaver",
|
||||
"goliath truck", "rockgrinder", "cult limo", "raider", "venom",
|
||||
"spartan assault", "predator", "vindicator", "whirlwind",
|
||||
"basilisk", "hydra", "wyvern", "manticore", "deathstrike",
|
||||
"onager dunecrawler", "kastelan", "hammerhead", "broadside",
|
||||
"stormsurge", "r'varna", "soul grinder", "defiler", "forgefiend",
|
||||
"maulerfiend", "plagueburst crawler", "annihilation barge",
|
||||
"doomsday ark", "fire prism", "night spinner", "exorcist",
|
||||
"immolator", "caladius", "coronus grav", "pallas grav",
|
||||
"impulsor", "gladiator", "baal predator", "land speeder",
|
||||
"ironstrider", "sydonian", "sentinel", "taurox", "kataphron",
|
||||
]),
|
||||
("Walker", [
|
||||
"dreadnought", "redemptor", "contemptor", "leviathan pattern",
|
||||
"brutalis", "ballistus", "deredeo", "wulfen dreadnought",
|
||||
"killa kan", "deff dread", "wraithlord", "wraithknight",
|
||||
"telemon", "achillus", "galatus", "paragon warsuit",
|
||||
"nemesis dreadknight", "invictor", "talos", "cronos",
|
||||
]),
|
||||
("Monster", [
|
||||
"monster", "carnifex", "exocrine", "haruspex", "mawloc", "trygon",
|
||||
"tervigon", "hierodule", "hierophant", "mutalith", "bloat-drone",
|
||||
"kharibdyss", "gorkanaut", "morkanaut", "magnus", "mortalith",
|
||||
"fiend of slaanesh", "plague drone", "cave troll", "balrog",
|
||||
"carnosaur", "stegadon", "bastiladon", "grorgeous",
|
||||
]),
|
||||
("Mounted", [
|
||||
"mounted", "thunderwolf", "outrider", "bike squad", "scout biker",
|
||||
"reaver jetbike", "windrider", "skyrunner", "warbiker",
|
||||
"allarus custodian on", "vertus praetor", "drakeling",
|
||||
"boarboy", "gruntaz", "khan on", "dracothian guard",
|
||||
]),
|
||||
("Battleline", [
|
||||
"battleline", "troops", "tactical squad", "intercessor",
|
||||
"incursor", "infiltrator", "infantry squad", "conscript",
|
||||
"krieg", "catachan", "cadian", "vostroyan", "penal legion",
|
||||
"ratling", "veteran squad", "fire warrior", "pathfinder team",
|
||||
"kroot", "termagant", "hormagaunt", "necron warrior",
|
||||
"necron immortal", "guardian defender", "storm guardian",
|
||||
"kabalite warrior", "boyz", "gretchin", "scion", "scout squad",
|
||||
"cultist", "plague marine squad",
|
||||
]),
|
||||
("Infantry", [
|
||||
"infantry", "veteran", "terminator", "sternguard", "bladeguard",
|
||||
"aggressor", "reiver", "wulfen", "wolf guard", "deathshroud",
|
||||
"blightlord", "poxwalker", "rubric marine", "rubricae",
|
||||
"scarab occult", "tzaangor", "chosen", "possessed", "obliterator",
|
||||
"daemonette", "bloodletter", "plaguebearer", "nurgling",
|
||||
"juggernaut", "chaos cultist", "acolyte hybrid", "aberrant",
|
||||
"neophyte", "purestrain", "genestealer", "lictor", "deathmark",
|
||||
"flayed one", "lychguard", "wraithguard", "wraithblades",
|
||||
"howling banshee", "striking scorpion", "warp spider",
|
||||
"fire dragon", "incubi", "wrack", "grotesque", "assault marine",
|
||||
"assault intercessor", "inceptor", "raptor", "warp talon",
|
||||
"flesh hound", "screamer", "hellion", "scourge", "mandrake",
|
||||
"swooping hawk", "shining spears", "canoptek wraith",
|
||||
"canoptek scarab", "canoptek spyder", "ravener", "gargoyle",
|
||||
"spore mine", "pyrovore", "khymera", "beast pack",
|
||||
"razorwing flock", "zephyrim", "seraphim", "stormboy",
|
||||
"deffkopta", "skorpekh destroyer", "skorpekh", "retributor",
|
||||
"dark reaper", "sicarian", "ruststalker", "skitarii",
|
||||
"hormagaunt brood", "termagant brood", "vanguard",
|
||||
]),
|
||||
("Hero", [
|
||||
"hero", "hq", "captain", "chapter master", "primarch", "lieutenant",
|
||||
"librarian", "chaplain", "apothecary", "farseer", "warlock",
|
||||
"spiritseer", "autarch", "warboss", "big mek", "weirdboy",
|
||||
"painboy", "commander", "ethereal", "hive tyrant", "swarmlord",
|
||||
"tyranid prime", "neurothrope", "deathleaper", "patriarch",
|
||||
"primus", "magus", "sanctus", "kelermorph", "overlord",
|
||||
"imotekh", "szarekh", "zahndrekh", "obyron", "plaguecaster",
|
||||
"necrosius", "chaos lord", "chaos sorcerer", "daemon prince",
|
||||
"exalted champion", "herald", "skulltaker", "kairos", "rotigus",
|
||||
"be'lakor", "great unclean one", "lord of change",
|
||||
"keeper of secrets", "skarbrand", "kharn", "abaddon", "typhus",
|
||||
"mortarion", "magnus", "ahriman", "lucius", "fabius bile",
|
||||
"vashtorr", "canoness", "morvenn", "junith", "immersa",
|
||||
"dialogus", "hospitaller", "trajann", "valdor", "inquisitor",
|
||||
"coteaz", "eisenhorn", "ravenor", "draxus", "celestine",
|
||||
"commissar", "priest", "enginseer", "dominus", "manipulus",
|
||||
"archmagos", "belisarius cawl", "farsight", "shadowsun",
|
||||
"aun'va", "darkstrider", "longstrike", "shade runner",
|
||||
]),
|
||||
("Rules & Accessories", [
|
||||
"codex", "rulebook", "rules", "data cards", "datacards",
|
||||
"warscroll", "index card", "cheat sheet", "reference",
|
||||
]),
|
||||
]
|
||||
|
||||
# Flattened per-system faction keyword index for quick matching
|
||||
FACTION_ALIASES: dict[str, dict[str, str]] = {
|
||||
system: {alias: faction for faction, aliases in factions.items() for alias in aliases}
|
||||
for system, factions in SYSTEMS.items()
|
||||
}
|
||||
|
||||
UNIT_CLASSES: dict[str, str] = {kw: klass for klass, kws in CLASSES for kw in kws}
|
||||
|
||||
# GW group of each faction (matches the GameSystemsRoot.lvl1 values in the
|
||||
# Algolia index). Used when a keyword-classified file cannot be fuzzy-matched.
|
||||
GROUP_OF_FACTION: dict[str, str] = {
|
||||
faction: group
|
||||
for group, factions in {
|
||||
"Space Marines": [
|
||||
"Space Marines", "Ultramarines", "Blood Angels", "Dark Angels",
|
||||
"Space Wolves", "Black Templars", "Imperial Fists", "Raven Guard",
|
||||
"Iron Hands", "White Scars", "Salamanders", "Grey Knights",
|
||||
"Deathwatch", "Crimson Fists", "Flesh Tearers",
|
||||
],
|
||||
"Armies of the Imperium": [
|
||||
"Astra Militarum", "Adeptus Mechanicus", "Adepta Sororitas",
|
||||
"Adeptus Custodes", "Imperial Knights", "Imperial Agents",
|
||||
],
|
||||
"Armies of Chaos": [
|
||||
"Chaos Space Marines", "Death Guard", "Thousand Sons",
|
||||
"Chaos Daemons", "Chaos Knights", "World Eaters",
|
||||
"Emperor's Children", "Iron Warriors", "Red Corsairs",
|
||||
],
|
||||
"Xenos Armies": [
|
||||
"Orks", "T'au Empire", "Tyranids", "Genestealer Cults", "Necrons",
|
||||
"Aeldari", "Drukhari", "Harlequins", "Leagues of Votann",
|
||||
],
|
||||
"Grand Alliance Order": [
|
||||
"Stormcast Eternals", "Seraphon", "Sylvaneth",
|
||||
"Lumineth Realm-lords", "Daughters of Khaine", "Fyreslayers",
|
||||
"Kharadron Overlords", "Idoneth Deepkin", "Cities of Sigmar",
|
||||
],
|
||||
"Grand Alliance Destruction": [
|
||||
"Orruk Warclans", "Gloomspite Gitz",
|
||||
],
|
||||
"Grand Alliance Death": [
|
||||
"Ossiarch Bonereapers", "Nighthaunt", "Soulblight Gravelords",
|
||||
],
|
||||
"Grand Alliance Chaos": [
|
||||
"Slaves to Darkness", "Blades of Khorne", "Maggotkin of Nurgle",
|
||||
"Skaven",
|
||||
],
|
||||
"Space Marine Legions": ["Space Marine Legions", "Mechanicum"],
|
||||
"Kill Team": ["Kill Team"],
|
||||
"Necromunda": ["Necromunda"],
|
||||
"Adeptus Titanicus": ["Adeptus Titanicus"],
|
||||
"Warhammer Underworlds": ["Warhammer Underworlds"],
|
||||
"Aeronautica Imperialis": ["Aeronautica Imperialis"],
|
||||
"Armies of the Old World": [
|
||||
"Kingdom of Bretonnia", "Orc and Goblin Tribes",
|
||||
"Tomb Kings of Khemri", "Empire of Man", "Warriors of Chaos",
|
||||
"High Elf Realms", "Wood Elf Realms", "Beastmen Brayherds",
|
||||
"Dwarfen Mountain Holds",
|
||||
],
|
||||
"Rivendell": ["Rivendell"],
|
||||
"Minas Tirith": ["Minas Tirith"],
|
||||
"Mordor": ["Mordor"],
|
||||
"Iron Hills": ["Iron Hills"],
|
||||
"Moria": ["Moria"],
|
||||
}.items()
|
||||
for faction in factions
|
||||
}
|
||||
@@ -0,0 +1,92 @@
|
||||
"""Downloader for 0807.st direct links via curl.
|
||||
|
||||
0807.st serves the files directly but rejects requests that don't look like a
|
||||
browser (requires a User-Agent and the Referer of the site page), and it
|
||||
rejects multi-connection/range abuse patterns differently than curl - which is
|
||||
why curl is used instead of aria2c here (see chat history: curl worked,
|
||||
aria2c's default request got 403).
|
||||
|
||||
Downloads go to a .part file first and are renamed on success, so an
|
||||
interrupted download is resumable (--continue-at -) and partial files are
|
||||
never mistaken for finished ones.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import os
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
from urllib.parse import urlparse
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
USER_AGENT = "Mozilla/5.0"
|
||||
REFERER = "https://0807.st/"
|
||||
|
||||
|
||||
class Zero807Error(Exception):
|
||||
"""Base error for 0807 downloads."""
|
||||
|
||||
|
||||
class Zero807NotFound(Zero807Error):
|
||||
"""The file no longer exists (HTTP 404/410). Not retryable."""
|
||||
|
||||
|
||||
def target_name(url: str) -> str:
|
||||
"""Derive the filename from the URL path (e.g. /6saCNie.zip -> 6saCNie.zip)."""
|
||||
name = Path(urlparse(url).path).name
|
||||
if not name:
|
||||
raise Zero807Error(f"cannot derive a filename from URL: {url}")
|
||||
return name
|
||||
|
||||
|
||||
def download(
|
||||
url: str,
|
||||
out_dir: Path,
|
||||
timeout_seconds: int = 7200,
|
||||
resume: bool = True,
|
||||
) -> tuple[Path, int]:
|
||||
"""Download one URL with curl; returns (final_path, size_bytes)."""
|
||||
name = target_name(url)
|
||||
out_dir.mkdir(parents=True, exist_ok=True)
|
||||
dest = out_dir / name
|
||||
tmp = dest.with_name(dest.name + ".part")
|
||||
|
||||
if dest.exists() and dest.stat().st_size > 0:
|
||||
log.info("Skip existing: %s", dest)
|
||||
return dest, dest.stat().st_size
|
||||
|
||||
cmd = [
|
||||
"curl", "-fSL",
|
||||
"--connect-timeout", "20",
|
||||
"--max-time", str(timeout_seconds),
|
||||
"--retry", "2", "--retry-delay", "5",
|
||||
"-A", USER_AGENT,
|
||||
"-e", REFERER,
|
||||
"-o", str(tmp),
|
||||
]
|
||||
if resume:
|
||||
cmd.append("--continue-at")
|
||||
cmd.append("-")
|
||||
cmd.append(url)
|
||||
|
||||
log.info("Downloading %s -> %s", url, dest.name)
|
||||
proc = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
|
||||
encoding="utf-8", errors="replace")
|
||||
if proc.returncode != 0:
|
||||
tail = [line for line in proc.stderr.strip().splitlines() if line][-3:]
|
||||
stderr = " | ".join(tail)
|
||||
# -f makes curl exit 22 on any HTTP error; dead links fail for good
|
||||
if "404" in stderr or "410" in stderr:
|
||||
raise Zero807NotFound(f"dead link (HTTP 404/410): {stderr}")
|
||||
raise Zero807Error(f"curl exit code {proc.returncode}: {stderr}")
|
||||
|
||||
size = tmp.stat().st_size if tmp.exists() else 0
|
||||
if size == 0:
|
||||
tmp.unlink(missing_ok=True)
|
||||
raise Zero807Error(f"empty file for {name}")
|
||||
if dest.suffix.lower() == ".zip" and not tmp.read_bytes()[:2] == b"PK":
|
||||
log.warning("%s does not look like a ZIP file (first bytes are not PK)", name)
|
||||
os.replace(tmp, dest)
|
||||
return dest, size
|
||||
Reference in New Issue
Block a user