feat(chunks): aggiunge validator.py — invarianti + metriche

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 <noreply@anthropic.com>
This commit is contained in:
2026-06-09 08:02:35 +02:00
parent d2b8baa844
commit ed42e1ba7b
2 changed files with 153 additions and 0 deletions
+63
View File
@@ -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)