diff --git a/README.md b/README.md
index ad6a5f8..5d90a4c 100644
--- a/README.md
+++ b/README.md
@@ -39,10 +39,12 @@ dedups each bucket. Peak disk is source + tree, so it fits where a global sort
 would not.
 
 ```bash
+# --scratch is a fast disk for per-batch files, sort temp, and build state.
+# --batch_gb caps each batch (peak disk = source + tree).
 collectionparser build \
   --source /path/to/extracted \
   --dataset /path/to/dataset \
-  --scratch /path/to/scratch \   # fast disk for batch files + sort temp
+  --scratch /path/to/scratch \
   --batch_gb 120
 ```
 
@@ -50,14 +52,18 @@ collectionparser build \
 same command. Finished batches are recorded in `build_state.json`, and parsed
 files within the current batch in `progress_<n>.txt`; only unparsed files are
 re-parsed. The sort+split step of an interrupted batch is redone (duplicates are
-removed by the final dedup), so nothing is lost. Stats are written to
-`--scratch/stats.md`.
+removed by the final dedup), so nothing is lost.
+
+State and stats live in `--scratch`: `build_state.json`, `stats.md`, and
+`extract_failures.txt` (archives that failed to extract, never silently skipped).
+`sort` temp is directed to `--scratch` via `TMPDIR`.
 
 **Archives** (`.rar`/`.zip`/`.7z`/`.tar.*`/`.gz`/`.bz2`/`.xz`) are extracted
 automatically into `--scratch` for the duration of their batch and removed once
-parsed; the source is never modified. `.zip`/`.tar.*`/single-file `.gz`/`.bz2`/`.xz`
-use the Python standard library; `.rar`/`.7z` need `7z` or `unrar` on `PATH` (use
-`--no-extract` to skip archives entirely).
+parsed; the source is never modified. `.zip`/`.tar.*`/single-file
+`.gz`/`.bz2`/`.xz` use the Python standard library; `.rar`/`.7z` need `7z` or
+`unrar` on `PATH`. Pass `--no-extract` to skip archives, or `--no-dedup` to
+skip the final per-bucket de-duplication.
 
 ## Processing (manual)
 
@@ -90,6 +96,29 @@ collectionparser query --dataset data --pattern '*@gmail.com'
 collectionparser export --dataset data --out records.jsonl
 ```
 
+## Crack hashed passwords
+
+`parse`/`build` keep hashed passwords (MD5/SHA1/SHA256) separate from plaintext
+ones instead of discarding them. `build` collects them in `--scratch/hashes.txt`;
+`parse` writes them to `--hashes_file` when given.
+
+```bash
+# Wordlist omitted -> rockyou is downloaded automatically; add the dataset's
+# own plaintext passwords as extra candidates:
+collectionparser crack \
+  --hashes hashes.txt \
+  --passwords-from dataset/ \
+  --out cracked.txt
+# Or bring your own wordlist(s):
+collectionparser crack --hashes hashes.txt --wordlist words.txt --out cracked.txt
+```
+
+Pure standard library: `hashlib` digests with one process per CPU core, each
+attacking a strided slice of the wordlist, so memory stays bounded; candidate
+sources are merged through `sort -u`. Only unsalted MD5/SHA1/SHA256 are
+attackable — bcrypt is ignored. Feed `cracked.txt` back into the index with
+`split`/`merge`.
+
 ## Measured performance
 
 Single large-file test on Collections 2-5 (3 files, 19.94 GB, 3 workers):
@@ -114,7 +143,8 @@ Run `collectionparser --help` or `collectionparser <command> --help`.
 | `build` | Batched, resumable: parse + dedup + split into the query tree. |
 | `query` | Exact or wildcard search over a dataset. |
 | `export` | Export a dataset as JSON Lines. |
-| `parse` | Parse an unstructured folder into `success`/`failure` files. |
+| `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. |
@@ -123,3 +153,11 @@ Run `collectionparser --help` or `collectionparser <command> --help`.
 | `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
+
+```bash
+pip install -e '.[dev]'
+ruff check .      # lint
+pytest            # tests
+```
diff --git a/SKILL.md b/SKILL.md
index d3689db..c0dcaf7 100644
--- a/SKILL.md
+++ b/SKILL.md
@@ -15,8 +15,8 @@ python -m venv venv && source venv/bin/activate
 pip install -e /path/to/collectionparser
 ```
 
-External tools it shells out to: `sort`, `split` (coreutils) and `rg`
-(ripgrep, required only by `test`/`evaluate`).
+External tools: `sort` and `split` (coreutils); `rg` (ripgrep) for
+`query`/`test`/`evaluate`; optional `7z`/`unrar` to extract `.rar`/`.7z` archives.
 
 ## How to run it for the user
 
@@ -40,7 +40,6 @@ sort -T tmp -u -S 90% --parallel=12 success.txt > success_sorted.txt
 collectionparser split --file success_sorted.txt --out data
 
 # 4. Optional repair pass (moves records to the right bucket, re-sorts).
-ulimit -n 60000
 collectionparser clean --path data
 
 # 5. Query.
@@ -58,9 +57,10 @@ collectionparser merge --src other_data --dest big_dataset
 
 | Command | Purpose | Key options |
 | --- | --- | --- |
-| `build` | Batched, resumable parse + dedup + split into the query tree. | `--source`, `--dataset`, `--scratch`, `--batch_gb` |
+| `build` | Batched, resumable parse + dedup + split into the query tree. | `--source`, `--dataset`, `--scratch`, `--batch_gb`, `--extract/--no-extract` |
 | `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` |
 | `split` | One sorted file -> `a/b/c` tree. | `--file`, `--out`, `--restart_from N` |
 | `chunk` | Split oversized TXT files. | `--path`, `--size MB` (default 50) |
@@ -71,33 +71,38 @@ collectionparser merge --src other_data --dest big_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` |
 
-Global flag: `--debug` (verbose; also bypasses the open-files check below).
+Global flag: `--debug` (verbose).
 
 ## Rules an agent must follow
 
 - **Prefer `build` for large sources.** It parses in size-bounded batches so the
   disk never needs source + temp + tree at once. Stop it anytime and rerun the
   same command to resume: finished batches are in `--scratch/build_state.json`,
-  parsed files in `progress_<n>.txt`. Stats land in `--scratch/stats.md`.
+  parsed files in `progress_<n>.txt`. Stats land in `--scratch/stats.md`; any
+  archive that fails extraction is written to `--scratch/extract_failures.txt`.
 - **Archives are auto-extracted** (`.zip`/`.tar.*`/`.gz` via stdlib; `.rar`/`.7z`
   need `7z` or `unrar` on `PATH`). The source is never modified. Use
   `--no-extract` to skip archive handling.
 - **`split` requires a sorted, unique input.** Always run `sort -u` first; skip
   it and the index silently contains duplicates and unsorted buckets.
-- **`split` and `clean` need `ulimit -n >= 60000`.** The CLI refuses to run
-  otherwise. Raising the limit is the real fix; `--debug` only skips the check.
 - **Memory is already flat.** Don't "optimize" by loading files in Python, and
   don't pre-chunk unless a single file is unwieldy — the tools stream.
 - **`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`.
-- **`test`/`evaluate` need `rg` on `PATH`.** If `rg` is missing, queries return
-  nothing; install ripgrep instead of working around it.
+- **`query`/`test`/`evaluate` need `rg` on `PATH`.** If `rg` is missing, queries
+  return nothing; 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`
   in the current directory; rerun the same command to resume. Don't delete it
   mid-run.
-- `parse` counts a `success` only for plaintext passwords. Hashes
-  (MD5/SHA1/SHA256/bcrypt) are intentionally not written to `success.txt`.
+- `parse`/`build` write only plaintext passwords to `success`; hashed ones
+  (MD5/SHA1/SHA256) go to a separate hashes file (`--scratch/hashes.txt` from
+  `build`, `--hashes_file` from `parse`) so they can be attacked later:
+  `collectionparser crack --hashes hashes.txt --out cracked.txt` (omit
+  `--wordlist` to auto-download rockyou; add `--passwords-from dataset/` to reuse
+  known plaintext passwords). bcrypt is not attackable and is left untouched.
 
 ## Verify the environment
 
diff --git a/collectionparser/api.py b/collectionparser/api.py
index 84e04c0..8f60898 100644
--- a/collectionparser/api.py
+++ b/collectionparser/api.py
@@ -1,3 +1,5 @@
+"""Implementations behind the CLI commands: parse, split, sort, merge, query."""
+
 import json
 import os
 import random
@@ -10,8 +12,7 @@ from tqdm import tqdm
 from collectionparser.parser import parse_to_files
 from collectionparser.query import QueryAPI
 from collectionparser.splitter import get_output_file, write_line_to_file
-from collectionparser.utils import (create_dir_for_file, filter_printable,
-                                    iter_lines, shuffle)
+from collectionparser.utils import create_dir_for_file, filter_printable, iter_lines, shuffle
 
 # ponytail: buckets under this size sort byte-exactly in RAM; larger buckets
 # shell out to `sort -u`, which spills to disk.
@@ -19,16 +20,18 @@ _IN_MEMORY_SORT_LIMIT = 8 * 1024 * 1024
 
 
 class API:
+    """Command implementations shared by the CLI."""
 
     @staticmethod
-    def evaluate(old, new):
+    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 = []
-            with open(random.choice(files), 'r', encoding='utf8', errors='ignore') as r:
-                for count, line in enumerate(r):
+            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:
@@ -41,116 +44,127 @@ class API:
                 print(f'OLD: {random_line}, NEW: {hit}, HIT_RATE = {hit_count / total_count:.3f}.')
 
     @staticmethod
-    def merge(src, dest):
+    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)}
         for src_file in find_files(src_root):
             src_index = str(src_file).replace(str(src_root), '')
             if src_index in dest_files:
-                merge(src_file, dest_files[src_index])
+                append_file(src_file, dest_files[src_index])
             else:
                 dst = dest_root / src_index.lstrip('/')
                 create_dir_for_file(str(dst))
-                move(str(src_file), str(dst))
+                move_file(src_file, dst)
 
     @staticmethod
-    def parse(path, success_file, failure_file):
-        parse_to_files(path, success_file, failure_file)
+    def parse(path: str, success_file: str, failure_file: str,
+              hashes_file: str | None = None) -> None:
+        parse_to_files(path, success_file, failure_file, hashes_file)
 
     @staticmethod
-    def sort(path):
-        files = [p for p in Path(path).expanduser().glob('**/*') if p.is_file() and p.suffix != '.sorted']
-        for p in tqdm(files, desc='Unique > Sorting > Replacing existing files'):
-            output_path = p.with_suffix('.sorted')
-            sort_unique(p, output_path)
-            shutil.move(output_path, p)
+    def sort(path: str) -> None:
+        """Sort and de-duplicate every file under ``path`` in place."""
+        files = [p for p in Path(path).expanduser().glob('**/*')
+                 if p.is_file() and p.suffix != '.sorted']
+        for file in tqdm(files, desc='Unique > Sorting > Replacing existing files'):
+            output_path = file.with_suffix('.sorted')
+            sort_unique(file, output_path)
+            shutil.move(output_path, file)
 
     @staticmethod
-    def clean(path):
+    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
-        files_handlers = {}  # u limit here!
+        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), 'r', encoding='utf8', errors='ignore') as r, \
-                        open(unique_file, 'w', encoding='utf8') as w:
-                    for line in r:
+                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]
-                        ground_truth_file = get_output_file(email)
-                        if ground_truth_file != key:
-                            output_file = Path(path) / ground_truth_file
-                            write_line_to_file(files_handlers, line, output_file, buffer_scale=10)
+                        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. '
-                                                    f'move: {email}: {str(file).replace(path, "")}'
-                                                    f' -> {str(output_file).replace(path, "")}')
+                                bar.set_description(f'clean. {num_files_moved:,} moved.')
                         else:
-                            w.write(line)
+                            writer.write(line)
                 sort_unique(unique_file, file)
                 os.remove(unique_file)
 
     @staticmethod
-    def test(file: str, dataset: str):
-        cracked_rate = total = 0
+    def test(file: str, dataset: str) -> None:
+        """Look up every email in ``file`` and print the cracked rate."""
+        cracked = total = 0
         for line in iter_lines(file):
             results = QueryAPI.query(dataset, line.strip(), 'email')
             if results:
                 passwords = [p.partition(':')[2] for p in results]
                 print('FOUND:', line, ':'.join(passwords))
-                cracked_rate += 1
+                cracked += 1
             total += 1
-        print(f'Cracked rate: {cracked_rate / total if total else 0.0}')
+        print(f'Cracked rate: {cracked / total if total else 0.0}')
 
     @staticmethod
-    def chunk_files(path: str, max_size: int):
-        # -d (numeric suffixes) is not compatible with MacOS.
+    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 f in tqdm(files, desc='chunk large files'):
-            if f.stat().st_size / 1e6 > max_size:
-                cmd = ['split', '-b', str(int(max_size * 1e6)), str(f), str(f) + '_']
-                print(f'File size: {f.stat().st_size / 1e6:.2f} Mb. ' + ' '.join(cmd))
+        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)
-                f.unlink()
+                file.unlink()
 
     @staticmethod
-    def export_jsonl(dataset: str, out: str):
+    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 w:
+        with open(out, 'w', encoding='utf8') as writer:
             for path in tqdm(find_files(Path(dataset).expanduser()), desc='export jsonl'):
-                with open(path, 'r', encoding='utf8', errors='ignore') as r:
-                    for line in r:
+                with open(path, encoding='utf8', errors='ignore') as reader:
+                    for line in reader:
                         email, _, password = line.rstrip('\n').partition(':')
-                        w.write(json.dumps({'email': email, 'password': password},
-                                           ensure_ascii=False) + '\n')
+                        writer.write(json.dumps({'email': email, 'password': password},
+                                                ensure_ascii=False) + '\n')
                         total += 1
         print(f'Exported {total:,} records to {out}.')
 
     @staticmethod
-    def filter_lines(file, out, less_than, more_than):
+    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 w, \
-                open(file, 'r', encoding='utf8', errors='ignore') as r:
-            for line in r:
-                line2 = filter_printable(line)
-                if less_than <= len(line2.strip()) <= more_than:
-                    w.write(line2)
+        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):
+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):
+def sort_unique(src, dst) -> None:
+    """Write the sorted, de-duplicated lines of ``src`` to ``dst``.
+
+    In-memory for small files, ``sort -u`` for large ones. ``src`` may equal
+    ``dst`` (in-place).
+    """
     src = Path(src)
     if src.stat().st_size <= _IN_MEMORY_SORT_LIMIT:
         lines = src.read_bytes().split(b'\n')
@@ -159,19 +173,22 @@ def sort_unique(src, dst):
         lines = sorted(set(lines))
         Path(dst).write_bytes(b'\n'.join(lines) + b'\n' if lines else b'')
     else:
-        with open(dst, 'wb') as w:
-            subprocess.run(['sort', '-u', str(src)], stdout=w, check=True,
+        tmp = f'{dst}.tmp'
+        with open(tmp, 'wb') as writer:
+            subprocess.run(['sort', '-u', str(src)], stdout=writer, check=True,
                            env={**os.environ, 'LC_ALL': 'C'})
+        os.replace(tmp, dst)
 
 
-def merge(src, dst):
+def append_file(src, dst) -> None:
+    """Append ``src`` to ``dst`` and delete ``src`` (streams, constant memory)."""
     print(f'merge {src} {dst}')
-    with open(src, 'rb') as r, open(dst, 'ab') as w:
-        shutil.copyfileobj(r, w, length=1024 * 1024)
+    with open(src, 'rb') as reader, open(dst, 'ab') as writer:
+        shutil.copyfileobj(reader, writer, length=1024 * 1024)
     print(f'del {src}')
     os.remove(src)
 
 
-def move(src, dst):
+def move_file(src, dst) -> None:
     print(f'mv {src} {dst}')
     shutil.move(src, dst)
diff --git a/collectionparser/build.py b/collectionparser/build.py
index 10e8ce6..c658d53 100644
--- a/collectionparser/build.py
+++ b/collectionparser/build.py
@@ -27,7 +27,7 @@ import time
 import zipfile
 from pathlib import Path
 
-from collectionparser.api import API
+from collectionparser.api import API, find_files, sort_unique
 from collectionparser.parser import parse_files
 from collectionparser.splitter import split_file
 
@@ -41,7 +41,8 @@ def _is_archive(path: Path) -> bool:
     return path.name.lower().endswith(ARCHIVE_SUFFIXES)
 
 
-def _which(*names):
+def _which(*names: str) -> str | None:
+    """Return the path of the first available command among ``names``."""
     for name in names:
         found = shutil.which(name)
         if found:
@@ -56,7 +57,7 @@ def _safe_target(dest: Path, name: str) -> Path:
     return target
 
 
-def _extract_compressed(path: Path, dest: Path):
+def _extract_compressed(path: Path, dest: Path) -> None:
     """Try the available RAR/7z extractors in turn.
 
     No single tool wins: `unrar` handles real RARs, `7z` handles the many
@@ -65,10 +66,7 @@ def _extract_compressed(path: Path, dest: Path):
     """
     seven = _which('7z', '7za')
     unrar = _which('unrar')
-    if path.name.lower().endswith('.rar'):
-        candidates = [unrar, seven]
-    else:
-        candidates = [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)
@@ -77,14 +75,14 @@ def _extract_compressed(path: Path, dest: Path):
             cmd = [tool, 'x', '-o+', '-idq', str(path), str(dest) + '/']
         else:
             cmd = [tool, 'x', '-y', f'-o{dest}', str(path)]
-        last = subprocess.run(cmd, 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})')
 
 
-def _extract_tar(path: Path, dest: Path):
+def _extract_tar(path: Path, dest: Path) -> None:
     with tarfile.open(path) as t:
         try:
             t.extractall(dest, filter='data')  # py3.12+
@@ -92,7 +90,7 @@ def _extract_tar(path: Path, dest: Path):
             t.extractall(dest)
 
 
-def _extract_archive(path: Path, dest: Path):
+def _extract_archive(path: Path, dest: Path) -> None:
     name = path.name.lower()
     dest.mkdir(parents=True, exist_ok=True)
     if name.endswith('.zip'):
@@ -110,16 +108,16 @@ def _extract_archive(path: Path, dest: Path):
             shutil.copyfileobj(r, w, length=1 << 20)
 
 
-def _discover(source):
-    root = Path(source).expanduser()
-    files = [p for p in root.glob('**/*') if p.is_file()]
+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)
     return files
 
 
-def _batches(files, batch_bytes):
-    batch, size = [], 0
+def _batches(files: list[Path], batch_bytes: int):
+    batch: list[Path] = []
+    size = 0
     for f in files:
         sz = f.stat().st_size
         if batch and size + sz > batch_bytes:
@@ -131,7 +129,7 @@ def _batches(files, batch_bytes):
         yield batch
 
 
-def _count_lines(path):
+def _count_lines(path: str | Path) -> int:
     total = 0
     with open(path, 'rb') as r:
         for chunk in iter(lambda: r.read(1 << 20), b''):
@@ -139,7 +137,7 @@ def _count_lines(path):
     return total
 
 
-def _stats_markdown(state):
+def _stats_markdown(state: dict) -> str:
     stats = state['stats']
     total_in = sum(s['input_bytes'] for s in stats)
     total_pairs = sum(s['pairs'] for s in stats)
@@ -156,6 +154,8 @@ def _stats_markdown(state):
     ]
     if total_s:
         lines.append(f'- aggregate throughput: {total_in / 1e6 / total_s:.1f} MB/s')
+    if state.get('hashes') is not None:
+        lines.append(f'- hashed passwords collected: {state["hashes"]:,}')
     if state.get('dedup_s') is not None:
         lines.append(f'- dedup seconds: {state["dedup_s"]:,.0f}')
     if state.get('unique_records') is not None:
@@ -163,32 +163,42 @@ def _stats_markdown(state):
     return '\n'.join(lines) + '\n'
 
 
-def _extract_batch(archives, extract_dir, index):
+def _extract_batch(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.
+    """
     if not archives:
-        return []
+        return [], []
     t0 = time.time()
-    failed = 0
-    errors = []
+    failures: list[str] = []
     for i, archive in enumerate(archives):
         try:
             _extract_archive(archive, extract_dir / f'{i:05d}_{archive.name}')
         except Exception as exc:  # one bad archive must not stop the build
-            failed += 1
-            if len(errors) < 5:
-                errors.append(f'{archive.name}: {exc}')
+            failures.append(f'{archive}\t{exc}')
     files = [p for p in extract_dir.glob('**/*') if p.is_file()]
     print(f'batch {index}: {len(archives)} archives -> {len(files)} files, '
-          f'{failed} failed ({time.time() - t0:.0f}s)', flush=True)
-    if errors:
-        print(f'WARN: extract errors (first {len(errors)}): {errors}', flush=True)
-    return files
+          f'{len(failures)} failed ({time.time() - t0:.0f}s)', flush=True)
+    for failure in failures[:5]:
+        print(f'WARN: {failure}', flush=True)
+    return files, failures
 
 
-def build(source, dataset, scratch, batch_bytes=DEFAULT_BATCH_BYTES, dedup=True, extract=True):
+def build(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.
+    Safe to stop and rerun: completed batches are skipped and unparsed files of
+    the current batch are re-queued.
+    """
     dataset = Path(dataset).expanduser()
     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
     for stale in scratch.glob('extract_*'):
         shutil.rmtree(stale, ignore_errors=True)
 
@@ -204,6 +214,8 @@ def build(source, dataset, scratch, batch_bytes=DEFAULT_BATCH_BYTES, dedup=True,
     print(f'{len(files)} files, {len(batches)} batches, {len(done)} already done.', flush=True)
 
     failures = scratch / 'failures.txt'
+    extract_failures = scratch / 'extract_failures.txt'
+    hashes = scratch / 'hashes.txt'
     for index, batch in enumerate(batches):
         if index in done:
             continue
@@ -217,22 +229,27 @@ def build(source, dataset, scratch, batch_bytes=DEFAULT_BATCH_BYTES, dedup=True,
             progress_path.unlink(missing_ok=True)
         parsed = set()
         if success.exists() and progress_path.exists():
-            with open(progress_path, 'r', encoding='utf8', errors='ignore') as r:
+            with open(progress_path, encoding='utf8', errors='ignore') as r:
                 parsed = {line.rstrip('\n') for line in r}
 
         inputs = [f for f in batch if not _is_archive(f)]
         if extract:
-            inputs.extend(_extract_batch([f for f in batch if _is_archive(f)], extract_dir, index))
+            extracted, failures_ = _extract_batch(
+                [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:
+                    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), progress_filename=str(progress_path))
+            parse_files(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
-            subprocess.run(['sort', '-u', '-o', str(success), str(success)], check=True,
-                           env={**os.environ, 'LC_ALL': 'C', 'TMPDIR': str(scratch)})
+            sort_unique(success, success)
             split_file(str(success), str(dataset))
             success.unlink(missing_ok=True)
             if failure.exists():
@@ -255,8 +272,12 @@ def build(source, dataset, scratch, batch_bytes=DEFAULT_BATCH_BYTES, dedup=True,
         (scratch / 'stats.md').write_text(_stats_markdown(state))
         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_path.write_text(json.dumps(state, indent=2))
+
     if dedup and not state.get('deduped'):
-        os.environ['TMPDIR'] = str(scratch)
         t0 = time.time()
         API.sort(str(dataset))  # per-bucket sort -u dedups across batches
         state['dedup_s'] = round(time.time() - t0, 1)
diff --git a/collectionparser/cli.py b/collectionparser/cli.py
index cff4b62..e750488 100644
--- a/collectionparser/cli.py
+++ b/collectionparser/cli.py
@@ -1,24 +1,18 @@
+"""Command line entry point."""
+
 import click
 
 from collectionparser.api import API
 from collectionparser.query import QueryAPI
 from collectionparser.splitter import split_file
-from collectionparser.utils import (input_dir, input_file, num_open_files_limit,
-                                    output_dir, output_file)
-
-# Extraction: nohup find . -name '*.tar.gz' -execdir tar --keep-newer-files -xzvf '{}' -C ../extracted \; &
+from collectionparser.utils import input_dir, input_file, output_dir, output_file
 
 
 @click.group()
 @click.option('--debug/--no-debug')
-@click.pass_context
-def cli(ctx, debug):
+def cli(debug):
     if debug:
         click.echo('Debugging mode enabled.')
-    if not debug and num_open_files_limit() < 60_000 and ctx.invoked_subcommand in ('split', 'clean'):
-        click.echo('ERROR: Increase the limit of the number of open files to at least 60,000.')
-        click.echo('ERROR: ulimit -n 60000')
-        ctx.exit(1)
 
 
 @cli.command(short_help='Converts a large FILE to a query friendly folder OUT (e.g. a/b/c). '
@@ -58,11 +52,13 @@ def test(file: str, dataset: str):
 
 @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=output_dir())
+@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())
-def parse(path, success_file, failure_file):
-    API.parse(path, success_file, failure_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):
+    API.parse(path, success_file, failure_file, hashes_file)
 
 
 @cli.command(short_help='Merges dataset SRC into dataset DEST. Duplicates are removed in the second pass.')
@@ -109,6 +105,21 @@ def export(dataset: str, out: str):
     API.export_jsonl(dataset, out)
 
 
+@cli.command(short_help='Dictionary-attacks collected MD5/SHA1/SHA256 hashes.')
+@click.option('--hashes', 'hashes_path', required=True, type=input_file())
+@click.option('--wordlist', multiple=True, type=click.Path(),
+              help='Plain wordlist(s); missing files are downloaded from rockyou.')
+@click.option('--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)
+
+
 @cli.command(short_help='Builds the deduped query dataset from a raw source folder in batches. '
                         'Stop and rerun to resume.')
 @click.option('--source', required=True, type=input_dir())
diff --git a/collectionparser/crack.py b/collectionparser/crack.py
new file mode 100644
index 0000000..baf3f04
--- /dev/null
+++ b/collectionparser/crack.py
@@ -0,0 +1,146 @@
+"""Dictionary attacks over the unsalted hashes collected by ``parse``.
+
+Candidate sources: plain wordlists, the password column of existing
+``email:password`` datasets, and rockyou (downloaded on demand). They are merged
+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.
+"""
+
+import gzip
+import hashlib
+import os
+import shutil
+import subprocess
+import urllib.request
+from concurrent.futures import ProcessPoolExecutor
+from pathlib import Path
+
+from collectionparser.api import find_files
+
+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')
+
+
+def download_rockyou(dest: Path) -> Path:
+    """Download and gunzip the default rockyou wordlist if not already present."""
+    dest = Path(dest)
+    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)
+    with 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:
+        for line in reader:
+            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 = [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'
+    env = {**os.environ, 'LC_ALL': 'C', 'TMPDIR': str(work_dir)}
+    process = subprocess.Popen(['sort', '-u'], stdin=subprocess.PIPE,
+                               stdout=open(unique, 'wb'), env=env)  # noqa: SIM115
+    with process.stdin as sink:
+        for wordlist in wordlists:
+            with open(wordlist, 'rb') as reader:
+                shutil.copyfileobj(reader, sink, length=1 << 20)
+        for source in passwords_from:
+            root = Path(source).expanduser()
+            for path in ([root] if root.is_file() else find_files(root)):
+                _write_passwords(path, sink)
+    process.wait()
+    if process.returncode != 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:
+        for raw in reader:
+            line = raw.rstrip(b'\r\n')
+            email, separator, digest = line.rpartition(b':')
+            if not separator or len(digest) not in _ALGOS or not _HEX_DIGITS.issuperset(digest):
+                continue
+            try:
+                key = bytes.fromhex(digest.decode('ascii'))
+            except ValueError:
+                continue
+            targets.setdefault(len(digest), {}).setdefault(key, []).append(email)
+    return targets
+
+
+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)
+    checks = [(targets[length], _ALGOS[length]) for length in targets]
+    matches: list[bytes] = []
+    with open(wordlist_path, 'rb') as reader:
+        for n, raw in enumerate(reader):
+            if n % total != index:
+                continue
+            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)
+    return matches
+
+
+def crack(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)
+    if not targets:
+        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())
+    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:
+        if workers == 1:
+            results = map(_scan_stripe, jobs)
+        else:
+            executor = ProcessPoolExecutor(max_workers=workers)
+            results = executor.map(_scan_stripe, jobs)
+        try:
+            for matches in results:
+                for line in matches:
+                    writer.write(line)
+                    total += 1
+        finally:
+            if workers != 1:
+                executor.shutdown()
+    print(f'Cracked {total:,} hashes -> {out_path}.')
+    return total
diff --git a/collectionparser/parser.py b/collectionparser/parser.py
index 70ed96c..95dbdaa 100644
--- a/collectionparser/parser.py
+++ b/collectionparser/parser.py
@@ -4,6 +4,7 @@ import shutil
 import string
 import uuid
 from concurrent.futures import ProcessPoolExecutor
+from contextlib import ExitStack
 from glob import glob
 from pathlib import Path
 from time import time
@@ -38,7 +39,8 @@ def password_type(password: str) -> PasswordType:
     return PasswordType.DECRYPTED
 
 
-def parse_line(row, separator, flip=False):
+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 != ' ':
@@ -50,7 +52,12 @@ def parse_line(row, separator, flip=False):
     if flip:
         mail, password = password, mail
     password = password.replace(' ', '')
-    if not password or len(password) > 30 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) == {'?'}:  # gauntletskd@gmail.com:?????? ??? ?????
         return None
@@ -79,11 +86,12 @@ def parse_line(row, separator, flip=False):
     if mail.count('@hotmail.com') == 2:  # taa06@hotmail.com@hotmail.com:8765432f
         mail = mail.replace('@hotmail.com', '', 1)
     if valid_email(mail) and not valid_email(password):
-        return mail, password, password_type(password)
+        return mail, password, kind
     return None
 
 
 def parseline(s: str):
+    """Clean one raw line and return ``(email, password, type)`` or ``None``."""
     b = filter_printable(s.strip())
     while '..' in b:
         b = b.replace('..', '')
@@ -113,9 +121,11 @@ class Processor:
     per worker is a single read buffer regardless of file size.
     """
 
-    def __init__(self, success_filename=None, failure_filename=None):
+    def __init__(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
 
     # Each worker writes its own buffered part file; the parent concatenates
     # parts after every chunk. Buffered writes avoid the one-syscall-per-line
@@ -123,27 +133,39 @@ class Processor:
     # interleaving (a >4KB line would otherwise split across writes).
     # ponytail: parts cost one extra read+copy pass; merge in-worker only if
     # that pass ever dominates.
-    def process(self, filename: Path):
+    def process(self, filename: Path) -> tuple[int, int]:
         success = failure = 0
         # 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
-        with open(str(filename), 'r', encoding='utf8', errors='ignore') as r, \
+        with open(str(filename), encoding='utf8', errors='ignore') as r, \
                 open(f'{self.success_filename}.part.{part}', 'w', encoding='utf8') as sw, \
                 open(f'{self.failure_filename}.part.{part}', 'w', encoding='utf8') as fw:
-            for line in r:
-                result = parseline(line)
-                if result is None:
-                    fw.write(filter_printable(line))  # stop writing binary data to file.
-                    failure += 1
-                elif result[2] == PasswordType.DECRYPTED:
-                    mail, password, _ = result
-                    sw.write(f'{str(mail).lower().strip()}:{password.strip()}\n')
-                    success += 1
+            hashes = (open(f'{self.hashes_filename}.part.{part}', 'w', encoding='utf8')  # noqa: SIM115
+                      if self.hashes_filename else None)
+            try:
+                for line in r:
+                    result = parseline(line)
+                    if result is None:
+                        fw.write(filter_printable(line))  # stop writing binary data.
+                        failure += 1
+                        continue
+                    mail, password, kind = result
+                    if kind == PasswordType.DECRYPTED:
+                        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')
+                    else:
+                        fw.write(filter_printable(line))  # no hash sink: keep as reject
+                        failure += 1
+            finally:
+                if hashes is not None:
+                    hashes.close()
         return success, failure
 
 
-def _clear_parts(*finals):
+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:
@@ -151,7 +173,7 @@ def _clear_parts(*finals):
             os.remove(part)
 
 
-def _merge_parts(final):
+def _merge_parts(final: str) -> None:
     parts = glob(f'{final}.part.*')
     if not parts:
         return
@@ -162,15 +184,24 @@ def _merge_parts(final):
             os.remove(part)
 
 
-def parse_files(files, success_filename, failure_filename, progress_filename=None):
-    """Parse an explicit list of files; used by both `parse` and `build`."""
-    _clear_parts(success_filename, failure_filename)
+def parse_files(files: list[Path], 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,
+                 *([hashes_filename] if hashes_filename else []))
     success_total = failure_total = 0
     workers = os.cpu_count() or 1
-    processor = Processor(success_filename, failure_filename)
+    processor = Processor(success_filename, failure_filename, hashes_filename)
     start_time = time()
-    progress = open(progress_filename, 'a+', encoding='utf8') if progress_filename else None
-    try:
+    with ExitStack() as stack:
+        progress = (stack.enter_context(open(progress_filename, 'a', encoding='utf8'))
+                    if progress_filename else None)
         with ProcessPoolExecutor(max_workers=workers) as executor, tqdm(files) as bar:
             for i in range(0, len(files), workers):
                 chunk = files[i:i + workers]
@@ -181,25 +212,26 @@ def parse_files(files, success_filename, failure_filename, progress_filename=Non
                 # a crash never records a file whose output was not written.
                 _merge_parts(success_filename)
                 _merge_parts(failure_filename)
+                if hashes_filename:
+                    _merge_parts(hashes_filename)
                 bar.update(len(chunk))
                 bar.set_description(f'success = {success_total:,} failure = {failure_total:,}')
-                for path in chunk:
-                    if progress:
+                if progress:
+                    for path in chunk:
                         progress.write(f'{path}\n')
-    finally:
-        if progress:
-            progress.close()
     print(f'Time elapsed {time() - start_time:.3f} seconds.')
     return success_total, failure_total
 
 
-def parse_to_files(input_dir, success_filename, failure_filename):
-    # Restart from where we stopped. Stream the progress file; never build a
-    # second in-memory copy of it (it can grow to millions of lines).
-    done = set()
+def parse_to_files(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'):
-        with open('progress.txt', 'r', encoding='utf8', errors='ignore') as p:
-            done = {line.rstrip('\n') for line in p}
+        with open('progress.txt', encoding='utf8', errors='ignore') as 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]
     shuffle(files)
-    parse_files(files, success_filename, failure_filename, 'progress.txt')
+    parse_files(files, success_filename, failure_filename, 'progress.txt', hashes_filename)
diff --git a/collectionparser/query.py b/collectionparser/query.py
index 2b17f2c..1129deb 100644
--- a/collectionparser/query.py
+++ b/collectionparser/query.py
@@ -1,12 +1,15 @@
+"""Exact and wildcard queries over a bucketed dataset."""
+
 import re
 import subprocess
 from functools import lru_cache
 from pathlib import Path
 
 
-def extract_emails_and_passwords(txt_lines):
-    emails_passwords = []
-    for txt_line in txt_lines:
+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?
             continue
         if ':' in txt_line:  # '_---madc0w---_@live.com:iskandar89
@@ -19,46 +22,47 @@ def extract_emails_and_passwords(txt_lines):
             email, password = txt_line.strip().split(separator)
         except ValueError:
             continue
-        emails_passwords.append((email, password))
-    return emails_passwords
-
-
-def postprocess(results):
-    return sorted({':'.join(pair) for pair in extract_emails_and_passwords(results)})
+        emails_passwords.add(f'{email}:{password}')
+    return sorted(emails_passwords)
 
 
 def glob_to_regex(pattern: str) -> str:
-    """Glob -> regex anchored to the email field (`*`/`?` never cross the ':').
+    """Convert a glob to a regex anchored to the email field.
 
-    ponytail: `*` matches within the email only; a `*` that must cross the
-    separator needs a different index.
+    ``*`` and ``?`` never cross the ``:`` separator.
+    ponytail: ``*`` matches within the email only; a wildcard that must cross
+    the separator needs a different index.
     """
     parts = []
-    for ch in pattern:
-        if ch == '*':
+    for char in pattern:
+        if char == '*':
             parts.append('[^:]*')
-        elif ch == '?':
+        elif char == '?':
             parts.append('[^:]')
         else:
-            parts.append(re.escape(ch))
+            parts.append(re.escape(char))
     return '^' + ''.join(parts) + ':'
 
 
 class QueryAPI:
+    """Read-only lookups: O(1) exact match or a ripgrep wildcard scan."""
 
     @staticmethod
-    @lru_cache(maxsize=4096)  # ponytail: cached dir listing is stale if the index changes mid-run
-    def scanning(directory: Path):
-        file_list, dir_list = {}, {}
-        for x in directory.iterdir():
-            if x.is_file():
-                file_list[x.name] = x
+    @lru_cache(maxsize=4096)  # ponytail: stale if the index changes mid-run
+    def scanning(directory: Path) -> tuple[dict[str, Path], dict[str, Path]]:
+        """Split a directory's entries into (files, subdirectories)."""
+        file_list: dict[str, Path] = {}
+        dir_list: dict[str, Path] = {}
+        for entry in directory.iterdir():
+            if entry.is_file():
+                file_list[entry.name] = entry
             else:
-                dir_list[x.name] = x
+                dir_list[entry.name] = entry
         return file_list, dir_list
 
     @staticmethod
-    def search_in_file(path: Path, email):
+    def search_in_file(path: Path, email: str) -> list[str]:
+        """Return matching lines for ``email`` from a single bucket file."""
         assert path.is_file(), path
         try:
             output = subprocess.check_output(
@@ -68,21 +72,23 @@ class QueryAPI:
         return [line for line in output.splitlines() if line]
 
     @staticmethod
-    def recursive_search(path: Path, characters: list, index: int):
+    def recursive_search(path: Path, characters: list[str], index: int) -> list[str]:
+        """Descend the bucket tree one character at a time, then search."""
         if index >= len(characters):
             return []
         file_list, dir_list = QueryAPI.scanning(path)
-        cur = characters[index].lower()
-        if cur in file_list:
-            return QueryAPI.search_in_file(file_list[cur], ''.join(characters))
-        if cur in dir_list:
-            return QueryAPI.recursive_search(dir_list[cur], characters, index + 1)
+        current = characters[index].lower()
+        if current in file_list:
+            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:
             return QueryAPI.search_in_file(file_list['symbols'], ''.join(characters))
         return []
 
     @staticmethod
-    def query(data_dir: str, request: str, query_type: str):
+    def query(data_dir: str, request: str, query_type: str) -> list[str]:
+        """Exact email lookup using the bucket path (O(1) file read)."""
         if len(request) <= 3:
             return []
         if query_type == 'email' and '@' not in request:
@@ -90,8 +96,8 @@ class QueryAPI:
         return postprocess(QueryAPI.recursive_search(Path(data_dir), list(request), 0))
 
     @staticmethod
-    def wildcard(data_dir: str, pattern: str):
-        """Substring/wildcard search across the whole dataset with ripgrep.
+    def wildcard(data_dir: str, pattern: str) -> list[str]:
+        """Glob search across the whole dataset with ripgrep.
 
         ponytail: full scan of the tree (ripgrep is fast enough); add a
         domain-sharded index only if wildcard latency becomes a problem.
diff --git a/collectionparser/splitter.py b/collectionparser/splitter.py
index 313fbc2..66c4e0b 100644
--- a/collectionparser/splitter.py
+++ b/collectionparser/splitter.py
@@ -1,47 +1,63 @@
+"""Turn a large ``email:password`` file into a query-friendly ``a/b/c`` tree."""
+
 import io
 import os
 from functools import lru_cache
-from itertools import takewhile
 from time import time
 
-from collectionparser.utils import RingDictionary, create_dir_for_file
+from collectionparser.utils import create_dir_for_file
 
 _ALNUM = 'abcdefghijklmnopqrstuvwxyz0123456789'
 
 
-def split_file(breach_compilation_file, output_dir='out', restart_from=0):
-    # This can leave a few records in the wrong bucket; run `clean` afterwards.
-    # The cause is not fully understood, likely a limit on how many output files
-    # can stay open at once combined with buffering.
+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
+    the wrong bucket; run ``clean`` afterwards.
+    """
     os.makedirs(output_dir, exist_ok=True)
-    files_handlers = RingDictionary()
-    t1 = time()
-    with open(breach_compilation_file, 'r', encoding='utf8', errors='ignore') as r:
-        for i, line in enumerate(r):
+    handles: dict[str, object] = {}
+    tick = time()
+    with open(input_file, encoding='utf8', errors='ignore') as reader:
+        for i, line in enumerate(reader):
             if i % 10_000_000 == 0:
-                elapsed = time() - t1
-                t1 = time()
+                elapsed = time() - tick
+                tick = time()
                 print(f'|- Processed {i:,} rows in {elapsed:.2f} seconds.')
             if i < restart_from or ':' not in line:
                 continue
             email = line.strip().split(':', 1)[0]
-            write_line_to_file(files_handlers, 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(files_handlers, line, output_file, buffer_scale=100):
-    try:  # faster than if: https://stackoverflow.com/questions/1835756/using-try-vs-if-in-python
-        files_handlers[output_file].write(line)
+def write_line_to_file(handles: dict[str, object], line: str, output_file: str,
+                       buffer_scale: int = 100, max_open: int = 100) -> None:
+    """Append ``line`` to ``output_file``, keeping at most ``max_open`` handles.
+
+    Evicted handles are flushed and closed so the process stays under the file
+    descriptor limit.
+    """
+    try:
+        handles[output_file].write(line)  # type: ignore[attr-defined]
     except KeyError:
+        if len(handles) >= max_open:
+            handles.pop(next(iter(handles))).flush()  # type: ignore[attr-defined]
         create_dir_for_file(output_file)
-        files_handlers[output_file] = open(output_file, mode='a+', encoding='utf8',
-                                           buffering=buffer_scale * io.DEFAULT_BUFFER_SIZE)
-        files_handlers[output_file].write(line)
+        handles[output_file] = open(  # noqa: SIM115 - cached handle, not scope-bound
+            output_file, mode='a+', encoding='utf8',
+            buffering=buffer_scale * io.DEFAULT_BUFFER_SIZE)
+        handles[output_file].write(line)  # type: ignore[attr-defined]
 
 
 @lru_cache(maxsize=8192)
 def get_output_file(email: str) -> str:
-    """Map an email to a b/c/d path using its first three alphanumeric chars."""
-    letters = list(takewhile(_ALNUM.__contains__, email.lower()[:3]))
-    if len(letters) < 3:
-        letters.append('symbols')
-    return '/'.join(letters)
+    """Map an email to its bucket path from its first three alphanumeric chars."""
+    parts: list[str] = []
+    for char in email.lower()[:3]:
+        if char not in _ALNUM:
+            break
+        parts.append(char)
+    if len(parts) < 3:
+        parts.append('symbols')
+    return '/'.join(parts)
diff --git a/collectionparser/utils.py b/collectionparser/utils.py
index dd1966a..ad7ff19 100644
--- a/collectionparser/utils.py
+++ b/collectionparser/utils.py
@@ -1,89 +1,64 @@
+"""Small shared helpers: deterministic shuffling, line streaming, path handling."""
+
 import os
 import random
 import re
 import string
 from enum import StrEnum
-from resource import RLIMIT_NOFILE, getrlimit
 
 import click
 
 _NON_PRINTABLE = re.compile(f'[^{re.escape(string.printable)}]')
 
 
-def shuffle(items):
-    random.seed(123)
-    random.shuffle(items)
-
-
-def num_open_files_limit() -> int:
-    return getrlimit(RLIMIT_NOFILE)[0]
+def shuffle(items: list) -> None:
+    """Shuffle ``items`` in place with a fixed seed (reproducible ordering)."""
+    random.Random(123).shuffle(items)
 
 
 def filter_printable(text: str) -> str:
+    """Drop non-printable characters so no binary data is written to text files."""
     return _NON_PRINTABLE.sub('', text)
 
 
-def iter_lines(filename):
-    with open(filename, 'r', encoding='utf8', errors='ignore') as r:
-        for line in r:
+def iter_lines(filename: str):
+    """Yield lines from ``filename`` without the trailing newline, streaming."""
+    with open(filename, encoding='utf8', errors='ignore') as reader:
+        for line in reader:
             yield line.rstrip('\n')
 
 
-def create_dir_for_file(filename):
+def create_dir_for_file(filename: str) -> None:
+    """Create the parent directory of ``filename`` if it is missing."""
     if parent := os.path.dirname(filename):
         os.makedirs(parent, exist_ok=True)
 
 
-def input_file():
+def input_file() -> click.Path:
     return click.Path(exists=True, file_okay=True, dir_okay=False,
                       writable=False, readable=True, resolve_path=True)
 
 
-def input_dir():
+def input_dir() -> click.Path:
     return click.Path(exists=True, file_okay=False, dir_okay=True,
                       writable=False, readable=True, resolve_path=True)
 
 
-def output_file():
+def output_file() -> click.Path:
     return click.Path(exists=False, file_okay=True, dir_okay=False,
                       writable=True, readable=True, resolve_path=True)
 
 
-def output_dir():
+def output_dir() -> click.Path:
     return click.Path(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'
     SHA1 = 'SHA1'
     SHA256 = 'SHA256'
     B_CRYPT = 'B_CRYPT'
     DECRYPTED = 'DECRYPTED'
-
-
-class RingDictionary:
-    """Dict of at most ``max_num_keys`` entries; evicting flushes the oldest.
-
-    Used to keep a bounded number of open file handles while splitting a large
-    file into many output files.
-    """
-
-    def __init__(self, max_num_keys=100):
-        self.d = {}
-        self.k = []
-        self.max_num_keys = max_num_keys
-
-    def __getitem__(self, key):
-        return self.d[key]
-
-    def __setitem__(self, key, item):
-        if key not in self.d:
-            if len(self.k) >= self.max_num_keys:
-                oldest = self.k.pop(0)
-                try:
-                    self.d.pop(oldest).flush()
-                except Exception:
-                    pass
-            self.k.append(key)
-            self.d[key] = item
diff --git a/pyproject.toml b/pyproject.toml
index 5baa7c0..6dbdefe 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -5,8 +5,18 @@ build-backend = "setuptools.build_meta"
 [project]
 name = "collectionparser"
 version = "3.0.0"
-description = "Command line tools to parse and query multi-billion password collections."
+description = "Parse, deduplicate, and query multi-billion password collections."
+readme = "README.md"
 requires-python = ">=3.11"
+keywords = ["passwords", "breach", "email", "dataset", "search"]
+classifiers = [
+    "Environment :: Console",
+    "Intended Audience :: Developers",
+    "Programming Language :: Python :: 3",
+    "Programming Language :: Python :: 3.11",
+    "Topic :: Security",
+    "Topic :: Utilities",
+]
 dependencies = [
     "click>=8.1",
     "tqdm>=4.66",
@@ -14,9 +24,18 @@ dependencies = [
 
 [project.optional-dependencies]
 test = ["pytest>=8"]
+dev = ["pytest>=8", "ruff>=0.6"]
 
 [project.scripts]
 collectionparser = "collectionparser.cli:cli"
 
 [tool.setuptools.packages.find]
 include = ["collectionparser*"]
+
+[tool.ruff]
+target-version = "py311"
+line-length = 120
+src = ["collectionparser", "tests"]
+
+[tool.ruff.lint]
+select = ["E4", "E7", "E9", "F", "I", "UP", "B", "SIM"]
diff --git a/tests/crack_test.py b/tests/crack_test.py
new file mode 100644
index 0000000..376d0c2
--- /dev/null
+++ b/tests/crack_test.py
@@ -0,0 +1,59 @@
+import hashlib
+import tempfile
+import unittest
+from glob import glob
+from pathlib import Path
+
+from collectionparser.crack import crack
+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.write_text('a@x.com:5f4dcc3b5aa765d61d8327deb882cf99\nb@y.com:plainpw\n',
+                           encoding='utf8')
+            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')
+                             for p in glob(str(Path(tmp) / 'h.txt.part.*')))
+            success = ''.join(Path(p).read_text(encoding='utf8')
+                              for p in glob(str(Path(tmp) / 's.txt.part.*')))
+            self.assertEqual('a@x.com:5f4dcc3b5aa765d61d8327deb882cf99\n', hashes)
+            self.assertEqual('b@y.com:plainpw\n', success)
+
+    def test_crack_md5_and_sha1(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            hashes = Path(tmp) / 'hashes.txt'
+            hashes.write_bytes(
+                b'a@x.com:' + hashlib.md5(b'hunter2').hexdigest().encode() + b'\n'
+                b'b@y.com:' + hashlib.sha1(b'secret').hexdigest().encode() + b'\n'
+                b'c@z.com:not-a-hash\n')
+            wordlist = Path(tmp) / 'words.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'a@x.com:hunter2', b'b@y.com:secret'},
+                             set(out.read_bytes().splitlines()))
+
+    def test_crack_from_plaintext_passwords(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            hashes = Path(tmp) / 'hashes.txt'
+            hashes.write_bytes(b'a@x.com:' + hashlib.md5(b'reused').hexdigest().encode() + b'\n')
+            empty = Path(tmp) / 'empty.txt'
+            empty.write_bytes(b'')
+            plaintext = Path(tmp) / 'plain.txt'
+            plaintext.write_bytes(b'other@x.com:reused\n')
+            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'a@x.com:reused'], out.read_bytes().splitlines())
+
+
+if __name__ == '__main__':
+    unittest.main()
diff --git a/tests/parser_test.py b/tests/parser_test.py
index eaabe7b..8f1099a 100644
--- a/tests/parser_test.py
+++ b/tests/parser_test.py
@@ -3,6 +3,7 @@ from time import time
 
 from collectionparser.parser import parseline as p
 from collectionparser.splitter import get_output_file
+from collectionparser.utils import PasswordType
 
 
 class ParserTest(unittest.TestCase):
@@ -97,7 +98,10 @@ class ParserTest(unittest.TestCase):
 
         self.assertIsNone(p('300396	matelymaly	rune_scape_player@hotmail.com	222.155.173.78	'
                             '3d2c5414a041f518b03b750e62e4ebc3:Nhc'))
-        self.assertIsNone(p('05-02-2012;5721869;melanie2032003@yahoo.fr:f02368945726d5fc2a14eb576f7276c0'))
+        # 32-char hex is a hash, not junk: kept (typed) so `crack` can attack it.
+        hashed = p('05-02-2012;5721869;melanie2032003@yahoo.fr:f02368945726d5fc2a14eb576f7276c0')
+        self.assertIsNotNone(hashed)
+        self.assertEqual(PasswordType.MD5, hashed[2])
         self.assertIsNotNone(p('diet3rix@gmail.com::g1u3ck'))
         self.assertIsNotNone(p('asuss_85@mail.ru:asuss_85@mail.ru:2904198'))
         self.assertIsNotNone(p('sanda ostojic.todor@gmail.com:vasafultv'))
diff --git a/tests/utils_test.py b/tests/utils_test.py
deleted file mode 100644
index bfe065b..0000000
--- a/tests/utils_test.py
+++ /dev/null
@@ -1,36 +0,0 @@
-import unittest
-
-from collectionparser.utils import RingDictionary
-
-
-class UtilsTest(unittest.TestCase):
-
-    def test_ring_dictionary_1(self):
-        rd = RingDictionary(max_num_keys=3)
-        rd['a'] = 2
-        self.assertListEqual(sorted(rd.d), ['a'])
-        self.assertListEqual(sorted(rd.k), ['a'])
-        rd['b'] = 2
-        self.assertListEqual(sorted(rd.d), ['a', 'b'])
-        self.assertListEqual(sorted(rd.k), ['a', 'b'])
-        rd['c'] = 2
-        self.assertListEqual(sorted(rd.d), ['a', 'b', 'c'])
-        self.assertListEqual(sorted(rd.k), ['a', 'b', 'c'])
-        rd['d'] = 2
-        self.assertListEqual(sorted(rd.d), ['b', 'c', 'd'])
-        self.assertListEqual(sorted(rd.k), ['b', 'c', 'd'])
-
-        rd = RingDictionary(max_num_keys=1)
-        rd['c'] = 1
-        rd['c'] = 2
-        rd['c'] = 3
-        self.assertListEqual(sorted(rd.d), ['c'])
-        self.assertListEqual(sorted(rd.k), ['c'])
-
-        rd['d'] = 3
-        self.assertListEqual(sorted(rd.d), ['d'])
-        self.assertListEqual(sorted(rd.k), ['d'])
-
-
-if __name__ == '__main__':
-    unittest.main()