From d7b33d4d8a95b81d85afc76f00ac2a9fb85ff376 Mon Sep 17 00:00:00 2001 From: StasonJatham Date: Sat, 19 Sep 2026 14:33:52 +0200 Subject: [PATCH] Friendlier for LLMs --- README.md | 13 +++++- SKILL.md | 18 +++++++++ collectionparser/build.py | 52 ++++++++++++++++++++++++ collectionparser/cli.py | 10 +++++ collectionparser/parser.py | 82 +++++++++++++++++++++++--------------- 5 files changed, 141 insertions(+), 34 deletions(-) diff --git a/README.md b/README.md index 5d90a4c..ac484db 100644 --- a/README.md +++ b/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_.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 --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. | diff --git a/SKILL.md b/SKILL.md index c0dcaf7..4e2eed9 100644 --- a/SKILL.md +++ b/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 diff --git a/collectionparser/build.py b/collectionparser/build.py index c658d53..4d95a57 100644 --- a/collectionparser/build.py +++ b/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 diff --git a/collectionparser/cli.py b/collectionparser/cli.py index e750488..f11da25 100644 --- a/collectionparser/cli.py +++ b/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()) diff --git a/collectionparser/parser.py b/collectionparser/parser.py index 95dbdaa..10ffaca 100644 --- a/collectionparser/parser.py +++ b/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,15 +181,21 @@ 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: - for part in parts: - with open(part, 'rb') as r: - shutil.copyfileobj(r, w, length=1024 * 1024) - os.remove(part) +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: + 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) def parse_files(files: list[Path], success_filename: str, failure_filename: str, @@ -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): - 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: 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 + 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