diff --git a/chunks/chunker.py b/chunks/chunker.py deleted file mode 100644 index 9d79d9e..0000000 --- a/chunks/chunker.py +++ /dev/null @@ -1,413 +0,0 @@ -#!/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) - -Uso: - python chunks/chunker.py --stem - python chunks/chunker.py # tutti gli stem con .md in sources/ - python chunks/chunker.py --stem --force -""" - -import argparse -import json -import re -import sys -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 - - out_dir.mkdir(parents=True, exist_ok=True) - out_file.write_text( - json.dumps(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", - }, 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 - - -# ─── Entry point ────────────────────────────────────────────────────────────── - -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") - 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) - - 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) diff --git a/chunks/config.py b/chunks/config.py deleted file mode 100644 index b9cabd8..0000000 --- a/chunks/config.py +++ /dev/null @@ -1,64 +0,0 @@ -#!/usr/bin/env python3 -""" -Parametri della pipeline di chunking. - -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 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