cleeeeeeeeen

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

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):
+        with open(file, encoding='utf8', errors='ignore') as reader:
-            results = QueryAPI.query(dataset, line.strip(), 'email')
+            for line in reader:
-            if results:
+                results = QueryAPI.query(dataset, line.strip())
-                passwords = [p.partition(':')[2] for p in results]
+                if results:
-                print('FOUND:', line, ':'.join(passwords))
+                    passwords = [p.partition(':')[2] for p in results]
-                cracked += 1
+                    print('FOUND:', line, ':'.join(passwords))
-            total += 1
+                    cracked += 1
+                total += 1
         print(f'Cracked rate: {cracked / total if total else 0.0}')
 
     @staticmethod

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)
+    totals = _totals(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'- input: {totals["input_bytes"] / 1e9:.1f} GB across {totals["files"]:,} files',
-        f'- valid pairs: {total_pairs:,}',
+        f'- valid pairs: {totals["pairs"]:,}',
-        f'- rejects: {total_rej:,}',
+        f'- rejects: {totals["rejects"]:,}',
         f'- batches: {len(stats)}',
-        f'- build seconds: {total_s:,.0f}',
+        f'- build seconds: {totals["seconds"]:,.0f}',
     ]
-    if total_s:
+    if totals['seconds']:
-        lines.append(f'- aggregate throughput: {total_in / 1e6 / total_s:.1f} MB/s')
+        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),
+        'input_gb': round(totals['input_bytes'] / 1e9, 2),
-        'pairs': sum(s['pairs'] for s in stats),
+        'pairs': totals['pairs'],
-        'rejects': sum(s['rejects'] for s in stats),
+        'rejects': totals['rejects'],
         'hashes': state.get('hashes'),
         'unique_records': state.get('unique_records'),
     }

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():
-def cli(debug):
+    pass
-    if debug:
-        click.echo('Debugging mode enabled.')
 
 
 @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))
 

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:
+        if len(request) <= 3 or '@' not in request:
-            return []
-        if query_type == 'email' and '@' not in request:
             return []
         return postprocess(QueryAPI.recursive_search(Path(data_dir), list(request), 0))
 

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:
+                       buffer_scale: int = 100) -> None:
-    """Append ``line`` to ``output_file``, keeping at most ``max_open`` handles.
+    """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

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):

pyproject.toml

@@ -23,7 +23,6 @@ dependencies = [
 ]
 
 [project.optional-dependencies]
-test = ["pytest>=8"]
 dev = ["pytest>=8", "ruff>=0.6"]
 
 [project.scripts]

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('[email protected]: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('[email protected]')
-            p('[email protected]:?????? ??? ?????')
-            p('[email protected]:85.242.201.81,felix')
-            p('[email protected]:125.26.106.158,nopparat')
-            p('[email protected]:INSERT INTO `users'
-              '`(`id`, `verify`, `username`, `email`, `password')
-            p('[email protected]:Login')
-        print(time() - t)
-
-    def test_parse_3(self):
         self.assertEqual('j/o/s', get_output_file('[email protected]'))
         self.assertEqual('c/e/o', get_output_file('[email protected]'))
         self.assertEqual('x/symbols', get_output_file('[email protected]'))