feat(chunks): pipeline AST-based completa — parser → segmenter → packer → validator
Riscrittura completa del chunker con architettura modulare basata su markdown-it-py.
Moduli:
- parser.py — markdown-it-py + dollarmath plugin → token stream con source map
- segmenter.py — token stream → Block[] (paragraph, code, table, list, math, html, thematic_break)
- packer.py — Block[] → Chunk[] con packing min/target/max e content_for_embedding
- validator.py — invarianti + metriche (size_compliance, overflow, sparse)
- chunker.py — CLI + orchestrazione pipeline, scrive chunks.json/meta.json/report.json
Flag universali: has_math rileva math_block, $$ e \begin{; has_table rileva pipe table e <table> HTML.
65 test passing.
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
+99
-372
@@ -1,413 +1,140 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
chunker.py — Chunking semantico da Markdown pulito
|
||||
|
||||
Input: sources/<stem>.md — Markdown già strutturato (H1/H2/H3 + paragrafi)
|
||||
Output: chunks/<stem>/chunks.json
|
||||
chunks/<stem>/meta.json
|
||||
|
||||
Regole:
|
||||
- ogni paragrafo diventa un chunk; paragrafi di contesti diversi non si mescolano
|
||||
- se un paragrafo supera MAX_CHARS viene spezzato a confine di frase (mai a metà)
|
||||
- paragrafi brevi (< MIN_CHARS) vengono fusi col successivo finché non raggiungono
|
||||
MIN_CHARS, a patto che abbiano lo stesso contesto heading
|
||||
- tabelle, liste e codice sono blocchi atomici (non si spezzano)
|
||||
chunker.py — CLI + orchestrazione pipeline AST-based
|
||||
|
||||
Uso:
|
||||
python chunks/chunker.py --stem <stem>
|
||||
python chunks/chunker.py # tutti gli stem con .md in sources/
|
||||
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 re
|
||||
import sys
|
||||
from dataclasses import asdict
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
_HERE = Path(__file__).resolve().parent
|
||||
if str(_HERE) not in sys.path:
|
||||
sys.path.insert(0, str(_HERE))
|
||||
import config as cfg
|
||||
_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
|
||||
|
||||
|
||||
# ─── Utilità ──────────────────────────────────────────────────────────────────
|
||||
|
||||
def split_sentences(text: str) -> list[str]:
|
||||
parts = re.split(cfg.SENTENCE_SPLIT_RE, text.strip())
|
||||
return [p.strip() for p in parts if p.strip()]
|
||||
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 context_to_meta(context: str) -> tuple[str, str]:
|
||||
parts = [p.strip() for p in context.split(" > ") if p.strip()]
|
||||
if len(parts) >= 2:
|
||||
return " > ".join(parts[:-1]), parts[-1]
|
||||
return (parts[0] if parts else ""), ""
|
||||
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([], [], {}))
|
||||
|
||||
# ─── Parser Markdown ──────────────────────────────────────────────────────────
|
||||
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={}))
|
||||
|
||||
_SKIP_HEADINGS_LOWER = {h.lower() for h in cfg.SKIP_HEADINGS}
|
||||
source = source_path.read_text(encoding="utf-8")
|
||||
tokens, lines = parse(source)
|
||||
blocks = segment(tokens, lines, config)
|
||||
chunks = pack(blocks, config, stem)
|
||||
|
||||
|
||||
def _is_skip_heading(title: str) -> bool:
|
||||
t = title.lower()
|
||||
return any(t == s or t.startswith(s) for s in _SKIP_HEADINGS_LOWER)
|
||||
|
||||
|
||||
def parse_paragraphs(text: str) -> list[dict]:
|
||||
"""Estrae blocchi dal Markdown con il loro contesto heading.
|
||||
|
||||
Restituisce: [{"context": "H1 > H2 > H3", "text": "...", "kind": "text|table|list|code"}]
|
||||
|
||||
- Heading senza corpo non emettono chunk: aggiornano solo il contesto.
|
||||
- Tabelle (righe |), liste (righe -/*) e code block (```) sono atomici.
|
||||
- Sezioni in SKIP_HEADINGS vengono saltate completamente.
|
||||
- Contenuto pre-heading saltato se SKIP_PRE_HEADING è True.
|
||||
"""
|
||||
headings = ["", "", ""] # H1, H2, H3
|
||||
result: list[dict] = []
|
||||
buf: list[str] = []
|
||||
cur_kind = "text"
|
||||
in_code = False
|
||||
skip_level: int | None = None # livello heading che ha attivato lo skip
|
||||
|
||||
def current_context() -> str:
|
||||
parts = [h for h in headings if h]
|
||||
return " > ".join(parts[:cfg.CONTEXT_DEPTH]) if parts else "documento"
|
||||
|
||||
def is_skipping() -> bool:
|
||||
return skip_level is not None
|
||||
|
||||
def flush() -> None:
|
||||
if is_skipping():
|
||||
buf.clear()
|
||||
return
|
||||
if cfg.SKIP_PRE_HEADING and current_context() == "documento":
|
||||
buf.clear()
|
||||
return
|
||||
body = "\n".join(buf).strip()
|
||||
if body:
|
||||
result.append({"context": current_context(), "text": body, "kind": cur_kind})
|
||||
buf.clear()
|
||||
|
||||
for line in text.splitlines():
|
||||
# ── Code block toggle ─────────────────────────────────────────────────
|
||||
if line.strip().startswith("```"):
|
||||
if not in_code:
|
||||
flush()
|
||||
cur_kind = "code"
|
||||
in_code = True
|
||||
if not is_skipping():
|
||||
buf.append(line)
|
||||
else:
|
||||
if not is_skipping():
|
||||
buf.append(line)
|
||||
in_code = False
|
||||
flush()
|
||||
cur_kind = "text"
|
||||
continue
|
||||
|
||||
if in_code:
|
||||
if not is_skipping():
|
||||
buf.append(line)
|
||||
continue
|
||||
|
||||
# ── Heading ───────────────────────────────────────────────────────────
|
||||
m = re.match(r"^(#{1,3}) (.+)", line)
|
||||
if m:
|
||||
flush()
|
||||
level = len(m.group(1))
|
||||
title = m.group(2).strip()
|
||||
|
||||
# chiudi skip se torniamo a livello pari o superiore
|
||||
if skip_level is not None and level <= skip_level:
|
||||
skip_level = None
|
||||
|
||||
headings[level - 1] = title
|
||||
for i in range(level, 3):
|
||||
headings[i] = ""
|
||||
cur_kind = "text"
|
||||
|
||||
# apri skip se questo heading è nella lista
|
||||
if _is_skip_heading(title):
|
||||
skip_level = level
|
||||
|
||||
continue
|
||||
|
||||
# ── Tabella ───────────────────────────────────────────────────────────
|
||||
if line.strip().startswith("|"):
|
||||
if cur_kind != "table":
|
||||
flush()
|
||||
cur_kind = "table"
|
||||
buf.append(line)
|
||||
continue
|
||||
|
||||
# ── Lista ─────────────────────────────────────────────────────────────
|
||||
if re.match(r"^\s*[-*]\s", line):
|
||||
if cur_kind != "list":
|
||||
flush()
|
||||
cur_kind = "list"
|
||||
buf.append(line)
|
||||
continue
|
||||
|
||||
# ── Riga vuota: chiude il paragrafo corrente ──────────────────────────
|
||||
if line.strip() == "":
|
||||
flush()
|
||||
cur_kind = "text"
|
||||
continue
|
||||
|
||||
# ── Testo normale ─────────────────────────────────────────────────────
|
||||
if cur_kind in ("table", "list", "code"):
|
||||
flush()
|
||||
cur_kind = "text"
|
||||
buf.append(line)
|
||||
|
||||
flush()
|
||||
return result
|
||||
|
||||
|
||||
# ─── Merge paragrafi brevi ────────────────────────────────────────────────────
|
||||
|
||||
def merge_short(paragraphs: list[dict]) -> list[dict]:
|
||||
"""Fonde paragrafi di testo consecutivi sotto MIN_CHARS con il successivo,
|
||||
purché condividano lo stesso contesto heading."""
|
||||
if not cfg.MERGE_SHORT_PARAGRAPHS:
|
||||
return paragraphs
|
||||
|
||||
result: list[dict] = []
|
||||
buf_para: dict | None = None
|
||||
buf_text: str = ""
|
||||
|
||||
def flush_buf() -> None:
|
||||
nonlocal buf_para, buf_text
|
||||
if buf_para is not None:
|
||||
result.append({**buf_para, "text": buf_text})
|
||||
buf_para = None
|
||||
buf_text = ""
|
||||
|
||||
for para in paragraphs:
|
||||
if para["kind"] != "text":
|
||||
flush_buf()
|
||||
result.append(para)
|
||||
continue
|
||||
|
||||
if buf_para is None:
|
||||
buf_para = para
|
||||
buf_text = para["text"]
|
||||
elif buf_para["context"] == para["context"] and len(buf_text) < cfg.MIN_CHARS:
|
||||
buf_text = buf_text + "\n\n" + para["text"]
|
||||
else:
|
||||
flush_buf()
|
||||
buf_para = para
|
||||
buf_text = para["text"]
|
||||
|
||||
flush_buf()
|
||||
return result
|
||||
|
||||
|
||||
# ─── Chunking ─────────────────────────────────────────────────────────────────
|
||||
|
||||
def make_chunks(paragraphs: list[dict]) -> list[dict]:
|
||||
"""Genera chunk da parse_paragraphs (dopo eventuale merge).
|
||||
|
||||
- Blocchi atomici (table, list, code): un chunk, mai spezzato.
|
||||
- Testo: un chunk per paragrafo; se supera MAX_CHARS, spezza a confine frase.
|
||||
"""
|
||||
chunks: list[dict] = []
|
||||
idx = 0
|
||||
|
||||
for para in paragraphs:
|
||||
text = para["text"]
|
||||
context = para["context"]
|
||||
kind = para["kind"]
|
||||
sezione, titolo = context_to_meta(context)
|
||||
|
||||
def emit(body: str) -> None:
|
||||
nonlocal idx
|
||||
chunk_text = f"[{context}]\n{body}"
|
||||
chunks.append({
|
||||
"chunk_id": f"c{idx}",
|
||||
"text": chunk_text,
|
||||
"sezione": sezione,
|
||||
"titolo": titolo,
|
||||
"context": context,
|
||||
"n_chars": len(chunk_text),
|
||||
})
|
||||
idx += 1
|
||||
|
||||
# ── Atomici ───────────────────────────────────────────────────────────
|
||||
if kind in ("table", "list", "code"):
|
||||
emit(text)
|
||||
continue
|
||||
|
||||
# ── Testo: split a confine di frase se supera MAX_CHARS ───────────────
|
||||
sents = split_sentences(text)
|
||||
if not sents:
|
||||
continue
|
||||
|
||||
current: list[str] = []
|
||||
for sent in sents:
|
||||
projected = len(" ".join(current + [sent]))
|
||||
if projected <= cfg.MAX_CHARS or not current:
|
||||
current.append(sent)
|
||||
else:
|
||||
emit(" ".join(current))
|
||||
current = [sent]
|
||||
|
||||
if current:
|
||||
emit(" ".join(current))
|
||||
|
||||
return chunks
|
||||
|
||||
|
||||
# ─── Merge frasi spezzate ─────────────────────────────────────────────────────
|
||||
|
||||
_SENT_END = re.compile(
|
||||
"[.!?;:"
|
||||
+ chr(0xBB) + ")\\\\"
|
||||
+ chr(0x2018) + chr(0x2019)
|
||||
+ chr(0x201C) + chr(0x201D)
|
||||
+ "\"'"
|
||||
+ chr(0x2014) + chr(0x2013) + chr(0x2026) + chr(0xB7)
|
||||
+ "]$"
|
||||
+ r"|\d[\d.,/]*$" # numero o versione
|
||||
+ r"|\$$" # formula LaTeX inline
|
||||
+ r"|\}$" # blocco LaTeX
|
||||
+ r"|>$" # tag HTML
|
||||
+ r"|\\\\$" # \\ LaTeX
|
||||
+ r"|\|$" # riga tabella
|
||||
result = ChunkingResult(
|
||||
stem=stem,
|
||||
source_path=str(source_path.relative_to(root)),
|
||||
chunks=chunks,
|
||||
diagnostics=Diagnostics([], [], {}),
|
||||
)
|
||||
|
||||
|
||||
def _body(chunk: dict) -> str:
|
||||
text = chunk["text"]
|
||||
nl = text.find("\n")
|
||||
return text[nl + 1:] if nl != -1 else text
|
||||
|
||||
|
||||
def merge_broken_sentences(chunks: list[dict]) -> list[dict]:
|
||||
"""Fonde chunk consecutivi con lo stesso contesto quando il primo termina
|
||||
senza punteggiatura di fine frase (frase spezzata dal sorgente)."""
|
||||
result: list[dict] = []
|
||||
i = 0
|
||||
while i < len(chunks):
|
||||
c = dict(chunks[i])
|
||||
body = _body(c)
|
||||
|
||||
while (
|
||||
i + 1 < len(chunks)
|
||||
and chunks[i + 1]["context"] == c["context"]
|
||||
and not _SENT_END.search(body.rstrip())
|
||||
):
|
||||
i += 1
|
||||
next_body = _body(chunks[i])
|
||||
body = body.rstrip() + " " + next_body.lstrip()
|
||||
|
||||
chunk_text = f"[{c['context']}]\n{body}"
|
||||
c["text"] = chunk_text
|
||||
c["n_chars"] = len(chunk_text)
|
||||
result.append(c)
|
||||
i += 1
|
||||
|
||||
for idx, c in enumerate(result):
|
||||
c["chunk_id"] = f"c{idx}"
|
||||
|
||||
return result
|
||||
|
||||
|
||||
# ─── Pipeline per documento ───────────────────────────────────────────────────
|
||||
|
||||
def process_stem(stem: str, project_root: Path, force: bool) -> bool:
|
||||
md_path = project_root / "sources" / f"{stem}_output" / "auto" / f"{stem}.md"
|
||||
if not md_path.exists():
|
||||
md_path = project_root / "sources" / f"{stem}.md"
|
||||
if not md_path.exists():
|
||||
print(f" ✗ {stem}.md non trovato (cercato in sources/{stem}_output/auto/ e sources/)")
|
||||
return False
|
||||
|
||||
out_dir = project_root / "chunks" / stem
|
||||
out_file = out_dir / "chunks.json"
|
||||
|
||||
if out_file.exists() and not force:
|
||||
print(f" ↩ chunks/{stem}/chunks.json già presente — skip (usa --force per rigenerare)")
|
||||
return True
|
||||
|
||||
print(f"[chunker] {stem}")
|
||||
|
||||
text = md_path.read_text(encoding="utf-8")
|
||||
paragraphs = parse_paragraphs(text)
|
||||
|
||||
if not paragraphs:
|
||||
print(f" ✗ Nessun paragrafo estratto da {md_path.name}")
|
||||
return False
|
||||
|
||||
if cfg.MERGE_SHORT_PARAGRAPHS:
|
||||
paragraphs = merge_short(paragraphs)
|
||||
|
||||
chunks = make_chunks(paragraphs)
|
||||
chunks = merge_broken_sentences(chunks)
|
||||
|
||||
if not chunks:
|
||||
print(f" ✗ Nessun chunk generato")
|
||||
return False
|
||||
result.diagnostics = validate(result, source, config)
|
||||
|
||||
out_dir.mkdir(parents=True, exist_ok=True)
|
||||
out_file.write_text(
|
||||
json.dumps(chunks, ensure_ascii=False, indent=2), encoding="utf-8"
|
||||
|
||||
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": str(md_path.relative_to(project_root)),
|
||||
"max_chars": cfg.MAX_CHARS,
|
||||
"min_chars": cfg.MIN_CHARS,
|
||||
"merge_short": cfg.MERGE_SHORT_PARAGRAPHS,
|
||||
"strategy": "one_paragraph_per_chunk",
|
||||
"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",
|
||||
)
|
||||
|
||||
lengths = [c["n_chars"] for c in chunks]
|
||||
over_max = sum(1 for l in lengths if l > cfg.MAX_CHARS)
|
||||
under_min = sum(1 for l in lengths if l < cfg.MIN_CHARS)
|
||||
avg = int(sum(lengths) / len(lengths))
|
||||
|
||||
print(f" ✅ {len(chunks)} chunk | media {avg} char | max {max(lengths)} char")
|
||||
if over_max:
|
||||
print(f" ⚠️ {over_max} chunk superano MAX_CHARS={cfg.MAX_CHARS}")
|
||||
if under_min:
|
||||
print(f" ℹ️ {under_min} chunk sotto MIN_CHARS={cfg.MIN_CHARS}")
|
||||
print(f" → chunks/{stem}/chunks.json")
|
||||
return True
|
||||
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
|
||||
|
||||
|
||||
# ─── Entry point ──────────────────────────────────────────────────────────────
|
||||
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)
|
||||
|
||||
if __name__ == "__main__":
|
||||
project_root = Path(__file__).parent.parent
|
||||
sources_dir = project_root / "sources"
|
||||
|
||||
parser = argparse.ArgumentParser(description="Markdown pulito → chunks.json")
|
||||
parser.add_argument("--stem", help="Nome documento (es. analisi2)")
|
||||
parser.add_argument("--force", action="store_true",
|
||||
help="Rigenera chunks.json anche se già presente")
|
||||
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()
|
||||
|
||||
if args.stem:
|
||||
stems = [args.stem]
|
||||
else:
|
||||
found = set()
|
||||
for p in sources_dir.glob("*_output/auto/*.md"):
|
||||
found.add(p.stem)
|
||||
for p in sources_dir.glob("*.md"):
|
||||
found.add(p.stem)
|
||||
stems = sorted(found)
|
||||
stems = [args.stem] if args.stem else _discover_stems(_ROOT)
|
||||
if not stems:
|
||||
print("Nessun file .md trovato in sources/")
|
||||
print("Nessun sorgente trovato in sources/.", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
results = [process_stem(s, project_root, args.force) for s in stems]
|
||||
ok = sum(results)
|
||||
print(f"\n{'✅' if all(results) else '⚠️ '} {ok}/{len(results)} documenti processati")
|
||||
sys.exit(0 if all(results) else 1)
|
||||
for stem in stems:
|
||||
run_pipeline(stem=stem, force=args.force)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
||||
+17
-57
@@ -1,64 +1,24 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Parametri della pipeline di chunking.
|
||||
from __future__ import annotations
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
Input atteso: sources/<stem>/<stem>.md — Markdown già pulito e ben strutturato.
|
||||
"""
|
||||
|
||||
# ─── Dimensione chunk ─────────────────────────────────────────────────────────
|
||||
|
||||
# Caratteri massimi per chunk (prefisso di contesto incluso).
|
||||
# Paragrafi più lunghi vengono spezzati a confine di frase.
|
||||
# Una singola frase che supera MAX_CHARS non viene mai spezzata.
|
||||
MAX_CHARS: int = 1200
|
||||
|
||||
# Soglia minima attesa (usata da verify_chunks come warning, non blocca).
|
||||
MIN_CHARS: int = 80
|
||||
|
||||
# ─── Spezzatura frasi ─────────────────────────────────────────────────────────
|
||||
|
||||
# Regex per rilevare il confine di fine frase.
|
||||
# Split solo prima di lettera maiuscola o virgolette — evita split su abbreviazioni.
|
||||
SENTENCE_SPLIT_RE: str = r"(?<=[.!?»])\s+(?=[A-ZÀÈÉÌÒÙ\"])"
|
||||
|
||||
# ─── Blocchi atomici ──────────────────────────────────────────────────────────
|
||||
|
||||
# Blocchi Markdown che non vengono mai spezzati, anche se superano MAX_CHARS.
|
||||
ATOMIC_TYPES: set[str] = {"table", "code", "list"}
|
||||
|
||||
# ─── Contesto heading ─────────────────────────────────────────────────────────
|
||||
|
||||
# Profondità massima del percorso heading incluso nel prefisso di ogni chunk.
|
||||
# 1 = solo H1, 2 = H1 > H2, 3 = H1 > H2 > H3.
|
||||
CONTEXT_DEPTH: int = 3
|
||||
|
||||
# ─── Sezioni da escludere ────────────────────────────────────────────────────
|
||||
|
||||
# Heading (case-insensitive) le cui sezioni vengono saltate completamente.
|
||||
# Il match è su prefisso: "indice" salta anche "Indice delle figure".
|
||||
# Lasciare vuoto per non escludere nulla.
|
||||
SKIP_HEADINGS: set[str] = {
|
||||
@dataclass
|
||||
class ChunkerConfig:
|
||||
max_chars: int = 1200
|
||||
min_chars: int = 80
|
||||
target_chars: int = 800
|
||||
context_depth: int = 3
|
||||
skip_headings: set[str] = field(default_factory=lambda: {
|
||||
"indice",
|
||||
"sommario",
|
||||
"bibliografia",
|
||||
"ringraziamenti",
|
||||
"abbreviazioni",
|
||||
}
|
||||
|
||||
# Se True, salta il contenuto che precede il primo heading (frontespizio, copertina).
|
||||
SKIP_PRE_HEADING: bool = True
|
||||
|
||||
# ─── Merge paragrafi corti ────────────────────────────────────────────────────
|
||||
|
||||
# Paragrafi consecutivi più corti di MIN_CHARS vengono fusi fino a raggiungerlo,
|
||||
# purché appartengano allo stesso contesto heading.
|
||||
MERGE_SHORT_PARAGRAPHS: bool = True
|
||||
|
||||
# ─── verify_chunks ────────────────────────────────────────────────────────────
|
||||
|
||||
# Numero minimo di simboli matematici perché un chunk incompleto sia classificato
|
||||
# come "matematico" (warning meno grave rispetto a frase spezzata normale).
|
||||
MATH_SYMS_MIN: int = 3
|
||||
|
||||
PROTECT_TABLES: bool = True
|
||||
PROTECT_MATH: bool = True
|
||||
})
|
||||
skip_pre_heading: bool = True
|
||||
merge_short: bool = True
|
||||
atomic_types: set[str] = field(default_factory=lambda: {
|
||||
"table", "code", "list", "math", "html",
|
||||
})
|
||||
fail_on_broken_fence: bool = True
|
||||
fail_on_content_loss: bool = False
|
||||
|
||||
@@ -0,0 +1,47 @@
|
||||
from __future__ import annotations
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
|
||||
@dataclass
|
||||
class Block:
|
||||
id: str
|
||||
kind: str # paragraph|heading|table|code|list|math|blockquote|html|thematic_break
|
||||
content: str
|
||||
plain_text: str
|
||||
atomic: bool
|
||||
start_line: int
|
||||
end_line: int
|
||||
header_path: list[dict]
|
||||
chars: int
|
||||
|
||||
|
||||
@dataclass
|
||||
class Chunk:
|
||||
chunk_id: str
|
||||
chunk_index: int
|
||||
content_original: str
|
||||
content_for_embedding: str
|
||||
content_type: str # section_fragment | atomic_block | overflow
|
||||
chars: int
|
||||
start_line: int
|
||||
end_line: int
|
||||
header_path: list[dict]
|
||||
block_ids: list[str]
|
||||
flags: dict
|
||||
neighbors: dict
|
||||
assets: list = field(default_factory=list)
|
||||
|
||||
|
||||
@dataclass
|
||||
class Diagnostics:
|
||||
errors: list[str]
|
||||
warnings: list[str]
|
||||
metrics: dict
|
||||
|
||||
|
||||
@dataclass
|
||||
class ChunkingResult:
|
||||
stem: str
|
||||
source_path: str
|
||||
chunks: list[Chunk]
|
||||
diagnostics: Diagnostics
|
||||
@@ -0,0 +1,181 @@
|
||||
from __future__ import annotations
|
||||
import re
|
||||
from chunks.models import Block, Chunk
|
||||
from chunks.config import ChunkerConfig
|
||||
|
||||
_SENTENCE_RE = re.compile(r"(?<=[.!?»])\s+(?=[A-ZÀÈÉÌÒÙ\"])")
|
||||
_CHUNK_COUNTER = 0
|
||||
|
||||
|
||||
def _reset_counter() -> None:
|
||||
global _CHUNK_COUNTER
|
||||
_CHUNK_COUNTER = 0
|
||||
|
||||
|
||||
def _make_chunk_id() -> str:
|
||||
global _CHUNK_COUNTER
|
||||
_CHUNK_COUNTER += 1
|
||||
return f"chk_{_CHUNK_COUNTER:06d}"
|
||||
|
||||
|
||||
def _header_prefix(header_path: list[dict], depth: int) -> str:
|
||||
return " > ".join(h["text"] for h in header_path[:depth])
|
||||
|
||||
|
||||
def _build_chunk(blocks: list[Block], index: int, config: ChunkerConfig,
|
||||
content_type: str = "section_fragment") -> Chunk:
|
||||
content = "\n\n".join(b.content for b in blocks)
|
||||
header_path = blocks[0].header_path
|
||||
prefix = _header_prefix(header_path, config.context_depth)
|
||||
embedding = f"{prefix}\n\n{content}" if prefix else content
|
||||
total_chars = len(content)
|
||||
return Chunk(
|
||||
chunk_id=_make_chunk_id(),
|
||||
chunk_index=index,
|
||||
content_original=content,
|
||||
content_for_embedding=embedding,
|
||||
content_type=content_type,
|
||||
chars=total_chars,
|
||||
start_line=blocks[0].start_line,
|
||||
end_line=blocks[-1].end_line,
|
||||
header_path=header_path,
|
||||
block_ids=[b.id for b in blocks],
|
||||
flags={
|
||||
"has_code": any(b.kind == "code" for b in blocks),
|
||||
"has_table": any(
|
||||
b.kind == "table"
|
||||
or (b.kind == "html" and "<table" in b.content.lower())
|
||||
for b in blocks
|
||||
),
|
||||
"has_math": any(
|
||||
b.kind == "math"
|
||||
or "$$" in b.content
|
||||
or r"\begin{" in b.content
|
||||
for b in blocks
|
||||
),
|
||||
"is_overflow": total_chars > config.max_chars,
|
||||
"is_sparse": total_chars < config.min_chars,
|
||||
},
|
||||
neighbors={"previous_chunk_id": None, "next_chunk_id": None},
|
||||
assets=[],
|
||||
)
|
||||
|
||||
|
||||
def _split_paragraph(block: Block, config: ChunkerConfig) -> list[Block]:
|
||||
sentences = _SENTENCE_RE.split(block.content)
|
||||
sub_blocks: list[Block] = []
|
||||
accumulated = ""
|
||||
|
||||
for sent in sentences:
|
||||
candidate = (accumulated + " " + sent).strip() if accumulated else sent
|
||||
if len(candidate) > config.max_chars and accumulated:
|
||||
text = accumulated.strip()
|
||||
sub_blocks.append(Block(
|
||||
id=f"{block.id}_s{len(sub_blocks) + 1}", kind="paragraph",
|
||||
content=text, plain_text=text, atomic=False,
|
||||
start_line=block.start_line, end_line=block.end_line,
|
||||
header_path=block.header_path, chars=len(text),
|
||||
))
|
||||
accumulated = sent
|
||||
else:
|
||||
accumulated = candidate
|
||||
|
||||
if accumulated:
|
||||
text = accumulated.strip()
|
||||
sub_blocks.append(Block(
|
||||
id=f"{block.id}_s{len(sub_blocks) + 1}", kind="paragraph",
|
||||
content=text, plain_text=text, atomic=False,
|
||||
start_line=block.start_line, end_line=block.end_line,
|
||||
header_path=block.header_path, chars=len(text),
|
||||
))
|
||||
return sub_blocks if sub_blocks else [block]
|
||||
|
||||
|
||||
def pack(blocks: list[Block], config: ChunkerConfig, stem: str) -> list[Chunk]:
|
||||
_reset_counter()
|
||||
chunks: list[Chunk] = []
|
||||
chunk_index = 0
|
||||
|
||||
# Espandi paragrafi sovradimensionati
|
||||
expanded: list[Block] = []
|
||||
for b in blocks:
|
||||
if not b.atomic and b.kind == "paragraph" and b.chars > config.max_chars:
|
||||
expanded.extend(_split_paragraph(b, config))
|
||||
else:
|
||||
expanded.append(b)
|
||||
|
||||
# Raggruppa per header_path
|
||||
groups: list[list[Block]] = []
|
||||
cur_group: list[Block] = []
|
||||
cur_key: str | None = None
|
||||
for b in expanded:
|
||||
key = str(b.header_path)
|
||||
if key != cur_key:
|
||||
if cur_group:
|
||||
groups.append(cur_group)
|
||||
cur_group = [b]
|
||||
cur_key = key
|
||||
else:
|
||||
cur_group.append(b)
|
||||
if cur_group:
|
||||
groups.append(cur_group)
|
||||
|
||||
for group in groups:
|
||||
accumulated: list[Block] = []
|
||||
accumulated_chars = 0
|
||||
|
||||
def flush() -> None:
|
||||
nonlocal accumulated, accumulated_chars, chunk_index
|
||||
if not accumulated:
|
||||
return
|
||||
chunks.append(_build_chunk(accumulated, chunk_index, config))
|
||||
chunk_index += 1
|
||||
accumulated = []
|
||||
accumulated_chars = 0
|
||||
|
||||
for b in group:
|
||||
if b.kind == "thematic_break":
|
||||
flush()
|
||||
continue
|
||||
|
||||
if b.atomic:
|
||||
if accumulated_chars + b.chars <= config.max_chars:
|
||||
accumulated.append(b)
|
||||
accumulated_chars += b.chars
|
||||
else:
|
||||
flush()
|
||||
ctype = "overflow" if b.chars > config.max_chars else "atomic_block"
|
||||
chunks.append(_build_chunk([b], chunk_index, config, content_type=ctype))
|
||||
chunk_index += 1
|
||||
else:
|
||||
# Flush preventivo se aggiungere questo blocco supererebbe max_chars
|
||||
# oppure supererebbe il target (con abbastanza contenuto già accumulato)
|
||||
if accumulated and accumulated_chars >= config.min_chars:
|
||||
if (accumulated_chars + b.chars > config.max_chars
|
||||
or accumulated_chars + b.chars > config.target_chars):
|
||||
flush()
|
||||
accumulated.append(b)
|
||||
accumulated_chars += b.chars
|
||||
|
||||
# Flush residuo — merge con precedente se troppo piccolo
|
||||
if accumulated:
|
||||
if (accumulated_chars < config.min_chars and chunks
|
||||
and chunks[-1].header_path == accumulated[0].header_path):
|
||||
prev = chunks[-1]
|
||||
merged = prev.content_original + "\n\n" + "\n\n".join(b.content for b in accumulated)
|
||||
prefix = _header_prefix(prev.header_path, config.context_depth)
|
||||
prev.content_original = merged
|
||||
prev.content_for_embedding = f"{prefix}\n\n{merged}" if prefix else merged
|
||||
prev.chars = len(merged)
|
||||
prev.end_line = accumulated[-1].end_line
|
||||
prev.block_ids.extend(b.id for b in accumulated)
|
||||
prev.flags["is_sparse"] = prev.chars < config.min_chars
|
||||
else:
|
||||
flush()
|
||||
|
||||
# Popola neighbors
|
||||
for idx, chunk in enumerate(chunks):
|
||||
chunk.neighbors["previous_chunk_id"] = chunks[idx - 1].chunk_id if idx > 0 else None
|
||||
chunk.neighbors["next_chunk_id"] = chunks[idx + 1].chunk_id if idx < len(chunks) - 1 else None
|
||||
|
||||
return chunks
|
||||
@@ -0,0 +1,24 @@
|
||||
from __future__ import annotations
|
||||
from markdown_it import MarkdownIt
|
||||
from markdown_it.token import Token
|
||||
|
||||
try:
|
||||
from mdit_py_plugins.dollarmath import dollarmath_plugin as _dollarmath
|
||||
_HAS_DOLLARMATH = True
|
||||
except ImportError:
|
||||
_HAS_DOLLARMATH = False
|
||||
|
||||
|
||||
def parse(source: str) -> tuple[list[Token], list[str]]:
|
||||
"""Parsa Markdown in token stream con source map.
|
||||
|
||||
Returns:
|
||||
tokens: lista Token con .map = [start_line, end_line] (0-indexed, end escluso)
|
||||
lines: righe sorgente (0-indexed) per ricostruzione testo esatto
|
||||
"""
|
||||
md = MarkdownIt().enable("table")
|
||||
if _HAS_DOLLARMATH:
|
||||
md = md.use(_dollarmath, allow_labels=False, allow_space=False)
|
||||
tokens = md.parse(source)
|
||||
lines = source.splitlines()
|
||||
return tokens, lines
|
||||
@@ -0,0 +1,234 @@
|
||||
from __future__ import annotations
|
||||
from markdown_it.token import Token
|
||||
from chunks.models import Block
|
||||
from chunks.config import ChunkerConfig
|
||||
|
||||
_COUNTER = 0
|
||||
|
||||
|
||||
def _reset_counter() -> None:
|
||||
global _COUNTER
|
||||
_COUNTER = 0
|
||||
|
||||
|
||||
def _make_id() -> str:
|
||||
global _COUNTER
|
||||
_COUNTER += 1
|
||||
return f"blk_{_COUNTER:04d}"
|
||||
|
||||
|
||||
def _inline_to_plain(token: Token) -> str:
|
||||
if token.children:
|
||||
parts = []
|
||||
for c in token.children:
|
||||
if c.type in ("text", "code_inline"):
|
||||
parts.append(c.content)
|
||||
elif c.type == "softbreak":
|
||||
parts.append(" ")
|
||||
return "".join(parts).strip()
|
||||
return (token.content or "").strip()
|
||||
|
||||
|
||||
def _lines_content(lines: list[str], start: int, end: int) -> str:
|
||||
return "\n".join(lines[start:end]).strip()
|
||||
|
||||
|
||||
def _current_path(stack: dict[int, str], depth: int) -> list[dict]:
|
||||
path = []
|
||||
for level in sorted(stack.keys()):
|
||||
path.append({"level": level, "text": stack[level]})
|
||||
return path[:depth]
|
||||
|
||||
|
||||
def _is_skip_heading(text: str, skip_set: set[str]) -> bool:
|
||||
t = text.lower().strip()
|
||||
return any(t == s or t.startswith(s) for s in {h.lower() for h in skip_set})
|
||||
|
||||
|
||||
def _find_close(tokens: list[Token], start: int, open_type: str, close_type: str) -> int:
|
||||
"""Ritorna indice del token close_type corrispondente a tokens[start]."""
|
||||
depth = 1
|
||||
i = start + 1
|
||||
while i < len(tokens) and depth > 0:
|
||||
if tokens[i].type == open_type:
|
||||
depth += 1
|
||||
elif tokens[i].type == close_type:
|
||||
depth -= 1
|
||||
i += 1
|
||||
return i - 1
|
||||
|
||||
|
||||
def segment(tokens: list[Token], lines: list[str], config: ChunkerConfig) -> list[Block]:
|
||||
_reset_counter()
|
||||
blocks: list[Block] = []
|
||||
heading_stack: dict[int, str] = {}
|
||||
pre_heading_done = False
|
||||
skip_mode = False
|
||||
skip_level: int | None = None
|
||||
|
||||
i = 0
|
||||
while i < len(tokens):
|
||||
tok = tokens[i]
|
||||
|
||||
# ── Heading ──────────────────────────────────────────────────────────
|
||||
if tok.type == "heading_open":
|
||||
level = int(tok.tag[1])
|
||||
inline = tokens[i + 1] if i + 1 < len(tokens) else None
|
||||
text = (inline.content if inline and inline.type == "inline" else "").strip()
|
||||
|
||||
if _is_skip_heading(text, config.skip_headings):
|
||||
skip_mode = True
|
||||
skip_level = level
|
||||
elif skip_mode and skip_level is not None and level <= skip_level:
|
||||
skip_mode = False
|
||||
skip_level = None
|
||||
|
||||
for lvl in [l for l in list(heading_stack.keys()) if l >= level]:
|
||||
del heading_stack[lvl]
|
||||
heading_stack[level] = text
|
||||
pre_heading_done = True
|
||||
i += 3 # heading_open, inline, heading_close
|
||||
continue
|
||||
|
||||
# ── Skip pre-heading ─────────────────────────────────────────────────
|
||||
if config.skip_pre_heading and not pre_heading_done:
|
||||
i += 1
|
||||
continue
|
||||
|
||||
# ── Skip section ─────────────────────────────────────────────────────
|
||||
if skip_mode:
|
||||
i += 1
|
||||
continue
|
||||
|
||||
header_path = _current_path(heading_stack, config.context_depth)
|
||||
|
||||
# ── Paragraph ────────────────────────────────────────────────────────
|
||||
if tok.type == "paragraph_open":
|
||||
inline = tokens[i + 1] if i + 1 < len(tokens) else None
|
||||
if tok.map:
|
||||
start, end = tok.map
|
||||
content = _lines_content(lines, start, end)
|
||||
plain = _inline_to_plain(inline) if inline and inline.type == "inline" else content
|
||||
blocks.append(Block(
|
||||
id=_make_id(),
|
||||
kind="paragraph",
|
||||
content=content,
|
||||
plain_text=plain,
|
||||
atomic=False,
|
||||
start_line=start,
|
||||
end_line=end,
|
||||
header_path=header_path,
|
||||
chars=len(content),
|
||||
))
|
||||
i += 3 # paragraph_open, inline, paragraph_close
|
||||
continue
|
||||
|
||||
# ── Code fence ───────────────────────────────────────────────────────
|
||||
if tok.type == "fence":
|
||||
if tok.map:
|
||||
start, end = tok.map
|
||||
content = _lines_content(lines, start, end)
|
||||
plain = f"[codice {tok.info.strip()}]" if tok.info.strip() else "[codice]"
|
||||
blocks.append(Block(
|
||||
id=_make_id(),
|
||||
kind="code",
|
||||
content=content,
|
||||
plain_text=plain,
|
||||
atomic=True,
|
||||
start_line=start,
|
||||
end_line=end,
|
||||
header_path=header_path,
|
||||
chars=len(content),
|
||||
))
|
||||
i += 1
|
||||
continue
|
||||
|
||||
# ── Container blocks ─────────────────────────────────────────────────
|
||||
_CONTAINERS = {
|
||||
"table_open": ("table_close", "table", True),
|
||||
"bullet_list_open": ("bullet_list_close", "list", True),
|
||||
"ordered_list_open": ("ordered_list_close", "list", True),
|
||||
"blockquote_open": ("blockquote_close", "blockquote", False),
|
||||
}
|
||||
if tok.type in _CONTAINERS:
|
||||
close_type, kind, atomic = _CONTAINERS[tok.type]
|
||||
atomic = atomic or kind in config.atomic_types
|
||||
close_idx = _find_close(tokens, i, tok.type, close_type)
|
||||
close_tok = tokens[close_idx]
|
||||
|
||||
start = tok.map[0] if tok.map else 0
|
||||
end = (close_tok.map[1] if close_tok.map else None) or (tok.map[1] if tok.map else start + 1)
|
||||
content = _lines_content(lines, start, end)
|
||||
blocks.append(Block(
|
||||
id=_make_id(),
|
||||
kind=kind,
|
||||
content=content,
|
||||
plain_text=content,
|
||||
atomic=atomic,
|
||||
start_line=start,
|
||||
end_line=end,
|
||||
header_path=header_path,
|
||||
chars=len(content),
|
||||
))
|
||||
i = close_idx + 1
|
||||
continue
|
||||
|
||||
# ── Thematic break ────────────────────────────────────────────────────
|
||||
if tok.type == "hr":
|
||||
start = tok.map[0] if tok.map else 0
|
||||
end = tok.map[1] if tok.map else start + 1
|
||||
blocks.append(Block(
|
||||
id=_make_id(),
|
||||
kind="thematic_break",
|
||||
content="---",
|
||||
plain_text="",
|
||||
atomic=False,
|
||||
start_line=start,
|
||||
end_line=end,
|
||||
header_path=header_path,
|
||||
chars=3,
|
||||
))
|
||||
i += 1
|
||||
continue
|
||||
|
||||
# ── Math block (dollarmath plugin) ───────────────────────────────────
|
||||
if tok.type == "math_block":
|
||||
start = tok.map[0] if tok.map else 0
|
||||
end = tok.map[1] if tok.map else start + 1
|
||||
content = _lines_content(lines, start, end)
|
||||
blocks.append(Block(
|
||||
id=_make_id(),
|
||||
kind="math",
|
||||
content=content,
|
||||
plain_text="[formula]",
|
||||
atomic=True,
|
||||
start_line=start,
|
||||
end_line=end,
|
||||
header_path=header_path,
|
||||
chars=len(content),
|
||||
))
|
||||
i += 1
|
||||
continue
|
||||
|
||||
# ── HTML block ────────────────────────────────────────────────────────
|
||||
if tok.type == "html_block":
|
||||
start = tok.map[0] if tok.map else 0
|
||||
end = tok.map[1] if tok.map else start + 1
|
||||
content = tok.content.strip()
|
||||
blocks.append(Block(
|
||||
id=_make_id(),
|
||||
kind="html",
|
||||
content=content,
|
||||
plain_text="",
|
||||
atomic="html" in config.atomic_types,
|
||||
start_line=start,
|
||||
end_line=end,
|
||||
header_path=header_path,
|
||||
chars=len(content),
|
||||
))
|
||||
i += 1
|
||||
continue
|
||||
|
||||
i += 1
|
||||
|
||||
return blocks
|
||||
@@ -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)
|
||||
@@ -1,493 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Verifica chunk
|
||||
|
||||
Analizza chunks/<stem>/chunks.json e segnala ogni anomalia che potrebbe
|
||||
degradare la qualità del retrieval. Non modifica nulla.
|
||||
|
||||
Input: chunks/<stem>/chunks.json
|
||||
Output: report a schermo + chunks/<stem>/report.json + exit code (0 = OK, 1 = problemi)
|
||||
|
||||
Uso:
|
||||
python chunks/verify_chunks.py --stem documento
|
||||
python chunks/verify_chunks.py # tutti i documenti in chunks/
|
||||
python chunks/verify_chunks.py --min 200 --max 800
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import re
|
||||
import sys
|
||||
from collections import Counter
|
||||
from pathlib import Path
|
||||
|
||||
_HERE = Path(__file__).resolve().parent
|
||||
if str(_HERE) not in sys.path:
|
||||
sys.path.insert(0, str(_HERE))
|
||||
import config as cfg
|
||||
|
||||
|
||||
# ─── Soglie ───────────────────────────────────────────────────────────────────
|
||||
|
||||
MIN_CHARS = cfg.MIN_CHARS
|
||||
MAX_CHARS = cfg.MAX_CHARS
|
||||
|
||||
_PUNCT_CLS = (
|
||||
"[.!?"
|
||||
+ chr(0xBB) + ")\\]"
|
||||
+ chr(0x2018) + chr(0x2019)
|
||||
+ chr(0x201C) + chr(0x201D)
|
||||
+ "\"'"
|
||||
+ chr(0x2014) + chr(0x2013) + chr(0x2026)
|
||||
+ chr(0xB7) # punto centrato LaTeX
|
||||
+ "]$"
|
||||
)
|
||||
PUNCT_END = re.compile(
|
||||
_PUNCT_CLS
|
||||
+ r"|/$"
|
||||
+ r"|\|$"
|
||||
+ r"|;$"
|
||||
+ r"|:$"
|
||||
+ r"|\d[\d.,/]*$"
|
||||
+ r"|\$$"
|
||||
+ r"|\}$"
|
||||
+ r"|>$"
|
||||
+ r"|\\\\$"
|
||||
)
|
||||
_HEX_END = re.compile(r"[0-9a-fA-F]{8,}$")
|
||||
_URL_TAIL = re.compile(r"(https?://|www\.)\S+(\s+\S+){0,3}$")
|
||||
_MATH_SYMS = re.compile(r"[∈∑≤≥≠∀∃∫√∞∂±×÷→←↔⊂⊃⊆⊇∩∪·°]")
|
||||
_ROMAN_END = re.compile(r"\b(I{1,3}|IV|VI{0,3}|IX|XI{0,2}|XIV|XV|XVI{0,2}|XIX|XX{0,2})$")
|
||||
_TABLE_SEP = re.compile(r"^\s*\|[\s\-|:]+\|\s*$")
|
||||
|
||||
|
||||
def _load_thresholds(stem_dir: Path) -> tuple[int, int]:
|
||||
meta = stem_dir / "meta.json"
|
||||
if meta.exists():
|
||||
m = json.loads(meta.read_text(encoding="utf-8"))
|
||||
return m["min_chars"], m["max_chars"]
|
||||
return MIN_CHARS, MAX_CHARS
|
||||
|
||||
|
||||
def _strip_prefix(text: str) -> str:
|
||||
text = text.lstrip()
|
||||
if text.startswith("["):
|
||||
end = text.find("]")
|
||||
if end != -1:
|
||||
return text[end + 1:].lstrip("\n")
|
||||
return text
|
||||
|
||||
|
||||
# ─── Checks ───────────────────────────────────────────────────────────────────
|
||||
|
||||
def is_empty(chunk: dict) -> bool:
|
||||
return not chunk.get("text", "").strip()
|
||||
|
||||
|
||||
def has_prefix(chunk: dict) -> bool:
|
||||
return chunk.get("text", "").lstrip().startswith("[")
|
||||
|
||||
|
||||
def is_prefix_malformed(chunk: dict) -> bool:
|
||||
"""Inizia con [ ma il prefisso non chiude con ] o ha contenuto vuoto."""
|
||||
text = chunk.get("text", "").lstrip()
|
||||
if not text.startswith("["):
|
||||
return False
|
||||
first_line = text.split("\n")[0]
|
||||
end = first_line.find("]")
|
||||
if end == -1:
|
||||
return True
|
||||
return len(first_line[1:end].strip()) == 0
|
||||
|
||||
|
||||
def is_body_empty(chunk: dict) -> bool:
|
||||
"""Prefisso valido ma nessun testo nel corpo."""
|
||||
text = chunk.get("text", "").lstrip()
|
||||
if not text.startswith("["):
|
||||
return False
|
||||
end = text.find("]")
|
||||
if end == -1:
|
||||
return False
|
||||
return len(text[end + 1:].strip()) == 0
|
||||
|
||||
|
||||
def is_too_short(chunk: dict, min_chars: int) -> bool:
|
||||
return chunk.get("n_chars", 0) < min_chars
|
||||
|
||||
|
||||
def is_too_long(chunk: dict, max_chars: int) -> bool:
|
||||
return chunk.get("n_chars", 0) > max_chars
|
||||
|
||||
|
||||
def ends_incomplete(chunk: dict) -> bool:
|
||||
text = chunk.get("text", "").rstrip()
|
||||
if not text:
|
||||
return False
|
||||
text_check = re.sub(r"[_*]+$", "", text).rstrip()
|
||||
if not text_check:
|
||||
return False
|
||||
if PUNCT_END.search(text_check):
|
||||
return False
|
||||
if _HEX_END.search(text_check):
|
||||
return False
|
||||
if _ROMAN_END.search(text_check):
|
||||
return False
|
||||
if _URL_TAIL.search(text_check[-200:]):
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def is_math_incomplete(chunk: dict) -> bool:
|
||||
return ends_incomplete(chunk) and len(_MATH_SYMS.findall(chunk.get("text", ""))) >= cfg.MATH_SYMS_MIN
|
||||
|
||||
|
||||
def is_table_broken(chunk: dict) -> bool:
|
||||
"""Tabella Markdown (≥2 righe con |) senza riga separatore |---|."""
|
||||
text = chunk.get("text", "")
|
||||
pipe_lines = [l for l in text.splitlines() if "|" in l and l.strip().startswith("|")]
|
||||
if len(pipe_lines) < 2:
|
||||
return False
|
||||
return not any(_TABLE_SEP.match(l) for l in pipe_lines)
|
||||
|
||||
|
||||
def find_duplicate_bodies(chunks: list[dict]) -> list[dict]:
|
||||
"""Chunk con testo body identico (prefisso escluso). Ignora corpi < 30 char."""
|
||||
seen: dict[str, str] = {}
|
||||
dupes = []
|
||||
for c in chunks:
|
||||
body = _strip_prefix(c.get("text", "")).strip()
|
||||
if len(body) < 30:
|
||||
continue
|
||||
cid = c["chunk_id"]
|
||||
if body in seen:
|
||||
dupes.append({
|
||||
"chunk_id": cid,
|
||||
"duplicate_of": seen[body],
|
||||
"sezione": c.get("sezione", ""),
|
||||
"titolo": c.get("titolo", ""),
|
||||
"n_chars": c.get("n_chars", 0),
|
||||
"last_text": body[:120],
|
||||
})
|
||||
else:
|
||||
seen[body] = cid
|
||||
return dupes
|
||||
|
||||
|
||||
# ─── Istogramma ───────────────────────────────────────────────────────────────
|
||||
|
||||
def _ascii_histogram(lengths: list[int], min_t: int, max_t: int,
|
||||
n_bins: int = 10, bar_width: int = 28) -> list[str]:
|
||||
if not lengths:
|
||||
return []
|
||||
lo, hi = min(lengths), max(lengths)
|
||||
if lo == hi:
|
||||
return [f" {lo:>5}–{hi:<5} │{'█' * bar_width}│ {len(lengths)}"]
|
||||
step = (hi - lo) / n_bins
|
||||
bins = [0] * n_bins
|
||||
for l in lengths:
|
||||
idx = min(int((l - lo) / step), n_bins - 1)
|
||||
bins[idx] += 1
|
||||
max_count = max(bins) or 1
|
||||
lines = []
|
||||
for i, count in enumerate(bins):
|
||||
lo_b = int(lo + i * step)
|
||||
hi_b = int(lo + (i + 1) * step)
|
||||
bar = "█" * round(count / max_count * bar_width)
|
||||
note = ""
|
||||
if lo_b <= min_t < hi_b:
|
||||
note = " ← MIN"
|
||||
elif lo_b <= max_t < hi_b:
|
||||
note = " ← MAX"
|
||||
lines.append(f" {lo_b:>5}–{hi_b:<5} │{bar:<{bar_width}}│ {count}{note}")
|
||||
return lines
|
||||
|
||||
|
||||
# ─── Helpers output ───────────────────────────────────────────────────────────
|
||||
|
||||
def _fmt_chunk(c: dict) -> str:
|
||||
cid = c.get("chunk_id", "?")
|
||||
n = c.get("n_chars", 0)
|
||||
preview = c.get("text", "")[:60].replace("\n", " ")
|
||||
return f" [{cid}] ({n} char) «{preview}»"
|
||||
|
||||
|
||||
def _chunk_entry(c: dict) -> dict:
|
||||
return {
|
||||
"chunk_id": c.get("chunk_id", ""),
|
||||
"sezione": c.get("sezione", ""),
|
||||
"titolo": c.get("titolo", ""),
|
||||
"n_chars": c.get("n_chars", 0),
|
||||
"last_text": c.get("text", "").rstrip().split("\n")[-1][-120:],
|
||||
}
|
||||
|
||||
|
||||
def _print_list(items: list[dict], limit: int = 5) -> None:
|
||||
for c in items[:limit]:
|
||||
print(_fmt_chunk(c))
|
||||
if len(items) > limit:
|
||||
print(f" ... e altri {len(items) - limit}")
|
||||
|
||||
|
||||
# ─── Core ─────────────────────────────────────────────────────────────────────
|
||||
|
||||
def verify_stem(stem: str, project_root: Path, min_chars: int, max_chars: int) -> bool:
|
||||
stem_dir = project_root / "chunks" / stem
|
||||
chunks_path = stem_dir / "chunks.json"
|
||||
min_chars, max_chars = _load_thresholds(stem_dir)
|
||||
|
||||
print(f"\nDocumento: {stem}")
|
||||
|
||||
if not chunks_path.exists():
|
||||
print(f" ✗ chunks/{stem}/chunks.json non trovato")
|
||||
print(f" Esegui prima: python chunks/chunker.py --stem {stem}")
|
||||
return False
|
||||
|
||||
chunks: list[dict] = json.loads(chunks_path.read_text(encoding="utf-8"))
|
||||
|
||||
if not chunks:
|
||||
print(f" ✗ chunks.json è vuoto")
|
||||
return False
|
||||
|
||||
# ── Raccogli problemi ──────────────────────────────────────────────────────
|
||||
|
||||
empty_chunks = [c for c in chunks if is_empty(c)]
|
||||
no_prefix = [c for c in chunks if not is_empty(c) and not has_prefix(c)]
|
||||
malformed_prefix = [c for c in chunks
|
||||
if not is_empty(c) and has_prefix(c) and is_prefix_malformed(c)]
|
||||
body_empty = [c for c in chunks
|
||||
if not is_empty(c) and has_prefix(c)
|
||||
and not is_prefix_malformed(c) and is_body_empty(c)]
|
||||
too_short = [c for c in chunks if is_too_short(c, min_chars)]
|
||||
too_long = [c for c in chunks if is_too_long(c, max_chars)]
|
||||
_incomplete_all = [c for c in chunks if not is_empty(c) and ends_incomplete(c)]
|
||||
incomplete_math = [c for c in _incomplete_all if is_math_incomplete(c)]
|
||||
incomplete = [c for c in _incomplete_all if not is_math_incomplete(c)]
|
||||
broken_tables = [c for c in chunks if is_table_broken(c)]
|
||||
duplicates = find_duplicate_bodies(chunks)
|
||||
|
||||
# ── Statistiche ───────────────────────────────────────────────────────────
|
||||
|
||||
lengths = [c.get("n_chars", 0) for c in chunks]
|
||||
n_total = len(chunks)
|
||||
blocker_ids = set(
|
||||
c["chunk_id"]
|
||||
for lst in [empty_chunks, no_prefix, malformed_prefix, body_empty, incomplete]
|
||||
for c in lst
|
||||
)
|
||||
n_ok = n_total - len(blocker_ids)
|
||||
min_l = min(lengths)
|
||||
max_l = max(lengths)
|
||||
avg_l = int(sum(lengths) / n_total)
|
||||
p50 = sorted(lengths)[n_total // 2]
|
||||
n_under = sum(1 for l in lengths if l < min_chars)
|
||||
n_norm = sum(1 for l in lengths if min_chars <= l <= max_chars)
|
||||
n_over = sum(1 for l in lengths if l > max_chars)
|
||||
|
||||
section_counts = Counter(c.get("sezione", "—") or "—" for c in chunks)
|
||||
|
||||
# ── Output statistiche ────────────────────────────────────────────────────
|
||||
|
||||
print(f" Totale: {n_total} | ✅ OK: {n_ok}")
|
||||
print()
|
||||
print(f" Lunghezze — min {min_l} p50 {p50} media {avg_l} max {max_l}")
|
||||
print(f" Fasce — <{min_chars}: {n_under} | {min_chars}–{max_chars}: {n_norm} | >{max_chars}: {n_over}")
|
||||
print()
|
||||
print(" Istogramma:")
|
||||
for line in _ascii_histogram(lengths, min_chars, max_chars):
|
||||
print(line)
|
||||
print()
|
||||
print(" Top sezioni:")
|
||||
for sezione, count in section_counts.most_common(5):
|
||||
bar = "▪" * min(count, 35)
|
||||
print(f" {bar} {count:>4} {sezione[:65]}")
|
||||
|
||||
# ── Blockers ──────────────────────────────────────────────────────────────
|
||||
|
||||
if empty_chunks:
|
||||
print(f"\n 🔴 {len(empty_chunks)} chunk VUOTI:")
|
||||
for c in empty_chunks[:5]:
|
||||
print(f" [{c.get('chunk_id', '?')}]")
|
||||
if len(empty_chunks) > 5:
|
||||
print(f" ... e altri {len(empty_chunks) - 5}")
|
||||
|
||||
if no_prefix:
|
||||
print(f"\n 🔴 {len(no_prefix)} chunk SENZA PREFISSO DI CONTESTO:")
|
||||
_print_list(no_prefix)
|
||||
print(f" → Causa probabile: heading mancanti nel clean.md")
|
||||
|
||||
if malformed_prefix:
|
||||
print(f"\n 🔴 {len(malformed_prefix)} chunk con PREFISSO MALFORMATO ([ senza ] o vuoto):")
|
||||
_print_list(malformed_prefix)
|
||||
print(f" → Causa probabile: heading con caratteri speciali nel clean.md")
|
||||
|
||||
if body_empty:
|
||||
print(f"\n 🔴 {len(body_empty)} chunk con CORPO VUOTO (solo prefisso):")
|
||||
_print_list(body_empty)
|
||||
print(f" → Causa probabile: sezioni senza testo nel clean.md")
|
||||
|
||||
if incomplete:
|
||||
print(f"\n 🟡 {len(incomplete)} chunk con FRASE SPEZZATA (warning):")
|
||||
for c in incomplete[:5]:
|
||||
last_line = c.get("text", "").rstrip().split("\n")[-1][-80:]
|
||||
print(f" [{c.get('chunk_id', '?')}] ...{last_line!r}")
|
||||
if len(incomplete) > 5:
|
||||
print(f" ... e altri {len(incomplete) - 5}")
|
||||
print(f" → Soluzione: correggi il sorgente .md e rigenera con chunker.py --force")
|
||||
|
||||
# ── Warnings ──────────────────────────────────────────────────────────────
|
||||
|
||||
if too_short:
|
||||
print(f"\n 🟡 {len(too_short)} chunk SOTTO MIN_CHARS ({min_chars}):")
|
||||
_print_list(too_short)
|
||||
|
||||
if too_long:
|
||||
print(f"\n 🟡 {len(too_long)} chunk SOPRA MAX ({max_chars}):")
|
||||
_print_list(too_long)
|
||||
print(f" → Causa: frasi non suddivisibili o blocchi atomici (tabelle/liste)")
|
||||
|
||||
if incomplete_math:
|
||||
print(f"\n 🟡 {len(incomplete_math)} chunk MATEMATICI senza punteggiatura finale:")
|
||||
for c in incomplete_math[:3]:
|
||||
last_line = c.get("text", "").rstrip().split("\n")[-1][-80:]
|
||||
print(f" [{c.get('chunk_id', '?')}] ...{last_line!r}")
|
||||
if len(incomplete_math) > 3:
|
||||
print(f" ... e altri {len(incomplete_math) - 3}")
|
||||
|
||||
if broken_tables:
|
||||
print(f"\n 🟡 {len(broken_tables)} TABELLE senza riga separatore |---|:")
|
||||
_print_list(broken_tables, limit=3)
|
||||
print(f" → Le tabelle potrebbero non renderizzarsi nel retrieval")
|
||||
|
||||
if duplicates:
|
||||
print(f"\n 🟡 {len(duplicates)} DUPLICATI (corpo identico):")
|
||||
for e in duplicates[:5]:
|
||||
print(f" [{e['chunk_id']}] ≡ [{e['duplicate_of']}] «{e['last_text'][:60]}»")
|
||||
if len(duplicates) > 5:
|
||||
print(f" ... e altri {len(duplicates) - 5}")
|
||||
print(f" → Causa probabile: sezioni ripetute nel sorgente .md")
|
||||
|
||||
# ── Report.json ───────────────────────────────────────────────────────────
|
||||
|
||||
blockers = empty_chunks + no_prefix + malformed_prefix + body_empty
|
||||
warnings = too_short + too_long + incomplete + incomplete_math + broken_tables
|
||||
|
||||
verdict = "blocked" if blockers else ("warnings_only" if (warnings or duplicates) else "ok")
|
||||
|
||||
report = {
|
||||
"stem": stem,
|
||||
"verdict": verdict,
|
||||
"stats": {
|
||||
"total": n_total,
|
||||
"ok": n_ok,
|
||||
"min_chars": min_l,
|
||||
"max_chars": max_l,
|
||||
"avg_chars": avg_l,
|
||||
"p50_chars": p50,
|
||||
"under_min": n_under,
|
||||
"in_range": n_norm,
|
||||
"over_max": n_over,
|
||||
"sections": [{"sezione": s, "n_chunks": n}
|
||||
for s, n in section_counts.most_common()],
|
||||
},
|
||||
"thresholds": {
|
||||
"min_chars": min_chars,
|
||||
"max_chars": max_chars,
|
||||
"target_chars": cfg.MAX_CHARS,
|
||||
},
|
||||
"blockers": {
|
||||
"empty": [_chunk_entry(c) for c in empty_chunks],
|
||||
"no_prefix": [_chunk_entry(c) for c in no_prefix],
|
||||
"malformed_prefix": [_chunk_entry(c) for c in malformed_prefix],
|
||||
"body_empty": [_chunk_entry(c) for c in body_empty],
|
||||
"incomplete": [_chunk_entry(c) for c in incomplete],
|
||||
},
|
||||
"warnings": {
|
||||
"too_short": [_chunk_entry(c) for c in too_short],
|
||||
"too_long": [_chunk_entry(c) for c in too_long],
|
||||
"incomplete_math": [_chunk_entry(c) for c in incomplete_math],
|
||||
"broken_tables": [_chunk_entry(c) for c in broken_tables],
|
||||
"duplicate_bodies": duplicates,
|
||||
},
|
||||
}
|
||||
|
||||
out_dir = project_root / "chunks" / stem
|
||||
out_dir.mkdir(parents=True, exist_ok=True)
|
||||
(out_dir / "report.json").write_text(
|
||||
json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8"
|
||||
)
|
||||
print(f"\n report.json → chunks/{stem}/")
|
||||
|
||||
# ── Prossimi passi ────────────────────────────────────────────────────────
|
||||
|
||||
print(f"\n {'─' * 50}")
|
||||
print(f" Verdict: {verdict.upper()}")
|
||||
print(f" {'─' * 50}")
|
||||
|
||||
if verdict == "ok":
|
||||
print(f" ✅ Tutto OK — procedi alla vettorizzazione:")
|
||||
print(f" python ingestion/ingest.py --stem {stem}")
|
||||
|
||||
elif verdict == "warnings_only":
|
||||
print(f" 🟡 Solo avvisi — puoi procedere alla vettorizzazione:")
|
||||
print(f" python ingestion/ingest.py --stem {stem}")
|
||||
if too_short or too_long:
|
||||
print()
|
||||
print(f" Per ottimizzare: correggi il sorgente .md e rigenera con --force")
|
||||
|
||||
else:
|
||||
print(f" 🔴 {len(blockers)} problemi bloccanti — correggi prima di procedere:")
|
||||
if empty_chunks or body_empty:
|
||||
print(f" • chunk vuoti/senza corpo → controlla sources/{stem}/auto/{stem}_clean.md")
|
||||
if no_prefix or malformed_prefix:
|
||||
print(f" • prefisso mancante/malformato → controlla gli heading in sources/{stem}.md")
|
||||
if incomplete:
|
||||
print(f" • frasi spezzate → correggi il sorgente e rigenera con --force")
|
||||
print()
|
||||
print(f" Dopo le correzioni:")
|
||||
print(f" python chunks/chunker.py --stem {stem} --force")
|
||||
print(f" python chunks/verify_chunks.py --stem {stem}")
|
||||
if warnings:
|
||||
print()
|
||||
print(f" 🟡 Hai anche {len(warnings)} avvisi — affrontali dopo aver risolto i 🔴.")
|
||||
|
||||
return not blockers
|
||||
|
||||
|
||||
# ─── Entry point ──────────────────────────────────────────────────────────────
|
||||
|
||||
if __name__ == "__main__":
|
||||
project_root = Path(__file__).parent.parent
|
||||
|
||||
parser = argparse.ArgumentParser(description="Verifica chunk")
|
||||
parser.add_argument("--stem", help="Nome del documento (sottocartella di chunks/)")
|
||||
parser.add_argument(
|
||||
"--min", type=int, default=cfg.MIN_CHARS,
|
||||
help=f"Soglia minima caratteri (default: {cfg.MIN_CHARS})"
|
||||
)
|
||||
parser.add_argument(
|
||||
"--max", type=int, default=cfg.MAX_CHARS,
|
||||
help=f"Soglia massima caratteri (default: {cfg.MAX_CHARS})"
|
||||
)
|
||||
args = parser.parse_args()
|
||||
|
||||
if args.stem:
|
||||
stems = [args.stem]
|
||||
else:
|
||||
chunks_dir = project_root / "chunks"
|
||||
if not chunks_dir.exists():
|
||||
print(f"Errore: cartella chunks/ non trovata in {project_root}")
|
||||
sys.exit(1)
|
||||
stems = sorted(
|
||||
p.name for p in chunks_dir.iterdir()
|
||||
if p.is_dir() and (p / "chunks.json").exists()
|
||||
)
|
||||
if not stems:
|
||||
print("Errore: nessun chunks.json trovato in chunks/")
|
||||
sys.exit(1)
|
||||
|
||||
results = [verify_stem(s, project_root, args.min, args.max) for s in stems]
|
||||
|
||||
ok = sum(results)
|
||||
total = len(results)
|
||||
print(f"\n{'✅' if all(results) else '⚠️ '} {ok}/{total} documenti senza problemi bloccanti")
|
||||
sys.exit(0 if all(results) else 1)
|
||||
@@ -2,3 +2,5 @@ pdfplumber==0.11.9
|
||||
PyMuPDF>=1.24.0
|
||||
chromadb
|
||||
pytest>=8.0
|
||||
markdown-it-py>=4.0
|
||||
mdit-py-plugins>=0.4
|
||||
|
||||
@@ -0,0 +1,15 @@
|
||||
import pytest
|
||||
from chunks.parser import parse
|
||||
from chunks.config import ChunkerConfig
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def cfg():
|
||||
return ChunkerConfig()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def parse_md():
|
||||
def _parse(md: str):
|
||||
return parse(md)
|
||||
return _parse
|
||||
@@ -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
|
||||
@@ -0,0 +1,29 @@
|
||||
from chunks.config import ChunkerConfig
|
||||
|
||||
|
||||
def test_default_values():
|
||||
c = ChunkerConfig()
|
||||
assert c.max_chars == 1200
|
||||
assert c.min_chars == 80
|
||||
assert c.target_chars == 800
|
||||
assert c.context_depth == 3
|
||||
assert "indice" in c.skip_headings
|
||||
assert c.skip_pre_heading is True
|
||||
assert c.merge_short is True
|
||||
assert "table" in c.atomic_types
|
||||
assert "code" in c.atomic_types
|
||||
|
||||
|
||||
def test_custom_values():
|
||||
c = ChunkerConfig(max_chars=2000, min_chars=100, context_depth=2)
|
||||
assert c.max_chars == 2000
|
||||
assert c.min_chars == 100
|
||||
assert c.context_depth == 2
|
||||
|
||||
|
||||
def test_skip_headings_case_insensitive_check():
|
||||
c = ChunkerConfig()
|
||||
skip_lower = {h.lower() for h in c.skip_headings}
|
||||
assert "indice" in skip_lower
|
||||
assert "sommario" in skip_lower
|
||||
assert "bibliografia" in skip_lower
|
||||
@@ -0,0 +1,51 @@
|
||||
from chunks.models import Block, Chunk, Diagnostics, ChunkingResult
|
||||
|
||||
|
||||
def test_block_creation():
|
||||
b = Block(
|
||||
id="blk_0001",
|
||||
kind="paragraph",
|
||||
content="Testo di esempio.",
|
||||
plain_text="Testo di esempio.",
|
||||
atomic=False,
|
||||
start_line=0,
|
||||
end_line=1,
|
||||
header_path=[{"level": 1, "text": "Titolo"}],
|
||||
chars=17,
|
||||
)
|
||||
assert b.id == "blk_0001"
|
||||
assert b.kind == "paragraph"
|
||||
assert not b.atomic
|
||||
assert b.chars == 17
|
||||
|
||||
|
||||
def test_chunk_creation():
|
||||
c = Chunk(
|
||||
chunk_id="chk_000001",
|
||||
chunk_index=1,
|
||||
content_original="Testo.",
|
||||
content_for_embedding="Titolo\n\nTesto.",
|
||||
content_type="section_fragment",
|
||||
chars=6,
|
||||
start_line=0,
|
||||
end_line=1,
|
||||
header_path=[{"level": 1, "text": "Titolo"}],
|
||||
block_ids=["blk_0001"],
|
||||
flags={"has_code": False, "has_table": False, "has_math": False,
|
||||
"is_overflow": False, "is_sparse": False},
|
||||
neighbors={"previous_chunk_id": None, "next_chunk_id": None},
|
||||
assets=[],
|
||||
)
|
||||
assert c.chunk_id == "chk_000001"
|
||||
assert c.assets == []
|
||||
|
||||
|
||||
def test_diagnostics_empty():
|
||||
d = Diagnostics(errors=[], warnings=[], metrics={})
|
||||
assert d.errors == []
|
||||
|
||||
|
||||
def test_chunking_result():
|
||||
r = ChunkingResult(stem="doc", source_path="sources/doc.md", chunks=[], diagnostics=Diagnostics([], [], {}))
|
||||
assert r.stem == "doc"
|
||||
assert r.chunks == []
|
||||
@@ -0,0 +1,150 @@
|
||||
import pytest
|
||||
from chunks.models import Block
|
||||
from chunks.config import ChunkerConfig
|
||||
from chunks.packer import pack
|
||||
|
||||
|
||||
def make_block(idx: int, content: str, kind: str = "paragraph", atomic: bool = False,
|
||||
header_path=None) -> Block:
|
||||
hp = header_path or [{"level": 1, "text": "Titolo"}]
|
||||
return Block(
|
||||
id=f"blk_{idx:04d}", kind=kind, content=content, plain_text=content,
|
||||
atomic=atomic, start_line=idx * 2, end_line=idx * 2 + 1,
|
||||
header_path=hp, chars=len(content),
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def cfg():
|
||||
return ChunkerConfig(min_chars=10, target_chars=50, max_chars=100)
|
||||
|
||||
|
||||
def test_single_block_becomes_single_chunk(cfg):
|
||||
blocks = [make_block(1, "Testo breve.")]
|
||||
chunks = pack(blocks, cfg, "test")
|
||||
assert len(chunks) == 1
|
||||
assert chunks[0].content_original == "Testo breve."
|
||||
|
||||
|
||||
def test_chunk_id_format(cfg):
|
||||
blocks = [make_block(1, "Testo.")]
|
||||
chunks = pack(blocks, cfg, "test")
|
||||
assert chunks[0].chunk_id.startswith("chk_")
|
||||
|
||||
|
||||
def test_neighbors_populated(cfg):
|
||||
blocks = [make_block(i, "x" * 60, header_path=[{"level": 1, "text": "T"}]) for i in range(1, 4)]
|
||||
chunks = pack(blocks, cfg, "test")
|
||||
assert len(chunks) >= 2
|
||||
assert chunks[0].neighbors["next_chunk_id"] == chunks[1].chunk_id
|
||||
assert chunks[1].neighbors["previous_chunk_id"] == chunks[0].chunk_id
|
||||
assert chunks[-1].neighbors["next_chunk_id"] is None
|
||||
|
||||
|
||||
def test_blocks_merged_until_target(cfg):
|
||||
blocks = [make_block(i, "x" * 20) for i in range(1, 4)]
|
||||
chunks = pack(blocks, cfg, "test")
|
||||
assert len(chunks) >= 2
|
||||
|
||||
|
||||
def test_oversized_paragraph_split(cfg):
|
||||
long_content = ("Questa è la prima frase completa. " * 5).strip()
|
||||
blocks = [make_block(1, long_content)]
|
||||
chunks = pack(blocks, cfg, "test")
|
||||
for c in chunks:
|
||||
assert c.chars <= cfg.max_chars or c.flags["is_overflow"]
|
||||
|
||||
|
||||
def test_atomic_block_not_split(cfg):
|
||||
atomic_content = "```python\n" + "codice\n" * 20 + "```"
|
||||
blocks = [make_block(1, atomic_content, kind="code", atomic=True)]
|
||||
chunks = pack(blocks, cfg, "test")
|
||||
assert len(chunks) == 1
|
||||
assert chunks[0].flags["is_overflow"] is True
|
||||
assert chunks[0].content_type == "overflow"
|
||||
|
||||
|
||||
def test_heading_path_break_creates_new_chunk(cfg):
|
||||
hp1 = [{"level": 1, "text": "Sezione 1"}]
|
||||
hp2 = [{"level": 1, "text": "Sezione 2"}]
|
||||
blocks = [make_block(1, "x" * 30, header_path=hp1), make_block(2, "x" * 30, header_path=hp2)]
|
||||
chunks = pack(blocks, cfg, "test")
|
||||
assert len(chunks) == 2
|
||||
assert chunks[0].header_path == hp1
|
||||
assert chunks[1].header_path == hp2
|
||||
|
||||
|
||||
def test_content_for_embedding_has_header_prefix(cfg):
|
||||
blocks = [make_block(1, "Contenuto.", header_path=[{"level": 1, "text": "Guida"},
|
||||
{"level": 2, "text": "Intro"}])]
|
||||
chunks = pack(blocks, cfg, "test")
|
||||
assert "Guida" in chunks[0].content_for_embedding
|
||||
assert "Intro" in chunks[0].content_for_embedding
|
||||
assert "Contenuto." in chunks[0].content_for_embedding
|
||||
|
||||
|
||||
def test_thematic_break_flushes_chunk(cfg):
|
||||
blocks = [
|
||||
make_block(1, "x" * 30),
|
||||
make_block(2, "---", kind="thematic_break"),
|
||||
make_block(3, "x" * 30),
|
||||
]
|
||||
chunks = pack(blocks, cfg, "test")
|
||||
assert len(chunks) == 2
|
||||
|
||||
|
||||
def test_flags_has_code(cfg):
|
||||
blocks = [make_block(1, "```\ncode\n```", kind="code", atomic=True)]
|
||||
chunks = pack(blocks, cfg, "test")
|
||||
assert chunks[0].flags["has_code"] is True
|
||||
|
||||
|
||||
def test_flags_has_table(cfg):
|
||||
blocks = [make_block(1, "| A | B |\n|---|---|\n| 1 | 2 |", kind="table", atomic=True)]
|
||||
chunks = pack(blocks, cfg, "test")
|
||||
assert chunks[0].flags["has_table"] is True
|
||||
|
||||
|
||||
def test_chunk_index_sequential(cfg):
|
||||
blocks = [make_block(i, "x" * 60) for i in range(1, 5)]
|
||||
chunks = pack(blocks, cfg, "test")
|
||||
for i, c in enumerate(chunks):
|
||||
assert c.chunk_index == i
|
||||
|
||||
|
||||
def test_assets_empty(cfg):
|
||||
blocks = [make_block(1, "Testo.")]
|
||||
chunks = pack(blocks, cfg, "test")
|
||||
assert chunks[0].assets == []
|
||||
|
||||
|
||||
def test_flags_has_math_via_kind(cfg):
|
||||
blocks = [make_block(1, "$$\nE=mc^2\n$$", kind="math", atomic=True)]
|
||||
chunks = pack(blocks, cfg, "test")
|
||||
assert chunks[0].flags["has_math"] is True
|
||||
|
||||
|
||||
def test_flags_has_math_via_content_dollars(cfg):
|
||||
# Paragrafo con $$ inline (senza dollarmath plugin attivo)
|
||||
blocks = [make_block(1, "Vedi la formula:\n$$\nx^2\n$$")]
|
||||
chunks = pack(blocks, cfg, "test")
|
||||
assert chunks[0].flags["has_math"] is True
|
||||
|
||||
|
||||
def test_flags_has_math_via_content_begin(cfg):
|
||||
blocks = [make_block(1, r"Equazione: \begin{align} x = 1 \end{align}")]
|
||||
chunks = pack(blocks, cfg, "test")
|
||||
assert chunks[0].flags["has_math"] is True
|
||||
|
||||
|
||||
def test_flags_has_table_via_html(cfg):
|
||||
html_content = "<table><tr><td>A</td></tr></table>"
|
||||
blocks = [make_block(1, html_content, kind="html", atomic=True)]
|
||||
chunks = pack(blocks, cfg, "test")
|
||||
assert chunks[0].flags["has_table"] is True
|
||||
|
||||
|
||||
def test_flags_no_math_plain_text(cfg):
|
||||
blocks = [make_block(1, "Testo normale senza formule.")]
|
||||
chunks = pack(blocks, cfg, "test")
|
||||
assert chunks[0].flags["has_math"] is False
|
||||
@@ -0,0 +1,69 @@
|
||||
from chunks.parser import parse
|
||||
|
||||
|
||||
def test_parse_returns_tokens_and_lines():
|
||||
md = "# Titolo\n\nParagrafo.\n"
|
||||
tokens, lines = parse(md)
|
||||
assert len(tokens) > 0
|
||||
assert lines[0] == "# Titolo"
|
||||
assert lines[2] == "Paragrafo."
|
||||
|
||||
|
||||
def test_tokens_have_source_map():
|
||||
md = "# Titolo\n\nParagrafo.\n"
|
||||
tokens, lines = parse(md)
|
||||
heading = next(t for t in tokens if t.type == "heading_open")
|
||||
assert heading.map is not None
|
||||
assert heading.map[0] == 0
|
||||
|
||||
|
||||
def test_parse_code_fence():
|
||||
md = "# T\n\n```python\ncodice\n```\n"
|
||||
tokens, lines = parse(md)
|
||||
fence = next((t for t in tokens if t.type == "fence"), None)
|
||||
assert fence is not None
|
||||
assert "codice" in fence.content
|
||||
|
||||
|
||||
def test_parse_table():
|
||||
md = "| A | B |\n|---|---|\n| 1 | 2 |\n"
|
||||
tokens, lines = parse(md)
|
||||
types = [t.type for t in tokens]
|
||||
assert "table_open" in types
|
||||
|
||||
|
||||
def test_parse_list():
|
||||
md = "- item 1\n- item 2\n"
|
||||
tokens, lines = parse(md)
|
||||
types = [t.type for t in tokens]
|
||||
assert "bullet_list_open" in types
|
||||
|
||||
|
||||
def test_lines_preserves_source():
|
||||
md = "Riga 1\nRiga 2\nRiga 3\n"
|
||||
_, lines = parse(md)
|
||||
assert lines[0] == "Riga 1"
|
||||
assert lines[1] == "Riga 2"
|
||||
|
||||
|
||||
def test_parse_math_block():
|
||||
md = "# T\n\n$$\nE=mc^2\n$$\n"
|
||||
tokens, lines = parse(md)
|
||||
types = [t.type for t in tokens]
|
||||
assert "math_block" in types
|
||||
|
||||
|
||||
def test_math_block_has_source_map():
|
||||
md = "# T\n\n$$\nE=mc^2\n$$\n"
|
||||
tokens, lines = parse(md)
|
||||
mb = next(t for t in tokens if t.type == "math_block")
|
||||
assert mb.map is not None
|
||||
assert "E=mc^2" in mb.content
|
||||
|
||||
|
||||
def test_parse_math_inline_stays_in_paragraph():
|
||||
md = "Testo con $x^2$ inline.\n"
|
||||
tokens, lines = parse(md)
|
||||
types = [t.type for t in tokens]
|
||||
assert "paragraph_open" in types
|
||||
assert "math_block" not in types
|
||||
@@ -0,0 +1,155 @@
|
||||
import pytest
|
||||
from chunks.parser import parse
|
||||
from chunks.config import ChunkerConfig
|
||||
from chunks.segmenter import segment
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def cfg():
|
||||
return ChunkerConfig()
|
||||
|
||||
|
||||
def test_paragraph_block(cfg):
|
||||
tokens, lines = parse("# Titolo\n\nParagrafo di testo.\n")
|
||||
blocks = segment(tokens, lines, cfg)
|
||||
para = next(b for b in blocks if b.kind == "paragraph")
|
||||
assert "Paragrafo di testo." in para.content
|
||||
assert para.header_path == [{"level": 1, "text": "Titolo"}]
|
||||
|
||||
|
||||
def test_code_block_is_atomic(cfg):
|
||||
tokens, lines = parse("# T\n\n```python\nprint('hello')\n```\n")
|
||||
blocks = segment(tokens, lines, cfg)
|
||||
code = next(b for b in blocks if b.kind == "code")
|
||||
assert code.atomic is True
|
||||
|
||||
|
||||
def test_table_block_is_atomic(cfg):
|
||||
tokens, lines = parse("# T\n\n| A | B |\n|---|---|\n| 1 | 2 |\n")
|
||||
blocks = segment(tokens, lines, cfg)
|
||||
table = next(b for b in blocks if b.kind == "table")
|
||||
assert table.atomic is True
|
||||
|
||||
|
||||
def test_list_block_is_atomic(cfg):
|
||||
tokens, lines = parse("# T\n\n- item 1\n- item 2\n")
|
||||
blocks = segment(tokens, lines, cfg)
|
||||
lst = next(b for b in blocks if b.kind == "list")
|
||||
assert lst.atomic is True
|
||||
|
||||
|
||||
def test_heading_stack_depth(cfg):
|
||||
md = "# H1\n\n## H2\n\n### H3\n\nTesto.\n"
|
||||
tokens, lines = parse(md)
|
||||
blocks = segment(tokens, lines, cfg)
|
||||
para = next(b for b in blocks if b.kind == "paragraph")
|
||||
assert para.header_path == [
|
||||
{"level": 1, "text": "H1"},
|
||||
{"level": 2, "text": "H2"},
|
||||
{"level": 3, "text": "H3"},
|
||||
]
|
||||
|
||||
|
||||
def test_context_depth_limits_header_path():
|
||||
cfg = ChunkerConfig(context_depth=2)
|
||||
md = "# H1\n\n## H2\n\n### H3\n\nTesto.\n"
|
||||
tokens, lines = parse(md)
|
||||
blocks = segment(tokens, lines, cfg)
|
||||
para = next(b for b in blocks if b.kind == "paragraph")
|
||||
assert len(para.header_path) == 2
|
||||
|
||||
|
||||
def test_skip_headings(cfg):
|
||||
md = "# Titolo\n\n## Indice\n\nContenuto saltato.\n\n## Sezione Reale\n\nContenuto reale.\n"
|
||||
tokens, lines = parse(md)
|
||||
blocks = segment(tokens, lines, cfg)
|
||||
contents = [b.content for b in blocks]
|
||||
assert not any("saltato" in c for c in contents)
|
||||
assert any("reale" in c for c in contents)
|
||||
|
||||
|
||||
def test_skip_pre_heading(cfg):
|
||||
md = "Testo prima del primo heading.\n\n# Titolo\n\nTesto dopo.\n"
|
||||
tokens, lines = parse(md)
|
||||
blocks = segment(tokens, lines, cfg)
|
||||
contents = [b.content for b in blocks]
|
||||
assert not any("prima del primo" in c for c in contents)
|
||||
assert any("dopo" in c for c in contents)
|
||||
|
||||
|
||||
def test_skip_pre_heading_disabled():
|
||||
cfg = ChunkerConfig(skip_pre_heading=False)
|
||||
md = "Testo prima.\n\n# Titolo\n\nTesto dopo.\n"
|
||||
tokens, lines = parse(md)
|
||||
blocks = segment(tokens, lines, cfg)
|
||||
contents = [b.content for b in blocks]
|
||||
assert any("prima" in c for c in contents)
|
||||
|
||||
|
||||
def test_block_ids_unique(cfg):
|
||||
md = "# T\n\nPara 1.\n\nPara 2.\n\nPara 3.\n"
|
||||
tokens, lines = parse(md)
|
||||
blocks = segment(tokens, lines, cfg)
|
||||
ids = [b.id for b in blocks]
|
||||
assert len(ids) == len(set(ids))
|
||||
|
||||
|
||||
def test_heading_reset_on_same_level(cfg):
|
||||
md = "# H1a\n\n## H2a\n\n# H1b\n\nTesto.\n"
|
||||
tokens, lines = parse(md)
|
||||
blocks = segment(tokens, lines, cfg)
|
||||
para = next(b for b in blocks if b.kind == "paragraph")
|
||||
assert para.header_path == [{"level": 1, "text": "H1b"}]
|
||||
|
||||
|
||||
def test_thematic_break(cfg):
|
||||
md = "# T\n\nPara.\n\n---\n\nPara 2.\n"
|
||||
tokens, lines = parse(md)
|
||||
blocks = segment(tokens, lines, cfg)
|
||||
kinds = [b.kind for b in blocks]
|
||||
assert "thematic_break" in kinds
|
||||
|
||||
|
||||
def test_blockquote(cfg):
|
||||
md = "# T\n\n> Una citazione.\n"
|
||||
tokens, lines = parse(md)
|
||||
blocks = segment(tokens, lines, cfg)
|
||||
bq = next((b for b in blocks if b.kind == "blockquote"), None)
|
||||
assert bq is not None
|
||||
assert "citazione" in bq.content
|
||||
|
||||
|
||||
def test_source_line_numbers(cfg):
|
||||
md = "# Titolo\n\nParagrafo.\n"
|
||||
tokens, lines = parse(md)
|
||||
blocks = segment(tokens, lines, cfg)
|
||||
para = next(b for b in blocks if b.kind == "paragraph")
|
||||
assert para.start_line == 2
|
||||
assert para.end_line == 3
|
||||
|
||||
|
||||
def test_math_block_is_atomic(cfg):
|
||||
md = "# T\n\n$$\nE=mc^2\n$$\n"
|
||||
tokens, lines = parse(md)
|
||||
blocks = segment(tokens, lines, cfg)
|
||||
math = next((b for b in blocks if b.kind == "math"), None)
|
||||
assert math is not None
|
||||
assert math.atomic is True
|
||||
assert "E=mc^2" in math.content
|
||||
|
||||
|
||||
def test_math_block_plain_text(cfg):
|
||||
md = "# T\n\n$$\n\\int_0^1 f(x) dx\n$$\n"
|
||||
tokens, lines = parse(md)
|
||||
blocks = segment(tokens, lines, cfg)
|
||||
math = next(b for b in blocks if b.kind == "math")
|
||||
assert math.plain_text == "[formula]"
|
||||
|
||||
|
||||
def test_html_block_with_table(cfg):
|
||||
md = "# T\n\n<table><tr><td>A</td></tr></table>\n"
|
||||
tokens, lines = parse(md)
|
||||
blocks = segment(tokens, lines, cfg)
|
||||
html = next((b for b in blocks if b.kind == "html"), None)
|
||||
assert html is not None
|
||||
assert "<table" in html.content
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user