refactor: modularizza pipeline in conversione/_pipeline/

Sostituisce i file monolitici pipeline.py e validate.py con il package
_pipeline/ a responsabilità separate. Entry point unificato in __main__.py
(convert + validate dallo stesso comando).

Moduli aggiunti:
- __main__.py       — CLI unificata (--stem, --force, validate, --detail)
- _pipeline/__init__.py — re-export pubblico
- _pipeline/checker.py  — validazione PDF
- _pipeline/deps.py     — verifica dipendenze Java + opendataloader
- _pipeline/structure.py — analyze() + strategia chunking

Moduli già committati in precedenza:
- _pipeline/converter.py, transforms.py, report.py, runner.py, validator.py

Aggiornamenti collaterali:
- .gitignore: exception !conversione/_pipeline/** per tracciare il package
- CLAUDE.md: documentazione aggiornata alla nuova architettura; fix riferimenti
  obsoleti a conversione/pipeline.py → conversione/

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-04-30 14:59:55 +02:00
parent faa8acae84
commit e41fcae248
9 changed files with 464 additions and 1842 deletions
+111
View File
@@ -0,0 +1,111 @@
#!/usr/bin/env python3
"""
Pipeline PDF → clean Markdown per vettorizzazione RAG.
Uso:
# Converti
python conversione/ --stem <nome>
python conversione/ --stem <nome> --force
python conversione/ # tutti i PDF in sources/
# Valida
python conversione/ validate
python conversione/ validate <stem> [<stem> ...] --detail
Prerequisiti:
pip install opendataloader-pdf pdfplumber
Java 11+ sul PATH (https://adoptium.net/)
"""
import argparse
import sys
from pathlib import Path
# Rende _pipeline importabile da conversione/
sys.path.insert(0, str(Path(__file__).parent))
from _pipeline import _check_deps, run, validate
def _build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
prog="conversione",
description="PDF → clean Markdown strutturato, pronto per chunking RAG",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog=(
"Esempi:\n"
" python conversione/ --stem manuale\n"
" python conversione/ --stem manuale --force\n"
" python conversione/ validate\n"
" python conversione/ validate manuale --detail"
),
)
# ── Subcommand: validate ──────────────────────────────────────────────
sub = parser.add_subparsers(dest="cmd", metavar="comando")
val = sub.add_parser(
"validate",
help="valida i report.json prodotti dalla conversione",
description="Legge i report.json e assegna un voto 0-100 (A/B/C/D/F).",
)
val.add_argument(
"stems",
nargs="*",
metavar="STEM",
help="stem da validare. Ometti per tutti.",
)
val.add_argument(
"--detail", "-d",
action="store_true",
help="mostra il dettaglio delle penalità per ogni documento",
)
# ── Opzioni convert (modalità default) ───────────────────────────────
parser.add_argument(
"--stem",
metavar="NOME",
help="nome del PDF in sources/ (senza estensione). Ometti per tutti.",
)
parser.add_argument(
"--force",
action="store_true",
help="riesegui anche se clean.md è già presente",
)
return parser
def main() -> None:
parser = _build_parser()
args = parser.parse_args()
root = Path(__file__).parent.parent
# ── Validate ─────────────────────────────────────────────────────────
if args.cmd == "validate":
validate(args.stems, root, detail=args.detail)
return
# ── Convert (default) ────────────────────────────────────────────────
_check_deps()
if args.stem:
stems = [args.stem]
else:
sources_dir = root / "sources"
if not sources_dir.exists():
print("Errore: cartella sources/ non trovata.")
sys.exit(1)
stems = sorted(p.stem for p in sources_dir.glob("*.pdf"))
if not stems:
print("Errore: nessun PDF trovato in sources/.")
sys.exit(1)
results = [run(s, root, args.force) for s in stems]
ok = sum(results)
total = len(results)
print(f"\n{'' if all(results) else '⚠️ '} {ok}/{total} documenti convertiti")
sys.exit(0 if all(results) else 1)
if __name__ == "__main__":
main()
+19
View File
@@ -0,0 +1,19 @@
from .deps import _check_deps
from .checker import check_pdf
from .converter import convert_pdf
from .transforms import apply_transforms
from .structure import analyze
from .report import build_report
from .runner import run
from .validator import validate
__all__ = [
"_check_deps",
"check_pdf",
"convert_pdf",
"apply_transforms",
"analyze",
"build_report",
"run",
"validate",
]
+51
View File
@@ -0,0 +1,51 @@
from pathlib import Path
def check_pdf(pdf_path: Path) -> tuple[bool, str]:
"""Validazione rapida: esistenza, leggibilità, testo estraibile."""
if not pdf_path.exists():
return False, f"File non trovato: {pdf_path}"
if pdf_path.suffix.lower() != ".pdf":
return False, f"Non è un PDF: {pdf_path.name}"
size = pdf_path.stat().st_size
if size == 0:
return False, "File vuoto"
if size < 1024:
return False, f"File troppo piccolo ({size} byte) — probabilmente corrotto"
try:
import pdfplumber
with pdfplumber.open(pdf_path) as pdf:
n_pages = len(pdf.pages)
if n_pages == 0:
return False, "PDF senza pagine"
sample = min(5, n_pages)
pages_with_text = sum(
1 for i in range(sample)
if len((pdf.pages[i].extract_text() or "").strip()) > 50
)
if pages_with_text == 0:
extended = min(15, n_pages)
if extended > sample:
ext_with_text = sum(
1 for i in range(sample, extended)
if len((pdf.pages[i].extract_text() or "").strip()) > 50
)
if ext_with_text > 0:
return True, (
f"{n_pages} pagine — prime {sample} vuote, "
f"testo trovato in pagine successive "
f"(possibile copertina immagine)"
)
return False, (
f"Nessun testo nelle prime {extended} pagine "
f"— probabilmente scansionato (OCR non supportato)"
)
return True, f"{n_pages} pagine, testo digitale confermato"
except MemoryError:
return False, "Memoria esaurita durante l'apertura del PDF"
except Exception as e:
msg = str(e).lower()
if "password" in msg or "encrypted" in msg:
return False, "PDF protetto da password"
return False, f"Impossibile aprire: {e}"
+23
View File
@@ -0,0 +1,23 @@
import subprocess
import sys
def _check_deps() -> None:
try:
import opendataloader_pdf # noqa: F401
except ImportError:
print("Errore: opendataloader-pdf non installato.")
print(" pip install opendataloader-pdf")
sys.exit(1)
try:
result = subprocess.run(
["java", "-version"],
capture_output=True, text=True,
)
if result.returncode != 0:
raise FileNotFoundError
except FileNotFoundError:
print("Errore: Java 11+ non trovato sul PATH.")
print(" Installa da https://adoptium.net/")
sys.exit(1)
+141
View File
@@ -0,0 +1,141 @@
import re
from pathlib import Path
# ─── Rilevamento lingua ───────────────────────────────────────────────────────
_IT_WORDS = frozenset([
"il", "la", "di", "e", "che", "non", "per", "un", "una", "si",
"con", "da", "del", "della", "dei", "in", "ma", "se", "lo", "le",
"gli", "al", "alla", "ai", "alle", "sono", "ha", "hanno", "era",
"erano", "nel", "nella", "nei", "nelle", "questo", "questa", "così",
])
_EN_WORDS = frozenset([
"the", "of", "and", "to", "in", "is", "that", "it", "was", "for",
"on", "are", "as", "with", "his", "they", "at", "be", "this", "have",
"from", "or", "an", "but", "not", "by", "he", "she", "we", "you",
"which", "their", "been", "has", "would", "there", "when", "will",
])
_FR_WORDS = frozenset([
"le", "les", "de", "du", "des", "et", "un", "une", "est", "que",
"pour", "dans", "sur", "avec", "qui", "par", "pas", "plus", "au",
"ce", "se", "ou", "mais", "comme", "aussi",
])
_DE_WORDS = frozenset([
"der", "die", "das", "und", "in", "von", "zu", "den", "mit", "ist",
"auf", "eine", "als", "dem", "des", "sich", "nicht", "auch", "werden",
"bei", "nach", "oder", "wenn", "wird", "war",
])
_ES_WORDS = frozenset([
"el", "los", "las", "de", "en", "un", "una", "es", "que", "por",
"con", "del", "para", "como", "pero", "sus", "son", "los", "hay",
"todo", "esta", "este", "ser", "más", "ya",
])
def _detect_language(text: str) -> str:
words = re.findall(r"\b[a-zA-Z]{2,}\b", text.lower())
sample = words[:2000]
scores = {
"it": sum(1 for w in sample if w in _IT_WORDS),
"en": sum(1 for w in sample if w in _EN_WORDS),
"fr": sum(1 for w in sample if w in _FR_WORDS),
"de": sum(1 for w in sample if w in _DE_WORDS),
"es": sum(1 for w in sample if w in _ES_WORDS),
}
best = max(scores, key=scores.get)
return best if scores[best] > 0 else "unknown"
# ─── Analisi struttura ────────────────────────────────────────────────────────
def _count_headers(text: str, level: int) -> int:
prefix = "#" * level + " "
return len(re.findall(rf"(?m)^{re.escape(prefix)}", text))
def _count_paragraphs(text: str) -> int:
blocks = re.split(r"\n{2,}", text)
return sum(1 for b in blocks if b.strip() and not re.match(r"^#+\s", b.strip()))
def _split_sections(text: str, level: int) -> list[str]:
prefix = "#" * level + " "
parts = re.split(rf"(?m)^{re.escape(prefix)}.+", text)
return [p for p in parts[1:] if p.strip()]
def _parse_sections_with_body(text: str, level: int = 3) -> list[tuple[str, str]]:
"""Restituisce lista di (header_line, body_text) per tutti gli header al livello dato."""
prefix = "#" * level + " "
lines = text.split("\n")
sections: list[tuple[str, str]] = []
cur_hdr: str | None = None
cur_body: list[str] = []
for line in lines:
if line.startswith(prefix):
if cur_hdr is not None:
sections.append((cur_hdr, "\n".join(cur_body).strip()))
cur_hdr = line
cur_body = []
elif cur_hdr is not None:
cur_body.append(line)
if cur_hdr is not None:
sections.append((cur_hdr, "\n".join(cur_body).strip()))
return sections
def analyze(md_path: Path) -> dict:
text = md_path.read_text(encoding="utf-8")
n_h1 = _count_headers(text, 1)
n_h2 = _count_headers(text, 2)
n_h3 = _count_headers(text, 3)
n_paragrafi = _count_paragraphs(text)
if n_h3 >= 5:
livello, boundary, strategia = 3, "h3", "h3_aware"
section_bodies = _split_sections(text, 3)
# Se h3 sono enormi e h2 più brevi, h2 è il boundary corretto
if n_h2 >= 3:
h2_bodies = _split_sections(text, 2)
avg_h3 = sum(len(b) for b in section_bodies) / len(section_bodies) if section_bodies else 0
avg_h2 = sum(len(b) for b in h2_bodies) / len(h2_bodies) if h2_bodies else 0
if avg_h3 > 5000 and avg_h2 < avg_h3 * 0.7:
livello, boundary, strategia = 2, "h2", "h2_paragraph_split"
section_bodies = h2_bodies
elif n_h2 >= 3:
livello, boundary, strategia = 2, "h2", "h2_paragraph_split"
section_bodies = _split_sections(text, 2)
elif n_h1 + n_h2 + n_h3 >= 1:
livello, boundary, strategia = 1, "paragrafo", "paragraph"
section_bodies = [b for b in re.split(r"\n{2,}", text) if b.strip()]
elif n_paragrafi >= 3:
livello, boundary, strategia = 1, "paragrafo", "paragraph"
section_bodies = [b for b in re.split(r"\n{2,}", text) if b.strip()]
else:
livello, boundary, strategia = 0, "nessuno", "sliding_window"
section_bodies = [text] if text.strip() else []
lengths = [len(b) for b in section_bodies if b.strip()]
lunghezza_media = int(sum(lengths) / len(lengths)) if lengths else 0
lingua = _detect_language(text)
avvertenze = []
short = sum(1 for l in lengths if l < 200)
long_ = sum(1 for l in lengths if l > 800)
if short:
avvertenze.append(f"{short} sezioni sotto i 200 caratteri — verranno accorpate")
if long_:
avvertenze.append(f"{long_} sezioni sopra i 800 caratteri — verranno divise")
return {
"livello_struttura": livello,
"n_h1": n_h1,
"n_h2": n_h2,
"n_h3": n_h3,
"n_paragrafi": n_paragrafi,
"boundary_primario": boundary,
"lingua_rilevata": lingua,
"lunghezza_media_sezione": lunghezza_media,
"strategia_chunking": strategia,
"avvertenze": avvertenze,
}
File diff suppressed because it is too large Load Diff
-210
View File
@@ -1,210 +0,0 @@
#!/usr/bin/env python3
"""
conversione/validate.py Validazione qualità Markdown
Legge i report.json prodotti da pipeline.py, stampa una tabella di stato
e assegna un voto (0-100) a ogni documento.
90-100 A ottimo, pronto per il chunker
75-89 B buono, qualche sezione lunga ma accettabile
60-74 C accettabile, anomalie minori da verificare
40-59 D da rivedere, problemi strutturali o residui evidenti
0-39 F da riprocessare, struttura assente o testo corrotto
Uso:
python conversione/validate.py # tutti gli stem
python conversione/validate.py analisi1 # stem specifico
python conversione/validate.py a b c # stem multipli
python conversione/validate.py --detail analisi1 # mostra dettaglio penalità
"""
import argparse
import json
import sys
from pathlib import Path
# ─── Punteggio ───────────────────────────────────────────────────────────────
_GRADES = [(90, "A"), (75, "B"), (60, "C"), (40, "D"), (0, "F")]
def _score(r: dict) -> tuple[int, list[str]]:
"""
Calcola un punteggio 0-100 sulla qualità del clean.md ai fini della
suddivisione in chunk e vettorizzazione.
Restituisce (score, lista_penalità_applicate).
Penalità struttura (il chunker non può operare senza header):
struttura assente (livello 0) 40
struttura piatta (livello 1) 15
Penalità residui (finiscono nei vettori e degradano il retrieval):
backtick 2/cad (max 20)
dot-leader 5/cad (max 10)
URL / watermark 5/cad (max 15)
immagini residue 5/cad (max 10)
<br> inline (artefatti tabelle) 2/cad (max 15)
simboli encoding (!/" residui) → 1/cad (max 10)
formule inline [N.M] 1/cad (max 8)
Penalità anomalie:
bare headers 3/cad (max 15)
Non penalizzate (il chunker le normalizza):
sezioni corte, sezioni lunghe, mediana, p25
"""
score = 100
detail = []
structure = r.get("structure", {})
anomalie = r.get("anomalie", {})
residui = r.get("residui", {})
livello = structure.get("livello_struttura", 0)
# ── Struttura ─────────────────────────────────────────────────────────
if livello == 0:
score -= 40
detail.append("struttura assente 40")
elif livello == 1:
score -= 15
detail.append("struttura piatta 15")
# ── Residui ───────────────────────────────────────────────────────────
def _pen(key: str, per_item: int, cap: int, label: str) -> None:
n = residui.get(key, 0)
if n:
p = min(cap, n * per_item)
nonlocal score
score -= p
detail.append(f"{label} ×{n} {p}")
_pen("backtick", 2, 20, "backtick")
_pen("dotleader", 5, 10, "dot-leader")
_pen("url", 5, 15, "url")
_pen("immagini", 5, 10, "immagini")
_pen("br_inline", 2, 15, "<br> inline")
_pen("simboli_encoding", 1, 10, "simboli encoding")
_pen("formule_inline", 1, 8, "formule inline")
_pen("footnote_markers", 1, 8, "footnote residui")
_pen("pua_markers", 2, 20, "caratteri PUA font Symbol")
# ── Anomalie ──────────────────────────────────────────────────────────
n_bare = anomalie.get("bare_headers", 0)
if n_bare:
p = min(15, n_bare * 3)
score -= p
detail.append(f"bare headers ×{n_bare} {p}")
return max(0, score), detail
def _grade(score: int) -> str:
return next(g for threshold, g in _GRADES if score >= threshold)
# ─── Validazione ─────────────────────────────────────────────────────────────
def validate(stems: list[str], project_root: Path, detail: bool = False) -> None:
conv_dir = project_root / "conversione"
paths = (
[conv_dir / s / "report.json" for s in stems]
if stems
else sorted(conv_dir.glob("*/report.json"))
)
if not paths:
print("Nessun report.json trovato in conversione/*/")
sys.exit(0)
rows = [
json.loads(p.read_text(encoding="utf-8")) if p.exists()
else {"stem": p.parent.name, "_missing": True}
for p in paths
]
# ── Intestazione ─────────────────────────────────────────────────────
col = max(len(r.get("stem", "stem")) for r in rows) + 2
header = (
f"{'stem':<{col}}"
f"{'h2':>4}{'h3':>5} "
f"{'strategia':<18}"
f"{'bare':>5}{'corte':>6}{'lunghe':>7}"
f"{'btk':>5}{'br':>4}{'enc':>4}{'url':>4}"
f"{'med':>6}"
f" {'voto':>4} grade"
)
sep = "" * len(header)
print(f"\n{header}\n{sep}")
scores = []
# ── Righe ─────────────────────────────────────────────────────────────
for r in rows:
if r.get("_missing"):
print(f"{r['stem']:<{col}} (report.json non trovato)")
continue
st = r.get("structure", {})
an = r.get("anomalie", {})
res = r.get("residui", {})
dist = r.get("distribution", {})
s, pen = _score(r)
scores.append(s)
print(
f"{r['stem']:<{col}}"
f"{st.get('n_h2', 0):>4}"
f"{st.get('n_h3', 0):>5} "
f"{st.get('strategia_chunking','?'):<18}"
f"{an.get('bare_headers', 0):>5}"
f"{an.get('short_sections', 0):>6}"
f"{an.get('long_sections', 0):>7}"
f"{res.get('backtick', 0):>5}"
f"{res.get('br_inline', 0):>4}"
f"{res.get('simboli_encoding', 0):>4}"
f"{res.get('url', 0):>4}"
f"{dist.get('mediana', 0):>6}"
f" {s:>4} {_grade(s)}"
)
if detail and pen:
for p in pen:
print(f" {'':>{col}}{p}")
# ── Riepilogo ─────────────────────────────────────────────────────────
print(sep)
if scores:
media = sum(scores) / len(scores)
print(
f"Documenti: {len(scores)} "
f"Media: {media:.0f}/100 {_grade(int(media))} "
f"(A≥90 B≥75 C≥60 D≥40 F<40)"
)
print(
"\nColonne: bare=header vuoti corte=sez<150ch lunghe=sez>1500ch "
"btk=backtick br=<br>inline enc=simboli encoding med=mediana chars\n"
)
# ─── Entry point ─────────────────────────────────────────────────────────────
if __name__ == "__main__":
parser = argparse.ArgumentParser(
description="Valida i report Markdown prodotti da pipeline.py",
epilog="Senza argomenti valida tutti gli stem in conversione/*/",
)
parser.add_argument(
"stems",
nargs="*",
metavar="STEM",
help="stem da validare (es: analisi1). Ometti per tutti.",
)
parser.add_argument(
"--detail", "-d",
action="store_true",
help="mostra dettaglio penalità per ogni documento",
)
args = parser.parse_args()
validate(args.stems, Path(__file__).parent.parent, detail=args.detail)