From ce4c3e5c870f2c70ee129fb28a8e801ef52c40dd Mon Sep 17 00:00:00 2001 From: Davide Grilli Date: Thu, 4 Jun 2026 14:18:44 +0200 Subject: [PATCH] refactor(chunks): rimuove Stage 1 e riscrive config per MD pulito MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Eliminati md_optimizer.py e fix_chunks.py: la pipeline non parte più da _content_list_v2.json ma da un .md già pulito in sources/. config.py ridotto da 114 a 64 righe: rimossi tutti i parametri MinerU (NOISE_TYPES, FRONTMATTER_HEADINGS, MODEL_SKIP_LABELS, ecc.) e aggiunti i parametri effettivamente utili al chunking: SKIP_HEADINGS, SKIP_PRE_HEADING, MERGE_SHORT_PARAGRAPHS, ATOMIC_TYPES, CONTEXT_DEPTH. Co-Authored-By: Claude Sonnet 4.6 --- chunks/config.py | 143 ++++------- chunks/fix_chunks.py | 397 ------------------------------ chunks/md_optimizer.py | 540 ----------------------------------------- 3 files changed, 47 insertions(+), 1033 deletions(-) delete mode 100644 chunks/fix_chunks.py delete mode 100644 chunks/md_optimizer.py diff --git a/chunks/config.py b/chunks/config.py index f355959..b9cabd8 100644 --- a/chunks/config.py +++ b/chunks/config.py @@ -1,113 +1,64 @@ #!/usr/bin/env python3 """ -Parametri della pipeline chunks: chunker.py (+ md_optimizer interno) + verify/fix. +Parametri della pipeline di chunking. -La pipeline è unificata: chunker.py esegue prima l'ottimizzazione del Markdown -(Stage 1, equivalente a md_optimizer.py) e poi il chunking (Stage 2). - -I parametri sono pensati per essere generici rispetto agli output di MinerU: -i file *_content_list_v2.json e *_model.json hanno sempre la stessa struttura, -indipendentemente dal documento sorgente. +Input atteso: sources//.md — Markdown già pulito e ben strutturato. """ -# ─── Stage 1 — md_optimizer (pulizia Markdown) ─────────────────────────────── +# ─── Dimensione chunk ───────────────────────────────────────────────────────── -# Tipi MinerU da ignorare completamente. -NOISE_TYPES: set[str] = { - "page_header", "page_number", "page_footer", "index", "page_aside_text", -} - -# Paragrafi promossi a H3 se testo ≤ H3_MAX_CHARS e matcha H3_DETECTION_RE. -# Regex generica: riga che inizia con numero seguito da punto e spazio. -# Per disabilitare la promozione a H3: imposta H3_DETECTION_RE = r"(?!)" -H3_DETECTION_RE: str = r"^\d+\.\s+\S" -H3_MAX_CHARS: int = 120 - -# Se True, i blocchi immagine non vengono inclusi nel Markdown. -SKIP_IMAGES: bool = True - -# Heading le cui sezioni vengono rimosse completamente (titolo + tutto il contenuto). -# Match case-insensitive: il testo dell'heading deve essere uguale o iniziare -# con uno dei valori seguenti. -# Nota: specifici per documento — impostare set vuoto per documenti non italiani -# o senza sezioni di frontmatter note. -FRONTMATTER_HEADINGS: set[str] = { - "sommario", - "indice", - "autori", - "abbreviazioni", - "atti normativi", - "specifici provvedimenti normativi", - "abbreviazioni generiche", -} - -# Numero minimo di heading consecutivi senza testo per riconoscere un TOC. -# Abbassare se il documento ha molti capitoli corti senza sottosezioni. -MIN_TOC_HEADINGS: int = 5 - -# Caratteri minimi di testo reale sotto un heading perché sia considerato -# "contenuto vero" (non frontespizio/copyright). -# Impostare >= lunghezza massima del testo di copyright/copertina nel documento. -MIN_CONTENT_CHARS: int = 2500 - -# Pattern per riconoscere prefissi di capitolo in blocchi paragraph. -# MinerU talvolta produce il numero/identificatore di capitolo come paragraph -# anziché come title L1 (comportamento non uniforme). Questi pattern permettono -# di bufferizzare tali paragrafi e fonderli col titolo L1 successivo. -# Impostare lista vuota [] per disabilitare. -CHAPTER_PREFIX_PATTERNS: list[str] = [ - r"^(CAPITOLO|PARTE)\s+(\d+|[IVXLCDM]+)\b", # italiano - r"^(CHAPTER|PART|SECTION)\s+(\d+|[IVXLCDM]+)\b", # inglese - r"^(CHAPITRE|PARTIE)\s+(\d+|[IVXLCDM]+)\b", # francese - r"^(KAPITEL|TEIL)\s+(\d+|[IVXLCDM]+)\b", # tedesco -] - -# Pattern testuali (regex) per riconoscere paragrafi "sommario interno" da saltare. -# Usati come fallback quando _model.json non assegna label "abstract". -# Generici: un pattern per paragrafo che inizia con indice/sommario di sezione. -# Per disabilitare: impostare lista vuota []. -SOMMARIO_PATTERNS: list[str] = [ - r"^SOMMARIO\s*:", # italiano - r"^SUMMARY\s*:", # inglese - r"^RÉSUMÉ\s*:", # francese - r"^ÍNDICE\s*:", # spagnolo/portoghese - r"^INHALT\s*:", # tedesco -] - -# ─── _model.json label sets ─────────────────────────────────────────────────── - -# Label di layout da saltare completamente. -MODEL_SKIP_LABELS: set[str] = { - "header", "number", "footer_image", "ocr_text", "aside_text", -} - -# Label che identifica indici/sommari interni (da saltare). -MODEL_ABSTRACT_LABELS: set[str] = {"abstract"} - -# ─── Stage 2 — chunker ──────────────────────────────────────────────────────── - -# Lunghezza massima di un chunk (caratteri, prefisso incluso). -# Paragrafi che superano questo limite vengono spezzati a confine di frase. -# Una singola frase che supera MAX_CHARS viene emessa intera (non si spezza mai). +# 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 -# Lunghezza minima attesa (warning in verify_chunks, non blocker). +# Soglia minima attesa (usata da verify_chunks come warning, non blocca). MIN_CHARS: int = 80 -# Frasi di overlap: l'ultima frase del chunk N viene preposta al chunk N+1. -OVERLAP_SENTENCES: int = 1 +# ─── Spezzatura frasi ───────────────────────────────────────────────────────── -# Regex per rilevare il confine di fine frase per lo split. +# 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 ────────────────────────────────────────────────────────── -# ─── verify_chunks.py / fix_chunks.py ───────────────────────────────────────── +# Blocchi Markdown che non vengono mai spezzati, anche se superano MAX_CHARS. +ATOMIC_TYPES: set[str] = {"table", "code", "list"} -# fix_chunks spezza un chunk too_long solo se supera MAX_CHARS × questo fattore. -SPLIT_THRESHOLD_FACTOR: float = 1.5 +# ─── Contesto heading ───────────────────────────────────────────────────────── -MATH_SYMS_MIN: int = 3 -PROTECT_TABLES: bool = True -PROTECT_MATH: bool = True -FIX_MAX_ITERATIONS: int = 3 +# 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/fix_chunks.py b/chunks/fix_chunks.py deleted file mode 100644 index 9f94fca..0000000 --- a/chunks/fix_chunks.py +++ /dev/null @@ -1,397 +0,0 @@ -#!/usr/bin/env python3 -""" -Fix chunk - -Applica correzioni dirette su chunks//chunks.json basandosi sul -report.json prodotto da verify_chunks.py. Non tocca clean.md. - -Fixes applicati: - empty → rimuove il chunk - incomplete → fonde con il chunk successivo (la frase continua) - no_prefix → aggiunge prefisso [sezione > titolo] se mancante - too_short → fonde con il chunk adiacente nello stesso sezione - too_long → spezza all'ultimo confine di paragrafo/frase entro MAX_CHARS - -Input: chunks//chunks.json + chunks//report.json -Output: chunks//chunks.json (sovrascrive) - -Uso: - python chunks/fix_chunks.py --stem documento - python chunks/fix_chunks.py --stem documento --dry-run -""" - -import argparse -import contextlib -import io -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 -from verify_chunks import verify_stem as _verify_stem - -MAX_CHARS = cfg.MAX_CHARS - - -def _load_thresholds(stem_dir: Path) -> int: - """Legge max_chars da meta.json (scritto dal chunker) o usa il default da config.""" - meta = stem_dir / "meta.json" - if meta.exists(): - import json as _json - return _json.loads(meta.read_text(encoding="utf-8"))["max_chars"] - return MAX_CHARS -PUNCT_END = re.compile(r"[.!?»)\]'\u2019\"\u201c\u201d\u2018\u2014\u2013-]$") - - -# ─── Helpers ────────────────────────────────────────────────────────────────── - -def _prefix(chunk: dict) -> str: - sezione = chunk.get("sezione", "") - titolo = chunk.get("titolo", "") - if titolo: - return f"[{sezione} > {titolo}]" - return f"[{sezione}]" - - -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 - - -def _rebuild_text(chunk: dict, body: str) -> str: - return f"{_prefix(chunk)}\n{body}" - - -# Fine frase forte: . ! ? seguiti da spazio + maiuscola o virgolette. -# Non usare punteggiatura debole (,;:)>>]) per non creare chunk incompleti. -_STRONG_END = re.compile( - r'[.!?\xbb]\s+(?=[A-Z\xc0-\xd6\xd8-\xde\xc0-\xff\xab\x22\x27(])' -) -_SECONDARY_END = re.compile(r';\s+') - - -def _split_at_boundary(text: str, max_chars: int) -> list[str]: - """Spezza text in parti ≤ max_chars su confini di frase forti (.!?). - - Se non trova un confine forte entro max_chars, NON spezza: meglio un - chunk too_long (warning) che un chunk incompleto (blocker). - """ - if len(text) <= max_chars: - return [text] - - parts = [] - remaining = text - - while len(remaining) > max_chars: - candidate = remaining[:max_chars] - - last_pos = -1 - for m in _STRONG_END.finditer(candidate): - last_pos = m.start() + 1 # posizione dopo il carattere terminatore - - if last_pos > 0: - first = remaining[:last_pos].rstrip() - remaining = remaining[last_pos:].lstrip() - if first: - parts.append(first) - else: - # Prova confine secondario: ; + spazio (clausole legali) - sec_pos = -1 - for m in _SECONDARY_END.finditer(candidate): - sec_pos = m.start() + 1 - if sec_pos > 0: - first = remaining[:sec_pos].rstrip() - remaining = remaining[sec_pos:].lstrip() - if first: - parts.append(first) - else: - # Nessun confine: lascia il chunk intero (too_long > incomplete) - break - - if remaining: - parts.append(remaining) - - return [p for p in parts if p.strip()] - - -# ─── Operazioni sui chunk ───────────────────────────────────────────────────── - -def fix_empty(chunks: list[dict], empty_ids: set[str]) -> tuple[list[dict], int]: - before = len(chunks) - chunks = [c for c in chunks if c["chunk_id"] not in empty_ids] - return chunks, before - len(chunks) - - -def fix_no_prefix(chunks: list[dict], no_prefix_ids: set[str]) -> tuple[list[dict], int]: - count = 0 - for c in chunks: - if c["chunk_id"] in no_prefix_ids: - body = _strip_prefix(c["text"]) - c["text"] = _rebuild_text(c, body) - c["n_chars"] = len(c["text"]) - count += 1 - return chunks, count - - -def fix_incomplete_and_short(chunks: list[dict], - problem_ids: set[str]) -> tuple[list[dict], int]: - merged = 0 - i = 0 - result: list[dict] = [] - - while i < len(chunks): - c = chunks[i] - if c["chunk_id"] in problem_ids and i + 1 < len(chunks): - nxt = chunks[i + 1] - body_c = _strip_prefix(c["text"]) - body_nxt = _strip_prefix(nxt["text"]) - merged_body = body_c.rstrip() + "\n" + body_nxt.lstrip() - nxt["text"] = _rebuild_text(nxt, merged_body) - nxt["n_chars"] = len(nxt["text"]) - merged += 1 - i += 1 - continue - result.append(c) - i += 1 - - return result, merged - - -def fix_too_long(chunks: list[dict], - too_long_ids: set[str], - max_chars: int) -> tuple[list[dict], int]: - result: list[dict] = [] - split_count = 0 - - for c in chunks: - if c["chunk_id"] not in too_long_ids: - result.append(c) - continue - - body = _strip_prefix(c["text"]) - parts = _split_at_boundary(body, max_chars) - - if len(parts) == 1: - result.append(c) - continue - - base_id = re.sub(r"__s\d+$", "", c["chunk_id"]) - base_sub = c.get("sub_index", 0) - - for j, part in enumerate(parts): - new_chunk = dict(c) - new_chunk["sub_index"] = base_sub + j - new_chunk["chunk_id"] = f"{base_id}__s{base_sub + j}" - new_chunk["text"] = _rebuild_text(new_chunk, part) - new_chunk["n_chars"] = len(new_chunk["text"]) - result.append(new_chunk) - - split_count += 1 - - return result, split_count - - -def renumber_ids(chunks: list[dict]) -> list[dict]: - seen: dict[str, int] = {} - for c in chunks: - base = re.sub(r"__s\d+$", "", c["chunk_id"]) - idx = seen.get(base, 0) - c["chunk_id"] = f"{base}__s{idx}" - c["sub_index"] = idx - seen[base] = idx + 1 - return chunks - - -# ─── Core ───────────────────────────────────────────────────────────────────── - -def fix_stem(stem: str, project_root: Path, max_chars: int, dry_run: bool, - max_iter: int = 10) -> bool: - stem_dir = project_root / "chunks" / stem - chunks_path = stem_dir / "chunks.json" - report_path = stem_dir / "report.json" - max_chars = _load_thresholds(stem_dir) - - 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 - - if not report_path.exists(): - print(f"✗ chunks/{stem}/report.json non trovato.") - print(f" Esegui prima: python chunks/verify_chunks.py --stem {stem}") - return False - - chunks: list[dict] = json.loads(chunks_path.read_text(encoding="utf-8")) - report: dict = json.loads(report_path.read_text(encoding="utf-8")) - - verdict = report.get("verdict", "ok") - print(f"\nDocumento: {stem} (verdict: {verdict})") - - if verdict == "ok": - print(" ✅ Nessun problema - nulla da correggere.") - return True - - empty_ids = {e["chunk_id"] for e in report.get("blockers", {}).get("empty", [])} - no_prefix_ids = {e["chunk_id"] for e in report.get("blockers", {}).get("no_prefix", [])} - incomplete_ids = {e["chunk_id"] for e in report.get("blockers", {}).get("incomplete", [])} - too_short_ids = {e["chunk_id"] for e in report.get("warnings", {}).get("too_short", [])} - - # Spezza solo chunk che superano upper × SPLIT_THRESHOLD_FACTOR, - # non quelli appena oltre upper (che causerebbero split con chunk incompleti). - _split_limit = max_chars * cfg.SPLIT_THRESHOLD_FACTOR - too_long_ids = { - e["chunk_id"] - for e in report.get("warnings", {}).get("too_long", []) - if e.get("n_chars", 0) > _split_limit - } - - ops: list[str] = [] - if empty_ids: - ops.append(f" 🗑 rimuovi {len(empty_ids)} chunk vuoti") - if no_prefix_ids: - ops.append(f" 🔧 aggiungi prefisso a {len(no_prefix_ids)} chunk") - if incomplete_ids: - ops.append(f" 🔗 fondi {len(incomplete_ids)} chunk incompleti col successivo") - if too_short_ids: - ops.append(f" 🔗 fondi {len(too_short_ids)} chunk troppo corti col successivo") - if too_long_ids: - ops.append(f" ✂️ spezza {len(too_long_ids)} chunk troppo lunghi") - - if not ops: - print(" ✅ Nessuna correzione necessaria.") - return True - - print("\n Operazioni pianificate:") - for op in ops: - print(op) - - if dry_run: - print("\n [dry-run] Nessuna modifica applicata.") - return True - - n_before = len(chunks) - - def _fix_blockers(chunks: list[dict], report: dict) -> list[dict]: - """Risolve solo i blockers (incomplete, empty, no_prefix) senza toccare warnings.""" - empty_ids_ = {e["chunk_id"] for e in report.get("blockers", {}).get("empty", [])} - no_prefix_ids_ = {e["chunk_id"] for e in report.get("blockers", {}).get("no_prefix", [])} - incomplete_ids_ = {e["chunk_id"] for e in report.get("blockers", {}).get("incomplete", [])} - if empty_ids_: - chunks, n = fix_empty(chunks, empty_ids_) - print(f" 🗑 Rimossi {n} chunk vuoti.") - if no_prefix_ids_: - chunks, n = fix_no_prefix(chunks, no_prefix_ids_) - print(f" 🔧 Aggiunto prefisso a {n} chunk.") - if incomplete_ids_: - chunks, n = fix_incomplete_and_short(chunks, incomplete_ids_) - print(f" 🔗 Fusi {n} chunk incompleti.") - return renumber_ids(chunks) - - def _fix_warnings(chunks: list[dict], report: dict) -> list[dict]: - """Applica fix opzionali: merge too_short e split too_long.""" - too_short_ids_ = {e["chunk_id"] for e in report.get("warnings", {}).get("too_short", [])} - too_long_ids_ = { - e["chunk_id"] - for e in report.get("warnings", {}).get("too_long", []) - if e.get("n_chars", 0) > max_chars * cfg.SPLIT_THRESHOLD_FACTOR - } - if too_short_ids_: - chunks, n = fix_incomplete_and_short(chunks, too_short_ids_) - print(f" 🔗 Fusi {n} chunk troppo corti.") - if too_long_ids_: - chunks, n = fix_too_long(chunks, too_long_ids_, max_chars) - print(f" ✂️ Spezzati {n} chunk lunghi.") - return renumber_ids(chunks) - - # Fase 1: risolvi blockers a convergenza (solo merge incomplete) - chunks = _fix_blockers(chunks, report) - - _min = cfg.MIN_CHARS - _max = cfg.MAX_CHARS - prev_blockers = sum(len(v) for v in report.get("blockers", {}).values()) - - for iteration in range(1, max_iter + 1): - chunks_path.write_text( - json.dumps(chunks, ensure_ascii=False, indent=2), encoding="utf-8" - ) - with contextlib.redirect_stdout(io.StringIO()): - _verify_stem(stem, project_root, _min, _max) - report = json.loads(report_path.read_text(encoding="utf-8")) - new_verdict = report.get("verdict", "ok") - curr_blockers = sum(len(v) for v in report.get("blockers", {}).values()) - - if new_verdict in ("ok", "warnings_only") or curr_blockers == 0: - break - if curr_blockers >= prev_blockers: - print(f"\n ⚠️ Nessun miglioramento ({curr_blockers} blockers) - i restanti richiedono correzione manuale del clean.md.") - break - - print(f"\n Iterazione {iteration + 1} - {curr_blockers} blockers residui:") - prev_blockers = curr_blockers - chunks = _fix_blockers(chunks, report) - - # Fase 2: fix warnings (too_short merge + too_long split) - una sola passata finale - with contextlib.redirect_stdout(io.StringIO()): - _verify_stem(stem, project_root, _min, _max) - report = json.loads(report_path.read_text(encoding="utf-8")) - n_short = len(report.get("warnings", {}).get("too_short", [])) - n_long = sum( - 1 for e in report.get("warnings", {}).get("too_long", []) - if e.get("n_chars", 0) > max_chars * cfg.SPLIT_THRESHOLD_FACTOR - ) - if n_short or n_long: - print(f"\n Fix warnings: {n_short} corti, {n_long} lunghi da spezzare") - chunks = _fix_warnings(chunks, report) - - n_after = len(chunks) - print(f"\n Totale chunk: {n_before} → {n_after}") - - chunks_path.write_text( - json.dumps(chunks, ensure_ascii=False, indent=2), encoding="utf-8" - ) - print(f" ✅ Salvato: chunks/{stem}/chunks.json") - - final_verdict = report.get("verdict", "?") - if final_verdict == "ok": - print(f" ✅ Verdict finale: ok - procedi alla vettorizzazione.") - elif final_verdict == "warnings_only": - print(f" 🟡 Verdict finale: warnings_only - puoi procedere.") - else: - print(f" 🔴 Verdict finale: {final_verdict} - rilancia la verifica manualmente:") - print(f" python chunks/verify_chunks.py --stem {stem}") - - return True - - -# ─── Entry point ────────────────────────────────────────────────────────────── - -if __name__ == "__main__": - project_root = Path(__file__).parent.parent - - parser = argparse.ArgumentParser(description="Fix chunk") - parser.add_argument("--stem", required=True, help="Nome del documento (sottocartella di chunks/)") - _max_def = cfg.MAX_CHARS - parser.add_argument( - "--max", type=int, default=_max_def, - help=f"Soglia massima caratteri per lo split (default: TARGET×(1+TOL) = {_max_def})" - ) - parser.add_argument( - "--dry-run", action="store_true", - help="Mostra le operazioni pianificate senza applicarle" - ) - parser.add_argument( - "--max-iter", type=int, default=10, metavar="N", - help="Numero massimo di iterazioni automatiche (default: 10)" - ) - args = parser.parse_args() - - ok = fix_stem(args.stem, project_root, args.max, args.dry_run, args.max_iter) - sys.exit(0 if ok else 1) diff --git a/chunks/md_optimizer.py b/chunks/md_optimizer.py deleted file mode 100644 index e1db5e1..0000000 --- a/chunks/md_optimizer.py +++ /dev/null @@ -1,540 +0,0 @@ -#!/usr/bin/env python3 -""" -Stage 1 — Ottimizzatore Markdown (modulo interno, chiamato da chunker.py) - -Legge _content_list_v2.json (struttura primaria) e _model.json (label di -layout) di MinerU e produce un Markdown pulito con gerarchia H1/H2/H3. - -Progettato per essere generico rispetto al documento: sfrutta la struttura -comune di tutti gli output MinerU senza dipendere da pattern testuali -specifici del documento sorgente. - -Logica di costruzione blocchi: - - title L1 consecutivi senza contenuto tra loro → fusi in un H1 unico - (il primo frammento è sempre il numero/identificatore del capitolo) - - title L1 singolo → H1 - - title L2 → H2 - - paragraph con label "abstract" o che matcha SOMMARIO_PATTERNS → skip - - paragraph breve che matcha H3_DETECTION_RE → H3 - - paragraph normale → testo - - label MODEL_SKIP_LABELS → skip - -Filtri di pulizia: - - _remove_frontmatter : rimuove sezioni per nome (FRONTMATTER_HEADINGS) - - _remove_toc_runs : rimuove sequenze di heading senza contenuto (TOC) - - _remove_frontespizio : rimuove contenuto prima del primo heading "vero" - (>= MIN_CONTENT_CHARS di testo reale) - -Input: sources//auto/_content_list_v2.json - sources//auto/_model.json (opzionale) -Output: sources//auto/_clean.md - -Uso standalone: - python chunks/md_optimizer.py --stem [--force] - python chunks/md_optimizer.py # tutti gli stem in sources/ -""" - -import argparse -import json -import re -import sys -from dataclasses import dataclass -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 - - -# ─── Struttura dati interna ─────────────────────────────────────────────────── - -@dataclass -class Block: - kind: str # "h1" | "h2" | "h3" | "text" | "list" | "table" - text: str - - -_HEADING_LEVEL = {"h1": 1, "h2": 2, "h3": 3} - -# Pattern compilati da config (inizializzati lazy per permettere hot-reload in test) -_SOMMARIO_RES: list[re.Pattern] = [] -_CHAPTER_PREFIX_RES: list[re.Pattern] = [] - -def _init_patterns() -> None: - global _SOMMARIO_RES, _CHAPTER_PREFIX_RES - _SOMMARIO_RES = [re.compile(p, re.IGNORECASE) for p in cfg.SOMMARIO_PATTERNS] - _CHAPTER_PREFIX_RES = [re.compile(p, re.IGNORECASE) for p in cfg.CHAPTER_PREFIX_PATTERNS] - -_init_patterns() - - -def _is_sommario(text: str) -> bool: - return any(r.match(text) for r in _SOMMARIO_RES) - - -def _is_chapter_prefix(text: str) -> bool: - """True se il testo è un identificatore di capitolo (es. "CAPITOLO 1"). - - Usato come fallback quando MinerU produce il numero del capitolo come - paragraph anziché come title L1. - """ - return any(r.match(text) for r in _CHAPTER_PREFIX_RES) - - -# ─── Caricamento e indicizzazione _model.json ───────────────────────────────── - -def _load_label_map(model_path: Path) -> dict[int, list[tuple[float, float, str]]]: - """Restituisce {page_idx: [(cx_v2, cy_v2, label), ...]} - - Le coordinate cx/cy sono nel sistema di riferimento v2: - v2_coord = model_coord * 1000 / model_page_dim - """ - if not model_path.exists(): - return {} - - pages = json.loads(model_path.read_text(encoding="utf-8")) - label_map: dict[int, list[tuple[float, float, str]]] = {} - - for page in pages: - info = page.get("page_info", {}) - page_no = info.get("page_no", 0) - pw = info.get("width", 1350) - ph = info.get("height", 1891) - - entries: list[tuple[float, float, str]] = [] - for det in page.get("layout_dets", []): - label = det.get("label", "") - if label in cfg.MODEL_SKIP_LABELS: - continue - x0, y0, x1, y1 = det["bbox"] - cx = (x0 + x1) * 0.5 * 1000.0 / pw - cy = (y0 + y1) * 0.5 * 1000.0 / ph - entries.append((cx, cy, label)) - - label_map[page_no] = entries - - return label_map - - -def _get_label(page_idx: int, bbox: list[int], - label_map: dict[int, list]) -> str: - """Restituisce il label model.json il cui centro è più vicino al centro - del bbox v2 (tolleranza 80 unità v2 ≈ 8% della larghezza pagina).""" - entries = label_map.get(page_idx) - if not entries: - return "" - x0, y0, x1, y1 = bbox - cx = (x0 + x1) * 0.5 - cy = (y0 + y1) * 0.5 - - best_label = "" - best_dist = 80.0 - - for ex, ey, label in entries: - dist = ((cx - ex) ** 2 + (cy - ey) ** 2) ** 0.5 - if dist < best_dist: - best_dist = dist - best_label = label - - return best_label - - -# ─── Estrazione testo dai blocchi MinerU ────────────────────────────────────── - -def _text_para(content: dict) -> str: - return " ".join( - p["content"] for p in content.get("paragraph_content", []) - if p.get("type") == "text" - ).strip() - - -def _text_title(content: dict) -> str: - return " ".join( - p["content"] for p in content.get("title_content", []) - if p.get("type") == "text" - ).strip() - - -def _text_list(content: dict) -> str: - lines = [] - for item in content.get("list_content", []): - for block in item.get("blocks", []): - t = block.get("content", "").strip() - if t: - lines.append(f"- {t}") - return "\n".join(lines) - - -def _is_h3_candidate(text: str) -> bool: - return ( - len(text) <= cfg.H3_MAX_CHARS - and bool(re.match(cfg.H3_DETECTION_RE, text)) - ) - - -# ─── Build blocchi da JSON MinerU ───────────────────────────────────────────── - -def _build_blocks(pages: list, label_map: dict) -> list[Block]: - """Costruisce la lista di Block dalla struttura MinerU. - - Logica per i titoli H1 consecutivi (generica, senza pattern lingua-specifica): - - Ogni title L1 viene bufferizzato come "pending_h1". - - Se arriva un altro title L1 subito dopo (senza contenuto tra loro), - i due frammenti vengono fusi in un unico H1 con " — " come separatore. - Questo gestisce il pattern comune di MinerU dove il numero/identificatore - del capitolo e il suo titolo sono due blocchi separati. - - Quando arriva contenuto non-titolo (paragrafo, lista, H2), il pending_h1 - viene emesso così com'è. - """ - blocks: list[Block] = [] - pending_h1: str = "" # titolo L1 in attesa di conferma/merge - - def _flush_h1() -> None: - nonlocal pending_h1 - if pending_h1: - blocks.append(Block(kind="h1", text=pending_h1)) - pending_h1 = "" - - for page_idx, page in enumerate(pages): - for item in page: - kind = item.get("type", "") - content = item.get("content", {}) - bbox = item.get("bbox", [0, 0, 0, 0]) - - # ── Tipi MinerU rumorosi ───────────────────────────────────────── - if kind in cfg.NOISE_TYPES: - _flush_h1() - continue - - model_label = _get_label(page_idx, bbox, label_map) - - # ── Label model rumorosi ───────────────────────────────────────── - if model_label in cfg.MODEL_SKIP_LABELS: - continue - - # ── Sommari interni (abstract label o pattern testuale) ────────── - if model_label in cfg.MODEL_ABSTRACT_LABELS: - continue - - # ── Titoli ─────────────────────────────────────────────────────── - if kind == "title": - text = _text_title(content) - if not text: - continue - level = min(content.get("level", 2), 3) - - if level == 1: - if pending_h1: - # Due title L1 consecutivi: fondi il precedente col corrente - merged = f"{pending_h1} — {text}" - pending_h1 = merged - else: - pending_h1 = text - else: - # H2: emetti prima il pending H1 se esiste - _flush_h1() - blocks.append(Block(kind="h2", text=text)) - - # ── Paragrafi ──────────────────────────────────────────────────── - elif kind == "paragraph": - text = _text_para(content) - if not text: - continue - - # Sommario interno: salta (fallback testuale se label non copre) - if _is_sommario(text): - continue - - # Prefisso di capitolo come paragraph (es. "CAPITOLO 1"): - # bufferizza come pending H1, verrà fuso col titolo L1 successivo - if _is_chapter_prefix(text): - if pending_h1: - pending_h1 = f"{pending_h1} — {text}" - else: - pending_h1 = text - continue - - _flush_h1() - - if _is_h3_candidate(text): - blocks.append(Block(kind="h3", text=text)) - else: - blocks.append(Block(kind="text", text=text)) - - # ── Liste ──────────────────────────────────────────────────────── - elif kind == "list": - _flush_h1() - text = _text_list(content) - if text: - blocks.append(Block(kind="list", text=text)) - - # ── Tabelle ────────────────────────────────────────────────────── - elif kind == "table": - _flush_h1() - body = content.get("table_body", "") - if body: - blocks.append(Block(kind="table", text=body)) - - # ── Immagini (opzionale) ───────────────────────────────────────── - elif kind == "image" and not cfg.SKIP_IMAGES: - _flush_h1() - src = content.get("image_source", {}).get("path", "") - caption = " ".join( - c.get("content", "") for c in content.get("image_caption", []) - ).strip() - if src: - blocks.append(Block(kind="text", text=f"![{caption}]({src})")) - - else: - _flush_h1() - - _flush_h1() # flush finale - return blocks - - -# ─── Helpers content check ──────────────────────────────────────────────────── - -def _has_content(blocks: list[Block], idx: int) -> bool: - """True se esiste almeno un blocco testo/lista/tabella prima del prossimo - heading di livello uguale o superiore.""" - level = _HEADING_LEVEL.get(blocks[idx].kind) - if level is None: - return False - for b in blocks[idx + 1:]: - blevel = _HEADING_LEVEL.get(b.kind) - if blevel is not None and blevel <= level: - return False - if b.kind in ("text", "list", "table"): - return True - return False - - -def _has_real_content(blocks: list[Block], idx: int) -> bool: - """True se il totale caratteri di testo sotto questo heading >= - MIN_CONTENT_CHARS. Permette di distinguere frontespizi (copyright breve) - da sezioni con contenuto vero.""" - level = _HEADING_LEVEL.get(blocks[idx].kind) - if level is None: - return False - total = 0 - for b in blocks[idx + 1:]: - blevel = _HEADING_LEVEL.get(b.kind) - if blevel is not None and blevel <= level: - break - if b.kind in ("text", "list", "table"): - total += len(b.text) - if total >= cfg.MIN_CONTENT_CHARS: - return True - return False - - -# ─── Filtri di pulizia ──────────────────────────────────────────────────────── - -def _remove_frontmatter(blocks: list[Block]) -> list[Block]: - """Rimuove le sezioni il cui heading è in FRONTMATTER_HEADINGS, insieme - a tutto il loro contenuto. - - Il salto continua finché non si trova un heading non-frontmatter — - questo elimina anche sezioni TOC consecutive in un colpo solo. - """ - def _norm(text: str) -> str: - t = text.strip().lower() - # Rimuovi eventuale prefisso "Xxx N — " (identificatore capitolo) - return re.sub(r"^\S+\s+\S+\s+[—\-]\s*", "", t) - - def _is_fm(text: str) -> bool: - core = _norm(text) - return any( - core == fm or core.startswith(fm + " ") - for fm in cfg.FRONTMATTER_HEADINGS - ) - - if not cfg.FRONTMATTER_HEADINGS: - return blocks - - result: list[Block] = [] - i = 0 - while i < len(blocks): - b = blocks[i] - if b.kind in _HEADING_LEVEL and _is_fm(b.text): - level = _HEADING_LEVEL[b.kind] - i += 1 - while i < len(blocks): - nxt = blocks[i] - nxt_level = _HEADING_LEVEL.get(nxt.kind) - if nxt_level is not None and nxt_level <= level and not _is_fm(nxt.text): - break - i += 1 - continue - result.append(b) - i += 1 - return result - - -def _remove_toc_runs(blocks: list[Block]) -> list[Block]: - """Rimuove sequenze di MIN_TOC_HEADINGS o più heading consecutivi senza - testo reale tra loro (TOC residuo). - - "Consecutivi" tolera micro-testi brevi (≤ 120 chars) intercalati tra - i heading (es. attribuzioni autori nel TOC). - """ - def _is_toc_entry(idx: int) -> bool: - b = blocks[idx] - if b.kind not in _HEADING_LEVEL: - return False - level = _HEADING_LEVEL[b.kind] - for b2 in blocks[idx + 1:]: - blevel = _HEADING_LEVEL.get(b2.kind) - if blevel is not None and blevel <= level: - return True - if b2.kind in ("text", "list", "table") and len(b2.text) > 20: - return False - return True - - result: list[Block] = [] - i = 0 - while i < len(blocks): - b = blocks[i] - if b.kind in _HEADING_LEVEL and _is_toc_entry(i): - j = i + 1 - toc_count = 1 - while j < len(blocks): - bj = blocks[j] - if bj.kind in _HEADING_LEVEL: - if _is_toc_entry(j): - toc_count += 1 - j += 1 - continue - else: - break - if bj.kind in ("text", "list") and len(bj.text) <= 120: - j += 1 - continue - break - if toc_count >= cfg.MIN_TOC_HEADINGS: - i = j - continue - result.append(b) - i += 1 - return result - - -def _remove_frontespizio(blocks: list[Block]) -> list[Block]: - """Rimuove tutto il contenuto prima del primo heading con contenuto reale - (>= MIN_CONTENT_CHARS): copertine, copyright, pagine iniziali.""" - for i, b in enumerate(blocks): - if b.kind in _HEADING_LEVEL and _has_real_content(blocks, i): - return blocks[i:] - return blocks - - -def filter_blocks(blocks: list[Block]) -> list[Block]: - blocks = _remove_frontmatter(blocks) - blocks = _remove_toc_runs(blocks) - blocks = _remove_frontespizio(blocks) - return blocks - - -# ─── Rendering ──────────────────────────────────────────────────────────────── - -def _render(blocks: list[Block]) -> str: - lines: list[str] = [] - prev_was_heading = False - - for b in blocks: - if b.kind in ("h1", "h2", "h3"): - prefix = "#" * _HEADING_LEVEL[b.kind] - if lines and not prev_was_heading: - lines.append("") - lines.append(f"{prefix} {b.text}") - prev_was_heading = True - else: - lines.append("") - lines.append(b.text) - prev_was_heading = False - - md = "\n".join(lines).strip() + "\n" - return re.sub(r"\n{3,}", "\n\n", md) - - -# ─── Core ───────────────────────────────────────────────────────────────────── - -def optimize(stem: str, project_root: Path, force: bool = False) -> bool: - """Esegue Stage 1: _content_list_v2.json + _model.json → _clean.md. - - Restituisce True se il file è stato prodotto (o era già presente e - force=False), False in caso di errore. - """ - auto_dir = project_root / "sources" / stem / "auto" - json_path = auto_dir / f"{stem}_content_list_v2.json" - model_path = auto_dir / f"{stem}_model.json" - out_path = auto_dir / f"{stem}_clean.md" - - print(f"\n[Stage 1] Documento: {stem}") - - if not json_path.exists(): - print(f" ✗ {json_path.name} non trovato") - return False - - if out_path.exists() and not force: - print(f" ↩ {out_path.name} già presente — skip ottimizzazione") - return True - - pages = json.loads(json_path.read_text(encoding="utf-8")) - - if model_path.exists(): - label_map = _load_label_map(model_path) - n_labels = sum(len(v) for v in label_map.values()) - print(f" 📐 {model_path.name} ({n_labels} label)") - else: - label_map = {} - print(f" ℹ️ {model_path.name} non trovato — nessun enrichment layout") - - blocks = _build_blocks(pages, label_map) - n_raw = len(blocks) - blocks = filter_blocks(blocks) - n_filtered = n_raw - len(blocks) - - md = _render(blocks) - out_path.write_text(md, encoding="utf-8") - - n_h1 = len(re.findall(r"^# ", md, re.MULTILINE)) - n_h2 = len(re.findall(r"^## ", md, re.MULTILINE)) - n_h3 = len(re.findall(r"^### ", md, re.MULTILINE)) - print(f" ✅ {out_path.name} " - f"({md.count(chr(10))} righe — H1={n_h1} H2={n_h2} H3={n_h3} " - f"rimossi={n_filtered}/{n_raw})") - return True - - -# ─── Entry point standalone ─────────────────────────────────────────────────── - -if __name__ == "__main__": - project_root = Path(__file__).parent.parent - - parser = argparse.ArgumentParser( - description="Stage 1: _content_list_v2.json + _model.json → _clean.md" - ) - parser.add_argument("--stem", help="Nome documento (sottocartella di sources/)") - parser.add_argument("--force", action="store_true", - help="Rigenera anche se _clean.md esiste già") - 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() - ) - if not stems: - print("Errore: nessun documento MinerU trovato in sources/") - sys.exit(1) - - results = [optimize(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)