64 lines
2.3 KiB
Python
64 lines
2.3 KiB
Python
|
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)
|