feat(chunks): aggiunge chunker.py — CLI + orchestrazione pipeline AST-based
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,140 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
chunker.py — CLI + orchestrazione pipeline AST-based
|
||||
|
||||
Uso:
|
||||
python chunks/chunker.py --stem <stem>
|
||||
python chunks/chunker.py # tutti gli stem in sources/
|
||||
python chunks/chunker.py --stem <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()
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user