From 1980efb0d653f65233569a19f9f2bcebe4961a19 Mon Sep 17 00:00:00 2001 From: Davide Grilli Date: Tue, 9 Jun 2026 08:21:54 +0200 Subject: [PATCH] =?UTF-8?q?feat(chunks):=20aggiunge=20chunker.py=20?= =?UTF-8?q?=E2=80=94=20CLI=20+=20orchestrazione=20pipeline=20AST-based?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Sonnet 4.6 --- chunks/chunker.py | 140 +++++++++++++++++++++++++++++++++++ tests/chunks/test_chunker.py | 109 +++++++++++++++++++++++++++ 2 files changed, 249 insertions(+) create mode 100644 chunks/chunker.py create mode 100644 tests/chunks/test_chunker.py diff --git a/chunks/chunker.py b/chunks/chunker.py new file mode 100644 index 0000000..f1c0821 --- /dev/null +++ b/chunks/chunker.py @@ -0,0 +1,140 @@ +#!/usr/bin/env python3 +""" +chunker.py — CLI + orchestrazione pipeline AST-based + +Uso: + python chunks/chunker.py --stem + python chunks/chunker.py # tutti gli stem in sources/ + python chunks/chunker.py --stem --force +""" +from __future__ import annotations +import argparse +import json +import sys +from dataclasses import asdict +from datetime import datetime, timezone +from pathlib import Path + +_HERE = Path(__file__).resolve().parent +_ROOT = _HERE.parent +if str(_ROOT) not in sys.path: + sys.path.insert(0, str(_ROOT)) + +from chunks.config import ChunkerConfig +from chunks.parser import parse +from chunks.segmenter import segment +from chunks.packer import pack +from chunks.validator import validate +from chunks.models import ChunkingResult, Diagnostics + + +def find_source(stem: str, root: Path) -> Path | None: + candidates = [ + root / "sources" / f"{stem}_output" / "auto" / f"{stem}.md", + root / "sources" / f"{stem}.md", + ] + for p in candidates: + if p.exists(): + return p + return None + + +def run_pipeline(stem: str, root: Path = _ROOT, + config: ChunkerConfig | None = None, + force: bool = False) -> ChunkingResult: + config = config or ChunkerConfig() + out_dir = root / "chunks" / stem + chunks_path = out_dir / "chunks.json" + + if chunks_path.exists() and not force: + print(f"[{stem}] esiste già — usa --force per rigenerare. Skip.") + return ChunkingResult(stem=stem, source_path="", chunks=[], + diagnostics=Diagnostics([], [], {})) + + source_path = find_source(stem, root) + if source_path is None: + print(f"[{stem}] sorgente non trovata in sources/. Saltato.", file=sys.stderr) + return ChunkingResult(stem=stem, source_path="", chunks=[], + diagnostics=Diagnostics( + errors=[f"sorgente non trovata per stem '{stem}'"], + warnings=[], metrics={})) + + source = source_path.read_text(encoding="utf-8") + tokens, lines = parse(source) + blocks = segment(tokens, lines, config) + chunks = pack(blocks, config, stem) + + result = ChunkingResult( + stem=stem, + source_path=str(source_path.relative_to(root)), + chunks=chunks, + diagnostics=Diagnostics([], [], {}), + ) + result.diagnostics = validate(result, source, config) + + out_dir.mkdir(parents=True, exist_ok=True) + + chunks_path.write_text( + json.dumps([asdict(c) for c in chunks], ensure_ascii=False, indent=2), + encoding="utf-8", + ) + (out_dir / "meta.json").write_text( + json.dumps({ + "stem": stem, + "source_path": result.source_path, + "total_chunks": len(chunks), + "total_chars": sum(c.chars for c in chunks), + "created_at": datetime.now(timezone.utc).isoformat(), + "config": { + "max_chars": config.max_chars, + "min_chars": config.min_chars, + "target_chars": config.target_chars, + "context_depth": config.context_depth, + }, + }, ensure_ascii=False, indent=2), + encoding="utf-8", + ) + (out_dir / "report.json").write_text( + json.dumps(asdict(result.diagnostics), ensure_ascii=False, indent=2), + encoding="utf-8", + ) + + n = len(chunks) + e = len(result.diagnostics.errors) + w = len(result.diagnostics.warnings) + print(f"[{stem}] {n} chunk | errors={e} warnings={w} → {out_dir}") + return result + + +def _discover_stems(root: Path) -> list[str]: + sources = root / "sources" + stems: set[str] = set() + if not sources.exists(): + return [] + for p in sources.iterdir(): + if p.is_dir() and p.name.endswith("_output"): + stem = p.name[: -len("_output")] + if (p / "auto" / f"{stem}.md").exists(): + stems.add(stem) + elif p.is_file() and p.suffix == ".md": + stems.add(p.stem) + return sorted(stems) + + +def main() -> None: + parser = argparse.ArgumentParser(description="Chunker Markdown AST-based") + parser.add_argument("--stem", help="Stem documento (es. valute-virtuali)") + parser.add_argument("--force", action="store_true", help="Rigenera anche se già presente") + args = parser.parse_args() + + stems = [args.stem] if args.stem else _discover_stems(_ROOT) + if not stems: + print("Nessun sorgente trovato in sources/.", file=sys.stderr) + sys.exit(1) + + for stem in stems: + run_pipeline(stem=stem, force=args.force) + + +if __name__ == "__main__": + main() diff --git a/tests/chunks/test_chunker.py b/tests/chunks/test_chunker.py new file mode 100644 index 0000000..ca0bbb9 --- /dev/null +++ b/tests/chunks/test_chunker.py @@ -0,0 +1,109 @@ +import json +import pytest +from pathlib import Path +from chunks.chunker import run_pipeline, find_source + + +SAMPLE_MD = """# Sezione Principale + +## Introduzione + +Questo è il primo paragrafo della sezione introduttiva. Contiene testo sufficiente. + +## Contenuto + +Il secondo paragrafo parla di argomenti tecnici. Continua per alcune righe. + +```python +def esempio(): + return 42 +``` + +## Indice + +Questo contenuto deve essere saltato. + +## Conclusioni + +Paragrafo conclusivo del documento di test. +""" + + +@pytest.fixture +def tmp_stem(tmp_path): + stem = "test_doc" + source_dir = tmp_path / "sources" / f"{stem}_output" / "auto" + source_dir.mkdir(parents=True) + (source_dir / f"{stem}.md").write_text(SAMPLE_MD, encoding="utf-8") + (tmp_path / "chunks").mkdir() + return tmp_path, stem + + +def test_run_pipeline_produces_output_files(tmp_stem): + root, stem = tmp_stem + run_pipeline(stem=stem, root=root) + out_dir = root / "chunks" / stem + assert (out_dir / "chunks.json").exists() + assert (out_dir / "meta.json").exists() + assert (out_dir / "report.json").exists() + + +def test_chunks_json_schema(tmp_stem): + root, stem = tmp_stem + run_pipeline(stem=stem, root=root) + data = json.loads((root / "chunks" / stem / "chunks.json").read_text()) + assert isinstance(data, list) + assert len(data) > 0 + c = data[0] + for field in ("chunk_id", "content_original", "content_for_embedding", + "header_path", "block_ids", "flags", "neighbors", "assets"): + assert field in c, f"campo mancante: {field}" + + +def test_indice_section_skipped(tmp_stem): + root, stem = tmp_stem + run_pipeline(stem=stem, root=root) + data = json.loads((root / "chunks" / stem / "chunks.json").read_text()) + all_content = " ".join(c["content_original"] for c in data) + assert "saltato" not in all_content + + +def test_code_block_preserved(tmp_stem): + root, stem = tmp_stem + run_pipeline(stem=stem, root=root) + data = json.loads((root / "chunks" / stem / "chunks.json").read_text()) + all_content = " ".join(c["content_original"] for c in data) + assert "def esempio" in all_content + + +def test_force_flag_regenerates(tmp_stem): + root, stem = tmp_stem + run_pipeline(stem=stem, root=root) + first = json.loads((root / "chunks" / stem / "chunks.json").read_text()) + run_pipeline(stem=stem, root=root, force=True) + second = json.loads((root / "chunks" / stem / "chunks.json").read_text()) + assert len(first) == len(second) + + +def test_no_force_skips_existing(tmp_stem, capsys): + root, stem = tmp_stem + run_pipeline(stem=stem, root=root) + run_pipeline(stem=stem, root=root, force=False) + captured = capsys.readouterr() + assert "skip" in captured.out.lower() or "esiste" in captured.out.lower() + + +def test_find_source_output_auto(tmp_path): + stem = "doc" + path = tmp_path / "sources" / f"{stem}_output" / "auto" / f"{stem}.md" + path.parent.mkdir(parents=True) + path.write_text("# T\n") + assert find_source(stem, tmp_path) == path + + +def test_find_source_flat(tmp_path): + stem = "doc" + path = tmp_path / "sources" / f"{stem}.md" + path.parent.mkdir(parents=True) + path.write_text("# T\n") + assert find_source(stem, tmp_path) == path