Rename project to collectionparser

.github/FUNDING.yml

@@ -1 +0,0 @@
-github: [philipperemy]

.gitignore

@@ -0,0 +1,21 @@
+__pycache__/
+*.py[cod]
+.pytest_cache/
+.coverage
+htmlcov/
+.tox/
+.nox/
+.venv/
+venv/
+build/
+dist/
+*.egg-info/
+
+# Local datasets and pipeline scratch/state files
+data/
+scratch/
+*.jsonl
+build_state.json
+progress_*.txt
+
+.DS_Store

Makefile

@@ -1,16 +0,0 @@
-all:
-	make install
-
-install:
-	pip install cython pytest
-	python setup.py build_ext --inplace
-	pip install . --upgrade
-
-uninstall:
-	pip uninstall -y breach
-
-test:
-	pytest tests
-
-clean:
-	rm -rf *.out breach/*.c *.bin *.exe *.o *.a breach/*.so build *.html __pycache__ breach/__pycache__

README.md

@@ -1,177 +1,125 @@
-# Collections 1, Collections 2-5 and AntiPublic Parser
+# collectionparser
 
-Command line tools to manipulate the data from those multi-billion passwords collections.
+Command line tools to parse, deduplicate, and query multi-billion password
+collections (Collections 1-5, AntiPublic, Breach Compilation).
 
-The full processing will take a couple of days and will generate a file structure that can be queried almost in o(1).
+Processing produces a deduped `a/b/c` bucket tree that answers exact lookups in
+O(1) and wildcard queries with a parallel ripgrep scan:
 
 ```
-$ <query> john.applessed@apple.com
-john.applessed@apple.com:toto123
+$ collectionparser query --dataset /path/to/data --pattern john.smith@gmail.com
+john.smith@gmail.com:toto123
+$ collectionparser query --dataset /path/to/data --pattern '*@gmail.com'
 ```
 
-The total number of unique records in the final dataset (Collection 1 to 5 + AntiPublic + Breach Compilation) is around 3.72 billions (_3,372,591,561_ to be precise).
+## Install
 
-
-### Set Up
-
-Create a virtual environment and install the package.
-
-```
-virtualenv -p python3 venv
-source venv/bin/activate
-make install
-```
-
-## Extraction
-
-https://superuser.com/questions/1308374/how-to-extract-all-tar-gz-files-present-in-multiple-folders-at-a-time-to-another
+Requires Python 3.11+ and [`ripgrep`](https://github.com/BurntSushi/ripgrep) (`rg`) for queries.
 
 ```bash
-find "$1" -name '*.tar.gz' -execdir tar -xzvf '{}' -C extracted \;
-find . -name "*.rar" -exec unrar x -o+ {} \;
+python -m venv venv && source venv/bin/activate
+pip install -e .
 ```
 
-## Processing
+## Memory
 
-Processing the Collection 1 is much faster than the Collections 2-5. The estimates for Collections 2-5 are reported below.
+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.
 
-The parsing took around 20 hours on my server (CPU i7-8700K, 32GB of memory). I didn't have a large enough SSD to store all the temporary computations so everything
-was done on a standard HDD. A faster disk will surely make the processing faster.
+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
+available stayed at 15.3 GB, swap unchanged.
 
-The sorting/removing duplicates step took 15 hours in total.
+## Build the query dataset
 
-The splitting into the smaller files (this file struct makes every query almost instantaneous) took a couple of hours at most.
-
-In total, expect around 2 days to process the Collections 2-5.
+`build` is the resumable, disk-bounded pipeline. It parses the source in batches,
+sorts+dedups each batch on a scratch disk, splits it into the bucket tree, then
+dedups each bucket. Peak disk is source + tree, so it fits where a global sort
+would not.
 
 ```bash
-breach parse --path /path/to/extracted --success_file success.txt --failure_file failure.txt --cython_acceleration
-rm -rf tmp && mkdir tmp # you need like 750GB in tmp/. By default /tmp/ is not enough for this!
+collectionparser build \
+  --source /path/to/extracted \
+  --dataset /path/to/dataset \
+  --scratch /path/to/scratch \   # fast disk for batch files + sort temp
+  --batch_gb 120
+```
+
+**Pause / resume:** stop it any time (Ctrl-C or `systemctl stop`) and rerun the
+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`.
+
+**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).
+
+## Processing (manual)
+
+```bash
+collectionparser parse --path /path/to/extracted --success_file success.txt --failure_file failure.txt
+
+rm -rf tmp && mkdir tmp  # sort needs room for temp files; /tmp is usually too small
 cat success.txt | pv -cN cut | sort -T tmp -u -S 90% --parallel=12 | pv -cN cut > success_sorted.txt
-breach split --file success_sorted.txt --out data
+
+collectionparser split --file success_sorted.txt --out data
 ```
 
-## Converting the format of BreachCompilation to the new format
+## Merging datasets
 
-The dataset is available here: https://github.com/philipperemy/tensorflow-1.4-billion-password-analysis
-
-It's easy to convert the large `BreachCompilation` dataset to this format by running those commands.
-
-Expect those commands to take some time (less than a day).
+`merge` is destructive: back up the destination first.
 
 ```bash
-find /path/to/BreachCompilation/ -type f -exec cat {} + > breach_compilation.txt
-rm -rf tmp && mkdir tmp # By default /tmp/ is not enough for this!
-cat breach_compilation.txt | pv -cN cut | sort -T tmp -u -S 90% --parallel=12 | pv -cN cut > breach_compilation_sorted.txt
-breach split --file breach_compilation_sorted.txt --out data_breach_compilation_sorted
-```
-
-From there a simple `breach merge` will be enough to merge it to the Collections 1 & 2 to 5.
-
-## Merging all the datasets together
-
-Run the Collection 1 and Collections 2-5 through the processing step described above.
-
-You will have two directories: `/path/to/collections1_data` and `/path/to/collections2_5_data`.
-
-Additionally, if you have the other dataset BreachCompilation, you will have another directory `/path/to/data_breach_compilation_sorted`, generated by the step above.
-
-The merge is destructive so it's better to create a copy of the output first and then merge each one into the output.
-
-```
 cp -rf /path/to/collections1_data /path/to/big_dataset
-breach merge --src /path/to/collections2_5_data --dest /path/to/big_dataset
-breach merge --src /path/to/data_breach_compilation_sorted --dest /path/to/big_dataset
+collectionparser merge --src /path/to/collections2_5_data --dest /path/to/big_dataset
 ```
 
-## Usage
+## Query and export
 
-The manual of the command line tool can be fetched by running `breach dumphelp`.
+```bash
+# exact (O(1) bucket lookup) or wildcard (* and ?):
+collectionparser query --dataset data --pattern 'john*'
+collectionparser query --dataset data --pattern '*@gmail.com'
 
+# JSON Lines for OpenSearch / bulk ingest:
+collectionparser export --dataset data --out records.jsonl
 ```
-Usage: cli [OPTIONS] COMMAND [ARGS]...
 
-Options:
-  --debug / --no-debug
-  --help                Show this message and exit.
+## Measured performance
 
-Commands:
-  chunk     chunk large TXT files into smaller files.
-  clean     Cleans a query friendly folder PATH. Move incorrect records and
-            sort the files.
-  dumphelp
-  evaluate  Evaluates some metrics such as precision/recall (e.g. is OLD into
-            NEW).
-  merge     Merges dataset SRC into dataset DEST.
-  parse     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.
-  sort      Sorts a query friendly folder PATH. Target is itself.
-  split     Converts a large FILE to a query friendly folder OUT (e.g. a/b/c).
-            Use RESTART_FROM to resume from the i-th line.
-  test      Infers passwords of a list of emails defined in FILE with a query
-            friendly folder DATASET.
+Single large-file test on Collections 2-5 (3 files, 19.94 GB, 3 workers):
 
-Usage: cli dumphelp [OPTIONS]
+| metric | value |
+| --- | --- |
+| parse throughput | 20.5 MB/s aggregate (6.8 MB/s per worker) |
+| valid pairs | 628,342,073 |
+| rejects | 3,321,666 |
+| peak RSS | 160 MB |
+| `sort -u` | ~125 MB/s |
+| `split` (sorted input) | ~43 MB/s |
 
-Options:
-  --help  Show this message and exit.
+Split is ~12× slower on unsorted input (open-handle thrash); always sort first.
 
-Usage: cli split [OPTIONS]
+## Commands
 
-Options:
-  --file FILE             [required]
-  --out DIRECTORY         [required]
-  --restart_from INTEGER  [default: 0]
-  --help                  Show this message and exit.
+Run `collectionparser --help` or `collectionparser <command> --help`.
 
-Usage: cli chunk [OPTIONS]
-
-Options:
-  --path DIRECTORY  [required]
-  --size INTEGER    [default: 50]
-  --help            Show this message and exit.
-
-Usage: cli sort [OPTIONS]
-
-Options:
-  --path DIRECTORY  [required]
-  --help            Show this message and exit.
-
-Usage: cli clean [OPTIONS]
-
-Options:
-  --path DIRECTORY  [required]
-  --help            Show this message and exit.
-
-Usage: cli test [OPTIONS]
-
-Options:
-  --file FILE                     [required]
-  --dataset [breach_compilation|collections_1|collections_2_5|all]
-                                  [required]
-  --help                          Show this message and exit.
-
-Usage: cli parse [OPTIONS]
-
-Options:
-  --path DIRECTORY                [required]
-  --success_file FILE             [required]
-  --failure_file FILE             [required]
-  --cython_acceleration / --no-cython_acceleration
-  --help                          Show this message and exit.
-
-Usage: cli merge [OPTIONS]
-
-Options:
-  --src DIRECTORY   [required]
-  --dest DIRECTORY  [required]
-  --help            Show this message and exit.
-
-Usage: cli evaluate [OPTIONS]
-
-Options:
-  --old DIRECTORY  [required]
-  --new DIRECTORY  [required]
-  --help           Show this message and exit.
-  ```
+| command | description |
+| --- | --- |
+| `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. |
+| `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. |
+| `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. |

SKILL.md

@@ -0,0 +1,111 @@
+---
+name: collectionparser
+description: Parse, index, merge, and query multi-billion-row email:password breach collections (Collections 1-5, AntiPublic, Breach Compilation). Use when a task involves huge credential dumps, building a query-friendly a/b/c index, cracking a list of emails, or merging/sorting such datasets. Memory-safe on 100GB+ files.
+---
+
+# collectionparser
+
+Turns a folder of raw credential dumps into an `a/b/c` file tree that can be
+queried in O(1), then looks up passwords for a list of emails.
+
+Install (Python 3.11+):
+
+```bash
+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`).
+
+## How to run it for the user
+
+1. Find the input: either a single raw/sorted text file, or a folder of files.
+   Confirm the path — never guess and never write into the user's dataset
+   before backing up.
+2. Run the pipeline below, one step at a time, watching for the sanity checks.
+3. Report the output paths and the row counts printed by each step.
+
+## Pipeline (the intended order)
+
+```bash
+# 1. Raw dumps in a folder -> valid (email:password) and rejected lines.
+collectionparser parse --path /path/to/raw --success_file success.txt --failure_file failure.txt
+
+# 2. Sort + dedupe the whole success file. Keep tmp on a big disk.
+rm -rf tmp && mkdir tmp
+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).
+ulimit -n 60000
+collectionparser clean --path data
+
+# 5. Query.
+collectionparser test --file emails.txt --dataset data
+```
+
+To add another dataset to an existing index:
+
+```bash
+cp -rf data big_dataset                 # merge is destructive; work on a copy
+collectionparser merge --src other_data --dest big_dataset
+```
+
+## Commands
+
+| Command | Purpose | Key options |
+| --- | --- | --- |
+| `build` | Batched, resumable parse + dedup + split into the query tree. | `--source`, `--dataset`, `--scratch`, `--batch_gb` |
+| `query` | Exact or wildcard (`*`, `?`) search over a dataset. | `--dataset`, `--pattern` |
+| `export` | Export a dataset as JSON Lines (OpenSearch). | `--dataset`, `--out` |
+| `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) |
+| `sort` | `sort -u` every file in place. | `--path` |
+| `clean` | Move misplaced records, then sort each file. | `--path` |
+| `merge` | Merge `--src` into `--dest` (destructive). Runs `clean` after. | `--src`, `--dest` |
+| `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` |
+
+Global flag: `--debug` (verbose; also bypasses the open-files check below).
+
+## 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`.
+- **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.
+- **`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`.
+
+## Verify the environment
+
+```bash
+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
+```
+
+If that works, the package and its entry point are installed correctly.

breach/api.py

@@ -1,154 +0,0 @@
-import os
-import random
-import shutil
-import string
-import subprocess
-from glob import glob
-from pathlib import Path
-
-from tqdm import tqdm
-
-from breach.parser import parse_to_files
-from breach.query import QueryAPI, get_query_manager
-from breach.splitter import get_output_file, write_line_to_file
-from breach.utils import alpha_num_lookup, shuffle, read_lines, create_dir_for_file
-
-
-class API:
-
-    @staticmethod
-    def evaluate(old, new):
-        hit_count = 0
-        total_count = 0
-        files = [p for p in Path(old).expanduser().glob('**/*') if p.is_file()]
-        for i in range(1000):
-            lines = read_lines(random.choice(files))
-            shuffle(lines)
-            random_lines = lines[0:10]
-            for random_line in random_lines:
-                password_list = QueryAPI.query(new, random_line.split(':')[0], 'email')
-                total_count += 1
-                if len(password_list) == 0:
-                    print(f'OLD: {random_line}, NEW: miss, HIT_RATE = {hit_count / total_count:.3f}.')
-                else:
-                    hit_count += 1
-                    print(f'OLD: {random_line}, NEW: hit, HIT_RATE = {hit_count / total_count:.3f}.')
-
-    @staticmethod
-    def merge(src, dest):
-        src_files = find_files(src)
-        dest_files = find_files(dest)
-        for src_index, src_file in src_files.items():
-            if src_index in dest_files:
-                merge(src_file, dest_files[src_index])
-            else:
-                move(src_file, dest + src_index)
-
-    @staticmethod
-    def parse(path, success_file, failure_file, cython_acceleration=False):
-        parse_to_files(path, success_file, failure_file, cython_acceleration)
-
-    @staticmethod
-    def sort(path):
-        path = str(Path(path).expanduser())
-        paths = [Path(a) for a in glob(path + '/**/*', recursive=True) if Path(a).is_file()
-                 and Path(a).suffix != '.sorted']
-        for path in tqdm(paths, desc='Unique > Sorting > Replacing existing files'):
-            output_path = path.with_suffix('.sorted')
-            lines = read_lines(path)
-            unique_sorted_lines = sorted(set(lines))
-            with open(output_path, 'w', encoding='utf8') as w:
-                w.write('\n'.join(unique_sorted_lines) + '\n')
-            shutil.move(output_path, path)
-
-    @staticmethod
-    def clean(path):
-        files = list(find_files(Path(path).expanduser()).values())
-        shuffle(files)
-        alpha = alpha_num_lookup()
-        num_files_moved = 0
-        files_handlers = {}  # u limit here!
-        with tqdm(files, desc='clean') as bar:
-            for file in bar:
-                lines_to_delete = []
-                lines = read_lines(file)
-                for line in lines:
-                    line_split = line.split(':')
-                    ground_truth_file = get_output_file(alpha, line_split[0])
-                    key = str(file).replace(path, '')
-                    if key.startswith('/'):
-                        key = key[1:]
-                    if ground_truth_file != key:
-                        output_file = Path(path) / ground_truth_file
-                        # print(f'MOVE: {line_split[0]} -> {output_file}.')
-                        write_line_to_file(files_handlers, line_split, output_file, buffer_scale=10)
-                        lines_to_delete.append(line)
-                        num_files_moved += 1
-                        if num_files_moved % 100_000 == 0:
-                            truncated_file = str(file).replace(path, "")
-                            truncated_output_file = str(output_file).replace(path, "")
-                            bar.set_description(f'clean. {num_files_moved:,} moved. '
-                                                f'move: {line_split[0]}: {truncated_file}'
-                                                f' -> {truncated_output_file}')
-                lines2 = sorted(set(lines) - set(lines_to_delete))
-                if len(lines) != len(lines2):
-                    with open(str(file), 'w', encoding='utf8') as w:
-                        w.write('\n'.join(lines2) + '\n')
-
-    @staticmethod
-    def test(file: str, dataset: str):
-        qm = get_query_manager(dataset)
-        cracked_rate = 0
-        lines = read_lines(file)
-        for line in lines:
-            results = qm.perform_query(line.strip(), 'email')
-            if len(results) > 0:
-                passwords = [p.split(':', 1)[-1] for p in results]
-                print('FOUND:', line, ':'.join(passwords))
-                cracked_rate += 1
-        print(f'Cracked rate: {cracked_rate / len(lines)}')
-
-    @staticmethod
-    def chunk_files(path: str, max_size: int):
-        # split -b 50000000 file.txt file.txt.
-        # -d to have .00 and .01 is not compatible with MacOS...
-        files = list(find_files(Path(path).expanduser()).values())
-        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))
-                subprocess.check_output(cmd)
-                f.unlink()
-
-    @staticmethod
-    def filter_lines(file, out, less_than, more_than):
-        create_dir_for_file(out)
-        with open(out, 'w', encoding='utf8') as w:
-            with open(file, 'r', encoding='utf8', errors='ignore') as r:
-                for line in r:
-                    line2 = ''.join(list(filter(lambda x: x in string.printable, line)))
-                    if less_than <= len(line2.strip()) <= more_than:
-                        w.write(line2)
-
-
-def find_files(path: str):
-    path2 = Path(path).expanduser()
-    files = [p for p in path2.glob('**/*') if p.is_file()]
-    index = {str(f).replace(str(path2), ''): f for f in files}
-    assert len(index) == len(files)
-    return index
-
-
-def merge(src, dst):
-    print(f'merge {src} {dst}')
-    with open(src, 'r', encoding='utf8', errors='ignore') as r:
-        content1 = r.read()
-    with open(dst, 'a', encoding='utf8', errors='ignore') as w:
-        w.write(content1)
-    print(f'del {src}')
-    os.remove(src)
-
-
-def move(src, dst):
-    print(f'mv {src} {dst}')
-    shutil.move(src, dst)

breach/cli.py

@@ -1,111 +0,0 @@
-import logging
-from pathlib import Path
-
-import click
-
-from breach.api import API
-from breach.splitter import split_file
-from breach.utils import Ct, num_open_files_limit
-
-
-# Extraction: nohup find . -name '*.tar.gz' -execdir tar --keep-newer-files -xzvf '{}' -C ../extracted \; &
-
-def recursive_help(cmd, parent=None):
-    ctx = click.core.Context(cmd, info_name=cmd.name, parent=parent)
-    print(cmd.get_help(ctx))
-    print()
-    commands = getattr(cmd, 'commands', {})
-    for sub in commands.values():
-        recursive_help(sub, ctx)
-
-
-@click.group()
-@click.option('--debug/--no-debug')
-@click.pass_context
-def cli(ctx, debug):
-    if debug:
-        click.echo('Debugging mode enabled.')
-    logging.basicConfig(format='%(asctime)12s - %(levelname)s - %(message)s', level=logging.INFO)
-    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')
-        exit(1)
-
-
-@cli.command()
-def dumphelp():
-    recursive_help(cli)
-
-
-@cli.command(short_help='Converts a large FILE to a query friendly folder OUT (e.g. a/b/c). '
-                        'Use RESTART_FROM to resume from the i-th line.')
-@click.option('--file', required=True, type=Ct.input_file())
-@click.option('--out', required=True, type=Ct.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.')
-@click.option('--path', required=True, type=Ct.input_dir())
-@click.option('--size', default=50, type=int, show_default=True)  # MB.
-def chunk(path, size):
-    API.chunk_files(path, size)
-
-
-@cli.command(short_help='Sorts a query friendly folder PATH. Target is itself.')
-@click.option('--path', required=True, type=Ct.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.')
-@click.option('--path', required=True, type=Ct.input_dir())
-def clean(path):
-    API.clean(path)
-
-
-@cli.command(short_help='Infers passwords of a list of emails defined in FILE with a query friendly folder DATASET.')
-@click.option('--file', required=True, type=Ct.input_file())
-# @click.option('--dataset', required=True, type=click.Choice(DATASET_CHOICES, case_sensitive=False))
-@click.option('--dataset', required=True)
-def test(file: str, dataset: str):
-    API.test(file, dataset)
-
-
-@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=Ct.output_dir())
-@click.option('--success_file', required=True, type=Ct.output_file())
-@click.option('--failure_file', required=True, type=Ct.output_file())
-@click.option('--cython_acceleration/--no-cython_acceleration', default=False)
-def parse(path, success_file, failure_file, cython_acceleration):
-    API.parse(Path(path).expanduser(), success_file, failure_file, cython_acceleration)
-
-
-@cli.command(short_help='Merges dataset SRC into dataset DEST. Duplicates are removed in the second pass.')
-@click.option('--src', required=True, type=Ct.input_dir())
-@click.option('--dest', required=True, type=Ct.input_dir())
-def merge(src: str, dest: str):
-    API.merge(src, dest)
-    API.clean(dest)
-
-
-@cli.command(short_help='Evaluates some metrics such as precision/recall (e.g. is OLD into NEW).')
-@click.option('--old', required=True, type=Ct.input_dir())
-@click.option('--new', required=True, type=Ct.input_dir())
-def evaluate(old: str, new: str):
-    API.evaluate(old, new)
-
-
-@cli.command(short_help='Filters lines less than N and more than M.')
-@click.option('--file', required=True, type=Ct.input_file())
-@click.option('--out', required=True, type=Ct.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)
-
-
-if __name__ == '__main__':
-    cli()

breach/cparser.pyx

@@ -1,179 +0,0 @@
-import re
-import string
-
-import cython
-from validate_email import validate_email
-
-from breach.utils import PasswordType
-
-cdef inline c_is_md5(str password):
-    if len(password) == 32:
-        for e in password:
-            if not ((ord('0') <= ord(e) <= ord('9')) or
-                    (ord('A') <= ord(e) <= ord('F')) or
-                    (ord('a') <= ord(e) <= ord('f'))):
-                return False
-        return True
-    else:
-        return False
-
-cdef inline c_is_sha1(str password):
-    if len(password) == 40:
-        for e in password:
-            if not ((ord('0') <= ord(e) <= ord('9')) or
-                    (ord('A') <= ord(e) <= ord('F')) or
-                    (ord('a') <= ord(e) <= ord('f'))):
-                return False
-        return True
-    else:
-        return False
-
-cdef inline c_is_sha_256(str password):
-    if len(password) == 64:
-        for e in password:
-            if not ((ord('0') <= ord(e) <= ord('9')) or
-                    (ord('A') <= ord(e) <= ord('F')) or
-                    (ord('a') <= ord(e) <= ord('f'))):
-                return False
-        return True
-    else:
-        return False
-
-cdef inline c_is_b_crypt(str password):
-    if 50 < len(password) < 70:
-        b_crypt_pattern_cond = password[0:4] == "$2a$" or password[0:4] == "$2b$" or password[0:4] == "$2y$"
-        if b_crypt_pattern_cond:
-            return True
-    return False
-
-cdef inline c_get_password_type(str password):
-    if c_is_md5(password):
-        return PasswordType.MD5
-    elif c_is_sha1(password):
-        return PasswordType.SHA1
-    elif c_is_sha_256(password):
-        return PasswordType.SHA256
-    elif c_is_b_crypt(password):
-        return PasswordType.B_CRYPT
-    else:
-        return PasswordType.DECRYPTED
-
-@cython.boundscheck(False)
-def c_parseline(str s):
-    # case   [email protected]:password
-    # case   [email protected];password
-    # case   [email protected] password
-    valid_separators = [':', ';', ' ', '\t', '  ', '   ', ', ', ',', ';;', '::', ',']
-    # https://github.com/swiftmailer/swiftmailer/issues/239
-    b = s.strip()
-    b = ''.join(list(filter(lambda x: x in string.printable, b)))
-    while '..' in b:  # TODO: could be move just for the email.
-        b = b.replace('..', '')
-    if b.startswith('mailto:'):
-        b = b[len('mailto:'):]
-    if b.startswith('E-mail:'):
-        b = b[len('E-mail:'):]
-    if b.startswith(':'):
-        b = b[1:]
-    if b.endswith(':'):
-        b = b[:-1]
-
-    # TODO: could be move just for the email. but we dont need to be 100% accurate here.
-    b = b.replace('.@', '@')  # just typo.
-    b = b.replace('@.', '@')  # just typo.
-    b = b.replace(' @ ', '@')  # just typo.
-    b = b.replace('@@', '@')  # just typo.
-
-    if ':' in b and ';' in b:
-        b = b.replace(';', ':')
-
-    # FLIP=FALSE
-    for valid_separator in valid_separators:
-        result = c_parse_line(b, valid_separator, flip=False)
-        if result is not None:
-            return result
-    # FLIP=TRUE
-    for valid_separator in valid_separators:
-        result = c_parse_line(b, valid_separator, flip=True)
-        if result is not None:
-            return result
-    # print(f'DISCARD: {s.strip()}')
-    return None
-
-@cython.boundscheck(False)
-cdef inline c_our_validate_email(str candidate):
-    parts = candidate.split('@')
-    if len(parts) != 2:
-        return False
-    if '.' not in parts[-1]:  # .com or .fr
-        return False
-    return validate_email(candidate)
-
-@cython.boundscheck(False)
-cdef c_parse_line(str row, str separator, bint flip=False):
-    cdef bint is_sep_in_row = separator in row
-    cdef bint sep_count_cond = False
-    cdef str mail = ''
-    cdef str password = ''
-    if not is_sep_in_row:  # return fast.
-        return None
-    if separator != ' ':
-        parts = row.split(separator)
-        sep_count_cond = len(parts) == 2
-        if len(parts) > 2:
-            row = separator.join(parts[-2:])
-            # [email protected]:kiyan1365:16
-            # [email protected]:godis2good:
-            sep_count_cond = True
-    elif is_sep_in_row:
-        sep_count_cond = True
-    else:
-        sep_count_cond = False
-    if sep_count_cond:
-        mail, password = row.split(separator, 1)
-        if flip:
-            password, mail = mail, password
-        password = password.replace(' ', '')
-        if len(password) > 30:  # Password of 30 chars look like this: adfsfdsgdfghkfdghdsdfgdfgdfgdf.
-            return None
-        if len(password) == 0:
-            return None
-        if '\n\n' in password:
-            return None
-        if set(password) == {'?'}:  # - [email protected]:?????? ??? ?????
-            return None
-        if password.count('.') >= 3 and \
-                re.search(r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}', password) is not None:
-            return None
-        mail = mail.replace(' ', '')
-        # https://grox.net/utils/encoding.html
-        if '@' not in mail:
-            mail = mail.replace('%40', '@')
-        # - [email protected]:125.26.106.158,nopparat
-        # count first because it's faster than compiling a regexp.
-        if len(mail) == 0:
-            return None
-        last_char = mail[-1]
-        first_char = mail[0]
-        # [email protected]\ ilovebois3
-        if last_char in ['<', '.', '\\', '>', '_']:
-            mail = mail[:-1]
-        # [email protected] atohiyamamot [exists]
-        if first_char in ['.', '<']:
-            mail = mail[1:]
-        if mail.endswith('@com'):
-            # iemongozalu@gmail@com mizo1210
-            mail = mail.replace('@com', '.com')
-        if not c_our_validate_email(mail):
-            # some people replace neanta,[email protected],jp
-            mail = mail.replace(',', '.')
-            parts = mail.split('@')
-            if len(parts) > 2:
-                mail = '@'.join(parts[0:2])
-        # TODO: do it for most common domain names.
-        if mail.count('@hotmail.com') == 2:  # [email protected]@hotmail.com:8765432f
-            mail = mail.replace('@hotmail.com', '', 1)
-        if c_our_validate_email(mail) and not c_our_validate_email(password):
-            password_type = c_get_password_type(password)
-            return mail, password, password_type
-    return None

breach/parser.py

@@ -1,314 +0,0 @@
-import os
-import re
-import string
-from glob import glob
-from pathlib import Path
-from time import time
-
-from breach.cparser import c_parseline
-from tqdm import tqdm
-from validate_email import validate_email
-
-from breach.utils import chunks, read_lines, PasswordType
-from breach.utils import parallel_function, shuffle
-
-
-# slow functions.
-
-def our_validate_email(candidate: str):
-    if candidate.count('@') != 1:
-        return False
-    if '.' not in candidate.split('@')[-1]:  # .com or .fr
-        return False
-    return validate_email(candidate)
-
-
-class LineParser:
-
-    @staticmethod
-    def _parse_line(row, separator, flip=False):
-        if separator not in row:  # return fast.
-            return None
-        if separator != ' ':
-            parts = row.split(separator)
-            sep_count_cond = len(parts) == 2
-            if len(parts) > 2:
-                row = separator.join(parts[-2:])
-                # [email protected]:kiyan1365:16
-                # [email protected]:godis2good:
-                sep_count_cond = True
-        elif separator in row:
-            sep_count_cond = True
-        else:
-            sep_count_cond = False
-        if sep_count_cond:
-            mail, password = row.split(separator, 1)
-            if flip:
-                password, mail = mail, password
-            password = password.replace(' ', '')
-            if len(password) > 30:  # Password of 30 chars look like this: adfsfdsgdfghkfdghdsdfgdfgdfgdf.
-                return None
-            if len(password) == 0:
-                return None
-            if '\n\n' in password:
-                return None
-            if set(password) == {'?'}:  # - [email protected]:?????? ??? ?????
-                return None
-            if password.count('.') >= 3 and \
-                    re.search(r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}', password) is not None:
-                return None
-            mail = mail.replace(' ', '')
-            # https://grox.net/utils/encoding.html
-            if '@' not in mail:
-                mail = mail.replace('%40', '@')
-            # - [email protected]:125.26.106.158,nopparat
-            # count first because it's faster than compiling a regexp.
-
-            if len(mail) == 0:
-                return None
-
-            last_char = mail[-1]
-            first_char = mail[0]
-            # [email protected]\ ilovebois3
-            if last_char in ['<', '.', '\\', '>', '_']:
-                mail = mail[:-1]
-            # [email protected] atohiyamamot [exists]
-            if first_char in ['.', '<']:
-                mail = mail[1:]
-            if mail.endswith('@com'):
-                # iemongozalu@gmail@com mizo1210
-                mail = mail.replace('@com', '.com')
-            if not our_validate_email(mail):
-                # some people replace neanta,[email protected],jp
-                mail = mail.replace(',', '.')
-                parts = mail.split('@')
-                if len(parts) > 2:
-                    mail = '@'.join(parts[0:2])
-            # TODO: do it for most common domain names.
-            if mail.count('@hotmail.com') == 2:  # [email protected]@hotmail.com:8765432f
-                mail = mail.replace('@hotmail.com', '', 1)
-            if our_validate_email(mail) and not our_validate_email(password):
-                password_type = LineParser.get_password_type(password)
-                return mail, password, password_type
-        return None
-
-    @staticmethod
-    def is_md5(password):
-        if len(password) == 32:
-            for e in password:
-                if not ((ord('0') <= ord(e) <= ord('9')) or
-                        (ord('A') <= ord(e) <= ord('F')) or
-                        (ord('a') <= ord(e) <= ord('f'))):
-                    return False
-            return True
-        else:
-            return False
-
-    @staticmethod
-    def is_sha1(password):
-        if len(password) == 40:
-            for e in password:
-                if not ((ord('0') <= ord(e) <= ord('9')) or
-                        (ord('A') <= ord(e) <= ord('F')) or
-                        (ord('a') <= ord(e) <= ord('f'))):
-                    return False
-            return True
-        else:
-            return False
-
-    @staticmethod
-    def is_sha_256(password):
-        if len(password) == 64:
-            for e in password:
-                if not ((ord('0') <= ord(e) <= ord('9')) or
-                        (ord('A') <= ord(e) <= ord('F')) or
-                        (ord('a') <= ord(e) <= ord('f'))):
-                    return False
-            return True
-        else:
-            return False
-
-    @staticmethod
-    def is_b_crypt(password):
-        b_crypt_pattern_cond = password[0:4] == "$2a$" or password[0:4] == "$2b$" or password[0:4] == "$2y$"
-        if 50 < len(password) < 70 and b_crypt_pattern_cond:
-            return True
-        return False
-
-    @staticmethod
-    def get_password_type(password):
-        if LineParser.is_md5(password):
-            return PasswordType.MD5
-        elif LineParser.is_sha1(password):
-            return PasswordType.SHA1
-        elif LineParser.is_sha_256(password):
-            return PasswordType.SHA256
-        elif LineParser.is_b_crypt(password):
-            return PasswordType.B_CRYPT
-        else:
-            return PasswordType.DECRYPTED
-
-    @staticmethod
-    def parseline(s: str):
-        # case   [email protected]:password
-        # case   [email protected];password
-        # case   [email protected] password
-        valid_separators = [':', ';', ' ', '\t', '  ', '   ', ', ', ',', ';;', '::', ',']
-        # https://github.com/swiftmailer/swiftmailer/issues/239
-        b = s.strip()
-        b = ''.join(list(filter(lambda x: x in string.printable, b)))
-        while '..' in b:  # TODO: could be move just for the email.
-            b = b.replace('..', '')
-        if b.startswith('mailto:'):
-            b = b[len('mailto:'):]
-        if b.startswith('E-mail:'):
-            b = b[len('E-mail:'):]
-        if b.startswith(':'):
-            b = b[1:]
-        if b.endswith(':'):
-            b = b[:-1]
-
-        # TODO: could be move just for the email. but we dont need to be 100% accurate here.
-        b = b.replace('.@', '@')  # just typo.
-        b = b.replace('@.', '@')  # just typo.
-        b = b.replace(' @ ', '@')  # just typo.
-        b = b.replace('@@', '@')  # just typo.
-
-        if ':' in b and ';' in b:
-            b = b.replace(';', ':')
-
-        # FLIP=FALSE
-        for valid_separator in valid_separators:
-            result = LineParser._parse_line(b, valid_separator, flip=False)
-            if result is not None:
-                return result
-        # FLIP=TRUE
-        for valid_separator in valid_separators:
-            result = LineParser._parse_line(b, valid_separator, flip=True)
-            if result is not None:
-                return result
-        # print(f'DISCARD: {s.strip()}')
-        return None
-
-
-class Processor:
-
-    def __init__(self, w=None, f=None, cython_acceleration=False):  # for big files -> writes to disk directly.
-        if cython_acceleration:
-            print('Cython acceleration enabled.')
-            self.bind = c_parseline
-        else:
-            print('Cython acceleration disabled.')
-            self.bind = LineParser.parseline
-        self.w = w  # cannot be serialized for multi processing.
-        self.f = f  # cannot be serialized for multi processing.
-
-    def process(self, filename: Path, is_big_file: bool = False):
-        success_list = []
-        failure_list = []
-        with open(str(filename), 'r', encoding='utf8', errors='ignore') as r:
-            buf_size = 10 ** 6  # for very large files.
-            while True:
-                lines = r.readlines(buf_size)
-                if len(lines) == 0:
-                    break
-                for line in lines:
-                    result = self.bind(line)
-                    if result is not None:
-                        mail, password, password_type = result
-                        if password_type == PasswordType.DECRYPTED:
-                            # mail is case insensitive.
-                            success_str = ':'.join([str(mail).lower().strip(), password.strip()]) + '\n'
-                            if is_big_file:
-                                self.w.write(success_str)
-                            else:
-                                success_list.append(success_str)
-                    else:
-                        # stop writing binary data to file...
-                        fail_line = ''.join(list(filter(lambda x: x in string.printable, line)))
-                        if is_big_file:
-                            self.f.write(fail_line)
-                        else:
-                            failure_list.append(fail_line)
-        return success_list, failure_list
-
-
-def parse_to_files(input_dir, success_filename, failure_filename, cython_acceleration):
-    files = [p for p in Path(input_dir).glob('**/*') if p.is_file()]
-    shuffle(files)
-    # restart from where we stopped.
-    progress_filename = 'progress.txt'
-    if os.path.exists(progress_filename):
-        lines = read_lines(progress_filename)
-    else:
-        lines = []
-    lines = {n: 1 for n in lines}
-    files = [f for f in files if str(f) not in lines]
-    shuffle(files)
-    # files.sort(key=lambda f: os.stat(f).st_size)  # smallest to largest
-
-    success_total = 0
-    failure_total = 0
-    num_threads = os.cpu_count()
-    start_time = time()
-    with open(progress_filename, 'a+') as p:
-        with open(failure_filename, 'a+', encoding='utf8') as f:
-            with open(success_filename, 'a+', encoding='utf8') as w:
-                large_files_processor = Processor(w=w, f=f, cython_acceleration=cython_acceleration)
-                small_files_processor = Processor(cython_acceleration=cython_acceleration)
-                with tqdm(files) as bar:
-                    for chunk in chunks(files, num_threads):
-                        # more than 300MB?
-                        any_large_files = any([c.stat().st_size / 1e6 > 300 for c in chunk])
-                        if any_large_files:
-                            # single thread and write directly to the files.
-                            for chunky in chunk:
-                                large_files_processor.process(chunky)
-                        else:
-                            # as many threads as we want.
-                            results = parallel_function(small_files_processor.process, chunk, len(chunk))
-
-                            for single_result in results:
-                                success_list, failure_list = single_result
-
-                                for single_success in success_list:
-                                    w.write(single_success)
-                                for single_success in failure_list:
-                                    f.write(single_success)
-
-                                success_total += len(success_list)
-                                failure_total += len(failure_list)
-
-                        bar.update(len(chunk))
-                        bar.set_description(f'success = {success_total:,} failure = {failure_total:,}')
-                        for single_chunk in chunk:
-                            p.write(str(single_chunk) + '\n')
-    print(f'Time elapsed {time() - start_time:.3f} seconds.')
-
-
-def statistics(input_dir):
-    files = glob(os.path.expanduser(input_dir) + '/**/*.*', recursive=True)
-    files = [f for f in files if not f.endswith('.tar.gz')]
-    shuffle(files)
-    parsed_lines = 0
-    with open('stats.txt', 'w') as w:
-        w.write(','.join(['success', 'failure', 'failure_rate', 'filename']) + '\n')
-        for filename in files:
-            success = 0
-            failure = 0
-            lines = read_lines(filename)
-            parsed_lines += len(lines)
-            for line in lines:
-                result = LineParser.parseline(line)
-                if result is not None:
-                    success += 1
-                else:
-                    failure += 1
-            try:
-                failure_rate = failure / (success + failure) * 100
-            except ZeroDivisionError:
-                failure_rate = 0.0
-            w.write(','.join([str(success), str(failure), f'{failure_rate:.3f}', filename]) + '\n')
-            print(f'Parsed lines = {parsed_lines:,}.')
-            print(f'filename = {filename}, success = {success:,}, '
-                  f'failure = {failure:,}, failure rate = {failure_rate:.3f}%')

breach/query.py

@@ -1,194 +0,0 @@
-import logging
-import os
-import subprocess
-from pathlib import Path
-
-logger = logging.getLogger(__name__)
-
-
-# https://pypi.org/project/ripgrepy/
-
-class BreachCompilationConstants:
-    EMAIL_INDEX = '/media/philippe/DATA/datasets/BreachCompilation/data'
-    PASSWORD_INDEX = '/media/philippe/DATA/datasets/BreachCompilation/passwords'
-
-
-class Collections1:
-    EMAIL_INDEX = '/media/philippe/8TB/datasets/passwords2/data'
-    PASSWORD_INDEX = None
-
-
-class Collections_2_5:
-    EMAIL_INDEX = '/media/philippe/8TB/datasets/passwords2/data'
-    PASSWORD_INDEX = None
-
-
-DATASET_CHOICES = ['breach_compilation', 'collections_1', 'collections_2_5', 'all']
-
-
-def get_query_manager(dataset_choice: str):
-    if dataset_choice.lower() == 'breach_compilation':
-        return QueryManager(BreachCompilationConstants)
-    elif dataset_choice.lower() == 'collections_1':
-        return QueryManager(Collections1)
-    elif dataset_choice.lower() == 'collections_2_5':
-        return QueryManager(Collections_2_5)
-    elif os.path.exists(dataset_choice):
-        class Custom:
-            EMAIL_INDEX = dataset_choice
-            PASSWORD_INDEX = None
-
-        return QueryManager(Custom)
-    else:
-        return QueryMultiManager()  # all
-
-
-def mask_pass_justin_rule(password: str, num_asterisks=-1) -> str:
-    if len(password) >= 8:
-        return password[0:2] + (len(password) - 3) * '*' + password[-1]
-    return password[0] + '*' * (len(password) - 2) + password[-1]
-
-
-def mask_pass(password: str, num_asterisks=2) -> str:
-    if len(password) <= num_asterisks:
-        return '*' * len(password)
-    if len(password) >= 4:
-        i = len(password) // 2
-        even = num_asterisks % 2 == 0
-        half = num_asterisks // 2
-        if even:
-            left = half
-            right = half
-        else:
-            left = half
-            right = half + 1
-        return password[0:i - left] + '*' * num_asterisks + password[i + right:]
-    else:
-        return password
-
-
-def extract_emails_and_passwords(txt_lines):
-    emails_passwords = []
-    for txt_line in txt_lines:
-        try:
-            if '@' in txt_line:  # does it contain an email address?
-                if all([char in txt_line for char in [':', ';']]):  # which separator is it? : or ;?
-                    separator = ':'
-                elif ':' in txt_line:  # '[email protected]:iskandar89
-                    separator = ':'
-                elif ';' in txt_line:  # '[email protected];ol1391ga
-                    separator = ';'
-                else:
-                    continue
-
-                strip_txt_line = txt_line.strip()
-                email, password = strip_txt_line.split(separator)
-                emails_passwords.append((email, password))
-        except Exception:
-            pass
-    return emails_passwords
-
-
-def postprocess(results, num_asterisks_to_mask_password=0):
-    response = [b for b in extract_emails_and_passwords(results)]
-    if num_asterisks_to_mask_password > 0:
-        response = [(b[0], mask_pass_justin_rule(b[1])) for b in response]
-    response = [':'.join(b) for b in response]
-    response = sorted(set(response))
-    return response
-
-
-class QueryAPI:
-
-    @staticmethod
-    def scanning(directory: Path):
-        file_list, dir_list = {}, {}
-        for x in directory.iterdir():
-            if x.is_file():
-                file_list[x.name] = x
-            else:
-                dir_list[x.name] = x
-        return file_list, dir_list
-
-    @staticmethod
-    def search_in_file(path: Path, email):
-        assert path.is_file(), path
-        try:
-            results = subprocess.check_output(['rg', '--no-line-number', '-ie', '^' +
-                                               email, str(path)]).decode('utf8').split('\n')
-            results = [a for a in results if a != '']
-        except subprocess.CalledProcessError:
-            results = []
-        return results
-
-    @staticmethod
-    def recursive_search(path: Path, characters: list, index: int):
-        fl, dl = QueryAPI.scanning(path)
-        cur = characters[index].lower()
-        if cur in fl:
-            return QueryAPI.search_in_file(fl[cur], ''.join(characters))
-        elif cur in dl:
-            return QueryAPI.recursive_search(dl[cur], characters, index + 1)
-        elif 'symbols' in fl:
-            return QueryAPI.search_in_file(fl['symbols'], ''.join(characters))
-        else:
-            raise Exception('Hello.')
-
-    @staticmethod
-    def query(data_dir: str, request: str, query_type: str, num_asterisks_to_mask_password: int = 0):
-        if len(request) <= 3:
-            return []
-        if query_type == 'email' and '@' not in request:
-            return []
-        response = QueryAPI.recursive_search(Path(data_dir), list(request), 0)
-        return postprocess(response, num_asterisks_to_mask_password)
-
-
-class QueryManagerStub:
-    def __init__(self, constants):
-        self.default_email_index = constants.EMAIL_INDEX
-        self.default_password_index = constants.PASSWORD_INDEX
-        self.counter = 0
-
-    def perform_query(self, request: str, query_type: str, num_asterisks_to_mask_password=0):
-        self.counter += 1
-        if self.counter % 2 == 0:
-            return [f'{request}:hello123', f'{request}:hello123456']
-        else:
-            return []
-
-
-class QueryManager:
-
-    def __init__(self, constants):
-        self.default_email_index = constants.EMAIL_INDEX
-        self.default_password_index = constants.PASSWORD_INDEX
-
-    def perform_query(self, request: str, query_type: str, num_asterisks_to_mask_password=0):
-        if query_type == 'email':
-            index = self.default_email_index
-        elif query_type == 'password':
-            index = self.default_password_index
-        else:
-            raise Exception({'error': 'Index is either email or password.'})
-        results = []
-        if ' ' in request:  # batch.
-            for single_req in request.split(' '):
-                results.extend(QueryAPI.query(index, single_req, query_type, num_asterisks_to_mask_password))
-        else:
-            results.extend(QueryAPI.query(index, request, query_type, num_asterisks_to_mask_password))
-        return results
-
-
-class QueryMultiManager:
-
-    def __init__(self):
-        self.qm1 = QueryManager(BreachCompilationConstants)
-        self.qm2 = QueryManager(Collections1)
-        self.qm3 = QueryManager(Collections_2_5)
-
-    def perform_query(self, request: str, query_type: str, num_asterisks_to_mask_password=0):
-        results = self.qm1.perform_query(request, query_type, num_asterisks_to_mask_password)
-        results.extend(self.qm2.perform_query(request, query_type, num_asterisks_to_mask_password))
-        results.extend(self.qm3.perform_query(request, query_type, num_asterisks_to_mask_password))
-        return sorted(set(results))

breach/splitter.py

@@ -1,80 +0,0 @@
-import csv
-import io
-import sys
-from time import time
-
-from breach.utils import alpha_num_lookup, create_dir, create_dir_for_file, RingDictionary
-
-csv.field_size_limit(sys.maxsize)
-
-
-def split_file(breach_compilation_file,
-               output_dir='out',
-               restart_from=0):
-    # It seems like this function is not 100% correct as we need to clean the dataset after this.
-    # It's not a big deal but I don't understand why we have to do that.
-    # Could probably be to a limitation to the large number of files we open at the same time x increase of buffer.
-    create_dir(output_dir)
-    alpha = alpha_num_lookup()
-    files_handlers = RingDictionary()
-    t1 = time()
-    with open(breach_compilation_file, 'r', encoding='utf8', errors='ignore') as r:
-        for i, line in enumerate(r):
-            if i % 10_000_000 == 0:
-                elapsed_time = time() - t1
-                t1 = time()
-                print(f'|- Processed {i:,} rows in {elapsed_time:.2f} seconds.')
-            if i < restart_from:
-                continue
-            try:  # faster than if: https://stackoverflow.com/questions/1835756/using-try-vs-if-in-python
-                email, password = line.strip().split(':', 1)
-            except ValueError:
-                continue
-
-            output_file = get_output_file(alpha, email)
-            output_file = output_dir + '/' + output_file
-            write_line_to_file(files_handlers, line, output_file)
-
-
-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)
-    except KeyError:
-        create_dir_for_file(output_file)
-        files_handlers[output_file] = open(file=output_file,
-                                           mode='a+',
-                                           encoding='utf8',
-                                           buffering=buffer_scale * io.DEFAULT_BUFFER_SIZE)
-        files_handlers[output_file].write(line)
-
-
-def get_output_file(alpha_lookup, email):
-    try:  # most likely case.
-        letter_0 = email[0].lower()
-        letter_1 = email[1].lower()
-        letter_2 = email[2].lower()
-    except IndexError:
-        email_len = len(email)
-        if email_len == 0:
-            letter_0 = ''
-            letter_1 = ''
-            letter_2 = ''
-        elif email_len == 1:
-            letter_0 = email[0].lower()
-            letter_1 = ''
-            letter_2 = ''
-        elif email_len == 2:
-            letter_0 = email[0].lower()
-            letter_1 = email[1].lower()
-            letter_2 = ''
-        else:
-            raise Exception('Error in program.')
-    if letter_0 not in alpha_lookup:
-        output_file = 'symbols'
-    elif letter_1 not in alpha_lookup:
-        output_file = letter_0 + '/symbols'
-    elif letter_2 not in alpha_lookup:
-        output_file = letter_0 + '/' + letter_1 + '/symbols'
-    else:
-        output_file = letter_0 + '/' + letter_1 + '/' + letter_2
-    return output_file

breach/utils.py

@@ -1,142 +0,0 @@
-import logging
-import os
-import random
-import subprocess
-from datetime import datetime
-from multiprocessing import Pool
-
-import click
-
-
-def shuffle(lst):
-    random.seed(123)
-    random.shuffle(lst)
-
-
-def num_open_files_limit():
-    return int(subprocess.check_output('ulimit -n', shell=True).decode().strip())
-
-
-def read_lines(filename: str):
-    with open(filename, 'r', encoding='utf8', errors='ignore') as r:
-        return r.read().strip().split('\n')
-
-
-def create_dir(output_dir: str):
-    if len(output_dir) > 0 and not os.path.exists(output_dir):
-        os.makedirs(output_dir)
-
-
-def create_dir_for_file(filename: str):
-    create_dir(os.path.dirname(filename))
-
-
-def alpha_num_lookup():
-    alpha = list('abcdefghijklmnopqrstuvwxyz0123456789')
-    alpha_lookup = {a: 1 for a in alpha}
-    assert len(alpha) == 36
-    return alpha_lookup
-
-
-def init_logger(name):
-    format_str = '%(asctime)12s - %(levelname)s - %(message)s'
-    formatter = logging.Formatter(format_str)
-    log_dir = 'logs'
-    create_dir(log_dir)
-    log_filename = os.path.join(log_dir, name + '_' + datetime.now().strftime('%Y-%m-%d_%H:%M:%S') + '.log')
-    logging.basicConfig(format=format_str, filename=log_filename, level=logging.INFO)
-    console = logging.StreamHandler()
-    console.setFormatter(formatter)
-    logging.getLogger('').addHandler(console)
-
-
-def parallel_function(f, sequence, num_threads=None):
-    if num_threads == 1:
-        return [f(x) for x in sequence]
-    pool = Pool(processes=num_threads)
-    result = pool.map(f, sequence)
-    cleaned = [x for x in result if x is not None]
-    pool.close()
-    pool.join()
-    return cleaned
-
-
-def chunks(lst, n):
-    """Yield successive n-sized chunks from l."""
-    for i in range(0, len(lst), n):
-        yield lst[i:i + n]
-
-
-class Ct:
-
-    @staticmethod
-    def input_file(writable=False):
-        return click.Path(exists=True, file_okay=True, dir_okay=False,
-                          writable=writable, readable=True, resolve_path=True)
-
-    @staticmethod
-    def input_dir(writable=False):
-        return click.Path(exists=True, file_okay=False, dir_okay=True,
-                          writable=writable, readable=True, resolve_path=True)
-
-    @staticmethod
-    def output_file():
-        return click.Path(exists=False, file_okay=True, dir_okay=False,
-                          writable=True, readable=True, resolve_path=True)
-
-    @staticmethod
-    def output_dir():
-        return click.Path(exists=False, file_okay=False, dir_okay=True,
-                          writable=True, readable=True, resolve_path=True)
-
-
-class PasswordType:
-    MD5 = 'MD5'
-    SHA1 = 'SHA1'
-    SHA256 = 'SHA256'
-    B_CRYPT = 'B_CRYPT'
-    DECRYPTED = 'DECRYPTED'
-
-
-class RingDictionary:
-
-    def __init__(self, max_num_keys=100, debug=False):
-        self.d = {}  # output_file -> file_handler.
-        self.k = []
-        self.max_num_keys = max_num_keys
-        self.debug = debug
-
-    def __getitem__(self, key):
-        return self.d[key]
-
-    def __del__(self):
-        for k in self.k:
-            self.d[k].flush()
-            if self.debug:
-                print(f'[RingDictionary] Flush stream: {k}.')
-
-    def __setitem__(self, key, item):
-        try:
-            self.d[key]
-        except KeyError:
-            if len(self.k) >= self.max_num_keys:
-                first_item = self.k.pop(0)
-                try:
-                    self.d[first_item].flush()  # for streams.
-                    if self.debug:
-                        print(f'[RingDictionary] Flush stream: {first_item}.')
-                except Exception:
-                    pass
-                del self.d[first_item]
-                if self.debug:
-                    print(f'[RingDictionary] Drop: {first_item}.')
-            if self.debug:
-                print(f'[RingDictionary] Add: {key}.')
-            self.k.append(key)
-            self.d[key] = item
-
-    def keys(self):
-        return self.d.keys()
-
-    def values(self):
-        return self.d.values()

collectionparser/api.py

@@ -0,0 +1,177 @@
+import json
+import os
+import random
+import shutil
+import subprocess
+from pathlib import Path
+
+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)
+
+# ponytail: buckets under this size sort byte-exactly in RAM; larger buckets
+# shell out to `sort -u`, which spills to disk.
+_IN_MEMORY_SORT_LIMIT = 8 * 1024 * 1024
+
+
+class API:
+
+    @staticmethod
+    def evaluate(old, new):
+        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):
+                    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], 'email')
+                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, dest):
+        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])
+            else:
+                dst = dest_root / src_index.lstrip('/')
+                create_dir_for_file(str(dst))
+                move(str(src_file), str(dst))
+
+    @staticmethod
+    def parse(path, success_file, failure_file):
+        parse_to_files(path, success_file, failure_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)
+
+    @staticmethod
+    def clean(path):
+        files = find_files(Path(path).expanduser())
+        shuffle(files)
+        num_files_moved = 0
+        files_handlers = {}  # u limit here!
+        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:
+                        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)
+                            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, "")}')
+                        else:
+                            w.write(line)
+                sort_unique(unique_file, file)
+                os.remove(unique_file)
+
+    @staticmethod
+    def test(file: str, dataset: str):
+        cracked_rate = 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
+            total += 1
+        print(f'Cracked rate: {cracked_rate / total if total else 0.0}')
+
+    @staticmethod
+    def chunk_files(path: str, max_size: int):
+        # -d (numeric suffixes) is not compatible with MacOS.
+        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))
+                subprocess.run(cmd, check=True)
+                f.unlink()
+
+    @staticmethod
+    def export_jsonl(dataset: str, out: str):
+        """Export the dataset as JSON Lines (`{"email":..., "password":...}`)."""
+        create_dir_for_file(out)
+        total = 0
+        with open(out, 'w', encoding='utf8') as w:
+            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:
+                        email, _, password = line.rstrip('\n').partition(':')
+                        w.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):
+        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)
+
+
+def find_files(path):
+    root = Path(path).expanduser()
+    return [p for p in root.glob('**/*') if p.is_file()]
+
+
+def sort_unique(src, dst):
+    src = Path(src)
+    if src.stat().st_size <= _IN_MEMORY_SORT_LIMIT:
+        lines = src.read_bytes().split(b'\n')
+        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'')
+    else:
+        with open(dst, 'wb') as w:
+            subprocess.run(['sort', '-u', str(src)], stdout=w, check=True,
+                           env={**os.environ, 'LC_ALL': 'C'})
+
+
+def merge(src, dst):
+    print(f'merge {src} {dst}')
+    with open(src, 'rb') as r, open(dst, 'ab') as w:
+        shutil.copyfileobj(r, w, length=1024 * 1024)
+    print(f'del {src}')
+    os.remove(src)
+
+
+def move(src, dst):
+    print(f'mv {src} {dst}')
+    shutil.move(src, dst)

collectionparser/build.py

@@ -0,0 +1,271 @@
+"""Resumable batched build of the deduped query dataset.
+
+Why batches: the raw collection is ~908 GB but the volume plus a sort temp
+cannot hold a global sort. Each batch is parsed, sorted+deduped on the scratch
+disk, and split into the same bucket tree; a final per-bucket sort dedups
+across batches. Peak disk = source + tree.
+
+Resume: completed batches are listed in ``build_state.json``; within a batch,
+each parsed input file is recorded in ``progress_<n>.txt``. Stop with Ctrl-C /
+``systemctl stop`` and rerun the same command to continue. The parse phase
+resumes per file; sort+split of an interrupted batch is redone (duplicates are
+removed by the final dedup), so an interrupted split does not lose data.
+
+Archives are extracted into the scratch disk for the duration of their batch and
+removed once parsed; the source is never modified.
+"""
+
+import bz2
+import gzip
+import json
+import lzma
+import os
+import shutil
+import subprocess
+import tarfile
+import time
+import zipfile
+from pathlib import Path
+
+from collectionparser.api import API
+from collectionparser.parser import parse_files
+from collectionparser.splitter import split_file
+
+DEFAULT_BATCH_BYTES = 120 * 1024 ** 3
+ARCHIVE_SUFFIXES = ('.rar', '.zip', '.7z', '.tar', '.tar.gz', '.tgz', '.tar.bz2',
+                    '.tbz2', '.tar.xz', '.txz', '.gz', '.bz2', '.xz')
+DEVNULL = subprocess.DEVNULL
+
+
+def _is_archive(path: Path) -> bool:
+    return path.name.lower().endswith(ARCHIVE_SUFFIXES)
+
+
+def _which(*names):
+    for name in names:
+        found = shutil.which(name)
+        if found:
+            return found
+    return 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}')
+    return target
+
+
+def _extract_compressed(path: Path, dest: Path):
+    """Try the available RAR/7z extractors in turn.
+
+    No single tool wins: `unrar` handles real RARs, `7z` handles the many
+    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')
+    unrar = _which('unrar')
+    if path.name.lower().endswith('.rar'):
+        candidates = [unrar, seven]
+    else:
+        candidates = [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:
+            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
+        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):
+    with tarfile.open(path) as t:
+        try:
+            t.extractall(dest, filter='data')  # py3.12+
+        except TypeError:  # py3.11
+            t.extractall(dest)
+
+
+def _extract_archive(path: Path, dest: Path):
+    name = path.name.lower()
+    dest.mkdir(parents=True, exist_ok=True)
+    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')):
+        _extract_compressed(path, dest)
+    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
+        with opener(path, 'rb') as r, open(dest / path.stem, 'wb') as w:
+            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()]
+    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
+    for f in files:
+        sz = f.stat().st_size
+        if batch and size + sz > batch_bytes:
+            yield batch
+            batch, size = [], 0
+        batch.append(f)
+        size += sz
+    if batch:
+        yield batch
+
+
+def _count_lines(path):
+    total = 0
+    with open(path, 'rb') as r:
+        for chunk in iter(lambda: r.read(1 << 20), b''):
+            total += chunk.count(b'\n')
+    return total
+
+
+def _stats_markdown(state):
+    stats = state['stats']
+    total_in = sum(s['input_bytes'] for s in stats)
+    total_pairs = sum(s['pairs'] for s in stats)
+    total_rej = sum(s['rejects'] for s in stats)
+    total_s = sum(s['total_s'] for s in stats)
+    lines = [
+        '# collectionparser build stats',
+        '',
+        f'- input: {total_in / 1e9:.1f} GB across {sum(s["files"] for s in stats):,} files',
+        f'- valid pairs: {total_pairs:,}',
+        f'- rejects: {total_rej:,}',
+        f'- batches: {len(stats)}',
+        f'- build seconds: {total_s:,.0f}',
+    ]
+    if total_s:
+        lines.append(f'- aggregate throughput: {total_in / 1e6 / total_s:.1f} MB/s')
+    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:
+        lines.append(f'- unique records: {state["unique_records"]:,}')
+    return '\n'.join(lines) + '\n'
+
+
+def _extract_batch(archives, extract_dir, index):
+    if not archives:
+        return []
+    t0 = time.time()
+    failed = 0
+    errors = []
+    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}')
+    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
+
+
+def build(source, dataset, scratch, batch_bytes=DEFAULT_BATCH_BYTES, dedup=True, extract=True):
+    dataset = Path(dataset).expanduser()
+    scratch = Path(scratch).expanduser()
+    dataset.mkdir(parents=True, exist_ok=True)
+    scratch.mkdir(parents=True, exist_ok=True)
+    for stale in scratch.glob('extract_*'):
+        shutil.rmtree(stale, ignore_errors=True)
+
+    if extract and not _which('7z', '7za', 'unrar'):
+        print('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 = json.loads(state_path.read_text()) if state_path.exists() else {'done': [], 'stats': []}
+    done = set(state['done'])
+    files = _discover(source)
+    batches = list(_batches(files, batch_bytes))
+    print(f'{len(files)} files, {len(batches)} batches, {len(done)} already done.', flush=True)
+
+    failures = scratch / 'failures.txt'
+    for index, batch in enumerate(batches):
+        if index in done:
+            continue
+        success = scratch / f'success_{index:05d}.txt'
+        failure = scratch / f'failure_{index:05d}.txt'
+        extract_dir = scratch / f'extract_{index:05d}'
+        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, 'r', 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))
+        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))
+        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)})
+            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:
+                    shutil.copyfileobj(r, w, length=1024 * 1024)
+                failure.unlink(missing_ok=True)
+            progress_path.unlink(missing_ok=True)
+        else:
+            pairs = rejects = 0
+        shutil.rmtree(extract_dir, ignore_errors=True)
+
+        elapsed = time.time() - t0
+        stat = {'batch': index, 'files': len(batch), '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_path.write_text(json.dumps(state, indent=2))
+        (scratch / 'stats.md').write_text(_stats_markdown(state))
+        print('BATCH ' + json.dumps(stat), flush=True)
+
+    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)
+        state['deduped'] = True
+        state_path.write_text(json.dumps(state, indent=2))
+
+    if state.get('unique_records') is None:
+        state['unique_records'] = sum(_count_lines(p) for p in dataset.glob('**/*') if p.is_file())
+        state_path.write_text(json.dumps(state, indent=2))
+
+    (scratch / 'stats.md').write_text(_stats_markdown(state))
+    print('BUILD COMPLETE', flush=True)

collectionparser/cli.py

@@ -0,0 +1,128 @@
+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 \; &
+
+
+@click.group()
+@click.option('--debug/--no-debug')
+@click.pass_context
+def cli(ctx, 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). '
+                        'Use RESTART_FROM to resume from the i-th line.')
+@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.')
+@click.option('--path', required=True, type=input_dir())
+@click.option('--size', default=50, type=int, show_default=True)  # MB.
+def chunk(path, size):
+    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.')
+@click.option('--path', required=True, type=input_dir())
+def clean(path):
+    API.clean(path)
+
+
+@cli.command(short_help='Infers passwords of a list of emails defined in FILE with a query friendly folder DATASET.')
+@click.option('--file', required=True, type=input_file())
+@click.option('--dataset', required=True)
+def test(file: str, dataset: str):
+    API.test(file, dataset)
+
+
+@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('--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)
+
+
+@cli.command(short_help='Merges dataset SRC into dataset DEST. Duplicates are removed in the second pass.')
+@click.option('--src', 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)
+
+
+@cli.command(short_help='Evaluates some metrics such as precision/recall (e.g. is OLD into NEW).')
+@click.option('--old', required=True, type=input_dir())
+@click.option('--new', required=True, type=input_dir())
+def evaluate(old: str, new: str):
+    API.evaluate(old, new)
+
+
+@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:
+        results = QueryAPI.wildcard(dataset, pattern)
+    else:
+        results = QueryAPI.query(dataset, pattern, 'email')
+    if results:
+        click.echo('\n'.join(results))
+
+
+@cli.command(short_help='Exports a dataset as JSON Lines for OpenSearch/bulk ingest.')
+@click.option('--dataset', required=True, type=input_dir())
+@click.option('--out', required=True, type=output_file())
+def export(dataset: str, out: str):
+    API.export_jsonl(dataset, out)
+
+
+@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())
+@click.option('--dataset', required=True, type=output_dir())
+@click.option('--scratch', required=True, type=output_dir())
+@click.option('--batch_gb', default=120, type=int, show_default=True)
+@click.option('--dedup/--no-dedup', default=True)
+@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)
+
+
+if __name__ == '__main__':
+    cli()

collectionparser/parser.py

@@ -0,0 +1,205 @@
+import os
+import re
+import shutil
+import string
+import uuid
+from concurrent.futures import ProcessPoolExecutor
+from glob import glob
+from pathlib import Path
+from time import time
+
+from tqdm import tqdm
+
+from collectionparser.utils import PasswordType, filter_printable, shuffle
+
+# Separators tried in order; the duplicate ',' of the original is dropped.
+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)
+_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
+
+
+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$'):
+        return PasswordType.B_CRYPT
+    return PasswordType.DECRYPTED
+
+
+def parse_line(row, separator, flip=False):
+    if separator not in row:
+        return None
+    if separator != ' ':
+        parts = row.split(separator)
+        if len(parts) > 2:
+            # [email protected]:kiyan1365:16 -> keep the last two fields.
+            row = separator.join(parts[-2:])
+    mail, password = row.split(separator, 1)
+    if flip:
+        mail, password = password, mail
+    password = password.replace(' ', '')
+    if not password or len(password) > 30 or '\n\n' in password:
+        return None
+    if set(password) == {'?'}:  # [email protected]:?????? ??? ?????
+        return None
+    if password.count('.') >= 3 and _IP_RE.search(password):
+        return None
+    mail = mail.replace(' ', '')
+    if '@' not in mail:  # https://grox.net/utils/encoding.html
+        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 '<.\\>_':
+        mail = mail[:-1]
+    if first_char in '.<':
+        mail = mail[1:]
+    if mail.endswith('@com'):  # iemongozalu@gmail@com mizo1210
+        mail = mail.replace('@com', '.com')
+    if not valid_email(mail):
+        # some people replace neanta,[email protected],jp
+        mail = mail.replace(',', '.')
+        parts = mail.split('@')
+        if len(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
+        mail = mail.replace('@hotmail.com', '', 1)
+    if valid_email(mail) and not valid_email(password):
+        return mail, password, password_type(password)
+    return None
+
+
+def parseline(s: str):
+    b = filter_printable(s.strip())
+    while '..' in b:
+        b = b.replace('..', '')
+    b = b.removeprefix('mailto:').removeprefix('E-mail:')
+    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('@@', '@')
+    if ':' in b and ';' in b:
+        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:
+        return None
+    for flip in (False, True):
+        for separator in VALID_SEPARATORS:
+            result = parse_line(b, separator, flip)
+            if result is not None:
+                return result
+    return None
+
+
+class Processor:
+    """Streams one input file line by line, appending results straight to disk.
+
+    Never holds parsed output in memory: workers append to the shared
+    success/failure files and return only (success, failure) counts, so memory
+    per worker is a single read buffer regardless of file size.
+    """
+
+    def __init__(self, success_filename=None, failure_filename=None):
+        self.success_filename = success_filename
+        self.failure_filename = failure_filename
+
+    # Each worker writes its own buffered part file; the parent concatenates
+    # parts after every chunk. Buffered writes avoid the one-syscall-per-line
+    # cost of shared O_APPEND, and separate files keep concurrent lines from
+    # 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):
+        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, \
+                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
+        return success, failure
+
+
+def _clear_parts(*finals):
+    # 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.*'):
+            os.remove(part)
+
+
+def _merge_parts(final):
+    parts = glob(f'{final}.part.*')
+    if not parts:
+        return
+    with open(final, 'ab') as w:
+        for part in parts:
+            with open(part, 'rb') as r:
+                shutil.copyfileobj(r, w, length=1024 * 1024)
+            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)
+    success_total = failure_total = 0
+    workers = os.cpu_count() or 1
+    processor = Processor(success_filename, failure_filename)
+    start_time = time()
+    progress = open(progress_filename, 'a+', encoding='utf8') if progress_filename else None
+    try:
+        with ProcessPoolExecutor(max_workers=workers) as executor, tqdm(files) as bar:
+            for i in range(0, len(files), workers):
+                chunk = files[i:i + workers]
+                for success, failure in executor.map(processor.process, chunk):
+                    success_total += success
+                    failure_total += failure
+                # Merge only after the whole chunk finished, then mark it done, so
+                # a crash never records a file whose output was not written.
+                _merge_parts(success_filename)
+                _merge_parts(failure_filename)
+                bar.update(len(chunk))
+                bar.set_description(f'success = {success_total:,} failure = {failure_total:,}')
+                for path in chunk:
+                    if progress:
+                        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()
+    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}
+    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')

collectionparser/query.py

@@ -0,0 +1,105 @@
+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:
+        if '@' not in txt_line:  # does it contain an email address?
+            continue
+        if ':' in txt_line:  # '[email protected]:iskandar89
+            separator = ':'
+        elif ';' in txt_line:  # '[email protected];ol1391ga
+            separator = ';'
+        else:
+            continue
+        try:
+            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)})
+
+
+def glob_to_regex(pattern: str) -> str:
+    """Glob -> regex anchored to the email field (`*`/`?` never cross the ':').
+
+    ponytail: `*` matches within the email only; a `*` that must cross the
+    separator needs a different index.
+    """
+    parts = []
+    for ch in pattern:
+        if ch == '*':
+            parts.append('[^:]*')
+        elif ch == '?':
+            parts.append('[^:]')
+        else:
+            parts.append(re.escape(ch))
+    return '^' + ''.join(parts) + ':'
+
+
+class QueryAPI:
+
+    @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
+            else:
+                dir_list[x.name] = x
+        return file_list, dir_list
+
+    @staticmethod
+    def search_in_file(path: Path, email):
+        assert path.is_file(), path
+        try:
+            output = subprocess.check_output(
+                ['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]
+
+    @staticmethod
+    def recursive_search(path: Path, characters: list, index: int):
+        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)
+        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):
+        if len(request) <= 3:
+            return []
+        if query_type == 'email' and '@' not in request:
+            return []
+        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.
+
+        ponytail: full scan of the tree (ripgrep is fast enough); add a
+        domain-sharded index only if wildcard latency becomes a problem.
+        """
+        try:
+            output = subprocess.check_output(
+                ['rg', '--no-filename', '--no-line-number', '-i', '-e',
+                 glob_to_regex(pattern), str(Path(data_dir))], text=True)
+        except subprocess.CalledProcessError:
+            return []
+        return postprocess(output.splitlines())

collectionparser/splitter.py

@@ -0,0 +1,47 @@
+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
+
+_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.
+    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):
+            if i % 10_000_000 == 0:
+                elapsed = time() - t1
+                t1 = 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)}')
+
+
+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)
+    except KeyError:
+        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)
+
+
+@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)

collectionparser/utils.py

@@ -0,0 +1,89 @@
+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 filter_printable(text: str) -> str:
+    return _NON_PRINTABLE.sub('', text)
+
+
+def iter_lines(filename):
+    with open(filename, 'r', encoding='utf8', errors='ignore') as r:
+        for line in r:
+            yield line.rstrip('\n')
+
+
+def create_dir_for_file(filename):
+    if parent := os.path.dirname(filename):
+        os.makedirs(parent, exist_ok=True)
+
+
+def input_file():
+    return click.Path(exists=True, file_okay=True, dir_okay=False,
+                      writable=False, readable=True, resolve_path=True)
+
+
+def input_dir():
+    return click.Path(exists=True, file_okay=False, dir_okay=True,
+                      writable=False, readable=True, resolve_path=True)
+
+
+def output_file():
+    return click.Path(exists=False, file_okay=True, dir_okay=False,
+                      writable=True, readable=True, resolve_path=True)
+
+
+def output_dir():
+    return click.Path(exists=False, file_okay=False, dir_okay=True,
+                      writable=True, readable=True, resolve_path=True)
+
+
+class PasswordType(StrEnum):
+    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

pyproject.toml

@@ -0,0 +1,22 @@
+[build-system]
+requires = ["setuptools>=68"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "collectionparser"
+version = "3.0.0"
+description = "Command line tools to parse and query multi-billion password collections."
+requires-python = ">=3.11"
+dependencies = [
+    "click>=8.1",
+    "tqdm>=4.66",
+]
+
+[project.optional-dependencies]
+test = ["pytest>=8"]
+
+[project.scripts]
+collectionparser = "collectionparser.cli:cli"
+
+[tool.setuptools.packages.find]
+include = ["collectionparser*"]

requirements.txt

@@ -1,4 +0,0 @@
-tqdm
-validate-email
-click
-cython

setup.py

@@ -1,26 +0,0 @@
-import sys
-
-from setuptools import setup, find_packages
-from Cython.Build import cythonize
-
-if sys.version_info < (3, 6):
-    raise Exception('Must be using Python 3.6 or above.')
-
-# Install cython first.
-setup(name='breach',
-      version='2.2',
-      author='Philippe Remy',
-      ext_modules=cythonize("breach/cparser.pyx", language_level="3"),
-      packages=find_packages(),
-      install_requires=[
-          'tqdm',
-          'validate-email',
-          'click'
-      ],
-      include_package_data=True,
-      entry_points={
-          'console_scripts': [
-              'breach=breach.cli:cli',
-          ],
-      }
-      )

tests/parser_test.py

@@ -1,16 +1,8 @@
 import unittest
 from time import time
 
-from breach.parser import LineParser
-from breach.splitter import get_output_file
-from breach.utils import alpha_num_lookup
-
-from breach.cparser import c_parseline
-
-p_fast = c_parseline
-p_slow = LineParser.parseline
-
-p = p_fast
+from collectionparser.parser import parseline as p
+from collectionparser.splitter import get_output_file
 
 
 class ParserTest(unittest.TestCase):
@@ -53,6 +45,10 @@ class ParserTest(unittest.TestCase):
         self.assertIsNone(p('ESSENTIALWAY'))
         self.assertIsNone(p('ESTARRON-ESPINOSA'))
         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('plain text line:with:colons'))
+        self.assertIsNotNone(p('user%40domain.com:password'))  # %40 still becomes '@'
         self.assertIsNotNone(p('[email protected]:[email protected]:victory'))
         self.assertIsNotNone(p('[email protected];;shanmuga6'))
         self.assertIsNotNone(p('[email protected]::shanmuga6'))
@@ -109,7 +105,7 @@ class ParserTest(unittest.TestCase):
 
     def test_parse_2(self):
         t = time()
-        for i in range(20000):  # 10,000 * 13 = 130,000 in 4.0 seconds.
+        for _ in range(20000):  # 10,000 * 13 = 130,000 in 4.0 seconds.
             # => 32B in in 10 days. That's a lot.
             p('[email protected]:123')
             p('hello@@gmail.com:123')
@@ -128,13 +124,12 @@ class ParserTest(unittest.TestCase):
         print(time() - t)
 
     def test_parse_3(self):
-        alpha = alpha_num_lookup()
-        self.assertEqual('j/o/s', get_output_file(alpha, 'josemariacanyero@hotmail.com'))
-        self.assertEqual('c/e/o', get_output_file(alpha, 'ceo21w@gmail.com'))
-        self.assertEqual('x/symbols', get_output_file(alpha, '[email protected]'))
-        self.assertEqual('x/a/symbols', get_output_file(alpha, 'xa_fear_238@hotmail.fr'))
-        self.assertEqual('x/symbols', get_output_file(alpha, 'x@hotmail.com'))
-        self.assertEqual('k/e/v', get_output_file(alpha, '[email protected]'))
+        self.assertEqual('j/o/s', get_output_file('[email protected]'))
+        self.assertEqual('c/e/o', get_output_file('ceo21w@gmail.com'))
+        self.assertEqual('x/symbols', get_output_file('x_fear_238@hotmail.fr'))
+        self.assertEqual('x/a/symbols', get_output_file('xa[email protected]'))
+        self.assertEqual('x/symbols', get_output_file('x@hotmail.com'))
+        self.assertEqual('k/e/v', get_output_file('kevmedic13@yahoo.com'))
 
 
 if __name__ == '__main__':

tests/query_test.py

@@ -1,33 +0,0 @@
-import unittest
-
-from breach.query import mask_pass, mask_pass_justin_rule
-
-
-class ParserTest(unittest.TestCase):
-
-    def test_mask_password(self):
-        self.assertEqual('123456**654321', mask_pass('12345688654321', num_asterisks=2))
-        self.assertEqual('123456***54321', mask_pass('12345688654321', num_asterisks=3))
-        self.assertEqual('12345****54321', mask_pass('12345688654321', num_asterisks=4))
-        self.assertEqual('12**********21', mask_pass('12345688654321', num_asterisks=10))
-        self.assertEqual('1************1', mask_pass('12345688654321', num_asterisks=12))
-        self.assertEqual('**************', mask_pass('12345688654321', num_asterisks=14))
-        self.assertEqual('**************', mask_pass('12345688654321', num_asterisks=18))
-        self.assertEqual('**************', mask_pass('12345688654321', num_asterisks=99))
-        self.assertEqual('1234568*654321', mask_pass('12345688654321', num_asterisks=1))
-        self.assertEqual('12345688654321', mask_pass('12345688654321', num_asterisks=0))
-
-    def test_mask_password_justin(self):
-        expected = 'il**********m'
-        initial = 'iloveicecream'
-        assert len(expected) == len(initial)
-        self.assertEqual(expected, mask_pass_justin_rule(initial, num_asterisks=-1))
-
-        expected = 'm****n'
-        initial = 'martin'
-        assert len(expected) == len(initial)
-        self.assertEqual(expected, mask_pass_justin_rule(initial, num_asterisks=-1))
-
-
-if __name__ == '__main__':
-    unittest.main()

tests/utils_test.py

@@ -1,6 +1,6 @@
 import unittest
 
-from breach.utils import RingDictionary
+from collectionparser.utils import RingDictionary
 
 
 class UtilsTest(unittest.TestCase):
@@ -8,27 +8,27 @@ class UtilsTest(unittest.TestCase):
     def test_ring_dictionary_1(self):
         rd = RingDictionary(max_num_keys=3)
         rd['a'] = 2
-        self.assertListEqual(sorted(rd.keys()), ['a'])
+        self.assertListEqual(sorted(rd.d), ['a'])
         self.assertListEqual(sorted(rd.k), ['a'])
         rd['b'] = 2
-        self.assertListEqual(sorted(rd.keys()), ['a', 'b'])
+        self.assertListEqual(sorted(rd.d), ['a', 'b'])
         self.assertListEqual(sorted(rd.k), ['a', 'b'])
         rd['c'] = 2
-        self.assertListEqual(sorted(rd.keys()), ['a', 'b', 'c'])
+        self.assertListEqual(sorted(rd.d), ['a', 'b', 'c'])
         self.assertListEqual(sorted(rd.k), ['a', 'b', 'c'])
         rd['d'] = 2
-        self.assertListEqual(sorted(rd.keys()), ['b', 'c', 'd'])
+        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.keys()), ['c'])
+        self.assertListEqual(sorted(rd.d), ['c'])
         self.assertListEqual(sorted(rd.k), ['c'])
 
         rd['d'] = 3
-        self.assertListEqual(sorted(rd.keys()), ['d'])
+        self.assertListEqual(sorted(rd.d), ['d'])
         self.assertListEqual(sorted(rd.k), ['d'])