Friendlier for LLMs

README.md

@@ -51,8 +51,16 @@ collectionparser build \
 **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.
+re-parsed (resume is **per file**, not per line). The sort+split step of an
+interrupted batch is redone (duplicates are removed by the final dedup), so
+nothing is lost.
+
+**Progress:** one machine-readable command.
+
+```bash
+collectionparser status --scratch /path/to/scratch            # one-line JSON
+collectionparser status --scratch /path/to/scratch --pretty
+```
 
 State and stats live in `--scratch`: `build_state.json`, `stats.md`, and
 `extract_failures.txt` (archives that failed to extract, never silently skipped).
@@ -141,6 +149,7 @@ Run `collectionparser --help` or `collectionparser <command> --help`.
 | command | description |
 | --- | --- |
 | `build` | Batched, resumable: parse + dedup + split into the query tree. |
+| `status` | Machine-readable build/dataset progress (JSON). |
 | `query` | Exact or wildcard search over a dataset. |
 | `export` | Export a dataset as JSON Lines. |
 | `crack` | Dictionary-attack collected hashes with a wordlist. |

SKILL.md

@@ -26,6 +26,24 @@ External tools: `sort` and `split` (coreutils); `rg` (ripgrep) for
 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.
 
+## Checking progress (when the user asks)
+
+One machine-readable command, no log spelunking:
+
+```bash
+collectionparser status --scratch /path/to/scratch        # one-line JSON
+collectionparser status --scratch /path/to/scratch --pretty
+```
+
+Fields: `phase`, `complete`, `batches_done`/`batches_total`, `current_batch`,
+`current_files_done`/`current_files_total`, `input_gb`, `pairs`, `rejects`,
+`hashes`, `dataset_files`, `dataset_bytes`. Report those. Only if `status`
+errors (e.g. no `build_state.json`) fall back to
+`systemctl is-active collectionparser-build.service` and the build log.
+
+Pause / resume is **per input file**, not per line: files already merged are
+skipped, and at most the files still in flight when it stopped are re-parsed.
+
 ## Pipeline (the intended order)
 
 ```bash

collectionparser/build.py

@@ -211,6 +211,9 @@ def build(source: str, dataset: str, scratch: str, batch_bytes: int = DEFAULT_BA
     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))
     print(f'{len(files)} files, {len(batches)} batches, {len(done)} already done.', flush=True)
 
     failures = scratch / 'failures.txt'
@@ -219,6 +222,8 @@ def build(source: str, dataset: str, scratch: str, batch_bytes: int = DEFAULT_BA
     for index, batch in enumerate(batches):
         if index in done:
             continue
+        state['phase'] = f'batch {index}'
+        state_path.write_text(json.dumps(state, indent=2))
         success = scratch / f'success_{index:05d}.txt'
         failure = scratch / f'failure_{index:05d}.txt'
         extract_dir = scratch / f'extract_{index:05d}'
@@ -278,6 +283,8 @@ def build(source: str, dataset: str, scratch: str, batch_bytes: int = DEFAULT_BA
         state_path.write_text(json.dumps(state, indent=2))
 
     if dedup and not state.get('deduped'):
+        state['phase'] = 'deduping'
+        state_path.write_text(json.dumps(state, indent=2))
         t0 = time.time()
         API.sort(str(dataset))  # per-bucket sort -u dedups across batches
         state['dedup_s'] = round(time.time() - t0, 1)
@@ -288,5 +295,50 @@ def build(source: str, dataset: str, scratch: str, batch_bytes: int = DEFAULT_BA
         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))
 
+    state['phase'] = 'complete'
+    state['complete'] = True
+    state_path.write_text(json.dumps(state, indent=2))
     (scratch / 'stats.md').write_text(_stats_markdown(state))
     print('BUILD COMPLETE', flush=True)
+
+
+def build_status(scratch: str) -> dict:
+    """Return a compact, machine-readable snapshot of a build under ``scratch``."""
+    scratch = Path(scratch).expanduser()
+    state_path = scratch / 'build_state.json'
+    if not state_path.exists():
+        return {'error': f'no build_state.json in {scratch}'}
+    state = json.loads(state_path.read_text())
+    done = state.get('done', [])
+    stats = state.get('stats', [])
+    total = state.get('batches_total')
+    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),
+        'hashes': state.get('hashes'),
+        'unique_records': state.get('unique_records'),
+    }
+    current = next((i for i in range(total or 0) if i not in done), None)
+    if current is not None:
+        progress = scratch / f'progress_{current:05d}.txt'
+        result['current_batch'] = current
+        result['current_files_done'] = sum(1 for _ in progress.open()) if progress.exists() else 0
+        batch_files = state.get('batch_files') or []
+        if current < len(batch_files):
+            result['current_files_total'] = batch_files[current]
+    for prefix in ('success', 'failure'):
+        result[f'{prefix}_bytes'] = sum(p.stat().st_size for p in scratch.glob(f'{prefix}_*.txt'))
+    hashes_file = scratch / 'hashes.txt'
+    result['hashes_bytes'] = hashes_file.stat().st_size if hashes_file.exists() else 0
+    dataset = state.get('dataset')
+    if dataset and Path(dataset).exists():
+        files = [p for p in Path(dataset).glob('**/*') if p.is_file()]
+        result['dataset_files'] = len(files)
+        result['dataset_bytes'] = sum(p.stat().st_size for p in files)
+    return result

collectionparser/cli.py

@@ -1,5 +1,7 @@
 """Command line entry point."""
 
+import json
+
 import click
 
 from collectionparser.api import API
@@ -120,6 +122,14 @@ def crack(hashes_path: str, wordlist: tuple, passwords_from: tuple, out: str, wo
               workers=workers or None)
 
 
+@cli.command(short_help='Print build/dataset progress as JSON (for scripts and agents).')
+@click.option('--scratch', required=True, type=click.Path(exists=True, file_okay=False, dir_okay=True))
+@click.option('--pretty', is_flag=True, help='Indent the JSON.')
+def status(scratch: str, pretty: bool):
+    from collectionparser.build import build_status
+    click.echo(json.dumps(build_status(scratch), indent=2 if pretty else None, default=str))
+
+
 @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())

collectionparser/parser.py

@@ -3,7 +3,7 @@ import re
 import shutil
 import string
 import uuid
-from concurrent.futures import ProcessPoolExecutor
+from concurrent.futures import ProcessPoolExecutor, as_completed
 from contextlib import ExitStack
 from glob import glob
 from pathlib import Path
@@ -133,16 +133,24 @@ class Processor:
     # interleaving (a >4KB line would otherwise split across writes).
     # ponytail: parts cost one extra read+copy pass; merge in-worker only if
     # that pass ever dominates.
-    def process(self, filename: Path) -> tuple[int, int]:
+    def process(self, filename: Path) -> tuple[int, int, list[str]]:
+        """Parse one file; return ``(success, failure, part_files)``.
+
+        Results go to uniquely-named part files that the parent merges once this
+        task is reported complete, so tasks can run without a chunk barrier.
+        """
         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
+        success_part = f'{self.success_filename}.part.{part}'
+        failure_part = f'{self.failure_filename}.part.{part}'
+        hashes_part = f'{self.hashes_filename}.part.{part}' if self.hashes_filename else None
+        parts = [success_part, failure_part] + ([hashes_part] if hashes_part else [])
         with open(str(filename), encoding='utf8', errors='ignore') as r, \
-                open(f'{self.success_filename}.part.{part}', 'w', encoding='utf8') as sw, \
-                open(f'{self.failure_filename}.part.{part}', 'w', encoding='utf8') as fw:
-            hashes = (open(f'{self.hashes_filename}.part.{part}', 'w', encoding='utf8')  # noqa: SIM115
-                      if self.hashes_filename else None)
+                open(success_part, 'w', encoding='utf8') as sw, \
+                open(failure_part, 'w', encoding='utf8') as fw:
+            hashes = open(hashes_part, 'w', encoding='utf8') if hashes_part else None  # noqa: SIM115
             try:
                 for line in r:
                     result = parseline(line)
@@ -162,7 +170,7 @@ class Processor:
             finally:
                 if hashes is not None:
                     hashes.close()
-        return success, failure
+        return success, failure, parts
 
 
 def _clear_parts(*finals: str) -> None:
@@ -173,14 +181,20 @@ def _clear_parts(*finals: str) -> None:
             os.remove(part)
 
 
-def _merge_parts(final: str) -> None:
-    parts = glob(f'{final}.part.*')
-    if not parts:
-        return
-    with open(final, 'ab') as w:
+def _merge_part_list(parts: list[str]) -> None:
+    """Append each completed task's part file to its final file, then remove it.
+
+    Only parts of tasks already reported complete are passed here, so an
+    in-flight task can never be truncated mid-write.
+    """
+    groups: dict[str, list[str]] = {}
     for part in parts:
-            with open(part, 'rb') as r:
-                shutil.copyfileobj(r, w, length=1024 * 1024)
+        groups.setdefault(part.rsplit('.part.', 1)[0], []).append(part)
+    for final, part_files in groups.items():
+        with open(final, 'ab') as writer:
+            for part in part_files:
+                with open(part, 'rb') as reader:
+                    shutil.copyfileobj(reader, writer, length=1024 * 1024)
                 os.remove(part)
 
 
@@ -200,25 +214,29 @@ def parse_files(files: list[Path], success_filename: str, failure_filename: str,
     processor = Processor(success_filename, failure_filename, hashes_filename)
     start_time = time()
     with ExitStack() as stack:
-        progress = (stack.enter_context(open(progress_filename, 'a', encoding='utf8'))
+        progress = (stack.enter_context(open(progress_filename, 'a', encoding='utf8', buffering=1))
                     if progress_filename else None)
-        with ProcessPoolExecutor(max_workers=workers) as executor, tqdm(files) as bar:
-            for i in range(0, len(files), workers):
-                chunk = files[i:i + workers]
-                for success, failure in executor.map(processor.process, chunk):
+        pending_parts: list[str] = []
+        with ProcessPoolExecutor(max_workers=workers) as executor, tqdm(total=len(files)) as bar:
+            # Submit everything and merge each task's own part as it completes:
+            # no per-chunk barrier, so one large file cannot idle the other
+            # workers. ponytail: all futures live at once (~tens of MB for
+            # 100k files); a bounded submission window is the upgrade path.
+            futures = {executor.submit(processor.process, f): f for f in files}
+            for future in as_completed(futures):
+                success, failure, parts = future.result()
                 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)
-                if hashes_filename:
-                    _merge_parts(hashes_filename)
-                bar.update(len(chunk))
+                pending_parts.extend(parts)
+                if len(pending_parts) >= 256:
+                    _merge_part_list(pending_parts)
+                    pending_parts.clear()
+                bar.update(1)
                 bar.set_description(f'success = {success_total:,} failure = {failure_total:,}')
                 if progress:
-                    for path in chunk:
-                        progress.write(f'{path}\n')
+                    progress.write(f'{futures[future]}\n')
+        if pending_parts:
+            _merge_part_list(pending_parts)
     print(f'Time elapsed {time() - start_time:.3f} seconds.')
     return success_total, failure_total