diff --git a/SKILL.md b/SKILL.md index 4e2eed9..a5a4e82 100644 --- a/SKILL.md +++ b/SKILL.md @@ -89,8 +89,6 @@ collectionparser merge --src other_data --dest big_dataset | `evaluate` | Hit-rate of one index against another. | `--old`, `--new` | | `filter-lines` | Keep lines within a length range. | `--file`, `--out`, `--less_than`, `--more_than` | -Global flag: `--debug` (verbose). - ## Rules an agent must follow - **Prefer `build` for large sources.** It parses in size-bounded batches so the diff --git a/collectionparser/api.py b/collectionparser/api.py index 8f60898..0d3e00c 100644 --- a/collectionparser/api.py +++ b/collectionparser/api.py @@ -9,10 +9,9 @@ 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 +from collectionparser.utils import create_dir_for_file, filter_printable, shuffle # ponytail: buckets under this size sort byte-exactly in RAM; larger buckets # shell out to `sort -u`, which spills to disk. @@ -37,7 +36,7 @@ class API: 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') + password_list = QueryAPI.query(new, random_line.partition(':')[0]) total_count += 1 hit = 'hit' if password_list else 'miss' hit_count += bool(password_list) @@ -58,11 +57,6 @@ class API: create_dir_for_file(str(dst)) move_file(src_file, dst) - @staticmethod - def parse(path: str, success_file: str, failure_file: str, - hashes_file: str | None = None) -> None: - parse_to_files(path, success_file, failure_file, hashes_file) - @staticmethod def sort(path: str) -> None: """Sort and de-duplicate every file under ``path`` in place.""" @@ -105,13 +99,14 @@ class API: def test(file: str, dataset: str) -> None: """Look up every email in ``file`` and print the cracked rate.""" cracked = total = 0 - for line in iter_lines(file): - results = QueryAPI.query(dataset, line.strip(), 'email') - if results: - passwords = [p.partition(':')[2] for p in results] - print('FOUND:', line, ':'.join(passwords)) - cracked += 1 - total += 1 + with open(file, encoding='utf8', errors='ignore') as reader: + for line in reader: + results = QueryAPI.query(dataset, line.strip()) + if results: + passwords = [p.partition(':')[2] for p in results] + print('FOUND:', line, ':'.join(passwords)) + cracked += 1 + total += 1 print(f'Cracked rate: {cracked / total if total else 0.0}') @staticmethod diff --git a/collectionparser/build.py b/collectionparser/build.py index 4d95a57..c24dbc0 100644 --- a/collectionparser/build.py +++ b/collectionparser/build.py @@ -137,23 +137,30 @@ def _count_lines(path: str | Path) -> int: return total +def _totals(stats: list[dict]) -> dict: + return { + 'input_bytes': sum(s['input_bytes'] for s in stats), + 'files': sum(s['files'] for s in stats), + 'pairs': sum(s['pairs'] for s in stats), + 'rejects': sum(s['rejects'] for s in stats), + 'seconds': sum(s['total_s'] for s in stats), + } + + def _stats_markdown(state: dict) -> str: stats = state['stats'] - total_in = sum(s['input_bytes'] for s in stats) - total_pairs = sum(s['pairs'] for s in stats) - total_rej = sum(s['rejects'] for s in stats) - total_s = sum(s['total_s'] for s in stats) + totals = _totals(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'- input: {totals["input_bytes"] / 1e9:.1f} GB across {totals["files"]:,} files', + f'- valid pairs: {totals["pairs"]:,}', + f'- rejects: {totals["rejects"]:,}', f'- batches: {len(stats)}', - f'- build seconds: {total_s:,.0f}', + f'- build seconds: {totals["seconds"]:,.0f}', ] - if total_s: - lines.append(f'- aggregate throughput: {total_in / 1e6 / total_s:.1f} MB/s') + if totals['seconds']: + lines.append(f'- aggregate throughput: {totals["input_bytes"] / 1e6 / totals["seconds"]:.1f} MB/s') if state.get('hashes') is not None: lines.append(f'- hashed passwords collected: {state["hashes"]:,}') if state.get('dedup_s') is not None: @@ -208,12 +215,16 @@ def build(source: str, dataset: str, scratch: str, batch_bytes: int = DEFAULT_BA state_path = scratch / 'build_state.json' state = json.loads(state_path.read_text()) if state_path.exists() else {'done': [], 'stats': []} + + def save(): + state_path.write_text(json.dumps(state, indent=2)) + done = set(state['done']) files = _discover(source) batches = list(_batches(files, batch_bytes)) state.update({'batches_total': len(batches), 'batch_files': [len(b) for b in batches], 'dataset': str(dataset), 'complete': False}) - state_path.write_text(json.dumps(state, indent=2)) + save() print(f'{len(files)} files, {len(batches)} batches, {len(done)} already done.', flush=True) failures = scratch / 'failures.txt' @@ -223,7 +234,7 @@ def build(source: str, dataset: str, scratch: str, batch_bytes: int = DEFAULT_BA if index in done: continue state['phase'] = f'batch {index}' - state_path.write_text(json.dumps(state, indent=2)) + save() success = scratch / f'success_{index:05d}.txt' failure = scratch / f'failure_{index:05d}.txt' extract_dir = scratch / f'extract_{index:05d}' @@ -273,31 +284,31 @@ def build(source: str, dataset: str, scratch: str, batch_bytes: int = DEFAULT_BA state['stats'].append(stat) done.add(index) state['done'] = sorted(done) - state_path.write_text(json.dumps(state, indent=2)) + save() (scratch / 'stats.md').write_text(_stats_markdown(state)) print('BATCH ' + json.dumps(stat), flush=True) if hashes.exists(): sort_unique(hashes, hashes) # de-duplicate collected hashed passwords state['hashes'] = _count_lines(hashes) - state_path.write_text(json.dumps(state, indent=2)) + save() if dedup and not state.get('deduped'): state['phase'] = 'deduping' - state_path.write_text(json.dumps(state, indent=2)) + save() 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)) + save() 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)) + save() state['phase'] = 'complete' state['complete'] = True - state_path.write_text(json.dumps(state, indent=2)) + save() (scratch / 'stats.md').write_text(_stats_markdown(state)) print('BUILD COMPLETE', flush=True) @@ -312,15 +323,16 @@ def build_status(scratch: str) -> dict: done = state.get('done', []) stats = state.get('stats', []) total = state.get('batches_total') + totals = _totals(stats) result = { 'phase': state.get('phase', 'unknown'), 'complete': state.get('complete', False), 'batches_done': len(done), 'batches_total': total, 'deduped': state.get('deduped', False), - 'input_gb': round(sum(s['input_bytes'] for s in stats) / 1e9, 2), - 'pairs': sum(s['pairs'] for s in stats), - 'rejects': sum(s['rejects'] for s in stats), + 'input_gb': round(totals['input_bytes'] / 1e9, 2), + 'pairs': totals['pairs'], + 'rejects': totals['rejects'], 'hashes': state.get('hashes'), 'unique_records': state.get('unique_records'), } diff --git a/collectionparser/cli.py b/collectionparser/cli.py index f11da25..0edf2bf 100644 --- a/collectionparser/cli.py +++ b/collectionparser/cli.py @@ -5,16 +5,15 @@ import json import click from collectionparser.api import API +from collectionparser.parser import parse_to_files from collectionparser.query import QueryAPI from collectionparser.splitter import split_file from collectionparser.utils import input_dir, input_file, output_dir, output_file @click.group() -@click.option('--debug/--no-debug') -def cli(debug): - if debug: - click.echo('Debugging mode enabled.') +def cli(): + pass @cli.command(short_help='Converts a large FILE to a query friendly folder OUT (e.g. a/b/c). ' @@ -60,7 +59,7 @@ def test(file: str, dataset: str): @click.option('--hashes_file', required=False, type=output_file(), help='Collect hashed passwords here for later `crack`.') def parse(path, success_file, failure_file, hashes_file): - API.parse(path, success_file, failure_file, hashes_file) + parse_to_files(path, success_file, failure_file, hashes_file) @cli.command(short_help='Merges dataset SRC into dataset DEST. Duplicates are removed in the second pass.') @@ -95,7 +94,7 @@ def query(dataset: str, pattern: str): if '*' in pattern or '?' in pattern: results = QueryAPI.wildcard(dataset, pattern) else: - results = QueryAPI.query(dataset, pattern, 'email') + results = QueryAPI.query(dataset, pattern) if results: click.echo('\n'.join(results)) diff --git a/collectionparser/query.py b/collectionparser/query.py index 1129deb..c05ddff 100644 --- a/collectionparser/query.py +++ b/collectionparser/query.py @@ -87,11 +87,9 @@ class QueryAPI: return [] @staticmethod - def query(data_dir: str, request: str, query_type: str) -> list[str]: + def query(data_dir: str, request: str) -> list[str]: """Exact email lookup using the bucket path (O(1) file read).""" - if len(request) <= 3: - return [] - if query_type == 'email' and '@' not in request: + if len(request) <= 3 or '@' not in request: return [] return postprocess(QueryAPI.recursive_search(Path(data_dir), list(request), 0)) diff --git a/collectionparser/splitter.py b/collectionparser/splitter.py index 66c4e0b..bd4e41e 100644 --- a/collectionparser/splitter.py +++ b/collectionparser/splitter.py @@ -32,8 +32,8 @@ def split_file(input_file: str, output_dir: str = 'out', restart_from: int = 0) def write_line_to_file(handles: dict[str, object], line: str, output_file: str, - buffer_scale: int = 100, max_open: int = 100) -> None: - """Append ``line`` to ``output_file``, keeping at most ``max_open`` handles. + buffer_scale: int = 100) -> None: + """Append ``line`` to ``output_file``, keeping at most 100 handles open. Evicted handles are flushed and closed so the process stays under the file descriptor limit. @@ -41,7 +41,7 @@ def write_line_to_file(handles: dict[str, object], line: str, output_file: str, try: handles[output_file].write(line) # type: ignore[attr-defined] except KeyError: - if len(handles) >= max_open: + if len(handles) >= 100: handles.pop(next(iter(handles))).flush() # type: ignore[attr-defined] create_dir_for_file(output_file) handles[output_file] = open( # noqa: SIM115 - cached handle, not scope-bound diff --git a/collectionparser/utils.py b/collectionparser/utils.py index ad7ff19..34026fa 100644 --- a/collectionparser/utils.py +++ b/collectionparser/utils.py @@ -21,13 +21,6 @@ def filter_printable(text: str) -> str: return _NON_PRINTABLE.sub('', text) -def iter_lines(filename: str): - """Yield lines from ``filename`` without the trailing newline, streaming.""" - with open(filename, encoding='utf8', errors='ignore') as reader: - for line in reader: - yield line.rstrip('\n') - - def create_dir_for_file(filename: str) -> None: """Create the parent directory of ``filename`` if it is missing.""" if parent := os.path.dirname(filename): diff --git a/pyproject.toml b/pyproject.toml index 6dbdefe..b05f122 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -23,7 +23,6 @@ dependencies = [ ] [project.optional-dependencies] -test = ["pytest>=8"] dev = ["pytest>=8", "ruff>=0.6"] [project.scripts] diff --git a/tests/parser_test.py b/tests/parser_test.py index 8f1099a..fdfbcc1 100644 --- a/tests/parser_test.py +++ b/tests/parser_test.py @@ -1,5 +1,4 @@ import unittest -from time import time from collectionparser.parser import parseline as p from collectionparser.splitter import get_output_file @@ -108,26 +107,6 @@ class ParserTest(unittest.TestCase): self.assertIsNone(p('jsjjd.@Com.:love melon')) def test_parse_2(self): - t = time() - for _ in range(20000): # 10,000 * 13 = 130,000 in 4.0 seconds. - # => 32B in in 10 days. That's a lot. - p('hello@gmail.com:123') - p('hello@@gmail.com:123') - p('hello @ gmail.com:123') - p('hello @@ gmail.com:123') - p('hello @@ gmail.com:123:') - p('hello @@ gmail.com,123') - p('123,hello @@ gmail.com') - p('hello@gmail.com123') - p('gauntletskd@gmail.com:?????? ??? ?????') - p('geoffroy.fogeoff4q@yahoo.co.uk.depth66charge:85.242.201.81,felix') - p('thakhuy.step129@hotmail.com.kkk111:125.26.106.158,nopparat') - p('bitebackbulldog@gmail.com:INSERT INTO `users' - '`(`id`, `verify`, `username`, `email`, `password') - p('aferrer@e-vende.com:Login') - print(time() - t) - - def test_parse_3(self): self.assertEqual('j/o/s', get_output_file('josemariacanyero@hotmail.com')) self.assertEqual('c/e/o', get_output_file('ceo21w@gmail.com')) self.assertEqual('x/symbols', get_output_file('x_fear_238@hotmail.fr'))