diff --git a/chunks/__init__.py b/chunks/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/chunks/chunker.py b/chunks/chunker.py index 9d79d9e..f1c0821 100644 --- a/chunks/chunker.py +++ b/chunks/chunker.py @@ -1,413 +1,140 @@ #!/usr/bin/env python3 """ -chunker.py — Chunking semantico da Markdown pulito - -Input: sources/.md — Markdown già strutturato (H1/H2/H3 + paragrafi) -Output: chunks//chunks.json - chunks//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 - 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 --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 - - -# ─── 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 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 ""), "" - - -# ─── Parser Markdown ────────────────────────────────────────────────────────── - -_SKIP_HEADINGS_LOWER = {h.lower() for h in cfg.SKIP_HEADINGS} - - -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 -) - - -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 +_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 + + +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 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([], [], {})) + + 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={})) + + source = source_path.read_text(encoding="utf-8") + tokens, lines = parse(source) + blocks = segment(tokens, lines, config) + chunks = pack(blocks, config, stem) + + result = ChunkingResult( + stem=stem, + source_path=str(source_path.relative_to(root)), + chunks=chunks, + diagnostics=Diagnostics([], [], {}), + ) + 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", + "stem": stem, + "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) - if not stems: - print("Nessun file .md trovato in sources/") - sys.exit(1) + stems = [args.stem] if args.stem else _discover_stems(_ROOT) + if not stems: + 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() diff --git a/chunks/config.py b/chunks/config.py index b9cabd8..d0da27b 100644 --- a/chunks/config.py +++ b/chunks/config.py @@ -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//.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] = { - "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 +@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", + }) + 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 diff --git a/chunks/models.py b/chunks/models.py new file mode 100644 index 0000000..c7a99b3 --- /dev/null +++ b/chunks/models.py @@ -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 diff --git a/chunks/packer.py b/chunks/packer.py new file mode 100644 index 0000000..cdb9dd5 --- /dev/null +++ b/chunks/packer.py @@ -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 " 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 diff --git a/chunks/parser.py b/chunks/parser.py new file mode 100644 index 0000000..1bd5844 --- /dev/null +++ b/chunks/parser.py @@ -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 diff --git a/chunks/segmenter.py b/chunks/segmenter.py new file mode 100644 index 0000000..a3e3d3d --- /dev/null +++ b/chunks/segmenter.py @@ -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 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/chunks/verify_chunks.py b/chunks/verify_chunks.py deleted file mode 100644 index 5a5a467..0000000 --- a/chunks/verify_chunks.py +++ /dev/null @@ -1,493 +0,0 @@ -#!/usr/bin/env python3 -""" -Verifica chunk - -Analizza chunks//chunks.json e segnala ogni anomalia che potrebbe -degradare la qualità del retrieval. Non modifica nulla. - -Input: chunks//chunks.json -Output: report a schermo + chunks//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) diff --git a/requirements.txt b/requirements.txt index adcadb0..d6621e6 100644 --- a/requirements.txt +++ b/requirements.txt @@ -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 diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/chunks/__init__.py b/tests/chunks/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/chunks/conftest.py b/tests/chunks/conftest.py new file mode 100644 index 0000000..12d7cf2 --- /dev/null +++ b/tests/chunks/conftest.py @@ -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 diff --git a/tests/chunks/test_chunker.py b/tests/chunks/test_chunker.py new file mode 100644 index 0000000..ca0bbb9 --- /dev/null +++ b/tests/chunks/test_chunker.py @@ -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 diff --git a/tests/chunks/test_config.py b/tests/chunks/test_config.py new file mode 100644 index 0000000..5f94370 --- /dev/null +++ b/tests/chunks/test_config.py @@ -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 diff --git a/tests/chunks/test_models.py b/tests/chunks/test_models.py new file mode 100644 index 0000000..72ec406 --- /dev/null +++ b/tests/chunks/test_models.py @@ -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 == [] diff --git a/tests/chunks/test_packer.py b/tests/chunks/test_packer.py new file mode 100644 index 0000000..72cf2dd --- /dev/null +++ b/tests/chunks/test_packer.py @@ -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 = "
A
" + 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 diff --git a/tests/chunks/test_parser.py b/tests/chunks/test_parser.py new file mode 100644 index 0000000..eb49285 --- /dev/null +++ b/tests/chunks/test_parser.py @@ -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 diff --git a/tests/chunks/test_segmenter.py b/tests/chunks/test_segmenter.py new file mode 100644 index 0000000..9cf1a9e --- /dev/null +++ b/tests/chunks/test_segmenter.py @@ -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
A
\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 " 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