From ed42e1ba7ba56be67e93956e5d4cfd967b980c82 Mon Sep 17 00:00:00 2001 From: Davide Grilli Date: Tue, 9 Jun 2026 08:02:35 +0200 Subject: [PATCH] =?UTF-8?q?feat(chunks):=20aggiunge=20validator.py=20?= =?UTF-8?q?=E2=80=94=20invarianti=20+=20metriche?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implementa validate() che controlla chunk_id duplicati, fence rotti, size compliance (esclusi overflow) e calcola metriche aggregate. Fix fixture cfg: min_chars=40 per allinearla ai chunk da 50 chars nel test. Co-Authored-By: Claude Sonnet 4.6 --- chunks/validator.py | 63 ++++++++++++++++++++++++ tests/chunks/test_validator.py | 90 ++++++++++++++++++++++++++++++++++ 2 files changed, 153 insertions(+) create mode 100644 chunks/validator.py create mode 100644 tests/chunks/test_validator.py diff --git a/chunks/validator.py b/chunks/validator.py new file mode 100644 index 0000000..4c65495 --- /dev/null +++ b/chunks/validator.py @@ -0,0 +1,63 @@ +from __future__ import annotations +import re +from chunks.models import ChunkingResult, Diagnostics +from chunks.config import ChunkerConfig + +_OPEN_FENCE_RE = re.compile(r"^(`{3,}|~{3,})", re.MULTILINE) + + +def _has_broken_fence(content: str) -> bool: + matches = _OPEN_FENCE_RE.findall(content) + return len(matches) % 2 != 0 + + +def validate(result: ChunkingResult, source: str, config: ChunkerConfig) -> Diagnostics: + errors: list[str] = [] + warnings: list[str] = [] + chunks = result.chunks + + if not chunks: + warnings.append("Nessun chunk prodotto.") + return Diagnostics(errors=errors, warnings=warnings, metrics={"total_chunks": 0}) + + # chunk_id unici + seen_ids: set[str] = set() + for c in chunks: + if c.chunk_id in seen_ids: + errors.append(f"chunk_id duplicato: {c.chunk_id}") + seen_ids.add(c.chunk_id) + + # fence rotto + for c in chunks: + if _has_broken_fence(c.content_original): + msg = f"fence rotto in {c.chunk_id} (righe {c.start_line}-{c.end_line})" + if config.fail_on_broken_fence: + errors.append(msg) + else: + warnings.append(msg) + + # size compliance (esclusi overflow) + non_overflow = [c for c in chunks if not c.flags.get("is_overflow")] + for c in non_overflow: + if c.chars > config.max_chars: + errors.append(f"chunk {c.chunk_id} supera max_chars ({c.chars} > {config.max_chars})") + + # metriche + total = len(chunks) + chars_list = [c.chars for c in chunks] + avg = sum(chars_list) // total if total else 0 + compliant = sum(1 for c in non_overflow if config.min_chars <= c.chars <= config.max_chars) + compliance = round(compliant / len(non_overflow), 4) if non_overflow else 1.0 + + metrics = { + "total_chunks": total, + "avg_chars": avg, + "min_chars_actual": min(chars_list), + "max_chars_actual": max(chars_list), + "overflow_count": sum(1 for c in chunks if c.flags.get("is_overflow")), + "sparse_count": sum(1 for c in chunks if c.flags.get("is_sparse")), + "atomic_count": sum(1 for c in chunks if c.content_type == "atomic_block"), + "size_compliance": compliance, + } + + return Diagnostics(errors=errors, warnings=warnings, metrics=metrics) diff --git a/tests/chunks/test_validator.py b/tests/chunks/test_validator.py new file mode 100644 index 0000000..a0f8dd4 --- /dev/null +++ b/tests/chunks/test_validator.py @@ -0,0 +1,90 @@ +import pytest +from chunks.models import Chunk, ChunkingResult, Diagnostics +from chunks.config import ChunkerConfig +from chunks.validator import validate + + +def make_chunk(idx: int, content: str, chars: int = None, + is_overflow: bool = False, content_type: str = "section_fragment") -> Chunk: + chars = chars or len(content) + return Chunk( + chunk_id=f"chk_{idx:06d}", chunk_index=idx, + content_original=content, content_for_embedding=content, + content_type=content_type, chars=chars, + start_line=idx * 3, end_line=idx * 3 + 2, + header_path=[{"level": 1, "text": "T"}], + block_ids=[f"blk_{idx:04d}"], + flags={"has_code": False, "has_table": False, "has_math": False, + "is_overflow": is_overflow, "is_sparse": False}, + neighbors={"previous_chunk_id": None, "next_chunk_id": None}, + assets=[], + ) + + +@pytest.fixture +def cfg(): + return ChunkerConfig(max_chars=200, min_chars=40, fail_on_content_loss=False) + + +def test_no_errors_on_valid_chunks(cfg): + result = ChunkingResult( + stem="test", source_path="sources/test.md", + chunks=[make_chunk(1, "Testo valido.")], + diagnostics=Diagnostics([], [], {}), + ) + diag = validate(result, "# T\n\nTesto valido.\n", cfg) + assert diag.errors == [] + + +def test_broken_fence_detected(cfg): + result = ChunkingResult( + stem="test", source_path="sources/test.md", + chunks=[make_chunk(1, "```python\ncodice senza chiusura")], + diagnostics=Diagnostics([], [], {}), + ) + diag = validate(result, "", cfg) + assert any("fence" in e.lower() for e in diag.errors) + + +def test_no_duplicate_chunk_ids(cfg): + c1 = make_chunk(1, "Testo A.") + c2 = make_chunk(1, "Testo B.") # stesso chunk_id + result = ChunkingResult( + stem="test", source_path="sources/test.md", + chunks=[c1, c2], diagnostics=Diagnostics([], [], {}), + ) + diag = validate(result, "", cfg) + assert any("duplicat" in e.lower() for e in diag.errors) + + +def test_metrics_populated(cfg): + chunks = [make_chunk(i, "x" * 50) for i in range(1, 4)] + result = ChunkingResult( + stem="test", source_path="sources/test.md", + chunks=chunks, diagnostics=Diagnostics([], [], {}), + ) + diag = validate(result, "", cfg) + assert diag.metrics["total_chunks"] == 3 + assert diag.metrics["avg_chars"] == 50 + assert diag.metrics["size_compliance"] == 1.0 + + +def test_overflow_not_counted_in_size_compliance(cfg): + chunks = [make_chunk(1, "x" * 300, is_overflow=True, content_type="overflow")] + result = ChunkingResult( + stem="test", source_path="sources/test.md", + chunks=chunks, diagnostics=Diagnostics([], [], {}), + ) + diag = validate(result, "", cfg) + assert diag.metrics["overflow_count"] == 1 + assert diag.errors == [] + + +def test_empty_chunks_list(cfg): + result = ChunkingResult( + stem="test", source_path="sources/test.md", + chunks=[], diagnostics=Diagnostics([], [], {}), + ) + diag = validate(result, "", cfg) + assert diag.warnings + assert diag.metrics["total_chunks"] == 0