refactor(chunks): rimuove Stage 1 e riscrive config per MD pulito
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 <noreply@anthropic.com>
This commit is contained in:
+44
-93
@@ -1,113 +1,64 @@
|
|||||||
#!/usr/bin/env python3
|
#!/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
|
Input atteso: sources/<stem>/<stem>.md — Markdown già pulito e ben strutturato.
|
||||||
(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.
|
|
||||||
"""
|
"""
|
||||||
|
|
||||||
# ─── Stage 1 — md_optimizer (pulizia Markdown) ───────────────────────────────
|
# ─── Dimensione chunk ─────────────────────────────────────────────────────────
|
||||||
|
|
||||||
# Tipi MinerU da ignorare completamente.
|
# Caratteri massimi per chunk (prefisso di contesto incluso).
|
||||||
NOISE_TYPES: set[str] = {
|
# Paragrafi più lunghi vengono spezzati a confine di frase.
|
||||||
"page_header", "page_number", "page_footer", "index", "page_aside_text",
|
# Una singola frase che supera MAX_CHARS non viene mai spezzata.
|
||||||
}
|
|
||||||
|
|
||||||
# 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).
|
|
||||||
MAX_CHARS: int = 1200
|
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
|
MIN_CHARS: int = 80
|
||||||
|
|
||||||
# Frasi di overlap: l'ultima frase del chunk N viene preposta al chunk N+1.
|
# ─── Spezzatura frasi ─────────────────────────────────────────────────────────
|
||||||
OVERLAP_SENTENCES: int = 1
|
|
||||||
|
|
||||||
# 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.
|
# Split solo prima di lettera maiuscola o virgolette — evita split su abbreviazioni.
|
||||||
SENTENCE_SPLIT_RE: str = r"(?<=[.!?»])\s+(?=[A-ZÀÈÉÌÒÙ\"])"
|
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.
|
# ─── Contesto heading ─────────────────────────────────────────────────────────
|
||||||
SPLIT_THRESHOLD_FACTOR: float = 1.5
|
|
||||||
|
|
||||||
|
# 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
|
MATH_SYMS_MIN: int = 3
|
||||||
|
|
||||||
PROTECT_TABLES: bool = True
|
PROTECT_TABLES: bool = True
|
||||||
PROTECT_MATH: bool = True
|
PROTECT_MATH: bool = True
|
||||||
FIX_MAX_ITERATIONS: int = 3
|
|
||||||
|
|||||||
@@ -1,397 +0,0 @@
|
|||||||
#!/usr/bin/env python3
|
|
||||||
"""
|
|
||||||
Fix chunk
|
|
||||||
|
|
||||||
Applica correzioni dirette su chunks/<stem>/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/<stem>/chunks.json + chunks/<stem>/report.json
|
|
||||||
Output: chunks/<stem>/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)
|
|
||||||
@@ -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/<stem>/auto/<stem>_content_list_v2.json
|
|
||||||
sources/<stem>/auto/<stem>_model.json (opzionale)
|
|
||||||
Output: sources/<stem>/auto/<stem>_clean.md
|
|
||||||
|
|
||||||
Uso standalone:
|
|
||||||
python chunks/md_optimizer.py --stem <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""))
|
|
||||||
|
|
||||||
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)
|
|
||||||
Reference in New Issue
Block a user