From f0c6bad0464f3f5d8132992fc808cdc3207db2c1 Mon Sep 17 00:00:00 2001 From: Davide Grilli Date: Thu, 4 Jun 2026 14:18:17 +0200 Subject: [PATCH] feat(chunker): riscrittura completa per input MD pulito MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Input: sources/_output/auto/.md (fallback: sources/.md) Output: chunks//chunks.json + meta.json Regole implementate: - 1 paragrafo = 1 chunk; paragrafi di contesti diversi non si mescolano - split a confine di frase se paragrafo > MAX_CHARS (mai a metà frase) - merge_short(): paragrafi < MIN_CHARS fusi col successivo (stesso contesto) - merge_broken_sentences(): chunk consecutivi con frase spezzata vengono uniti automaticamente se stesso contesto e corpo senza punteggiatura finale - parse_paragraphs(): skip sezioni via SKIP_HEADINGS (prefisso case-insensitive) e skip contenuto pre-heading via SKIP_PRE_HEADING - blocchi atomici: tabelle, liste, code block non vengono mai spezzati - nessun overlap tra chunk Co-Authored-By: Claude Sonnet 4.6 --- chunks/chunker.py | 398 +++++++++++++++++++++++++++++----------------- 1 file changed, 252 insertions(+), 146 deletions(-) diff --git a/chunks/chunker.py b/chunks/chunker.py index 480a5ff..9d79d9e 100644 --- a/chunks/chunker.py +++ b/chunks/chunker.py @@ -1,29 +1,22 @@ #!/usr/bin/env python3 """ -Pipeline di chunking unificata (Stage 1 + Stage 2) +chunker.py — Chunking semantico da Markdown pulito -Stage 1 — Ottimizzazione Markdown (md_optimizer): - Legge _content_list_v2.json + _model.json di MinerU e produce _clean.md - con gerarchia H1/H2/H3 pulita (TOC, frontespizi e sommari rimossi). - -Stage 2 — Chunking semantico: - Divide il _clean.md in chunk semantici: - - un chunk per paragrafo (mai due paragrafi nello stesso chunk) - - split a confine di frase se il paragrafo supera MAX_CHARS - - overlap di OVERLAP_SENTENCES frasi tra chunk consecutivi - - tabelle e liste sono blocchi atomici (non si spezzano) - -Input: sources//auto/_content_list_v2.json - sources//auto/_model.json (opzionale) -Output: sources//auto/_clean.md - chunks//chunks.json +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 in sources/ + python chunks/chunker.py # tutti gli stem con .md in sources/ python chunks/chunker.py --stem --force - python chunks/chunker.py --stem --skip-optimize # salta Stage 1 """ import argparse @@ -36,7 +29,6 @@ _HERE = Path(__file__).resolve().parent if str(_HERE) not in sys.path: sys.path.insert(0, str(_HERE)) import config as cfg -from md_optimizer import optimize as _optimize_md # ─── Utilità ────────────────────────────────────────────────────────────────── @@ -47,7 +39,6 @@ def split_sentences(text: str) -> list[str]: def context_to_meta(context: str) -> tuple[str, str]: - """Divide 'H1 > H2 > H3' in (sezione, titolo) per ingest/verify.""" parts = [p.strip() for p in context.split(" > ") if p.strip()] if len(parts) >= 2: return " > ".join(parts[:-1]), parts[-1] @@ -56,77 +47,175 @@ def context_to_meta(context: str) -> tuple[str, str]: # ─── 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 _clean.md con il loro contesto heading. + """Estrae blocchi dal Markdown con il loro contesto heading. - Restituisce: [{"context": "H1 > H2 > H3", "text": "...", "kind": "text|table|list"}] + Restituisce: [{"context": "H1 > H2 > H3", "text": "...", "kind": "text|table|list|code"}] - Ogni riga vuota chiude il paragrafo corrente. Tabelle (righe con |) e - liste (righe con -) vengono accumulate come blocchi atomici. + - 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. """ - h1 = h2 = h3 = "" - result: list[dict] = [] - buf: list[str] = [] - cur_kind = "text" + 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: - parts = [p for p in [h1, h2, h3] if p] - context = " > ".join(parts) if parts else "documento" - result.append({"context": context, "text": body, "kind": cur_kind}) + result.append({"context": current_context(), "text": body, "kind": cur_kind}) buf.clear() for line in text.splitlines(): - if re.match(r"^# ", line): + # ── 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() - h1, h2, h3 = line[2:].strip(), "", "" + 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" - elif re.match(r"^## ", line): - flush() - h2, h3 = line[3:].strip(), "" - cur_kind = "text" - elif re.match(r"^### ", line): - flush() - h3 = line[4:].strip() - cur_kind = "text" - elif line.strip().startswith("|"): + + # 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) - elif line.strip().startswith("- "): + continue + + # ── Lista ───────────────────────────────────────────────────────────── + if re.match(r"^\s*[-*]\s", line): if cur_kind != "list": flush() cur_kind = "list" buf.append(line) - elif line.strip() == "": + continue + + # ── Riga vuota: chiude il paragrafo corrente ────────────────────────── + if line.strip() == "": flush() cur_kind = "text" - else: - if cur_kind in ("table", "list"): - flush() - cur_kind = "text" - buf.append(line) + 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 dal risultato di parse_paragraphs. + """Genera chunk da parse_paragraphs (dopo eventuale merge). - Regole: - - un chunk = un paragrafo (o sotto-parte se > MAX_CHARS) - - split solo a confine di frase; una frase che supera MAX_CHARS è emessa intera - - l'ultima frase del chunk N viene preposta al chunk N+1 (overlap) - - tabelle e liste: blocco atomico (mai spezzato) + - 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] = [] - overlap_tail: list[str] = [] - idx = 0 + chunks: list[dict] = [] + idx = 0 for para in paragraphs: text = para["text"] @@ -134,10 +223,8 @@ def make_chunks(paragraphs: list[dict]) -> list[dict]: kind = para["kind"] sezione, titolo = context_to_meta(context) - # ── Blocchi atomici (tabelle, liste) ────────────────────────────────── - if kind in ("table", "list"): - prefix = " ".join(overlap_tail) + " " if overlap_tail else "" - body = (prefix + text).strip() + def emit(body: str) -> None: + nonlocal idx chunk_text = f"[{context}]\n{body}" chunks.append({ "chunk_id": f"c{idx}", @@ -147,95 +234,119 @@ def make_chunks(paragraphs: list[dict]) -> list[dict]: "context": context, "n_chars": len(chunk_text), }) - idx += 1 - sents = split_sentences(text) - overlap_tail = sents[-cfg.OVERLAP_SENTENCES:] if cfg.OVERLAP_SENTENCES else [] + idx += 1 + + # ── Atomici ─────────────────────────────────────────────────────────── + if kind in ("table", "list", "code"): + emit(text) continue - # ── Paragrafo testo: split a confine di frase ───────────────────────── + # ── Testo: split a confine di frase se supera MAX_CHARS ─────────────── sents = split_sentences(text) if not sents: continue - current: list[str] = list(overlap_tail) - has_primary: bool = False - + current: list[str] = [] for sent in sents: - candidate_len = len(" ".join(current + [sent])) - - if candidate_len <= cfg.MAX_CHARS or not has_primary: + projected = len(" ".join(current + [sent])) + if projected <= cfg.MAX_CHARS or not current: current.append(sent) - has_primary = True else: - body = " ".join(current) - 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 - overlap_tail = current[-cfg.OVERLAP_SENTENCES:] if cfg.OVERLAP_SENTENCES else [] - current = list(overlap_tail) + [sent] - has_primary = True + emit(" ".join(current)) + current = [sent] - if has_primary: - body = " ".join(current) - 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 - overlap_tail = current[-cfg.OVERLAP_SENTENCES:] if cfg.OVERLAP_SENTENCES else [] + 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, skip_optimize: bool) -> bool: - """Esegue Stage 1 (ottimizzazione MD) + Stage 2 (chunking) per un 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 - # ── Stage 1: ottimizzazione Markdown ────────────────────────────────────── - if not skip_optimize: - ok = _optimize_md(stem, project_root, force=force) - if not ok: - return False - else: - print(f"\n[Stage 1] skip (--skip-optimize)") - - # ── Stage 2: chunking ───────────────────────────────────────────────────── - clean_md = project_root / "sources" / stem / "auto" / f"{stem}_clean.md" out_dir = project_root / "chunks" / stem out_file = out_dir / "chunks.json" - print(f"[Stage 2] Chunking: {stem}") - - if not clean_md.exists(): - print(f" ✗ {stem}_clean.md non trovato") - return False - if out_file.exists() and not force: - print(f" ↩ chunks.json già presente — skip chunking") + print(f" ↩ chunks/{stem}/chunks.json già presente — skip (usa --force per rigenerare)") return True - text = clean_md.read_text(encoding="utf-8") + 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 {clean_md.name}") + 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") @@ -247,12 +358,13 @@ def process_stem(stem: str, project_root: Path, ) (out_dir / "meta.json").write_text( json.dumps({ - "min_chars": cfg.MIN_CHARS, - "max_chars": cfg.MAX_CHARS, - "target_chars": cfg.MAX_CHARS, - "overlap": cfg.OVERLAP_SENTENCES, - "strategy": "paragraph_overlap", - }, ensure_ascii=False), + "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", ) @@ -274,34 +386,28 @@ def process_stem(stem: str, project_root: Path, if __name__ == "__main__": project_root = Path(__file__).parent.parent + sources_dir = project_root / "sources" - parser = argparse.ArgumentParser( - description="Pipeline unificata MinerU → _clean.md → chunks.json" - ) - parser.add_argument("--stem", help="Nome documento (sottocartella di 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 _clean.md e chunks.json anche se esistono") - parser.add_argument("--skip-optimize", action="store_true", - help="Salta Stage 1 (usa _clean.md già presente)") + help="Rigenera chunks.json anche se già presente") args = parser.parse_args() if args.stem: stems = [args.stem] else: - sources_dir = project_root / "sources" - stems = sorted( - p.name for p in sources_dir.iterdir() - if p.is_dir() - and (p / "auto" / f"{p.name}_content_list_v2.json").exists() - ) + 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("Errore: nessun documento MinerU trovato in sources/") + print("Nessun file .md trovato in sources/") sys.exit(1) - results = [ - process_stem(s, project_root, args.force, args.skip_optimize) - for s in stems - ] + 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)