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)