mem safe & format

2026-09-19 14:48:59 +02:00
1 parent 2cc8e1d788
commit 8f92e64ffd
@@ -25,7 +25,13 @@ pip install -e .
 Every command streams input and writes incrementally. Peak memory does not grow
 with file size: a single 100 GB+ file is parsed with a small constant footprint.
-`sort` and `clean` shell out to the OS `sort -u`, which spills to disk.
+`sort` shells out to the OS `sort -u`, which spills to disk.
 - `crack` builds the hash index once and shares it with the workers by forking
  (copy-on-write) — one copy, not one per worker (verified: ~450 MB total for a
  1M-hash index across 4 workers).
 - `query` streams wildcard matches line by line from ripgrep, so a broad pattern
  over a huge dataset stays bounded.
 Measured on real Collections 2-5 data (19.94 GB, three files): **160 MB peak RSS**
 for the whole process tree, versus **15.4 GB** for the original parser. Min RAM
@@ -155,13 +161,8 @@ Run `collectionparser --help` or `collectionparser <command> --help`.
 | `crack` | Dictionary-attack collected hashes with a wordlist. |
 | `parse` | Parse an unstructured folder into `success`/`failure`/`hashes` files. |
 | `split` | Convert a large sorted file into a query-friendly `a/b/c` folder. |
 | `chunk` | Split large TXT files into smaller ones. |
 | `sort` | Sort and deduplicate every file in a query-friendly folder. |
-| `clean` | Move misplaced records to the right bucket and sort. |
+| `merge` | Merge one query-friendly folder into another, then sort/dedup. |
 | `merge` | Merge one query-friendly folder into another. |
 | `test` | Look up passwords for the emails in a file. |
 | `evaluate` | Measure hit rate of one dataset against another. |
 | `filter-lines` | Keep lines whose length is within a range. |
 ## Development
@@ -57,11 +57,8 @@ sort -T tmp -u -S 90% --parallel=12 success.txt > success_sorted.txt
 # 3. Sorted unique file -> query-friendly a/b/c tree.
 collectionparser split --file success_sorted.txt --out data
-# 4. Optional repair pass (moves records to the right bucket, re-sorts).
+# 4. Query.
-collectionparser clean --path data
+collectionparser query --dataset data --pattern '[email protected]'
 # 5. Query.
 collectionparser test --file emails.txt --dataset data
 ```
 To add another dataset to an existing index:
@@ -79,15 +76,11 @@ collectionparser merge --src other_data --dest big_dataset
 | `query` | Exact or wildcard (`*`, `?`) search over a dataset. | `--dataset`, `--pattern` |
 | `export` | Export a dataset as JSON Lines (OpenSearch). | `--dataset`, `--out` |
 | `crack` | Dictionary-attack collected MD5/SHA1/SHA256 hashes. | `--hashes`, `--wordlist`, `--out`, `--workers` |
-| `parse` | Folder of raw files -> `success`/`failure` files. Streams. | `--path`, `--success_file`, `--failure_file` |
+| `parse` | Folder of raw files -> `success`/`failure`/`hashes` files. Streams. | `--path`, `--success_file`, `--failure_file`, `--hashes_file` |
 | `split` | One sorted file -> `a/b/c` tree. | `--file`, `--out`, `--restart_from N` |
 | `chunk` | Split oversized TXT files. | `--path`, `--size MB` (default 50) |
 | `sort` | `sort -u` every file in place. | `--path` |
-| `clean` | Move misplaced records, then sort each file. | `--path` |
+| `merge` | Merge `--src` into `--dest` (destructive), then sort/dedup. | `--src`, `--dest` |
-| `merge` | Merge `--src` into `--dest` (destructive). Runs `clean` after. | `--src`, `--dest` |
+| `status` | JSON build/dataset progress. | `--scratch`, `--pretty` |
 | `test` | Print passwords found for the emails in a file. | `--file`, `--dataset` |
 | `evaluate` | Hit-rate of one index against another. | `--old`, `--new` |
 | `filter-lines` | Keep lines within a length range. | `--file`, `--out`, `--less_than`, `--more_than` |
 ## Rules an agent must follow
@@ -106,8 +99,8 @@ collectionparser merge --src other_data --dest big_dataset
 - **`merge` is destructive**: it appends into `--dest` and deletes each `--src`
  file. Back up `--dest` before running.
 - **`--dataset` is a path** to a query-friendly folder built by `split`.
- **`query`/`test`/`evaluate` need `rg` on `PATH`.** If `rg` is missing, queries
+- **`query` needs `rg` on `PATH`.** If `rg` is missing, queries return nothing;
-  return nothing; install ripgrep instead of working around it.
+  install ripgrep instead of working around it.
 - **`query` wildcards are anchored to the email field**: `*`/`?` never cross the
  `:` separator (`*@gmail.com` works, matching into the password does not).
 - **`parse` restarts safely.** Completed inputs are recorded in `progress.txt`
@@ -126,7 +119,8 @@ collectionparser merge --src other_data --dest big_dataset
 collectionparser --help
 # round trip on toy data (~1s)
 printf '[email protected]:pw\[email protected]:pw\n' > /tmp/raw.txt
-collectionparser filter-lines --file /tmp/raw.txt --out /tmp/o.txt --less_than 1 --more_than 50
+collectionparser split --file /tmp/raw.txt --out /tmp/tree
 test -f /tmp/tree/a/symbols && echo OK
 ```
 If that works, the package and its entry point are installed correctly.
@@ -1,17 +1,14 @@
-"""Implementations behind the CLI commands: parse, split, sort, merge, query."""
+"""Implementations behind the CLI commands: parse, split, sort, merge, export."""
 import json
 import os
 import random
 import shutil
 import subprocess
 from pathlib import Path
 from tqdm import tqdm
-from collectionparser.query import QueryAPI
+from collectionparser.utils import create_dir_for_file, find_files
 from collectionparser.splitter import get_output_file, write_line_to_file
 from collectionparser.utils import create_dir_for_file, filter_printable, shuffle
 # ponytail: buckets under this size sort byte-exactly in RAM; larger buckets
 # shell out to `sort -u`, which spills to disk.
@@ -21,137 +18,51 @@ _IN_MEMORY_SORT_LIMIT = 8 * 1024 * 1024
 class API:
    """Command implementations shared by the CLI."""
    @staticmethod
    def evaluate(old: str, new: str) -> None:
        """Report the hit rate of ``new`` against a sample of ``old``."""
        hit_count = total_count = 0
        files = [p for p in Path(old).expanduser().glob('**/*') if p.is_file()]
        for _ in range(1000):
            # reservoir sample 10 lines so arbitrarily large files stay O(1) in RAM.
            random_lines: list[str] = []
            with open(random.choice(files), encoding='utf8', errors='ignore') as reader:
                for count, line in enumerate(reader):
                    if count < 10:
                        random_lines.append(line.rstrip('\n'))
                    elif (j := random.randint(0, count)) < 10:
                        random_lines[j] = line.rstrip('\n')
            for random_line in random_lines:
                password_list = QueryAPI.query(new, random_line.partition(':')[0])
                total_count += 1
                hit = 'hit' if password_list else 'miss'
                hit_count += bool(password_list)
                print(f'OLD: {random_line}, NEW: {hit}, HIT_RATE = {hit_count / total_count:.3f}.')
    @staticmethod
    def merge(src: str, dest: str) -> None:
        """Merge the bucket tree ``src`` into ``dest`` (destructive)."""
        src_root = Path(src).expanduser()
        dest_root = Path(dest).expanduser()
-        dest_files = {str(f).replace(str(dest_root), ''): f for f in find_files(dest_root)}
+        dest_files = {
            str(f).replace(str(dest_root), ""): f for f in find_files(dest_root)
        }
        for src_file in find_files(src_root):
-            src_index = str(src_file).replace(str(src_root), '')
+            src_index = str(src_file).replace(str(src_root), "")
            if src_index in dest_files:
                append_file(src_file, dest_files[src_index])
            else:
-                dst = dest_root / src_index.lstrip('/')
+                dst = dest_root / src_index.lstrip("/")
                create_dir_for_file(str(dst))
                move_file(src_file, dst)
    @staticmethod
    def sort(path: str) -> None:
        """Sort and de-duplicate every file under ``path`` in place."""
-        files = [p for p in Path(path).expanduser().glob('**/*')
+        files = [p for p in find_files(path) if p.suffix != ".sorted"]
-                 if p.is_file() and p.suffix != '.sorted']
+        for file in tqdm(files, desc="Unique > Sorting > Replacing existing files"):
-        for file in tqdm(files, desc='Unique > Sorting > Replacing existing files'):
+            output_path = file.with_suffix(".sorted")
            output_path = file.with_suffix('.sorted')
            sort_unique(file, output_path)
            shutil.move(output_path, file)
    @staticmethod
    def clean(path: str) -> None:
        """Move misplaced records to their correct bucket and sort each file."""
        files = find_files(Path(path).expanduser())
        shuffle(files)
        num_files_moved = 0
        handles: dict[str, object] = {}
        with tqdm(files, desc='clean') as bar:
            for file in bar:
                key = str(file).replace(path, '').removeprefix('/')
                unique_file = str(file) + '.unique'
                # stream: moved lines go to their target file, kept lines to a
                # temp file. Nothing is held in RAM.
                with open(str(file), encoding='utf8', errors='ignore') as reader, \
                        open(unique_file, 'w', encoding='utf8') as writer:
                    for line in reader:
                        email = line.partition(':')[0]
                        if get_output_file(email) != key:
                            write_line_to_file(handles, line, Path(path) / get_output_file(email),
                                               buffer_scale=10)
                            num_files_moved += 1
                            if num_files_moved % 100_000 == 0:
                                bar.set_description(f'clean. {num_files_moved:,} moved.')
                        else:
                            writer.write(line)
                sort_unique(unique_file, file)
                os.remove(unique_file)
    @staticmethod
    def test(file: str, dataset: str) -> None:
        """Look up every email in ``file`` and print the cracked rate."""
        cracked = total = 0
        with open(file, encoding='utf8', errors='ignore') as reader:
            for line in reader:
                results = QueryAPI.query(dataset, line.strip())
                if results:
                    passwords = [p.partition(':')[2] for p in results]
                    print('FOUND:', line, ':'.join(passwords))
                    cracked += 1
                total += 1
        print(f'Cracked rate: {cracked / total if total else 0.0}')
    @staticmethod
    def chunk_files(path: str, max_size: int) -> None:
        """Split files larger than ``max_size`` MB and delete the originals."""
        files = find_files(Path(path).expanduser())
        for file in tqdm(files, desc='chunk large files'):
            if file.stat().st_size / 1e6 > max_size:
                # -d (numeric suffixes) is not compatible with MacOS.
                cmd = ['split', '-b', str(int(max_size * 1e6)), str(file), str(file) + '_']
                print(f'File size: {file.stat().st_size / 1e6:.2f} Mb. ' + ' '.join(cmd))
                subprocess.run(cmd, check=True)
                file.unlink()
    @staticmethod
    def export_jsonl(dataset: str, out: str) -> None:
        """Export the dataset as JSON Lines (`{"email":..., "password":...}`)."""
        create_dir_for_file(out)
        total = 0
-        with open(out, 'w', encoding='utf8') as writer:
+        with open(out, "w", encoding="utf8") as writer:
-            for path in tqdm(find_files(Path(dataset).expanduser()), desc='export jsonl'):
+            for path in tqdm(find_files(dataset), desc="export jsonl"):
-                with open(path, encoding='utf8', errors='ignore') as reader:
+                with open(path, encoding="utf8", errors="ignore") as reader:
                    for line in reader:
-                        email, _, password = line.rstrip('\n').partition(':')
+                        email, _, password = line.rstrip("\n").partition(":")
-                        writer.write(json.dumps({'email': email, 'password': password},
+                        writer.write(
-                                                ensure_ascii=False) + '\n')
+                            json.dumps(
                                {"email": email, "password": password},
                                ensure_ascii=False,
                            )
                            + "\n"
                        )
                        total += 1
-        print(f'Exported {total:,} records to {out}.')
+        print(f"Exported {total:,} records to {out}.")
    @staticmethod
    def filter_lines(file: str, out: str, less_than: int, more_than: int) -> None:
        """Keep lines whose printable length is within ``[less_than, more_than]``."""
        create_dir_for_file(out)
        with open(out, 'w', encoding='utf8') as writer, \
                open(file, encoding='utf8', errors='ignore') as reader:
            for line in reader:
                printable = filter_printable(line)
                if less_than <= len(printable.strip()) <= more_than:
                    writer.write(printable)
 def find_files(path) -> list[Path]:
    """Return every regular file under ``path`` (recursively)."""
    root = Path(path).expanduser()
    return [p for p in root.glob('**/*') if p.is_file()]
 def sort_unique(src, dst) -> None:
@@ -162,28 +73,32 @@ def sort_unique(src, dst) -> None:
    """
    src = Path(src)
    if src.stat().st_size <= _IN_MEMORY_SORT_LIMIT:
-        lines = src.read_bytes().split(b'\n')
+        lines = src.read_bytes().split(b"\n")
-        if lines and lines[-1] == b'':
+        if lines and lines[-1] == b"":
            lines.pop()
        lines = sorted(set(lines))
-        Path(dst).write_bytes(b'\n'.join(lines) + b'\n' if lines else b'')
+        Path(dst).write_bytes(b"\n".join(lines) + b"\n" if lines else b"")
    else:
-        tmp = f'{dst}.tmp'
+        tmp = f"{dst}.tmp"
-        with open(tmp, 'wb') as writer:
+        with open(tmp, "wb") as writer:
-            subprocess.run(['sort', '-u', str(src)], stdout=writer, check=True,
+            subprocess.run(
-                           env={**os.environ, 'LC_ALL': 'C'})
+                ["sort", "-u", str(src)],
                stdout=writer,
                check=True,
                env={**os.environ, "LC_ALL": "C"},
            )
        os.replace(tmp, dst)
 def append_file(src, dst) -> None:
    """Append ``src`` to ``dst`` and delete ``src`` (streams, constant memory)."""
-    print(f'merge {src} {dst}')
+    print(f"merge {src} {dst}")
-    with open(src, 'rb') as reader, open(dst, 'ab') as writer:
+    with open(src, "rb") as reader, open(dst, "ab") as writer:
        shutil.copyfileobj(reader, writer, length=1024 * 1024)
-    print(f'del {src}')
+    print(f"del {src}")
    os.remove(src)
 def move_file(src, dst) -> None:
-    print(f'mv {src} {dst}')
+    print(f"mv {src} {dst}")
    shutil.move(src, dst)
@@ -31,9 +31,22 @@ from collectionparser.api import API, find_files, sort_unique
 from collectionparser.parser import parse_files
 from collectionparser.splitter import split_file
-DEFAULT_BATCH_BYTES = 120 * 1024 ** 3
+DEFAULT_BATCH_BYTES = 120 * 1024**3
-ARCHIVE_SUFFIXES = ('.rar', '.zip', '.7z', '.tar', '.tar.gz', '.tgz', '.tar.bz2',
+ARCHIVE_SUFFIXES = (
-                    '.tbz2', '.tar.xz', '.txz', '.gz', '.bz2', '.xz')
+    ".rar",
    ".zip",
    ".7z",
    ".tar",
    ".tar.gz",
    ".tgz",
    ".tar.bz2",
    ".tbz2",
    ".tar.xz",
    ".txz",
    ".gz",
    ".bz2",
    ".xz",
 )
 DEVNULL = subprocess.DEVNULL
@@ -53,7 +66,7 @@ def _which(*names: str) -> str | None:
 def _safe_target(dest: Path, name: str) -> Path:
    target = (dest / name).resolve()
    if not str(target).startswith(str(dest.resolve()) + os.sep):
-        raise ValueError(f'unsafe archive member: {name}')
+        raise ValueError(f"unsafe archive member: {name}")
    return target
@@ -64,28 +77,32 @@ def _extract_compressed(path: Path, dest: Path) -> None:
    mislabeled archives in the wild. Try both, first rc in (0, 1) wins.
    ponytail: 7z may follow archive paths outside dest; trusted local dumps only.
    """
-    seven = _which('7z', '7za')
+    seven = _which("7z", "7za")
-    unrar = _which('unrar')
+    unrar = _which("unrar")
-    candidates = [unrar, seven] if path.name.lower().endswith('.rar') else [seven, unrar]
+    candidates = (
        [unrar, seven] if path.name.lower().endswith(".rar") else [seven, unrar]
    )
    last = None
    for tool in (c for c in candidates if c):
        shutil.rmtree(dest, ignore_errors=True)
        dest.mkdir(parents=True, exist_ok=True)
-        if 'unrar' in tool:
+        if "unrar" in tool:
-            cmd = [tool, 'x', '-o+', '-idq', str(path), str(dest) + '/']
+            cmd = [tool, "x", "-o+", "-idq", str(path), str(dest) + "/"]
        else:
-            cmd = [tool, 'x', '-y', f'-o{dest}', str(path)]
+            cmd = [tool, "x", "-y", f"-o{dest}", str(path)]
-        last = subprocess.run(cmd, check=False, stdout=DEVNULL, stderr=DEVNULL).returncode
+        last = subprocess.run(
            cmd, check=False, stdout=DEVNULL, stderr=DEVNULL
        ).returncode
        if last in (0, 1):
            return
    shutil.rmtree(dest, ignore_errors=True)
-    raise RuntimeError(f'extractors failed (last rc={last})')
+    raise RuntimeError(f"extractors failed (last rc={last})")
 def _extract_tar(path: Path, dest: Path) -> None:
    with tarfile.open(path) as t:
        try:
-            t.extractall(dest, filter='data')  # py3.12+
+            t.extractall(dest, filter="data")  # py3.12+
        except TypeError:  # py3.11
            t.extractall(dest)
@@ -93,25 +110,36 @@ def _extract_tar(path: Path, dest: Path) -> None:
 def _extract_archive(path: Path, dest: Path) -> None:
    name = path.name.lower()
    dest.mkdir(parents=True, exist_ok=True)
-    if name.endswith('.zip'):
+    if name.endswith(".zip"):
        with zipfile.ZipFile(path) as z:
            for member in z.namelist():
                _safe_target(dest, member)
            z.extractall(dest)
-    elif name.endswith(('.rar', '.7z')):
+    elif name.endswith((".rar", ".7z")):
        _extract_compressed(path, dest)
-    elif name.endswith(('.tar', '.tar.gz', '.tgz', '.tar.bz2', '.tbz2', '.tar.xz', '.txz')):
+    elif name.endswith(
        (".tar", ".tar.gz", ".tgz", ".tar.bz2", ".tbz2", ".tar.xz", ".txz")
    ):
        _extract_tar(path, dest)
    else:  # single-file .gz/.bz2/.xz
-        opener = gzip.open if name.endswith('.gz') else bz2.open if name.endswith('.bz2') else lzma.open
+        opener = (
-        with opener(path, 'rb') as r, open(dest / path.stem, 'wb') as w:
+            gzip.open
            if name.endswith(".gz")
            else bz2.open
            if name.endswith(".bz2")
            else lzma.open
        )
        with opener(path, "rb") as r, open(dest / path.stem, "wb") as w:
            shutil.copyfileobj(r, w, length=1 << 20)
 def _discover(source: str) -> list[Path]:
    files = find_files(source)
    files.sort(key=lambda p: p.stat().st_size)  # small files first
-    print(f'{len(files)} files, {sum(1 for p in files if _is_archive(p))} archives.', flush=True)
+    print(
        f"{len(files)} files, {sum(1 for p in files if _is_archive(p))} archives.",
        flush=True,
    )
    return files
@@ -131,47 +159,50 @@ def _batches(files: list[Path], batch_bytes: int):
 def _count_lines(path: str | Path) -> int:
    total = 0
-    with open(path, 'rb') as r:
+    with open(path, "rb") as r:
-        for chunk in iter(lambda: r.read(1 << 20), b''):
+        for chunk in iter(lambda: r.read(1 << 20), b""):
-            total += chunk.count(b'\n')
+            total += chunk.count(b"\n")
    return total
 def _totals(stats: list[dict]) -> dict:
    return {
-        'input_bytes': sum(s['input_bytes'] for s in stats),
+        "input_bytes": sum(s["input_bytes"] for s in stats),
-        'files': sum(s['files'] for s in stats),
+        "files": sum(s["files"] for s in stats),
-        'pairs': sum(s['pairs'] for s in stats),
+        "pairs": sum(s["pairs"] for s in stats),
-        'rejects': sum(s['rejects'] for s in stats),
+        "rejects": sum(s["rejects"] for s in stats),
-        'seconds': sum(s['total_s'] for s in stats),
+        "seconds": sum(s["total_s"] for s in stats),
    }
 def _stats_markdown(state: dict) -> str:
-    stats = state['stats']
+    stats = state["stats"]
    totals = _totals(stats)
    lines = [
-        '# collectionparser build stats',
+        "# collectionparser build stats",
-        '',
+        "",
-        f'- input: {totals["input_bytes"] / 1e9:.1f} GB across {totals["files"]:,} files',
+        f"- input: {totals['input_bytes'] / 1e9:.1f} GB across {totals['files']:,} files",
-        f'- valid pairs: {totals["pairs"]:,}',
+        f"- valid pairs: {totals['pairs']:,}",
-        f'- rejects: {totals["rejects"]:,}',
+        f"- rejects: {totals['rejects']:,}",
-        f'- batches: {len(stats)}',
+        f"- batches: {len(stats)}",
-        f'- build seconds: {totals["seconds"]:,.0f}',
+        f"- build seconds: {totals['seconds']:,.0f}",
    ]
-    if totals['seconds']:
+    if totals["seconds"]:
-        lines.append(f'- aggregate throughput: {totals["input_bytes"] / 1e6 / totals["seconds"]:.1f} MB/s')
+        lines.append(
-    if state.get('hashes') is not None:
+            f"- aggregate throughput: {totals['input_bytes'] / 1e6 / totals['seconds']:.1f} MB/s"
-        lines.append(f'- hashed passwords collected: {state["hashes"]:,}')
+        )
-    if state.get('dedup_s') is not None:
+    if state.get("hashes") is not None:
-        lines.append(f'- dedup seconds: {state["dedup_s"]:,.0f}')
+        lines.append(f"- hashed passwords collected: {state['hashes']:,}")
-    if state.get('unique_records') is not None:
+    if state.get("dedup_s") is not None:
-        lines.append(f'- unique records: {state["unique_records"]:,}')
+        lines.append(f"- dedup seconds: {state['dedup_s']:,.0f}")
-    return '\n'.join(lines) + '\n'
+    if state.get("unique_records") is not None:
        lines.append(f"- unique records: {state['unique_records']:,}")
    return "\n".join(lines) + "\n"
-def _extract_batch(archives: list[Path], extract_dir: Path,
+def _extract_batch(
-                   index: int) -> tuple[list[Path], list[str]]:
+    archives: list[Path], extract_dir: Path, index: int
 ) -> tuple[list[Path], list[str]]:
    """Extract archives; return ``(files, failures)``.
    Failures are logged, never silently dropped, so they can be recovered later.
@@ -182,19 +213,28 @@ def _extract_batch(archives: list[Path], extract_dir: Path,
    failures: list[str] = []
    for i, archive in enumerate(archives):
        try:
-            _extract_archive(archive, extract_dir / f'{i:05d}_{archive.name}')
+            _extract_archive(archive, extract_dir / f"{i:05d}_{archive.name}")
        except Exception as exc:  # one bad archive must not stop the build
-            failures.append(f'{archive}\t{exc}')
+            failures.append(f"{archive}\t{exc}")
-    files = [p for p in extract_dir.glob('**/*') if p.is_file()]
+    files = [p for p in extract_dir.glob("**/*") if p.is_file()]
-    print(f'batch {index}: {len(archives)} archives -> {len(files)} files, '
+    print(
-          f'{len(failures)} failed ({time.time() - t0:.0f}s)', flush=True)
+        f"batch {index}: {len(archives)} archives -> {len(files)} files, "
        f"{len(failures)} failed ({time.time() - t0:.0f}s)",
        flush=True,
    )
    for failure in failures[:5]:
-        print(f'WARN: {failure}', flush=True)
+        print(f"WARN: {failure}", flush=True)
    return files, failures
-def build(source: str, dataset: str, scratch: str, batch_bytes: int = DEFAULT_BATCH_BYTES,
+def build(
-          dedup: bool = True, extract: bool = True) -> None:
+    source: str,
    dataset: str,
    scratch: str,
    batch_bytes: int = DEFAULT_BATCH_BYTES,
    dedup: bool = True,
    extract: bool = True,
 ) -> None:
    """Build the deduped query dataset from ``source`` into ``dataset``.
    ``scratch`` (a fast disk) holds per-batch files, sort temp and build state.
@@ -205,63 +245,85 @@ def build(source: str, dataset: str, scratch: str, batch_bytes: int = DEFAULT_BA
    scratch = Path(scratch).expanduser()
    dataset.mkdir(parents=True, exist_ok=True)
    scratch.mkdir(parents=True, exist_ok=True)
-    os.environ['TMPDIR'] = str(scratch)  # `sort -u` spills here, not into the dataset
+    os.environ["TMPDIR"] = str(scratch)  # `sort -u` spills here, not into the dataset
-    for stale in scratch.glob('extract_*'):
+    for stale in scratch.glob("extract_*"):
        shutil.rmtree(stale, ignore_errors=True)
-    if extract and not _which('7z', '7za', 'unrar'):
+    if extract and not _which("7z", "7za", "unrar"):
-        print('WARN: no 7z/7za/unrar found; .rar/.7z archives will be skipped '
+        print(
-              '(.zip/.tar/.gz still work).', flush=True)
+            "WARN: no 7z/7za/unrar found; .rar/.7z archives will be skipped "
            "(.zip/.tar/.gz still work).",
            flush=True,
        )
-    state_path = scratch / 'build_state.json'
+    state_path = scratch / "build_state.json"
-    state = json.loads(state_path.read_text()) if state_path.exists() else {'done': [], 'stats': []}
+    state = (
        json.loads(state_path.read_text())
        if state_path.exists()
        else {"done": [], "stats": []}
    )
    def save():
        state_path.write_text(json.dumps(state, indent=2))
-    done = set(state['done'])
+    done = set(state["done"])
    files = _discover(source)
    batches = list(_batches(files, batch_bytes))
-    state.update({'batches_total': len(batches), 'batch_files': [len(b) for b in batches],
+    state.update(
-                  'dataset': str(dataset), 'complete': False})
+        {
            "batches_total": len(batches),
            "batch_files": [len(b) for b in batches],
            "dataset": str(dataset),
            "complete": False,
        }
    )
    save()
-    print(f'{len(files)} files, {len(batches)} batches, {len(done)} already done.', flush=True)
+    print(
        f"{len(files)} files, {len(batches)} batches, {len(done)} already done.",
        flush=True,
    )
-    failures = scratch / 'failures.txt'
+    failures = scratch / "failures.txt"
-    extract_failures = scratch / 'extract_failures.txt'
+    extract_failures = scratch / "extract_failures.txt"
-    hashes = scratch / 'hashes.txt'
+    hashes = scratch / "hashes.txt"
    for index, batch in enumerate(batches):
        if index in done:
            continue
-        state['phase'] = f'batch {index}'
+        state["phase"] = f"batch {index}"
        save()
-        success = scratch / f'success_{index:05d}.txt'
+        success = scratch / f"success_{index:05d}.txt"
-        failure = scratch / f'failure_{index:05d}.txt'
+        failure = scratch / f"failure_{index:05d}.txt"
-        extract_dir = scratch / f'extract_{index:05d}'
+        extract_dir = scratch / f"extract_{index:05d}"
-        progress_path = scratch / f'progress_{index:05d}.txt'
+        progress_path = scratch / f"progress_{index:05d}.txt"
        input_bytes = sum(f.stat().st_size for f in batch)
        if not success.exists():
            progress_path.unlink(missing_ok=True)
        parsed = set()
        if success.exists() and progress_path.exists():
-            with open(progress_path, encoding='utf8', errors='ignore') as r:
+            with open(progress_path, encoding="utf8", errors="ignore") as r:
-                parsed = {line.rstrip('\n') for line in r}
+                parsed = {line.rstrip("\n") for line in r}
        inputs = [f for f in batch if not _is_archive(f)]
        if extract:
            extracted, failures_ = _extract_batch(
-                [f for f in batch if _is_archive(f)], extract_dir, index)
+                [f for f in batch if _is_archive(f)], extract_dir, index
            )
            inputs.extend(extracted)
            if failures_:
-                with open(extract_failures, 'a', encoding='utf8') as w:
+                with open(extract_failures, "a", encoding="utf8") as w:
-                    w.write('\n'.join(failures_) + '\n')
+                    w.write("\n".join(failures_) + "\n")
        pending = [f for f in inputs if str(f) not in parsed]
        t0 = time.time()
        if pending:
-            parse_files(pending, str(success), str(failure),
+            parse_files(
-                        progress_filename=str(progress_path), hashes_filename=str(hashes))
+                pending,
                str(success),
                str(failure),
                progress_filename=str(progress_path),
                hashes_filename=str(hashes),
            )
        if success.exists():
            pairs = _count_lines(success)
            rejects = _count_lines(failure) if failure.exists() else 0
@@ -269,7 +331,7 @@ def build(source: str, dataset: str, scratch: str, batch_bytes: int = DEFAULT_BA
            split_file(str(success), str(dataset))
            success.unlink(missing_ok=True)
            if failure.exists():
-                with open(failures, 'ab') as w, open(failure, 'rb') as r:
+                with open(failures, "ab") as w, open(failure, "rb") as r:
                    shutil.copyfileobj(r, w, length=1024 * 1024)
                failure.unlink(missing_ok=True)
            progress_path.unlink(missing_ok=True)
@@ -278,79 +340,91 @@ def build(source: str, dataset: str, scratch: str, batch_bytes: int = DEFAULT_BA
        shutil.rmtree(extract_dir, ignore_errors=True)
        elapsed = time.time() - t0
-        stat = {'batch': index, 'files': len(batch), 'input_bytes': input_bytes,
+        stat = {
-                'pairs': pairs, 'rejects': rejects, 'total_s': round(elapsed, 1),
+            "batch": index,
-                'mb_per_s': round(input_bytes / 1e6 / elapsed, 2) if elapsed else 0.0}
+            "files": len(batch),
-        state['stats'].append(stat)
+            "input_bytes": input_bytes,
            "pairs": pairs,
            "rejects": rejects,
            "total_s": round(elapsed, 1),
            "mb_per_s": round(input_bytes / 1e6 / elapsed, 2) if elapsed else 0.0,
        }
        state["stats"].append(stat)
        done.add(index)
-        state['done'] = sorted(done)
+        state["done"] = sorted(done)
        save()
-        (scratch / 'stats.md').write_text(_stats_markdown(state))
+        (scratch / "stats.md").write_text(_stats_markdown(state))
-        print('BATCH ' + json.dumps(stat), flush=True)
+        print("BATCH " + json.dumps(stat), flush=True)
    if hashes.exists():
        sort_unique(hashes, hashes)  # de-duplicate collected hashed passwords
-        state['hashes'] = _count_lines(hashes)
+        state["hashes"] = _count_lines(hashes)
        save()
-    if dedup and not state.get('deduped'):
+    if dedup and not state.get("deduped"):
-        state['phase'] = 'deduping'
+        state["phase"] = "deduping"
        save()
        t0 = time.time()
        API.sort(str(dataset))  # per-bucket sort -u dedups across batches
-        state['dedup_s'] = round(time.time() - t0, 1)
+        state["dedup_s"] = round(time.time() - t0, 1)
-        state['deduped'] = True
+        state["deduped"] = True
        save()
-    if state.get('unique_records') is None:
+    if state.get("unique_records") is None:
-        state['unique_records'] = sum(_count_lines(p) for p in dataset.glob('**/*') if p.is_file())
+        state["unique_records"] = sum(
            _count_lines(p) for p in dataset.glob("**/*") if p.is_file()
        )
        save()
-    state['phase'] = 'complete'
+    state["phase"] = "complete"
-    state['complete'] = True
+    state["complete"] = True
    save()
-    (scratch / 'stats.md').write_text(_stats_markdown(state))
+    (scratch / "stats.md").write_text(_stats_markdown(state))
-    print('BUILD COMPLETE', flush=True)
+    print("BUILD COMPLETE", flush=True)
 def build_status(scratch: str) -> dict:
    """Return a compact, machine-readable snapshot of a build under ``scratch``."""
    scratch = Path(scratch).expanduser()
-    state_path = scratch / 'build_state.json'
+    state_path = scratch / "build_state.json"
    if not state_path.exists():
-        return {'error': f'no build_state.json in {scratch}'}
+        return {"error": f"no build_state.json in {scratch}"}
    state = json.loads(state_path.read_text())
-    done = state.get('done', [])
+    done = state.get("done", [])
-    stats = state.get('stats', [])
+    stats = state.get("stats", [])
-    total = state.get('batches_total')
+    total = state.get("batches_total")
    totals = _totals(stats)
    result = {
-        'phase': state.get('phase', 'unknown'),
+        "phase": state.get("phase", "unknown"),
-        'complete': state.get('complete', False),
+        "complete": state.get("complete", False),
-        'batches_done': len(done),
+        "batches_done": len(done),
-        'batches_total': total,
+        "batches_total": total,
-        'deduped': state.get('deduped', False),
+        "deduped": state.get("deduped", False),
-        'input_gb': round(totals['input_bytes'] / 1e9, 2),
+        "input_gb": round(totals["input_bytes"] / 1e9, 2),
-        'pairs': totals['pairs'],
+        "pairs": totals["pairs"],
-        'rejects': totals['rejects'],
+        "rejects": totals["rejects"],
-        'hashes': state.get('hashes'),
+        "hashes": state.get("hashes"),
-        'unique_records': state.get('unique_records'),
+        "unique_records": state.get("unique_records"),
    }
    current = next((i for i in range(total or 0) if i not in done), None)
    if current is not None:
-        progress = scratch / f'progress_{current:05d}.txt'
+        progress = scratch / f"progress_{current:05d}.txt"
-        result['current_batch'] = current
+        result["current_batch"] = current
-        result['current_files_done'] = sum(1 for _ in progress.open()) if progress.exists() else 0
+        result["current_files_done"] = (
-        batch_files = state.get('batch_files') or []
+            sum(1 for _ in progress.open()) if progress.exists() else 0
        )
        batch_files = state.get("batch_files") or []
        if current < len(batch_files):
-            result['current_files_total'] = batch_files[current]
+            result["current_files_total"] = batch_files[current]
-    for prefix in ('success', 'failure'):
+    for prefix in ("success", "failure"):
-        result[f'{prefix}_bytes'] = sum(p.stat().st_size for p in scratch.glob(f'{prefix}_*.txt'))
+        result[f"{prefix}_bytes"] = sum(
-    hashes_file = scratch / 'hashes.txt'
+            p.stat().st_size for p in scratch.glob(f"{prefix}_*.txt")
-    result['hashes_bytes'] = hashes_file.stat().st_size if hashes_file.exists() else 0
+        )
-    dataset = state.get('dataset')
+    hashes_file = scratch / "hashes.txt"
    result["hashes_bytes"] = hashes_file.stat().st_size if hashes_file.exists() else 0
    dataset = state.get("dataset")
    if dataset and Path(dataset).exists():
-        files = [p for p in Path(dataset).glob('**/*') if p.is_file()]
+        files = [p for p in Path(dataset).glob("**/*") if p.is_file()]
-        result['dataset_files'] = len(files)
+        result["dataset_files"] = len(files)
-        result['dataset_bytes'] = sum(p.stat().st_size for p in files)
+        result["dataset_bytes"] = sum(p.stat().st_size for p in files)
    return result
@@ -16,133 +16,149 @@ def cli():
    pass
-@cli.command(short_help='Converts a large FILE to a query friendly folder OUT (e.g. a/b/c). '
+@cli.command(
-                        'Use RESTART_FROM to resume from the i-th line.')
+    short_help="Converts a large FILE to a query friendly folder OUT (e.g. a/b/c). "
-@click.option('--file', required=True, type=input_file())
+    "Use RESTART_FROM to resume from the i-th line."
-@click.option('--out', required=True, type=output_dir())
+)
-@click.option('--restart_from', default=0, type=int, show_default=True)
+@click.option("--file", required=True, type=input_file())
@click.option("--out", required=True, type=output_dir())
@click.option("--restart_from", default=0, type=int, show_default=True)
 def split(file, out, restart_from):
    split_file(file, out, restart_from)
-@cli.command(short_help='chunk large TXT files into smaller files.')
+@cli.command(
-@click.option('--path', required=True, type=input_dir())
+    short_help="Sorts and de-duplicates every file in a query friendly folder PATH."
-@click.option('--size', default=50, type=int, show_default=True)  # MB.
+)
-def chunk(path, size):
+@click.option("--path", required=True, type=input_dir())
    API.chunk_files(path, size)
@cli.command(short_help='Sorts a query friendly folder PATH. Target is itself.')
@click.option('--path', required=True, type=input_dir())
 def sort(path):
    API.sort(path)
-@cli.command(short_help='Cleans a query friendly folder PATH. Move incorrect records and sort the files.')
+@cli.command(
-@click.option('--path', required=True, type=input_dir())
+    short_help="Parses an unstructured folder PATH into SUCCESS_FILE, FAILURE_FILE "
-def clean(path):
+    "and (optionally) HASHES_FILE."
-    API.clean(path)
+)
-
+@click.option("--path", required=True, type=input_dir())
-
+@click.option("--success_file", required=True, type=output_file())
-@cli.command(short_help='Infers passwords of a list of emails defined in FILE with a query friendly folder DATASET.')
+@click.option("--failure_file", required=True, type=output_file())
-@click.option('--file', required=True, type=input_file())
+@click.option(
-@click.option('--dataset', required=True)
+    "--hashes_file",
-def test(file: str, dataset: str):
+    required=False,
-    API.test(file, dataset)
+    type=output_file(),
-
+    help="Collect hashed passwords here for later `crack`.",
-
+)
@cli.command(short_help='Parses an unstructured folder PATH of many files and generates two files: '
                        'SUCCESS_FILE and FAILURE_FILE. All valid email:password will go to SUCCESS_FILE.')
@click.option('--path', required=True, type=input_dir())
@click.option('--success_file', required=True, type=output_file())
@click.option('--failure_file', required=True, type=output_file())
@click.option('--hashes_file', required=False, type=output_file(),
              help='Collect hashed passwords here for later `crack`.')
 def parse(path, success_file, failure_file, hashes_file):
    parse_to_files(path, success_file, failure_file, hashes_file)
-@cli.command(short_help='Merges dataset SRC into dataset DEST. Duplicates are removed in the second pass.')
+@cli.command(short_help="Merges dataset SRC into dataset DEST, then sorts/dedups DEST.")
-@click.option('--src', required=True, type=input_dir())
+@click.option("--src", required=True, type=input_dir())
-@click.option('--dest', required=True, type=input_dir())
+@click.option("--dest", required=True, type=input_dir())
 def merge(src: str, dest: str):
    API.merge(src, dest)
-    API.clean(dest)
+    API.sort(dest)
-@cli.command(short_help='Evaluates some metrics such as precision/recall (e.g. is OLD into NEW).')
+@cli.command(
-@click.option('--old', required=True, type=input_dir())
+    short_help="Searches a dataset by exact email or wildcard pattern "
-@click.option('--new', required=True, type=input_dir())
+    '(e.g. "*@gmail.com", "john*", "jo?n@*").'
-def evaluate(old: str, new: str):
+)
-    API.evaluate(old, new)
+@click.option("--dataset", required=True, type=input_dir())
-
+@click.option("--pattern", required=True)
@cli.command(short_help='Filters lines less than N and more than M.')
@click.option('--file', required=True, type=input_file())
@click.option('--out', required=True, type=output_file())
@click.option('--less_than', required=True, type=int)
@click.option('--more_than', required=True, type=int)
 def filter_lines(file: str, out: str, less_than: int, more_than: int):
    API.filter_lines(file, out, less_than, more_than)
@cli.command(short_help='Searches a dataset by exact email or wildcard pattern '
                        '(e.g. "*@gmail.com", "john*", "jo?n@*").')
@click.option('--dataset', required=True, type=input_dir())
@click.option('--pattern', required=True)
 def query(dataset: str, pattern: str):
-    if '*' in pattern or '?' in pattern:
+    if "*" in pattern or "?" in pattern:
-        results = QueryAPI.wildcard(dataset, pattern)
+        for line in QueryAPI.wildcard(dataset, pattern):
            click.echo(line)
    else:
        results = QueryAPI.query(dataset, pattern)
-    if results:
+        if results:
-        click.echo('\n'.join(results))
+            click.echo("\n".join(results))
-@cli.command(short_help='Exports a dataset as JSON Lines for OpenSearch/bulk ingest.')
+@cli.command(short_help="Exports a dataset as JSON Lines for OpenSearch/bulk ingest.")
-@click.option('--dataset', required=True, type=input_dir())
+@click.option("--dataset", required=True, type=input_dir())
-@click.option('--out', required=True, type=output_file())
+@click.option("--out", required=True, type=output_file())
 def export(dataset: str, out: str):
    API.export_jsonl(dataset, out)
-@cli.command(short_help='Dictionary-attacks collected MD5/SHA1/SHA256 hashes.')
+@cli.command(short_help="Dictionary-attacks collected MD5/SHA1/SHA256 hashes.")
-@click.option('--hashes', 'hashes_path', required=True, type=input_file())
+@click.option("--hashes", "hashes_path", required=True, type=input_file())
-@click.option('--wordlist', multiple=True, type=click.Path(),
+@click.option(
-              help='Plain wordlist(s); missing files are downloaded from rockyou.')
+    "--wordlist",
-@click.option('--passwords-from', multiple=True, type=click.Path(),
+    multiple=True,
-              help='Dataset file/folder whose plaintext passwords join the wordlist.')
+    type=click.Path(),
-@click.option('--out', required=True, type=output_file())
+    help="Plain wordlist(s); missing files are downloaded from rockyou.",
-@click.option('--workers', default=0, type=int, show_default=True,
+)
-              help='0 = all CPUs.')
+@click.option(
-def crack(hashes_path: str, wordlist: tuple, passwords_from: tuple, out: str, workers: int):
+    "--passwords-from",
    multiple=True,
    type=click.Path(),
    help="Dataset file/folder whose plaintext passwords join the wordlist.",
 )
@click.option("--out", required=True, type=output_file())
@click.option("--workers", default=0, type=int, show_default=True, help="0 = all CPUs.")
 def crack(
    hashes_path: str, wordlist: tuple, passwords_from: tuple, out: str, workers: int
 ):
    from collectionparser.crack import crack as run_crack
-    run_crack(hashes_path, out, wordlists=wordlist, passwords_from=passwords_from,
+
-              workers=workers or None)
+    run_crack(
        hashes_path,
        out,
        wordlists=wordlist,
        passwords_from=passwords_from,
        workers=workers or None,
    )
-@cli.command(short_help='Print build/dataset progress as JSON (for scripts and agents).')
+@cli.command(
-@click.option('--scratch', required=True, type=click.Path(exists=True, file_okay=False, dir_okay=True))
+    short_help="Print build/dataset progress as JSON (for scripts and agents)."
-@click.option('--pretty', is_flag=True, help='Indent the JSON.')
+)
@click.option(
    "--scratch",
    required=True,
    type=click.Path(exists=True, file_okay=False, dir_okay=True),
 )
@click.option("--pretty", is_flag=True, help="Indent the JSON.")
 def status(scratch: str, pretty: bool):
    from collectionparser.build import build_status
-    click.echo(json.dumps(build_status(scratch), indent=2 if pretty else None, default=str))
+
    click.echo(
        json.dumps(build_status(scratch), indent=2 if pretty else None, default=str)
    )
-@cli.command(short_help='Builds the deduped query dataset from a raw source folder in batches. '
+@cli.command(
-                        'Stop and rerun to resume.')
+    short_help="Builds the deduped query dataset from a raw source folder in batches. "
-@click.option('--source', required=True, type=input_dir())
+    "Stop and rerun to resume."
-@click.option('--dataset', required=True, type=output_dir())
+)
-@click.option('--scratch', required=True, type=output_dir())
+@click.option("--source", required=True, type=input_dir())
-@click.option('--batch_gb', default=120, type=int, show_default=True)
+@click.option("--dataset", required=True, type=output_dir())
-@click.option('--dedup/--no-dedup', default=True)
+@click.option("--scratch", required=True, type=output_dir())
-@click.option('--extract/--no-extract', default=True, show_default=True,
+@click.option("--batch_gb", default=120, type=int, show_default=True)
-              help='Extract archives (.rar/.zip/.7z/.tar.*/.gz) found in the source.')
+@click.option("--dedup/--no-dedup", default=True)
-def build(source: str, dataset: str, scratch: str, batch_gb: int, dedup: bool, extract: bool):
+@click.option(
    "--extract/--no-extract",
    default=True,
    show_default=True,
    help="Extract archives (.rar/.zip/.7z/.tar.*/.gz) found in the source.",
 )
 def build(
    source: str, dataset: str, scratch: str, batch_gb: int, dedup: bool, extract: bool
 ):
    from collectionparser.build import build as run_build
-    run_build(source, dataset, scratch, batch_bytes=batch_gb * 1024 ** 3,
+
-              dedup=dedup, extract=extract)
+    run_build(
        source,
        dataset,
        scratch,
        batch_bytes=batch_gb * 1024**3,
        dedup=dedup,
        extract=extract,
    )
-if __name__ == '__main__':
+if __name__ == "__main__":
    cli()
@@ -6,12 +6,16 @@ through ``sort -u`` so memory stays bounded for arbitrarily large inputs.
 Only unsalted MD5/SHA1/SHA256 are attackable; bcrypt is intentionally not
 supported (a wordlist attack on it is infeasible).
-ponytail: the target index is rebuilt per worker (RAM x workers); a shared mmap
+
-index is the upgrade path if the hash file is enormous.
+Memory: the target index is built once and shared with the workers by forking
 (copy-on-write), so it costs one copy, not one per worker. ponytail: platforms
 without ``fork`` fall back to rebuilding it in each worker (RAM x workers); a
 shared mmap index is the upgrade path for enormous hash files.
 """
 import gzip
 import hashlib
 import multiprocessing
 import os
 import shutil
 import subprocess
@@ -21,11 +25,14 @@ from pathlib import Path
 from collectionparser.api import find_files
-DEFAULT_WORDLIST_URL = 'https://weakpass.com/download/90/rockyou.txt.gz'
+DEFAULT_WORDLIST_URL = "https://weakpass.com/download/90/rockyou.txt.gz"
 # hex digest length -> hashlib constructor
 _ALGOS = {32: hashlib.md5, 40: hashlib.sha1, 64: hashlib.sha256}
-_HEX_DIGITS = frozenset(b'0123456789abcdefABCDEF')
+_HEX_DIGITS = frozenset(b"0123456789abcdefABCDEF")
 # Filled once in `crack`; forked workers inherit it instead of rebuilding it.
 _TARGETS: dict[int, dict[bytes, list[bytes]]] | None = None
 def download_rockyou(dest: Path) -> Path:
@@ -34,58 +41,68 @@ def download_rockyou(dest: Path) -> Path:
    if dest.exists() and dest.stat().st_size:
        return dest
    dest.parent.mkdir(parents=True, exist_ok=True)
-    print(f'Downloading wordlist {DEFAULT_WORDLIST_URL} ...', flush=True)
+    print(f"Downloading wordlist {DEFAULT_WORDLIST_URL} ...", flush=True)
-    with urllib.request.urlopen(DEFAULT_WORDLIST_URL) as response, \
+    with (
-            gzip.GzipFile(fileobj=response) as gz, open(dest, 'wb') as out:
+        urllib.request.urlopen(DEFAULT_WORDLIST_URL) as response,
        gzip.GzipFile(fileobj=response) as gz,
        open(dest, "wb") as out,
    ):
        shutil.copyfileobj(gz, out, length=1 << 20)
    return dest
 def _write_passwords(source: Path, out) -> None:
    """Append the password column of an ``email:password`` file to ``out``."""
-    with open(source, encoding='utf8', errors='ignore') as reader:
+    with open(source, encoding="utf8", errors="ignore") as reader:
        for line in reader:
-            out.write(line.partition(':')[2].encode('utf8'))
+            out.write(line.partition(":")[2].encode("utf8"))
 def prepare_candidates(wordlists, passwords_from, work_dir: Path) -> Path:
    """Return a single, de-duplicated wordlist built from all candidate sources."""
    work_dir = Path(work_dir)
    work_dir.mkdir(parents=True, exist_ok=True)
-    wordlists = [Path(w) for w in wordlists] or [work_dir / 'rockyou.txt']
+    wordlists = [Path(w) for w in wordlists] or [work_dir / "rockyou.txt"]
    wordlists = [w if w.exists() else download_rockyou(w) for w in wordlists]
    if len(wordlists) == 1 and not passwords_from:
        return wordlists[0]
-    unique = work_dir / 'candidates.unique.txt'
+    unique = work_dir / "candidates.unique.txt"
-    env = {**os.environ, 'LC_ALL': 'C', 'TMPDIR': str(work_dir)}
+    env = {**os.environ, "LC_ALL": "C", "TMPDIR": str(work_dir)}
-    process = subprocess.Popen(['sort', '-u'], stdin=subprocess.PIPE,
+    with (
-                               stdout=open(unique, 'wb'), env=env)  # noqa: SIM115
+        open(unique, "wb") as sink_file,
-    with process.stdin as sink:
+        subprocess.Popen(
-        for wordlist in wordlists:
+            ["sort", "-u"], stdin=subprocess.PIPE, stdout=sink_file, env=env
-            with open(wordlist, 'rb') as reader:
+        ) as process,
-                shutil.copyfileobj(reader, sink, length=1 << 20)
+    ):
-        for source in passwords_from:
+        with process.stdin as sink:
-            root = Path(source).expanduser()
+            for wordlist in wordlists:
-            for path in ([root] if root.is_file() else find_files(root)):
+                with open(wordlist, "rb") as reader:
-                _write_passwords(path, sink)
+                    shutil.copyfileobj(reader, sink, length=1 << 20)
-    process.wait()
+            for source in passwords_from:
-    if process.returncode != 0:
+                root = Path(source).expanduser()
-        raise RuntimeError('failed to build the candidate wordlist')
+                for path in [root] if root.is_file() else find_files(root):
                    _write_passwords(path, sink)
        if process.wait() != 0:
            raise RuntimeError("failed to build the candidate wordlist")
    return unique
 def load_targets(hashes_path: str | Path) -> dict[int, dict[bytes, list[bytes]]]:
    """Return ``{hex_length: {digest_bytes: [email, ...]}}`` from a hashes file."""
    targets: dict[int, dict[bytes, list[bytes]]] = {}
-    with open(hashes_path, 'rb') as reader:
+    with open(hashes_path, "rb") as reader:
        for raw in reader:
-            line = raw.rstrip(b'\r\n')
+            line = raw.rstrip(b"\r\n")
-            email, separator, digest = line.rpartition(b':')
+            email, separator, digest = line.rpartition(b":")
-            if not separator or len(digest) not in _ALGOS or not _HEX_DIGITS.issuperset(digest):
+            if (
                not separator
                or len(digest) not in _ALGOS
                or not _HEX_DIGITS.issuperset(digest)
            ):
                continue
            try:
-                key = bytes.fromhex(digest.decode('ascii'))
+                key = bytes.fromhex(digest.decode("ascii"))
            except ValueError:
                continue
            targets.setdefault(len(digest), {}).setdefault(key, []).append(email)
@@ -95,44 +112,61 @@ def load_targets(hashes_path: str | Path) -> dict[int, dict[bytes, list[bytes]]]
 def _scan_stripe(args: tuple) -> list[bytes]:
    """Attack one residue class of the wordlist (``index`` of every ``total``)."""
    hashes_path, wordlist_path, index, total = args
-    targets = load_targets(hashes_path)
+    targets = _TARGETS if _TARGETS is not None else load_targets(hashes_path)
    checks = [(targets[length], _ALGOS[length]) for length in targets]
    matches: list[bytes] = []
-    with open(wordlist_path, 'rb') as reader:
+    with open(wordlist_path, "rb") as reader:
        for n, raw in enumerate(reader):
            if n % total != index:
                continue
-            word = raw.rstrip(b'\r\n')
+            word = raw.rstrip(b"\r\n")
            if not word:
                continue
            for target, constructor in checks:
                emails = target.get(constructor(word).digest())
                if emails:
-                    matches.extend(email + b':' + word + b'\n' for email in emails)
+                    matches.extend(email + b":" + word + b"\n" for email in emails)
    return matches
-def crack(hashes_path: str, out_path: str, wordlists=(), passwords_from=(),
+def crack(
-          workers: int | None = None, work_dir: str | None = None) -> int:
+    hashes_path: str,
    out_path: str,
    wordlists=(),
    passwords_from=(),
    workers: int | None = None,
    work_dir: str | None = None,
 ) -> int:
    """Attack the hashes with the candidate wordlists; write ``email:password``."""
-    targets = load_targets(hashes_path)
+    global _TARGETS
    _TARGETS = load_targets(hashes_path)
    targets = _TARGETS
    if not targets:
-        print('No crackable MD5/SHA1/SHA256 hashes found.')
+        print("No crackable MD5/SHA1/SHA256 hashes found.")
        return 0
    work_dir = Path(work_dir) if work_dir else Path(out_path).parent
    wordlist = prepare_candidates(list(wordlists), list(passwords_from), work_dir)
    workers = max(1, workers or os.cpu_count() or 1)
-    n_hashes = sum(len(emails) for by_digest in targets.values() for emails in by_digest.values())
+    n_hashes = sum(
-    print(f'Attacking {n_hashes:,} hashes with {wordlist.name} on {workers} worker(s).', flush=True)
+        len(emails) for by_digest in targets.values() for emails in by_digest.values()
    )
    print(
        f"Attacking {n_hashes:,} hashes with {wordlist.name} on {workers} worker(s).",
        flush=True,
    )
    jobs = [(str(hashes_path), str(wordlist), i, workers) for i in range(workers)]
    total = 0
-    with open(out_path, 'wb') as writer:
+    with open(out_path, "wb") as writer:
        if workers == 1:
            results = map(_scan_stripe, jobs)
        else:
-            executor = ProcessPoolExecutor(max_workers=workers)
+            # fork shares the parent's _TARGETS by copy-on-write.
            context = (
                multiprocessing.get_context("fork") if hasattr(os, "fork") else None
            )
            executor = ProcessPoolExecutor(max_workers=workers, mp_context=context)
            results = executor.map(_scan_stripe, jobs)
        try:
            for matches in results:
@@ -142,5 +176,5 @@ def crack(hashes_path: str, out_path: str, wordlists=(), passwords_from=(),
        finally:
            if workers != 1:
                executor.shutdown()
-    print(f'Cracked {total:,} hashes -> {out_path}.')
+    print(f"Cracked {total:,} hashes -> {out_path}.")
    return total
@@ -11,30 +11,31 @@ from time import time
 from tqdm import tqdm
-from collectionparser.utils import PasswordType, filter_printable, shuffle
+from collectionparser.utils import PasswordType, filter_printable, find_files, shuffle
 # Separators tried in order; the duplicate ',' of the original is dropped.
-VALID_SEPARATORS = (':', ';', ' ', '\t', '  ', '   ', ', ', ',', ';;', '::')
+VALID_SEPARATORS = (":", ";", " ", "\t", "  ", "   ", ", ", ",", ";;", "::")
 # Same shape as the old `validate_email` dependency: rejects '<', '/', '>' etc.
 _EMAIL_RE = re.compile(
    r"^[a-z0-9!#$%&'*+/=?^_`{|}~-]+(?:\.[a-z0-9!#$%&'*+/=?^_`{|}~-]+)*"
    r"@(?:[a-z0-9](?:[a-z0-9-]*[a-z0-9])?\.)+[a-z0-9](?:[a-z0-9-]*[a-z0-9])?$",
-    re.IGNORECASE)
+    re.IGNORECASE,
-_IP_RE = re.compile(r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}')
+)
 _IP_RE = re.compile(r"\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}")
 _HEX = frozenset(string.hexdigits)
 _HASH_TYPES = {32: PasswordType.MD5, 40: PasswordType.SHA1, 64: PasswordType.SHA256}
 def valid_email(candidate: str) -> bool:
-    return candidate.count('@') == 1 and _EMAIL_RE.match(candidate) is not None
+    return candidate.count("@") == 1 and _EMAIL_RE.match(candidate) is not None
 def password_type(password: str) -> PasswordType:
    hash_type = _HASH_TYPES.get(len(password))
    if hash_type is not None and _HEX.issuperset(password):
        return hash_type
-    if 50 < len(password) < 70 and password[:4] in ('$2a$', '$2b$', '$2y$'):
+    if 50 < len(password) < 70 and password[:4] in ("$2a$", "$2b$", "$2y$"):
        return PasswordType.B_CRYPT
    return PasswordType.DECRYPTED
@@ -43,7 +44,7 @@ def parse_line(row: str, separator: str, flip: bool = False):
    """Try to read one ``email<sep>password`` record; ``None`` if implausible."""
    if separator not in row:
        return None
-    if separator != ' ':
+    if separator != " ":
        parts = row.split(separator)
        if len(parts) > 2:
            # [email protected]:kiyan1365:16 -> keep the last two fields.
@@ -51,40 +52,40 @@ def parse_line(row: str, separator: str, flip: bool = False):
    mail, password = row.split(separator, 1)
    if flip:
        mail, password = password, mail
-    password = password.replace(' ', '')
+    password = password.replace(" ", "")
-    if not password or '\n\n' in password:
+    if not password or "\n\n" in password:
        return None
    kind = password_type(password)
    # Plaintext beyond 30 chars is junk; hashes are longer by nature and are kept
    # so they can be attacked later with `crack`.
    if kind == PasswordType.DECRYPTED and len(password) > 30:
        return None
-    if set(password) == {'?'}:  # [email protected]:?????? ??? ?????
+    if set(password) == {"?"}:  # [email protected]:?????? ??? ?????
        return None
-    if password.count('.') >= 3 and _IP_RE.search(password):
+    if password.count(".") >= 3 and _IP_RE.search(password):
        return None
-    mail = mail.replace(' ', '')
+    mail = mail.replace(" ", "")
-    if '@' not in mail:  # https://grox.net/utils/encoding.html
+    if "@" not in mail:  # https://grox.net/utils/encoding.html
-        mail = mail.replace('%40', '@')
+        mail = mail.replace("%40", "@")
    if not mail:
        return None
    last_char, first_char = mail[-1], mail[0]
    # [email protected]\ ilovebois3 / [email protected] atohiyamamot
-    if last_char in '<.\\>_':
+    if last_char in "<.\\>_":
        mail = mail[:-1]
-    if first_char in '.<':
+    if first_char in ".<":
        mail = mail[1:]
-    if mail.endswith('@com'):  # iemongozalu@gmail@com mizo1210
+    if mail.endswith("@com"):  # iemongozalu@gmail@com mizo1210
-        mail = mail.replace('@com', '.com')
+        mail = mail.replace("@com", ".com")
    if not valid_email(mail):
        # some people replace neanta,[email protected],jp
-        mail = mail.replace(',', '.')
+        mail = mail.replace(",", ".")
-        parts = mail.split('@')
+        parts = mail.split("@")
        if len(parts) > 2:
-            mail = '@'.join(parts[:2])
+            mail = "@".join(parts[:2])
    # TODO: do it for most common domain names.
-    if mail.count('@hotmail.com') == 2:  # [email protected]@hotmail.com:8765432f
+    if mail.count("@hotmail.com") == 2:  # [email protected]@hotmail.com:8765432f
-        mail = mail.replace('@hotmail.com', '', 1)
+        mail = mail.replace("@hotmail.com", "", 1)
    if valid_email(mail) and not valid_email(password):
        return mail, password, kind
    return None
@@ -93,17 +94,17 @@ def parse_line(row: str, separator: str, flip: bool = False):
 def parseline(s: str):
    """Clean one raw line and return ``(email, password, type)`` or ``None``."""
    b = filter_printable(s.strip())
-    while '..' in b:
+    while ".." in b:
-        b = b.replace('..', '')
+        b = b.replace("..", "")
-    b = b.removeprefix('mailto:').removeprefix('E-mail:')
+    b = b.removeprefix("mailto:").removeprefix("E-mail:")
-    b = b.removeprefix(':').removesuffix(':')
+    b = b.removeprefix(":").removesuffix(":")
    # TODO: could be moved just for the email, but we don't need 100% accuracy.
-    b = b.replace('.@', '@').replace('@.', '@').replace(' @ ', '@').replace('@@', '@')
+    b = b.replace(".@", "@").replace("@.", "@").replace(" @ ", "@").replace("@@", "@")
-    if ':' in b and ';' in b:
+    if ":" in b and ";" in b:
-        b = b.replace(';', ':')
+        b = b.replace(";", ":")
    # A valid email needs an '@'; '%40' is the only cleanup that can create one.
    # Skip the whole separator scan for lines that can never yield a pair.
-    if '@' not in b and '%40' not in b:
+    if "@" not in b and "%40" not in b:
        return None
    for flip in (False, True):
        for separator in VALID_SEPARATORS:
@@ -121,8 +122,12 @@ class Processor:
    per worker is a single read buffer regardless of file size.
    """
-    def __init__(self, success_filename: str, failure_filename: str,
+    def __init__(
-                 hashes_filename: str | None = None):
+        self,
        success_filename: str,
        failure_filename: str,
        hashes_filename: str | None = None,
    ):
        self.success_filename = success_filename
        self.failure_filename = failure_filename
        self.hashes_filename = hashes_filename
@@ -143,14 +148,18 @@ class Processor:
        # Unique per task, not per process: one worker may pull several files
        # from the queue, and a pid-named part would overwrite the previous one.
        part = uuid.uuid4().hex
-        success_part = f'{self.success_filename}.part.{part}'
+        success_part = f"{self.success_filename}.part.{part}"
-        failure_part = f'{self.failure_filename}.part.{part}'
+        failure_part = f"{self.failure_filename}.part.{part}"
-        hashes_part = f'{self.hashes_filename}.part.{part}' if self.hashes_filename else None
+        hashes_part = (
            f"{self.hashes_filename}.part.{part}" if self.hashes_filename else None
        )
        parts = [success_part, failure_part] + ([hashes_part] if hashes_part else [])
-        with open(str(filename), encoding='utf8', errors='ignore') as r, \
+        with (
-                open(success_part, 'w', encoding='utf8') as sw, \
+            open(str(filename), encoding="utf8", errors="ignore") as r,
-                open(failure_part, 'w', encoding='utf8') as fw:
+            open(success_part, "w", encoding="utf8") as sw,
-            hashes = open(hashes_part, 'w', encoding='utf8') if hashes_part else None  # noqa: SIM115
+            open(failure_part, "w", encoding="utf8") as fw,
        ):
            hashes = open(hashes_part, "w", encoding="utf8") if hashes_part else None  # noqa: SIM115
            try:
                for line in r:
                    result = parseline(line)
@@ -160,10 +169,10 @@ class Processor:
                        continue
                    mail, password, kind = result
                    if kind == PasswordType.DECRYPTED:
-                        sw.write(f'{str(mail).lower().strip()}:{password.strip()}\n')
+                        sw.write(f"{str(mail).lower().strip()}:{password.strip()}\n")
                        success += 1
                    elif hashes is not None:
-                        hashes.write(f'{str(mail).lower().strip()}:{password}\n')
+                        hashes.write(f"{str(mail).lower().strip()}:{password}\n")
                    else:
                        fw.write(filter_printable(line))  # no hash sink: keep as reject
                        failure += 1
@@ -177,7 +186,7 @@ def _clear_parts(*finals: str) -> None:
    # Leftovers mean a previous run died before merging. Those files were never
    # recorded in progress.txt, so they are reprocessed; drop the partial parts.
    for final in finals:
-        for part in glob(f'{final}.part.*'):
+        for part in glob(f"{final}.part.*"):
            os.remove(part)
@@ -189,35 +198,50 @@ def _merge_part_list(parts: list[str]) -> None:
    """
    groups: dict[str, list[str]] = {}
    for part in parts:
-        groups.setdefault(part.rsplit('.part.', 1)[0], []).append(part)
+        groups.setdefault(part.rsplit(".part.", 1)[0], []).append(part)
    for final, part_files in groups.items():
-        with open(final, 'ab') as writer:
+        with open(final, "ab") as writer:
            for part in part_files:
-                with open(part, 'rb') as reader:
+                with open(part, "rb") as reader:
                    shutil.copyfileobj(reader, writer, length=1024 * 1024)
                os.remove(part)
-def parse_files(files: list[Path], success_filename: str, failure_filename: str,
+def parse_files(
-                progress_filename: str | None = None,
+    files: list[Path],
-                hashes_filename: str | None = None) -> tuple[int, int]:
+    success_filename: str,
    failure_filename: str,
    progress_filename: str | None = None,
    hashes_filename: str | None = None,
 ) -> tuple[int, int]:
    """Parse an explicit list of files; used by both ``parse`` and ``build``.
    Returns the total ``(success, failure)`` counts. Hashed passwords go to
    ``hashes_filename`` when given. When ``progress_filename`` is given, each
    file is recorded after its chunk is safely merged, so a rerun can skip it.
    """
-    _clear_parts(success_filename, failure_filename,
+    _clear_parts(
-                 *([hashes_filename] if hashes_filename else []))
+        success_filename,
        failure_filename,
        *([hashes_filename] if hashes_filename else []),
    )
    success_total = failure_total = 0
    workers = os.cpu_count() or 1
    processor = Processor(success_filename, failure_filename, hashes_filename)
    start_time = time()
    with ExitStack() as stack:
-        progress = (stack.enter_context(open(progress_filename, 'a', encoding='utf8', buffering=1))
+        progress = (
-                    if progress_filename else None)
+            stack.enter_context(
                open(progress_filename, "a", encoding="utf8", buffering=1)
            )
            if progress_filename
            else None
        )
        pending_parts: list[str] = []
-        with ProcessPoolExecutor(max_workers=workers) as executor, tqdm(total=len(files)) as bar:
+        with (
            ProcessPoolExecutor(max_workers=workers) as executor,
            tqdm(total=len(files)) as bar,
        ):
            # Submit everything and merge each task's own part as it completes:
            # no per-chunk barrier, so one large file cannot idle the other
            # workers. ponytail: all futures live at once (~tens of MB for
@@ -232,24 +256,32 @@ def parse_files(files: list[Path], success_filename: str, failure_filename: str,
                    _merge_part_list(pending_parts)
                    pending_parts.clear()
                bar.update(1)
-                bar.set_description(f'success = {success_total:,} failure = {failure_total:,}')
+                bar.set_description(
                    f"success = {success_total:,} failure = {failure_total:,}"
                )
                if progress:
-                    progress.write(f'{futures[future]}\n')
+                    progress.write(f"{futures[future]}\n")
        if pending_parts:
            _merge_part_list(pending_parts)
-    print(f'Time elapsed {time() - start_time:.3f} seconds.')
+    print(f"Time elapsed {time() - start_time:.3f} seconds.")
    return success_total, failure_total
-def parse_to_files(input_dir: str, success_filename: str, failure_filename: str,
+def parse_to_files(
-                   hashes_filename: str | None = None) -> None:
+    input_dir: str,
    success_filename: str,
    failure_filename: str,
    hashes_filename: str | None = None,
 ) -> None:
    """Parse every file under ``input_dir``, resuming from ``progress.txt``."""
    # Stream the progress file; never build a second in-memory copy of it (it
    # can grow to millions of lines).
    done: set[str] = set()
-    if os.path.exists('progress.txt'):
+    if os.path.exists("progress.txt"):
-        with open('progress.txt', encoding='utf8', errors='ignore') as reader:
+        with open("progress.txt", encoding="utf8", errors="ignore") as reader:
-            done = {line.rstrip('\n') for line in reader}
+            done = {line.rstrip("\n") for line in reader}
-    files = [p for p in Path(input_dir).glob('**/*') if p.is_file() and str(p) not in done]
+    files = [p for p in find_files(input_dir) if str(p) not in done]
    shuffle(files)
-    parse_files(files, success_filename, failure_filename, 'progress.txt', hashes_filename)
+    parse_files(
        files, success_filename, failure_filename, "progress.txt", hashes_filename
    )
@@ -10,19 +10,19 @@ def postprocess(results: list[str]) -> list[str]:
    """Normalise ``email:password`` lines into a sorted, de-duplicated list."""
    emails_passwords: set[str] = set()
    for txt_line in results:
-        if '@' not in txt_line:  # does it contain an email address?
+        if "@" not in txt_line:  # does it contain an email address?
            continue
-        if ':' in txt_line:  # '[email protected]:iskandar89
+        if ":" in txt_line:  # '[email protected]:iskandar89
-            separator = ':'
+            separator = ":"
-        elif ';' in txt_line:  # '[email protected];ol1391ga
+        elif ";" in txt_line:  # '[email protected];ol1391ga
-            separator = ';'
+            separator = ";"
        else:
            continue
        try:
            email, password = txt_line.strip().split(separator)
        except ValueError:
            continue
-        emails_passwords.add(f'{email}:{password}')
+        emails_passwords.add(f"{email}:{password}")
    return sorted(emails_passwords)
@@ -35,13 +35,13 @@ def glob_to_regex(pattern: str) -> str:
    """
    parts = []
    for char in pattern:
-        if char == '*':
+        if char == "*":
-            parts.append('[^:]*')
+            parts.append("[^:]*")
-        elif char == '?':
+        elif char == "?":
-            parts.append('[^:]')
+            parts.append("[^:]")
        else:
            parts.append(re.escape(char))
-    return '^' + ''.join(parts) + ':'
+    return "^" + "".join(parts) + ":"
 class QueryAPI:
@@ -66,7 +66,9 @@ class QueryAPI:
        assert path.is_file(), path
        try:
            output = subprocess.check_output(
-                ['rg', '--no-line-number', '-ie', '^' + re.escape(email), str(path)], text=True)
+                ["rg", "--no-line-number", "-ie", "^" + re.escape(email), str(path)],
                text=True,
            )
        except subprocess.CalledProcessError:
            return []
        return [line for line in output.splitlines() if line]
@@ -79,31 +81,44 @@ class QueryAPI:
        file_list, dir_list = QueryAPI.scanning(path)
        current = characters[index].lower()
        if current in file_list:
-            return QueryAPI.search_in_file(file_list[current], ''.join(characters))
+            return QueryAPI.search_in_file(file_list[current], "".join(characters))
        if current in dir_list:
            return QueryAPI.recursive_search(dir_list[current], characters, index + 1)
-        if 'symbols' in file_list:
+        if "symbols" in file_list:
-            return QueryAPI.search_in_file(file_list['symbols'], ''.join(characters))
+            return QueryAPI.search_in_file(file_list["symbols"], "".join(characters))
        return []
    @staticmethod
    def query(data_dir: str, request: str) -> list[str]:
        """Exact email lookup using the bucket path (O(1) file read)."""
-        if len(request) <= 3 or '@' not in request:
+        if len(request) <= 3 or "@" not in request:
            return []
        return postprocess(QueryAPI.recursive_search(Path(data_dir), list(request), 0))
    @staticmethod
-    def wildcard(data_dir: str, pattern: str) -> list[str]:
+    def wildcard(data_dir: str, pattern: str):
-        """Glob search across the whole dataset with ripgrep.
+        """Stream matching lines across the dataset with ripgrep.
-        ponytail: full scan of the tree (ripgrep is fast enough); add a
+        Streamed line by line, so a broad wildcard over a huge dataset stays
-        domain-sharded index only if wildcard latency becomes a problem.
+        memory-bounded. ponytail: no sort/dedup (the dataset is already unique)
        and a full scan; add a domain-sharded index only if latency matters.
        """
-        try:
+        process = subprocess.Popen(
-            output = subprocess.check_output(
+            [
-                ['rg', '--no-filename', '--no-line-number', '-i', '-e',
+                "rg",
-                 glob_to_regex(pattern), str(Path(data_dir))], text=True)
+                "--no-filename",
-        except subprocess.CalledProcessError:
+                "--no-line-number",
-            return []
+                "-i",
-        return postprocess(output.splitlines())
+                "-e",
                glob_to_regex(pattern),
                str(Path(data_dir)),
            ],
            stdout=subprocess.PIPE,
            text=True,
        )
        with process:
            assert process.stdout is not None
            for line in process.stdout:
                line = line.rstrip("\n")
                if line:
                    yield line
@@ -7,32 +7,32 @@ from time import time
 from collectionparser.utils import create_dir_for_file
-_ALNUM = 'abcdefghijklmnopqrstuvwxyz0123456789'
+_ALNUM = "abcdefghijklmnopqrstuvwxyz0123456789"
-def split_file(input_file: str, output_dir: str = 'out', restart_from: int = 0) -> None:
+def split_file(input_file: str, output_dir: str = "out", restart_from: int = 0) -> None:
    """Bucket every ``email:password`` line of ``input_file`` by email prefix.
-    Resume from the ``restart_from``-th input line. A few records may land in
+    Resume from the ``restart_from``-th input line.
    the wrong bucket; run ``clean`` afterwards.
    """
    os.makedirs(output_dir, exist_ok=True)
    handles: dict[str, object] = {}
    tick = time()
-    with open(input_file, encoding='utf8', errors='ignore') as reader:
+    with open(input_file, encoding="utf8", errors="ignore") as reader:
        for i, line in enumerate(reader):
            if i % 10_000_000 == 0:
                elapsed = time() - tick
                tick = time()
-                print(f'|- Processed {i:,} rows in {elapsed:.2f} seconds.')
+                print(f"|- Processed {i:,} rows in {elapsed:.2f} seconds.")
-            if i < restart_from or ':' not in line:
+            if i < restart_from or ":" not in line:
                continue
-            email = line.strip().split(':', 1)[0]
+            email = line.strip().split(":", 1)[0]
-            write_line_to_file(handles, line, f'{output_dir}/{get_output_file(email)}')
+            write_line_to_file(handles, line, f"{output_dir}/{get_output_file(email)}")
-def write_line_to_file(handles: dict[str, object], line: str, output_file: str,
+def write_line_to_file(
-                       buffer_scale: int = 100) -> None:
+    handles: dict[str, object], line: str, output_file: str, buffer_scale: int = 100
 ) -> None:
    """Append ``line`` to ``output_file``, keeping at most 100 handles open.
    Evicted handles are flushed and closed so the process stays under the file
@@ -45,8 +45,11 @@ def write_line_to_file(handles: dict[str, object], line: str, output_file: str,
            handles.pop(next(iter(handles))).flush()  # type: ignore[attr-defined]
        create_dir_for_file(output_file)
        handles[output_file] = open(  # noqa: SIM115 - cached handle, not scope-bound
-            output_file, mode='a+', encoding='utf8',
+            output_file,
-            buffering=buffer_scale * io.DEFAULT_BUFFER_SIZE)
+            mode="a+",
            encoding="utf8",
            buffering=buffer_scale * io.DEFAULT_BUFFER_SIZE,
        )
        handles[output_file].write(line)  # type: ignore[attr-defined]
@@ -59,5 +62,5 @@ def get_output_file(email: str) -> str:
            break
        parts.append(char)
    if len(parts) < 3:
-        parts.append('symbols')
+        parts.append("symbols")
-    return '/'.join(parts)
+    return "/".join(parts)
@@ -1,14 +1,20 @@
 """Small shared helpers: deterministic shuffling, line streaming, path handling."""
 import os
 import random
 import re
 import string
 from enum import StrEnum
 from pathlib import Path
 import click
-_NON_PRINTABLE = re.compile(f'[^{re.escape(string.printable)}]')
+_NON_PRINTABLE = re.compile(f"[^{re.escape(string.printable)}]")
 def find_files(path) -> list[Path]:
    """Return every regular file under ``path`` (recursively)."""
    root = Path(path).expanduser()
    return [p for p in root.glob("**/*") if p.is_file()]
 def shuffle(items: list) -> None:
@@ -18,40 +24,65 @@ def shuffle(items: list) -> None:
 def filter_printable(text: str) -> str:
    """Drop non-printable characters so no binary data is written to text files."""
-    return _NON_PRINTABLE.sub('', text)
+    return _NON_PRINTABLE.sub("", text)
 def create_dir_for_file(filename: str) -> None:
    """Create the parent directory of ``filename`` if it is missing."""
-    if parent := os.path.dirname(filename):
+    parent = Path(filename).parent
-        os.makedirs(parent, exist_ok=True)
+    if parent != Path("."):
        parent.mkdir(parents=True, exist_ok=True)
 def input_file() -> click.Path:
-    return click.Path(exists=True, file_okay=True, dir_okay=False,
+    return click.Path(
-                      writable=False, readable=True, resolve_path=True)
+        exists=True,
        file_okay=True,
        dir_okay=False,
        writable=False,
        readable=True,
        resolve_path=True,
    )
 def input_dir() -> click.Path:
-    return click.Path(exists=True, file_okay=False, dir_okay=True,
+    return click.Path(
-                      writable=False, readable=True, resolve_path=True)
+        exists=True,
        file_okay=False,
        dir_okay=True,
        writable=False,
        readable=True,
        resolve_path=True,
    )
 def output_file() -> click.Path:
-    return click.Path(exists=False, file_okay=True, dir_okay=False,
+    return click.Path(
-                      writable=True, readable=True, resolve_path=True)
+        exists=False,
        file_okay=True,
        dir_okay=False,
        writable=True,
        readable=True,
        resolve_path=True,
    )
 def output_dir() -> click.Path:
-    return click.Path(exists=False, file_okay=False, dir_okay=True,
+    return click.Path(
-                      writable=True, readable=True, resolve_path=True)
+        exists=False,
        file_okay=False,
        dir_okay=True,
        writable=True,
        readable=True,
        resolve_path=True,
    )
 class PasswordType(StrEnum):
    """Classification of a password field; only ``DECRYPTED`` is queryable."""
-    MD5 = 'MD5'
+    MD5 = "MD5"
-    SHA1 = 'SHA1'
+    SHA1 = "SHA1"
-    SHA256 = 'SHA256'
+    SHA256 = "SHA256"
-    B_CRYPT = 'B_CRYPT'
+    B_CRYPT = "B_CRYPT"
-    DECRYPTED = 'DECRYPTED'
+    DECRYPTED = "DECRYPTED"
@@ -33,8 +33,13 @@ include = ["collectionparser*"]
 [tool.ruff]
 target-version = "py311"
-line-length = 120
+line-length = 88
 src = ["collectionparser", "tests"]
 [tool.ruff.lint]
-select = ["E4", "E7", "E9", "F", "I", "UP", "B", "SIM"]
+select = ["E", "W", "F", "I", "N", "UP", "B", "C4", "SIM"]
 ignore = ["E501"]  # long lines are handled by the formatter
 [tool.ruff.format]
 quote-style = "double"
 indent-style = "space"
@@ -9,51 +9,69 @@ from collectionparser.parser import Processor
 class CrackTest(unittest.TestCase):
    def test_hash_saved_by_processor(self):
        with tempfile.TemporaryDirectory() as tmp:
-            src = Path(tmp) / 'in.txt'
+            src = Path(tmp) / "in.txt"
-            src.write_text('[email protected]:5f4dcc3b5aa765d61d8327deb882cf99\n[email protected]:plainpw\n',
+            src.write_text(
-                           encoding='utf8')
+                "[email protected]:5f4dcc3b5aa765d61d8327deb882cf99\n[email protected]:plainpw\n",
-            processor = Processor(str(Path(tmp) / 's.txt'), str(Path(tmp) / 'f.txt'),
+                encoding="utf8",
-                                  str(Path(tmp) / 'h.txt'))
+            )
            processor = Processor(
                str(Path(tmp) / "s.txt"),
                str(Path(tmp) / "f.txt"),
                str(Path(tmp) / "h.txt"),
            )
            processor.process(src)
-            hashes = ''.join(Path(p).read_text(encoding='utf8')
+            hashes = "".join(
-                             for p in glob(str(Path(tmp) / 'h.txt.part.*')))
+                Path(p).read_text(encoding="utf8")
-            success = ''.join(Path(p).read_text(encoding='utf8')
+                for p in glob(str(Path(tmp) / "h.txt.part.*"))
-                              for p in glob(str(Path(tmp) / 's.txt.part.*')))
+            )
-            self.assertEqual('[email protected]:5f4dcc3b5aa765d61d8327deb882cf99\n', hashes)
+            success = "".join(
-            self.assertEqual('[email protected]:plainpw\n', success)
+                Path(p).read_text(encoding="utf8")
                for p in glob(str(Path(tmp) / "s.txt.part.*"))
            )
            self.assertEqual("[email protected]:5f4dcc3b5aa765d61d8327deb882cf99\n", hashes)
            self.assertEqual("[email protected]:plainpw\n", success)
    def test_crack_md5_and_sha1(self):
        with tempfile.TemporaryDirectory() as tmp:
-            hashes = Path(tmp) / 'hashes.txt'
+            hashes = Path(tmp) / "hashes.txt"
            hashes.write_bytes(
-                b'[email protected]:' + hashlib.md5(b'hunter2').hexdigest().encode() + b'\n'
+                b"[email protected]:" + hashlib.md5(b"hunter2").hexdigest().encode() + b"\n"
-                b'[email protected]:' + hashlib.sha1(b'secret').hexdigest().encode() + b'\n'
+                b"[email protected]:" + hashlib.sha1(b"secret").hexdigest().encode() + b"\n"
-                b'[email protected]:not-a-hash\n')
+                b"[email protected]:not-a-hash\n"
-            wordlist = Path(tmp) / 'words.txt'
+            )
-            wordlist.write_bytes(b'foo\nhunter2\nbar\nsecret\n')
+            wordlist = Path(tmp) / "words.txt"
-            out = Path(tmp) / 'cracked.txt'
+            wordlist.write_bytes(b"foo\nhunter2\nbar\nsecret\n")
            out = Path(tmp) / "cracked.txt"
            total = crack(str(hashes), str(out), wordlists=[str(wordlist)], workers=2)
            self.assertEqual(2, total)
-            self.assertEqual({b'[email protected]:hunter2', b'[email protected]:secret'},
+            self.assertEqual(
-                             set(out.read_bytes().splitlines()))
+                {b"[email protected]:hunter2", b"[email protected]:secret"},
                set(out.read_bytes().splitlines()),
            )
    def test_crack_from_plaintext_passwords(self):
        with tempfile.TemporaryDirectory() as tmp:
-            hashes = Path(tmp) / 'hashes.txt'
+            hashes = Path(tmp) / "hashes.txt"
-            hashes.write_bytes(b'[email protected]:' + hashlib.md5(b'reused').hexdigest().encode() + b'\n')
+            hashes.write_bytes(
-            empty = Path(tmp) / 'empty.txt'
+                b"[email protected]:" + hashlib.md5(b"reused").hexdigest().encode() + b"\n"
-            empty.write_bytes(b'')
+            )
-            plaintext = Path(tmp) / 'plain.txt'
+            empty = Path(tmp) / "empty.txt"
-            plaintext.write_bytes(b'[email protected]:reused\n')
+            empty.write_bytes(b"")
-            out = Path(tmp) / 'cracked.txt'
+            plaintext = Path(tmp) / "plain.txt"
-            total = crack(str(hashes), str(out), wordlists=[str(empty)],
+            plaintext.write_bytes(b"[email protected]:reused\n")
-                          passwords_from=[str(plaintext)], workers=1)
+            out = Path(tmp) / "cracked.txt"
            total = crack(
                str(hashes),
                str(out),
                wordlists=[str(empty)],
                passwords_from=[str(plaintext)],
                workers=1,
            )
            self.assertEqual(1, total)
-            self.assertEqual([b'[email protected]:reused'], out.read_bytes().splitlines())
+            self.assertEqual([b"[email protected]:reused"], out.read_bytes().splitlines())
-if __name__ == '__main__':
+if __name__ == "__main__":
    unittest.main()
@@ -6,114 +6,133 @@ from collectionparser.utils import PasswordType
 class ParserTest(unittest.TestCase):
    def test_parse_1(self):
-        self.assertIsNotNone(p('[email protected]:123'))
+        self.assertIsNotNone(p("[email protected]:123"))
-        self.assertIsNotNone(p('hello@@gmail.com:123'))
+        self.assertIsNotNone(p("hello@@gmail.com:123"))
-        self.assertIsNotNone(p('hello @ gmail.com:123'))
+        self.assertIsNotNone(p("hello @ gmail.com:123"))
-        self.assertIsNotNone(p('hello @@ gmail.com:123'))
+        self.assertIsNotNone(p("hello @@ gmail.com:123"))
-        self.assertIsNotNone(p('hello @@ gmail.com:123:'))
+        self.assertIsNotNone(p("hello @@ gmail.com:123:"))
-        self.assertIsNone(p('[email protected]:'))
+        self.assertIsNone(p("[email protected]:"))
-        self.assertIsNotNone(p('[email protected] 2296'))
+        self.assertIsNotNone(p("[email protected] 2296"))
-        self.assertIsNotNone(p('hello @@ gmail.com,123'))
+        self.assertIsNotNone(p("hello @@ gmail.com,123"))
-        self.assertIsNotNone(p('123,hello @@ gmail.com'))
+        self.assertIsNotNone(p("123,hello @@ gmail.com"))
-        self.assertIsNone(p('[email protected]'))
+        self.assertIsNone(p("[email protected]"))
-        self.assertIsNone(p('[email protected]:?????? ??? ?????'))
+        self.assertIsNone(p("[email protected]:?????? ??? ?????"))
-        self.assertIsNone(p('[email protected]:85.242.201.81,felix'))
+        self.assertIsNone(
-        self.assertIsNone(p('[email protected]:125.26.106.158,nopparat'))
+            p("[email protected]:85.242.201.81,felix")
-        self.assertIsNone(p('[email protected]:INSERT INTO `users'
+        )
-                            '`(`id`, `verify`, `username`, `email`, `password'))
+        self.assertIsNone(
-        self.assertIsNotNone(p('[email protected]:Login'))
+            p("[email protected]:125.26.106.158,nopparat")
-        self.assertIsNotNone(p('[email protected] ilovebois3'))
+        )
-        self.assertIsNotNone(p('[email protected] atohiyamamot'))
+        self.assertIsNone(
-        self.assertIsNotNone(p('[email protected]@hotmail.com:8765432f'))
+            p(
-        self.assertIsNotNone(p('neanta,[email protected],jp:123'))
+                "[email protected]:INSERT INTO `users"
-        self.assertIsNotNone(p('iemongozalu@gmail@com mizo1210'))
+                "`(`id`, `verify`, `username`, `email`, `password"
-        self.assertIsNotNone(p('[email protected]@hotmail.com:sayonara'))
+            )
-        self.assertIsNone(p('iemongozalu@gmail@com:'))
+        )
-        self.assertIsNone(p('iemongozalu@gmail@com,'))
+        self.assertIsNotNone(p("[email protected]:Login"))
-        self.assertIsNotNone(p('[email protected] chi chi 123'))
+        self.assertIsNotNone(p("[email protected] ilovebois3"))
-        self.assertIsNotNone(p('[email protected]@34589@@@@@@:19820429'))
+        self.assertIsNotNone(p("[email protected] atohiyamamot"))
-        self.assertIsNotNone(p('[email protected]@33584@@@@@@ 03260702'))
+        self.assertIsNotNone(p("[email protected]@hotmail.com:8765432f"))
        self.assertIsNotNone(p("neanta,[email protected],jp:123"))
        self.assertIsNotNone(p("iemongozalu@gmail@com mizo1210"))
        self.assertIsNotNone(p("[email protected]@hotmail.com:sayonara"))
        self.assertIsNone(p("iemongozalu@gmail@com:"))
        self.assertIsNone(p("iemongozalu@gmail@com,"))
        self.assertIsNotNone(p("[email protected] chi chi 123"))
        self.assertIsNotNone(p("[email protected]@34589@@@@@@:19820429"))
        self.assertIsNotNone(p("[email protected]@33584@@@@@@ 03260702"))
-        self.assertIsNotNone(p('<[email protected]:sedagirl'))
+        self.assertIsNotNone(p("<[email protected]:sedagirl"))
-        self.assertIsNotNone(p('<[email protected]>:<phoenix>'))
+        self.assertIsNotNone(p("<[email protected]>:<phoenix>"))
-        self.assertIsNotNone(p('[email protected]< gorillague'))
+        self.assertIsNotNone(p("[email protected]< gorillague"))
-        self.assertIsNone(p('</td>@bk.ru:peng7788525'))
+        self.assertIsNone(p("</td>@bk.ru:peng7788525"))
-        self.assertIsNone(p('<admin/>@mail.ru:andrew123'))
+        self.assertIsNone(p("<admin/>@mail.ru:andrew123"))
-        self.assertIsNone(p('<i>test</i>@bk.ru:lippens'))
+        self.assertIsNone(p("<i>test</i>@bk.ru:lippens"))
-        self.assertIsNone(p('ESSENTIALWAY'))
+        self.assertIsNone(p("ESSENTIALWAY"))
-        self.assertIsNone(p('ESTARRON-ESPINOSA'))
+        self.assertIsNone(p("ESTARRON-ESPINOSA"))
-        self.assertIsNone(p('ERLONMIRANDA:sha98ne'))
+        self.assertIsNone(p("ERLONMIRANDA:sha98ne"))
        # No '@' and no '%40' -> cannot yield a pair, must stay None.
-        self.assertIsNone(p('no at sign at all'))
+        self.assertIsNone(p("no at sign at all"))
-        self.assertIsNone(p('plain text line:with:colons'))
+        self.assertIsNone(p("plain text line:with:colons"))
-        self.assertIsNotNone(p('user%40domain.com:password'))  # %40 still becomes '@'
+        self.assertIsNotNone(p("user%40domain.com:password"))  # %40 still becomes '@'
-        self.assertIsNotNone(p('[email protected]:[email protected]:victory'))
+        self.assertIsNotNone(
-        self.assertIsNotNone(p('[email protected];;shanmuga6'))
+            p("[email protected]:[email protected]:victory")
-        self.assertIsNotNone(p('[email protected]::shanmuga6'))
+        )
-        self.assertIsNotNone(p('[email protected]::hell@'))
+        self.assertIsNotNone(p("[email protected];;shanmuga6"))
-        self.assertIsNotNone(p('[email protected]::hell@:'))
+        self.assertIsNotNone(p("[email protected]::shanmuga6"))
-        self.assertIsNotNone(p('Aliulfat [email protected]:4849040'))
+        self.assertIsNotNone(p("[email protected]::hell@"))
-        self.assertIsNotNone(p('harry gill@ yaho.com:7087096771'))
+        self.assertIsNotNone(p("[email protected]::hell@:"))
-        self.assertIsNotNone(p('shaheen .jafri @gmail.com:destinyuntold'))
+        self.assertIsNotNone(p("Aliulfat [email protected]:4849040"))
-        self.assertIsNotNone(p('mail@ottino snc.it:04921540011'))
+        self.assertIsNotNone(p("harry gill@ yaho.com:7087096771"))
-        self.assertIsNone(p('sssssssandhya@gmail.:puleyar1'))
+        self.assertIsNotNone(p("shaheen .jafri @gmail.com:destinyuntold"))
        self.assertIsNotNone(p("mail@ottino snc.it:04921540011"))
        self.assertIsNone(p("sssssssandhya@gmail.:puleyar1"))
-        self.assertIsNotNone(p('cjywzy%40eyou.com:Aebdd797'))
+        self.assertIsNotNone(p("cjywzy%40eyou.com:Aebdd797"))
-        self.assertIsNotNone(p('finanza.rapido2013%40yahoo.it:123456'))
+        self.assertIsNotNone(p("finanza.rapido2013%40yahoo.it:123456"))
-        self.assertIsNotNone(p('habitacasa3%40yahoo.it:cesira89'))
+        self.assertIsNotNone(p("habitacasa3%40yahoo.it:cesira89"))
-        self.assertIsNotNone(p('harry gill@ yaho.com:7087096771'))
+        self.assertIsNotNone(p("harry gill@ yaho.com:7087096771"))
-        self.assertIsNone(p('uwe kopka <[email protected]>:biochemie'))
+        self.assertIsNone(p("uwe kopka <[email protected]>:biochemie"))
        # multi-line.
-        self.assertIsNotNone(p('INLOVE12:::1920-05-02:[email protected]:INLOVE12'))
+        self.assertIsNotNone(p("INLOVE12:::1920-05-02:[email protected]:INLOVE12"))
-        self.assertIsNotNone(p('Nicesweetboobs:1983-10-05:[email protected]:ilovekiss'))
+        self.assertIsNotNone(
-        self.assertIsNotNone(p('[email protected]:[email protected];97tata1'))
+            p("Nicesweetboobs:1983-10-05:[email protected]:ilovekiss")
        )
        self.assertIsNotNone(
            p("[email protected]:[email protected];97tata1")
        )
        multi_lines = [
-            '[email protected];[email protected]:CAROLINAPIANTA',
+            "[email protected];[email protected]:CAROLINAPIANTA",
-            'Nicesweetboobs:1983-10-05:[email protected]:ilovekiss',
+            "Nicesweetboobs:1983-10-05:[email protected]:ilovekiss",
-            '190.40.252.191,XabelciberX,[email protected],gatita9685567',
+            "190.40.252.191,XabelciberX,[email protected],gatita9685567",
-            '201.245.164.50,andres vergara,[email protected],barcelona10',
+            "201.245.164.50,andres vergara,[email protected],barcelona10",
-            '65.94.11.229,stephane,[email protected],ste264',
+            "65.94.11.229,stephane,[email protected],ste264",
-            'INLOVE12:::1920-05-02:[email protected]:INLOVE12',
+            "INLOVE12:::1920-05-02:[email protected]:INLOVE12",
-            'agaliha84:::1969-07-02:[email protected]:agaliha84',
+            "agaliha84:::1969-07-02:[email protected]:agaliha84",
-            'happy8360:::1963-09-12:[email protected]:jdj407',
+            "happy8360:::1963-09-12:[email protected]:jdj407",
-            'tinkertailor99:1947-10-12:[email protected]:77880577',
+            "tinkertailor99:1947-10-12:[email protected]:77880577",
-            'Cash5178:1970-11-06:[email protected]:hulk1234',
+            "Cash5178:1970-11-06:[email protected]:hulk1234",
-            'Creecher0275:1972-04-22:[email protected]:complete',
+            "Creecher0275:1972-04-22:[email protected]:complete",
-            '[email protected]:[email protected];97tata1',
+            "[email protected]:[email protected];97tata1",
-            '[email protected]:[email protected];zwezda',
+            "[email protected]:[email protected];zwezda",
-            'SavvySummer:1973-03-17:[email protected]:sunshine0699',
+            "SavvySummer:1973-03-17:[email protected]:sunshine0699",
-            'RaWRrr:1990-12-02:[email protected]:nitsua',
+            "RaWRrr:1990-12-02:[email protected]:nitsua",
-            '[email protected];[email protected]:CAROLINAPIANTA',
+            "[email protected];[email protected]:CAROLINAPIANTA",
-            '[email protected];[email protected]:RRAHOTEL',
+            "[email protected];[email protected]:RRAHOTEL",
-            'silent4463:1980-10-14:[email protected]:salinas'
+            "silent4463:1980-10-14:[email protected]:salinas",
        ]
        for multi_line in multi_lines:
            self.assertIsNotNone(p(multi_line))
-        self.assertIsNone(p('300396	matelymaly	[email protected]	222.155.173.78	'
+        self.assertIsNone(
-                            '3d2c5414a041f518b03b750e62e4ebc3:Nhc'))
+            p(
                "300396	matelymaly	[email protected]	222.155.173.78	"
                "3d2c5414a041f518b03b750e62e4ebc3:Nhc"
            )
        )
        # 32-char hex is a hash, not junk: kept (typed) so `crack` can attack it.
-        hashed = p('05-02-2012;5721869;[email protected]:f02368945726d5fc2a14eb576f7276c0')
+        hashed = p(
            "05-02-2012;5721869;[email protected]:f02368945726d5fc2a14eb576f7276c0"
        )
        self.assertIsNotNone(hashed)
        self.assertEqual(PasswordType.MD5, hashed[2])
-        self.assertIsNotNone(p('[email protected]::g1u3ck'))
+        self.assertIsNotNone(p("[email protected]::g1u3ck"))
-        self.assertIsNotNone(p('[email protected]:[email protected]:2904198'))
+        self.assertIsNotNone(p("[email protected]:[email protected]:2904198"))
-        self.assertIsNotNone(p('sanda [email protected]:vasafultv'))
+        self.assertIsNotNone(p("sanda [email protected]:vasafultv"))
-        self.assertIsNone(p('jsjjd.@Com.:love melon'))
+        self.assertIsNone(p("jsjjd.@Com.:love melon"))
    def test_parse_2(self):
-        self.assertEqual('j/o/s', get_output_file('[email protected]'))
+        self.assertEqual("j/o/s", get_output_file("[email protected]"))
-        self.assertEqual('c/e/o', get_output_file('[email protected]'))
+        self.assertEqual("c/e/o", get_output_file("[email protected]"))
-        self.assertEqual('x/symbols', get_output_file('[email protected]'))
+        self.assertEqual("x/symbols", get_output_file("[email protected]"))
-        self.assertEqual('x/a/symbols', get_output_file('[email protected]'))
+        self.assertEqual("x/a/symbols", get_output_file("[email protected]"))
-        self.assertEqual('x/symbols', get_output_file('[email protected]'))
+        self.assertEqual("x/symbols", get_output_file("[email protected]"))
-        self.assertEqual('k/e/v', get_output_file('[email protected]'))
+        self.assertEqual("k/e/v", get_output_file("[email protected]"))
-if __name__ == '__main__':
+if __name__ == "__main__":
    unittest.main()