refactor: ottimizza pipeline PDF→Markdown — struttura piatta e verbosità

- Unifica deps.py + checker.py + converter.py in extract.py (fronte PDF)
- Sposta transforms/ in _pipeline/ (struttura piatta, no sottocartelle)
- Aggiunge spinner animato (thread) durante conversione opendataloader-pdf
- Aggiunge progresso step-by-step [i/37] per apply_transforms via callback
- Mostra punteggio qualità (score/100 grade) a fine elaborazione
- Fix: _DOTLEADER_RE spostata in _constants.py (non più definita inline)
- Fix: report.py importa regex da _constants invece di ridefinirle
- Fix: _t_remove_urls ora conta e ritorna le rimozioni effettive

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-05-07 14:30:41 +02:00
parent afbf29514d
commit 64dc403e80
20 changed files with 311 additions and 900 deletions
+20 -21
View File
@@ -69,37 +69,35 @@ conversione/
├── clear.sh # Rimuove output di uno stem
└── _pipeline/
├── __init__.py # Re-export pubblico
├── deps.py # _check_deps() verifica opendataloader-pdf e Java
├── checker.py # check_pdf() — validazione PDF
├── converter.py # convert_pdf() — wrapper opendataloader
├── extract.py # _check_deps() + validate_pdf() + convert_pdf()
├── runner.py # run() — orchestrazione 4 fasi
├── structure.py # analyze() + rilevamento lingua e struttura
├── report.py # build_report() → report.json
├── runner.py # run() — orchestrazione 4 fasi
├── validator.py # validate() + _score() + _grade()
── transforms/ # Package pulizia strutturale
├── __init__.py # Espone apply_transforms()
├── _apply.py # Orchestratore: lista _transforms in ordine semantico
├── _constants.py# Regex compilate e mappe statiche condivise
├── _encoding.py # Gruppo 1: PUA font Symbol, accenti LaTeX, simboli SI
├── _artifacts.py# Gruppo 2: immagini, BR, footnote, URL, righe ricorrenti, watermark
├── _headers.py # Gruppo 3: normalizzazione livelli, concat, bold, ALLCAPS
├── _structure.py# Gruppo 4: TOC, ALLCAPS→##, sezioni numerate, math, articoli
├── _text.py # Gruppo 5: merge paragrafi, whitespace, poesia, versi
├── _finish.py # Gruppo 6: header vuoti/garbage, formula-header, frontmatter
└── _helpers.py # Funzioni pure condivise (es. _sentence_case)
── _apply.py # apply_transforms() — orchestratore in ordine semantico
├── _constants.py # Regex compilate e mappe statiche condivise
├── _encoding.py # Gruppo 1: PUA font Symbol, accenti LaTeX, simboli SI
├── _artifacts.py # Gruppo 2: immagini, BR, footnote, URL, righe ricorrenti, watermark
├── _headers.py # Gruppo 3: normalizzazione livelli, concat, bold, ALLCAPS
├── _structure.py # Gruppo 4: TOC, ALLCAPS→##, sezioni numerate, math, articoli
├── _text.py # Gruppo 5: merge paragrafi, whitespace, poesia, versi
├── _finish.py # Gruppo 6: header vuoti/garbage, formula-header, frontmatter
└── _helpers.py # Funzioni pure condivise (_sentence_case, _is_allcaps_line, ecc.)
```
### `__main__.py` — entry point unificato
CLI con due modalità: conversione (default, `--stem`, `--force`) e validazione (subcommand `validate`, con stems opzionali e `--detail`). Aggiunge `conversione/` a `sys.path` e delega a `_pipeline`. Uso: `python conversione/ [--stem X] [--force]` oppure `python conversione/ validate [X] [--detail]`.
### `_pipeline/transforms/` — cuore della pipeline
### `_pipeline/extract.py` — fronte PDF
Package con ~35 trasformazioni atomiche (`_t_*`). L'orchestratore `_apply.py::apply_transforms(text) -> (text, stats)` le chiama in ordine semantico fisso — **non modificare l'ordine senza capire le dipendenze tra gruppi**.
Raggruppa in un unico modulo le tre responsabilità legate al PDF: `_check_deps()` verifica che `opendataloader-pdf` e Java 11+ siano disponibili; `validate_pdf(pdf_path) -> (bool, str)` controlla esistenza, dimensione e presenza di testo digitale; `convert_pdf(pdf_path, out_dir) -> Path` invoca opendataloader-pdf con i parametri RAG-ottimali e restituisce il percorso del `.md` prodotto. Auto-rileva i PDF taggati (Word/InDesign) per attivare `use_struct_tree`.
Ogni `_t_*` vive nel modulo corrispondente al suo gruppo. Le costanti regex (compilate una volta) stanno tutte in `_constants.py` e vengono importate dove servono.
### `_pipeline/_apply.py` — cuore della pipeline
Ordine logico dei gruppi in `_apply.py`:
Contiene `apply_transforms(text) -> (text, stats)` che chiama ~35 trasformazioni atomiche (`_t_*`) in ordine semantico fisso — **non modificare l'ordine senza capire le dipendenze tra gruppi**. Ogni `_t_*` vive nel modulo del suo gruppo; le costanti regex compilate stanno in `_constants.py`.
Ordine logico dei gruppi:
1. **Encoding** (`_encoding.py`) — PUA font Symbol, accenti backtick LaTeX, moltiplicazione, micro
2. **Pulizia artefatti** (`_artifacts.py`) — immagini, `<br>`, footnote superscript, URL, box symbol, righe ricorrenti, watermark
3. **Struttura header** (`_headers.py`) — fix header+body concatenati, Capitolo inline, normalizzazione livelli numerati, `####``###`, bold, ALL-CAPS
@@ -157,9 +155,10 @@ Assegna un voto 0100 (A/B/C/D/F) leggendo `report.json`. Penalità principali
Quando si aggiunge una trasformazione in `apply_transforms()`:
- Ogni `_t_*` deve restituire `(testo, n_modifiche)` — il contatore alimenta `report.json`.
- Implementare la funzione nel modulo del gruppo corretto (`_encoding.py`, `_artifacts.py`, ecc.), importarla in `_apply.py` e inserire la coppia `("stat_key", _t_nuova)` nella lista `_transforms` nel punto logicamente corretto.
- Compilare i pattern regex in `_constants.py` come costanti di modulo, non dentro la funzione.
- Compilare i pattern regex in `_constants.py` come costanti di modulo, mai dentro la funzione.
- Testare con `.venv/bin/python conversione/ --stem <stem> --force` e confrontare `report.json`.
- Un nuovo tipo di artefatto: prima aggiungerlo come residuo in `report.py` (`_scan`), poi implementare la `_t_*` che lo rimuove.
- Un nuovo tipo di artefatto: prima aggiungerlo come residuo in `report.py` (funzione `_scan`), poi implementare la `_t_*` che lo rimuove.
- I residui in `report.py` usano `_MATH_SYMBOLS_RE`, `_EXERCISE_TRIGGER_RE` e `_MATH_HDR_RE` da `transforms._constants` — non ridefinirli localmente.
---
+4 -6
View File
@@ -1,15 +1,13 @@
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 .extract import _check_deps, validate_pdf, convert_pdf
from ._apply 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",
"validate_pdf",
"convert_pdf",
"apply_transforms",
"analyze",
@@ -31,66 +31,72 @@ from ._finish import (
)
def apply_transforms(text: str) -> tuple[str, dict]:
def apply_transforms(text: str, on_step=None) -> tuple[str, dict]:
"""
Applica le trasformazioni strutturali al Markdown grezzo.
Restituisce (testo_modificato, statistiche).
L'ordine è semantico: encoding → artefatti → struttura header →
costruzione struttura testo rifinitura.
on_step(i, total, label) callback opzionale chiamato dopo ogni step.
"""
_has_ex = bool(re.search(r"\b(Esercizi|Exercises|Problems|Homework)\b", text, re.IGNORECASE))
_transforms: list[tuple[str | None, object]] = [
# (stat_key, fn, label)
_transforms: list[tuple[str | None, object, str]] = [
# 1. Encoding
("n_simboli_pua_corretti", _t_fix_symbol_font),
("n_accenti_corretti", _t_fix_accents),
("n_moltiplicazioni_corrette", _t_fix_multiplication),
("n_micro_corretti", _t_fix_micro),
("n_simboli_pua_corretti", _t_fix_symbol_font, "encoding PUA Symbol"),
("n_accenti_corretti", _t_fix_accents, "accenti backtick LaTeX"),
("n_moltiplicazioni_corrette", _t_fix_multiplication, "simbolo moltiplicazione"),
("n_micro_corretti", _t_fix_micro, "simbolo micro SI"),
# 2. Pulizia artefatti
("n_immagini_rimosse", _t_remove_images),
("n_br_rimossi", _t_fix_br),
("n_tabsep_rimossi", _t_fix_tabsep),
("n_note_rimosse", _t_remove_footnotes),
("n_simboli_math_rimossi", _t_fix_math_symbols),
("n_formule_rimossi", _t_remove_formula_labels),
("n_dotleader_rimossi", _t_remove_dotleaders),
("n_righe_ricorrenti_rimosse", _t_remove_recurring_lines),
("n_immagini_rimosse", _t_remove_images, "rimozione immagini"),
("n_br_rimossi", _t_fix_br, "fix <br> inline"),
("n_tabsep_rimossi", _t_fix_tabsep, "fix separatori tabella"),
("n_note_rimosse", _t_remove_footnotes, "rimozione footnote"),
("n_simboli_math_rimossi", _t_fix_math_symbols, "rimozione box math"),
("n_formule_rimossi", _t_remove_formula_labels, "rimozione label formula"),
("n_dotleader_rimossi", _t_remove_dotleaders, "rimozione dot-leader TOC"),
("n_righe_ricorrenti_rimosse", _t_remove_recurring_lines, "rimozione righe ricorrenti"),
# 3. Struttura header
("n_header_concat_fixati", _t_fix_header_concat),
(None, _t_extract_capitolo),
("n_header_numerati_normalizzati", _t_normalize_numbered_headings),
(None, _t_normalize_header_levels),
(None, _t_remove_header_bold),
(None, _t_normalize_allcaps_headers),
("n_header_concat_fixati", _t_fix_header_concat, "fix header+corpo concatenati"),
(None, _t_extract_capitolo, "estrazione Capitolo inline"),
("n_header_numerati_normalizzati", _t_normalize_numbered_headings, "normalizzazione livelli numerati"),
(None, _t_normalize_header_levels, "normalizzazione livelli ####→###"),
(None, _t_remove_header_bold, "rimozione bold negli header"),
(None, _t_normalize_allcaps_headers, "normalizzazione ALL-CAPS header"),
# 4. Costruzione struttura
("toc_rimosso", _t_remove_toc),
("n_toc_orfani_rimossi", _t_remove_orphan_toc),
("n_header_allcaps", _t_allcaps_to_headers),
("n_sezioni_numerate", partial(_t_numbered_sections, has_exercises=_has_ex)),
("n_ambienti_matematici", _t_extract_math),
("n_articoli_estratti", _t_extract_articles),
("toc_rimosso", _t_remove_toc, "rimozione TOC"),
("n_toc_orfani_rimossi", _t_remove_orphan_toc, "rimozione voci TOC orfane"),
("n_header_allcaps", _t_allcaps_to_headers, "ALL-CAPS → ##"),
("n_sezioni_numerate", partial(_t_numbered_sections, has_exercises=_has_ex), "sezioni numerate → ###"),
("n_ambienti_matematici", _t_extract_math, "estrazione ambienti matematici"),
("n_articoli_estratti", _t_extract_articles, "estrazione articoli → ###"),
# 5. Testo
("n_paragrafi_uniti", _t_merge_paragraphs),
(None, _t_normalize_whitespace),
(None, _t_collapse_blank_lines),
("n_versi_ripristinati", _t_restore_poetry_lines),
("n_header_verso_demotati", _t_demote_verse_headers),
(None, _t_remove_urls),
("n_paragrafi_uniti", _t_merge_paragraphs, "merge paragrafi spezzati"),
(None, _t_normalize_whitespace, "normalizzazione whitespace"),
(None, _t_collapse_blank_lines, "riduzione righe vuote"),
("n_versi_ripristinati", _t_restore_poetry_lines, "ripristino versi poesia"),
("n_header_verso_demotati", _t_demote_verse_headers, "demozione header-verso"),
("n_url_rimossi", _t_remove_urls, "rimozione URL"),
# 6. Rifinitura
(None, _t_remove_empty_headers),
("n_titoli_uniti", _t_merge_title_headers),
(None, lambda t: (re.sub(r"(?m)^(#{1,6}.+?)\s*\|\s*\d{1,3}\s*$", r"\1", t), 0)),
("n_garbage_headers_rimossi", _t_remove_garbage_headers),
("n_formula_headers_demotati", _t_math_header_demotion),
("n_frontmatter_rimossi", _t_remove_frontmatter),
("n_watermark_rimossi", _t_remove_watermarks),
(None, _t_remove_empty_headers, "rimozione header vuoti"),
("n_titoli_uniti", _t_merge_title_headers, "merge titoli isolati"),
(None, lambda t: (re.sub(r"(?m)^(#{1,6}.+?)\s*\|\s*\d{1,3}\s*$", r"\1", t), 0), "fix header|pagina"),
("n_garbage_headers_rimossi", _t_remove_garbage_headers, "rimozione garbage header"),
("n_formula_headers_demotati", _t_math_header_demotion, "demozione formula-header"),
("n_frontmatter_rimossi", _t_remove_frontmatter, "rimozione frontmatter"),
("n_watermark_rimossi", _t_remove_watermarks, "rimozione watermark"),
]
total = len(_transforms)
stats: dict = {}
for stat_key, fn in _transforms:
for i, (stat_key, fn, label) in enumerate(_transforms, 1):
text, n = fn(text)
if stat_key:
stats[stat_key] = stats.get(stat_key, 0) + n
if on_step:
on_step(i, total, label)
stats["toc_rimosso"] = bool(stats.get("toc_rimosso", 0))
return text, stats
@@ -3,7 +3,7 @@ import re
from collections import Counter
from ._constants import (
_WATERMARK_RE, _TABSEP_RE, _SUPERSCRIPT_RE, _FOOTNOTE_BODY_RE,
_WATERMARK_RE, _TABSEP_RE, _SUPERSCRIPT_RE, _FOOTNOTE_BODY_RE, _DOTLEADER_RE,
)
@@ -47,15 +47,9 @@ def _t_remove_formula_labels(text: str) -> tuple[str, int]:
def _t_remove_dotleaders(text: str) -> tuple[str, int]:
_DOTLEADER_RE = r"^[^\n]*(?:(?:\. ){3,}|\.{4,})[^\n]*$"
n = len(re.findall(_DOTLEADER_RE, text, re.MULTILINE))
text = re.sub(_DOTLEADER_RE, "", text, flags=re.MULTILINE)
text = re.sub(
r"(?m)^(i{1,3}|iv|vi{0,3}|ix|xi{0,2}|x)$",
"",
text,
flags=re.IGNORECASE,
)
n = len(_DOTLEADER_RE.findall(text))
text = _DOTLEADER_RE.sub("", text)
text = re.sub(r"(?m)^(i{1,3}|iv|vi{0,3}|ix|xi{0,2}|x)$", "", text, flags=re.IGNORECASE)
return text, n
@@ -103,4 +97,6 @@ def _t_remove_watermarks(text: str) -> tuple[str, int]:
def _t_remove_urls(text: str) -> tuple[str, int]:
return re.sub(r"(?m)^(https?://|www\.)\S+\s*$", "", text), 0
n = len(re.findall(r"(?m)^(https?://|www\.)\S+\s*$", text))
text = re.sub(r"(?m)^(https?://|www\.)\S+\s*$", "", text)
return text, n
@@ -132,6 +132,7 @@ _WATERMARK_RE = re.compile(
re.IGNORECASE | re.MULTILINE,
)
_TABSEP_RE = re.compile(r"(?m)^\|\s*\|\s*$|^\|---\|?\s*$")
_DOTLEADER_RE = re.compile(r"^[^\n]*(?:(?:\. ){3,}|\.{4,})[^\n]*$", re.MULTILINE)
_FM_RE = re.compile(
r"https?://|www\.|@[A-Za-z]|\bUniversit[àa]\b|\bDipartimento\b|"
r"\bCopyright\b|\bLicenza\b|\bEdizione\b|"
-51
View File
@@ -1,51 +0,0 @@
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}"
-62
View File
@@ -1,62 +0,0 @@
from pathlib import Path
def _is_tagged_pdf(pdf_path: Path) -> bool:
try:
import fitz
doc = fitz.open(str(pdf_path))
tagged = "StructTreeRoot" in doc.pdf_catalog()
doc.close()
return tagged
except Exception:
return False
def convert_pdf(pdf_path: Path, out_dir: Path) -> Path:
"""
Converte il PDF in Markdown tramite opendataloader-pdf.
Scrive il file nella out_dir e restituisce il percorso.
Parametri scelti per output RAG-ottimale:
- keep_line_breaks=False → testo fluente, no hard-wrap PDF
- reading_order="xycut" → corregge ordine multi-colonna (XY-Cut++)
- sanitize=False → preserva il testo originale
- image_output="off" → nessuna immagine estratta né referenziata
- table_method="cluster" → rileva tabelle senza bordi visibili
- content_safety_off → evita filtraggio di footnote (tiny) e layer OCG
- use_struct_tree → attivo solo se il PDF è taggato (Word/InDesign)
"""
import opendataloader_pdf
out_dir.mkdir(parents=True, exist_ok=True)
tagged = _is_tagged_pdf(pdf_path)
opendataloader_pdf.convert(
input_path=str(pdf_path),
output_dir=str(out_dir),
format="markdown",
keep_line_breaks=False,
reading_order="xycut",
sanitize=False,
image_output="off",
table_method="cluster",
content_safety_off=["tiny", "hidden-ocg"],
use_struct_tree=tagged,
quiet=True,
)
md_file = out_dir / f"{pdf_path.stem}.md"
if not md_file.exists():
candidates = list(out_dir.glob("*.md"))
if not candidates:
raise RuntimeError(f"Nessun file .md prodotto in {out_dir}")
md_file = candidates[0]
content = md_file.read_text(encoding="utf-8", errors="replace").strip()
if len(content) < 100:
raise RuntimeError(
f"opendataloader ha prodotto un file .md quasi vuoto ({len(content)} char) "
f"— il PDF potrebbe essere corrotto o non supportato"
)
return md_file
-23
View File
@@ -1,23 +0,0 @@
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)
+138
View File
@@ -0,0 +1,138 @@
"""Estrazione PDF: verifica dipendenze, validazione, conversione → raw Markdown."""
import subprocess
import sys
from pathlib import Path
# ─── Dipendenze ───────────────────────────────────────────────────────────────
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)
# ─── Validazione PDF ──────────────────────────────────────────────────────────
def validate_pdf(pdf_path: Path) -> tuple[bool, str]:
"""Verifica esistenza, leggibilità e presenza di testo digitale 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}"
# ─── Conversione PDF → Markdown ───────────────────────────────────────────────
def _is_tagged_pdf(pdf_path: Path) -> bool:
try:
import fitz
doc = fitz.open(str(pdf_path))
tagged = "StructTreeRoot" in doc.pdf_catalog()
doc.close()
return tagged
except Exception:
return False
def convert_pdf(pdf_path: Path, out_dir: Path) -> Path:
"""
Converte il PDF in Markdown tramite opendataloader-pdf (XY-Cut++).
Parametri per output RAG-ottimale:
keep_line_breaks=False → testo fluente, elimina hard-wrap del PDF
reading_order="xycut" → ricostruisce ordine di lettura multi-colonna
sanitize=False → preserva il testo originale senza filtri
image_output="off" → nessuna immagine estratta né referenziata
table_method="cluster" → rileva tabelle anche senza bordi visibili
content_safety_off → non scarta footnote (tiny) né layer OCG nascosti
use_struct_tree → attivo solo per PDF taggati (Word/InDesign)
"""
import opendataloader_pdf
out_dir.mkdir(parents=True, exist_ok=True)
tagged = _is_tagged_pdf(pdf_path)
opendataloader_pdf.convert(
input_path=str(pdf_path),
output_dir=str(out_dir),
format="markdown",
keep_line_breaks=False,
reading_order="xycut",
sanitize=False,
image_output="off",
table_method="cluster",
content_safety_off=["tiny", "hidden-ocg"],
use_struct_tree=tagged,
quiet=True,
)
md_file = out_dir / f"{pdf_path.stem}.md"
if not md_file.exists():
candidates = list(out_dir.glob("*.md"))
if not candidates:
raise RuntimeError(f"Nessun file .md prodotto in {out_dir}")
md_file = candidates[0]
content = md_file.read_text(encoding="utf-8", errors="replace").strip()
if len(content) < 100:
raise RuntimeError(
f"opendataloader ha prodotto un file .md quasi vuoto ({len(content)} char) "
f"— il PDF potrebbe essere corrotto o non supportato"
)
return md_file
+4 -13
View File
@@ -1,10 +1,10 @@
import json
import re
from collections import Counter
from datetime import datetime
from pathlib import Path
from .structure import _parse_sections_with_body
from ._constants import _MATH_SYMBOLS_RE, _EXERCISE_TRIGGER_RE, _MATH_HDR_RE
def build_report(
@@ -59,26 +59,17 @@ def build_report(
break
return hits
_math_sym_scan = re.compile(
r"[=+∈∀∃≤≥∞∑∫∂→↔⊂⊃∩∪αβγδεζηθικλμνξοπρστυφχψωΑΒΓΔΕΖΗΘΙΚΛΜΝΞΟΠΡΣΤΥΦΧΨΩ]"
)
_ex_trigger_scan = re.compile(
r"\b(Si dimostri|Si calcoli|Si provi|Si trovi|Trovare|Find|Prove|Show that"
r"|Compute|Calculate|Dimostrare|Verificare)\b",
re.IGNORECASE,
)
def _scan_formula_headers(max_n: int = 10) -> list[dict]:
hits = []
for i, line in enumerate(text_lines):
m = re.match(r"^(#{2,3})\s+(.+)$", line)
m = _MATH_HDR_RE.match(line)
if not m:
continue
body = m.group(2)
if len(body) <= 100:
continue
has_math = len(_math_sym_scan.findall(body)) >= 3
has_ex = bool(_ex_trigger_scan.search(body))
has_math = len(_MATH_SYMBOLS_RE.findall(body)) >= 3
has_ex = bool(_EXERCISE_TRIGGER_RE.search(body))
if has_math or has_ex:
hits.append({"riga": i + 1, "testo": line.strip()[:120]})
if len(hits) >= max_n:
+91 -29
View File
@@ -1,16 +1,51 @@
import json
import sys
import tempfile
import threading
import time
from pathlib import Path
from .checker import check_pdf
from .converter import convert_pdf
from .transforms import apply_transforms
from .extract import validate_pdf, convert_pdf
from ._apply import apply_transforms
from .structure import analyze
from .report import build_report
from .validator import _score, _grade
_LIVELLO_DESC = {3: "ricca (h3)", 2: "parziale (h2)", 1: "paragrafi", 0: "testo piatto"}
_SPIN_FRAMES = "⠋⠙⠹⠸⠼⠴⠦⠧⠇⠏"
class _Spinner:
"""Spinner animato in un thread separato — mostra frame + tempo trascorso."""
def __init__(self, prefix: str):
self._prefix = prefix
self._stop = threading.Event()
self._thread = threading.Thread(target=self._run, daemon=True)
self._t0 = 0.0
def __enter__(self):
self._t0 = time.perf_counter()
self._thread.start()
return self
def __exit__(self, *_):
self._stop.set()
self._thread.join()
sys.stdout.write("\r" + " " * 72 + "\r")
sys.stdout.flush()
def _run(self):
i = 0
while not self._stop.wait(0.1):
elapsed = time.perf_counter() - self._t0
frame = _SPIN_FRAMES[i % len(_SPIN_FRAMES)]
sys.stdout.write(f"\r {frame} {self._prefix} {elapsed:.0f}s")
sys.stdout.flush()
i += 1
def run(stem: str, project_root: Path, force: bool) -> bool:
pdf_path = project_root / "sources" / f"{stem}.pdf"
@@ -29,7 +64,9 @@ def run(stem: str, project_root: Path, force: bool) -> bool:
# [1] Validazione
print(" [1/4] Validazione PDF...")
ok, msg = check_pdf(pdf_path)
pdf_mb = pdf_path.stat().st_size / (1024 * 1024) if pdf_path.exists() else 0
print(f" File: {pdf_path.name} ({pdf_mb:.1f} MB)")
ok, msg = validate_pdf(pdf_path)
if not ok:
print(f"{msg}")
return False
@@ -37,47 +74,69 @@ def run(stem: str, project_root: Path, force: bool) -> bool:
# [2] Conversione
print(" [2/4] Conversione PDF → Markdown (opendataloader-pdf)...")
with tempfile.TemporaryDirectory() as tmp:
try:
md_file = convert_pdf(pdf_path, Path(tmp))
except MemoryError:
print(" ✗ Memoria esaurita durante la conversione")
return False
except Exception as e:
print(f" ✗ Conversione fallita: {e}")
return False
try:
raw_text = md_file.read_text(encoding="utf-8")
except UnicodeDecodeError as e:
print(f" ✗ Errore encoding nel file prodotto: {e}")
return False
with _Spinner("opendataloader-pdf in esecuzione...") as spinner:
t0 = time.perf_counter()
with tempfile.TemporaryDirectory() as tmp:
try:
md_file = convert_pdf(pdf_path, Path(tmp))
except MemoryError:
print(" ✗ Memoria esaurita durante la conversione")
return False
except Exception as e:
print(f" ✗ Conversione fallita: {e}")
return False
try:
raw_text = md_file.read_text(encoding="utf-8")
except UnicodeDecodeError as e:
print(f" ✗ Errore encoding nel file prodotto: {e}")
return False
elapsed = time.perf_counter() - t0
size_kb = len(raw_text.encode()) // 1024
n_lines = raw_text.count("\n")
print(f" ✅ Markdown grezzo: {size_kb} KB, {n_lines} righe")
print(f" ✅ Markdown grezzo: {size_kb} KB, {n_lines} righe ({elapsed:.1f}s)")
# [3] Pulizia strutturale
print(" [3/4] Pulizia strutturale...")
clean_text, t = apply_transforms(raw_text)
def _on_step(i: int, total: int, label: str) -> None:
sys.stdout.write(f"\r [{i}/{total}] {label:<45}")
sys.stdout.flush()
clean_text, t = apply_transforms(raw_text, on_step=_on_step)
sys.stdout.write("\r" + " " * 72 + "\r")
sys.stdout.flush()
reduction = 100 * (1 - len(clean_text) / len(raw_text)) if raw_text else 0
print(f"Simboli PUA corretti: {t['n_simboli_pua_corretti']}")
print(f" Immagini rimosse: {t['n_immagini_rimosse']}")
print(f" Note rimosse: {t['n_note_rimosse']}")
print(f"Encoding")
print(f" Simboli PUA corretti: {t['n_simboli_pua_corretti']}")
print(f" Accenti corretti: {t['n_accenti_corretti']}")
print(f" Artefatti")
print(f" Immagini rimosse: {t['n_immagini_rimosse']}")
print(f" <br> rimossi: {t['n_br_rimossi']}")
print(f" Note rimosse: {t['n_note_rimosse']}")
print(f" Dot-leader rimossi: {t['n_dotleader_rimossi']}")
print(f" Righe ricorrenti rim.: {t['n_righe_ricorrenti_rimosse']}")
print(f" URL rimossi: {t['n_url_rimossi']}")
print(f" Watermark rimossi: {t['n_watermark_rimossi']}")
print(f" Header")
print(f" Header concat fixati: {t['n_header_concat_fixati']}")
print(f" Header num. normaliz.: {t['n_header_numerati_normalizzati']}")
print(f" Articoli → ###: {t['n_articoli_estratti']}")
print(f" Ambienti matematici: {t['n_ambienti_matematici']}")
print(f" Titoli header uniti: {t['n_titoli_uniti']}")
print(f" Struttura")
print(f" TOC rimosso: {'' if t['toc_rimosso'] else 'no'}")
print(f" TOC orfani rimossi: {t['n_toc_orfani_rimossi']}")
print(f" Versi poesia riprist.: {t['n_versi_ripristinati']}")
print(f" Header verso demotati: {t['n_header_verso_demotati']}")
print(f" ALL-CAPS → ##: {t['n_header_allcaps']}")
print(f" Sezioni → ###: {t['n_sezioni_numerate']}")
print(f" Ambienti matematici: {t['n_ambienti_matematici']}")
print(f" Articoli → ###: {t['n_articoli_estratti']}")
print(f" Testo")
print(f" Paragrafi uniti: {t['n_paragrafi_uniti']}")
print(f" Versi poesia riprist.: {t['n_versi_ripristinati']}")
print(f" Header verso demotati: {t['n_header_verso_demotati']}")
print(f" Rifinitura")
print(f" Garbage header rim.: {t['n_garbage_headers_rimossi']}")
print(f" Titoli header uniti: {t['n_titoli_uniti']}")
print(f" Formula-hdr demotati: {t['n_formula_headers_demotati']}")
print(f" Frontmatter rimossi: {t['n_frontmatter_rimossi']}")
print(f" Riduzione testo: {reduction:.0f}%")
# [4] Profilo strutturale
@@ -104,8 +163,11 @@ def run(stem: str, project_root: Path, force: bool) -> bool:
for w in profile["avvertenze"]:
print(f" ⚠️ {w}")
build_report(stem, out_dir, clean_text, t, profile, reduction)
report_path = build_report(stem, out_dir, clean_text, t, profile, reduction)
report_data = json.loads(report_path.read_text(encoding="utf-8"))
score, _ = _score(report_data)
print(f"\n Output → conversione/{stem}/")
print(f" raw.md (immutabile) clean.md report.json")
print(f" Punteggio qualità: {score}/100 {_grade(score)}")
return True
@@ -1,4 +0,0 @@
"""Package transforms: pipeline di pulizia strutturale per Markdown RAG."""
from ._apply import apply_transforms
__all__ = ["apply_transforms"]
@@ -1,560 +0,0 @@
# Pipeline ottimizzazione PDF→Markdown — Implementation Plan
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
**Goal:** Eliminare la necessità di revisione manuale del `clean.md` ottimizzando i parametri di opendataloader-pdf e aggiungendo trasformazioni mirate per tutti i tipi di PDF.
**Architecture:** Quattro file modificati: `converter.py` (parametri adattivi + rilevamento PDF taggato), `transforms.py` (PUA bracket TeX + demozione header-formula), `report.py` (nuova metrica residua), `validator.py` (nuova penalità). Nessun cambio all'API pubblica di `_pipeline`.
**Tech Stack:** Python 3.12, opendataloader-pdf (Java), PyMuPDF (fitz), regex
---
## File modificati
| File | Tipo | Responsabilità |
|------|------|----------------|
| `conversione/_pipeline/converter.py` | Modify | `_is_tagged_pdf()` + nuovi parametri convert |
| `conversione/_pipeline/transforms.py` | Modify | PUA bracket TeX + `_t_math_header_demotion` |
| `conversione/_pipeline/report.py` | Modify | `formula_headers_residui` nella sezione residui |
| `conversione/_pipeline/validator.py` | Modify | Penalità formula headers |
---
## Task 1: Converter adattivo — `_is_tagged_pdf()` + nuovi parametri
**Files:**
- Modify: `conversione/_pipeline/converter.py`
- [ ] **Step 1: Leggi il file attuale**
```bash
cat conversione/_pipeline/converter.py
```
- [ ] **Step 2: Sostituisci interamente il contenuto**
Il nuovo `converter.py` aggiunge `_is_tagged_pdf()` (usa fitz per controllare `StructTreeRoot` nel catalog del PDF) e passa i nuovi parametri a `opendataloader_pdf.convert()`:
- `table_method="cluster"` — sempre attivo, migliora tabelle senza bordi
- `content_safety_off=["tiny", "hidden-ocg"]` — evita filtraggio di footnote e layer OCG
- `use_struct_tree=tagged` — attivo solo se PDF è taggato
```python
from pathlib import Path
def _is_tagged_pdf(pdf_path: Path) -> bool:
try:
import fitz
doc = fitz.open(str(pdf_path))
tagged = "StructTreeRoot" in doc.pdf_catalog()
doc.close()
return tagged
except Exception:
return False
def convert_pdf(pdf_path: Path, out_dir: Path) -> Path:
"""
Converte il PDF in Markdown tramite opendataloader-pdf.
Scrive il file nella out_dir e restituisce il percorso.
Parametri scelti per output RAG-ottimale:
- keep_line_breaks=False → testo fluente, no hard-wrap PDF
- reading_order="xycut" → corregge ordine multi-colonna (XY-Cut++)
- sanitize=False → preserva il testo originale
- image_output="off" → nessuna immagine estratta né referenziata
- table_method="cluster" → rileva tabelle senza bordi visibili
- content_safety_off → evita filtraggio di footnote e layer OCG
- use_struct_tree → attivo se PDF è taggato (Word/InDesign)
"""
import opendataloader_pdf
out_dir.mkdir(parents=True, exist_ok=True)
tagged = _is_tagged_pdf(pdf_path)
opendataloader_pdf.convert(
input_path=str(pdf_path),
output_dir=str(out_dir),
format="markdown",
keep_line_breaks=False,
reading_order="xycut",
sanitize=False,
image_output="off",
table_method="cluster",
content_safety_off=["tiny", "hidden-ocg"],
use_struct_tree=tagged,
quiet=True,
)
md_file = out_dir / f"{pdf_path.stem}.md"
if not md_file.exists():
candidates = list(out_dir.glob("*.md"))
if not candidates:
raise RuntimeError(f"Nessun file .md prodotto in {out_dir}")
md_file = candidates[0]
content = md_file.read_text(encoding="utf-8", errors="replace").strip()
if len(content) < 100:
raise RuntimeError(
f"opendataloader ha prodotto un file .md quasi vuoto ({len(content)} char) "
f"— il PDF potrebbe essere corrotto o non supportato"
)
return md_file
```
- [ ] **Step 3: Verifica sintattica**
```bash
.venv/bin/python -c "from conversione._pipeline.converter import convert_pdf, _is_tagged_pdf; print('OK')"
```
Atteso: `OK`
- [ ] **Step 4: Commit**
```bash
git add conversione/_pipeline/converter.py
git commit -m "feat(converter): parametri adattivi — use_struct_tree, cluster tables, content-safety"
```
---
## Task 2: Aggiunta PUA bracket TeX (U+F8EBU+F8FE)
**Files:**
- Modify: `conversione/_pipeline/transforms.py` (sezione `_SYMBOL_PUA_MAP`, righe ~28127)
Questi codepoint sono pezzi di parentesi/bracket grandi del font Computer Modern (TeX), non ricostruibili come singolo simbolo → mappati a `""`.
- [ ] **Step 1: Aggiungi le entries mancanti alla fine di `_SYMBOL_PUA_MAP`**
Individua la riga `"": "", # bracket extension piece (non ricostruibile)` (circa riga 122) e aggiungi **dopo** l'ultima entry esistente della mappa (prima della `}`):
```python
"": "", # TeX large paren left
"": "", # TeX large paren extension
"": "", # TeX large paren right
"": "", # TeX large paren right extension
"": "", # TeX large bracket left
"": "", # TeX large bracket extension
"": "", # TeX brace top-left
"": "", # TeX brace mid
"": "", # TeX brace mid-right
"": "", # TeX brace extension
"": "", # TeX brace right
"": "", # TeX bracket right large
"": "", # TeX bracket right extension
"": "", # TeX bracket right close
"": "", # TeX integral large
"": "", # TeX integral extension
"": "", # TeX integral top
"": "", # TeX radical top
"": "", # TeX radical extension
"": "", # TeX arrowhead
```
- [ ] **Step 2: Verifica che _SYMBOL_PUA_RE si aggiorni automaticamente**
```bash
.venv/bin/python -c "
from conversione._pipeline.transforms import _SYMBOL_PUA_MAP, _SYMBOL_PUA_RE
pua_chars = ['', '', '', '']
for c in pua_chars:
assert c in _SYMBOL_PUA_MAP, f'Manca {repr(c)}'
assert _SYMBOL_PUA_RE.search(c), f'Regex non cattura {repr(c)}'
print(f'OK — {len(_SYMBOL_PUA_MAP)} PUA chars mappati')
"
```
Atteso: `OK — N PUA chars mappati` (N > 90)
- [ ] **Step 3: Verifica sostituzione su testo di esempio**
```bash
.venv/bin/python -c "
from conversione._pipeline.transforms import apply_transforms
testo = 'Sia x = f(n) e n la parentesi grande.'
pulito, stats = apply_transforms(testo)
assert '' not in pulito
assert '' not in pulito
print('Testo pulito:', repr(pulito))
print('PUA corretti:', stats['n_simboli_pua_corretti'])
"
```
Atteso: nessun PUA nel testo pulito, `n_simboli_pua_corretti` > 0.
- [ ] **Step 4: Commit**
```bash
git add conversione/_pipeline/transforms.py
git commit -m "feat(transforms): aggiungi PUA bracket TeX U+F8EB-F8FE alla mappa simboli"
```
---
## Task 3: Nuova trasformazione `_t_math_header_demotion`
**Files:**
- Modify: `conversione/_pipeline/transforms.py`
Demota a testo semplice gli header `##`/`###` che sono enunciati di esercizi o formule lunghe (non titoli di sezione reali).
**Criteri di demozione** (almeno uno tra math e exercise deve valere):
- Livello `##` o `###`
- Lunghezza testo (senza `#`) > 100 caratteri
- `math`: ≥ 3 simboli matematici nell'header (da set: `=`, `+`, `∈`, `∀`, `∃`, `≤`, `≥`, `∞`, `∑`, `∫`, `∂`, `→`, `↔`, `⊂`, `⊃`, `∩`, ``, lettere greche Unicode U+03B1U+03C9 e U+0391U+03A9)
- `exercise`: matcha pattern traccia (`\b(Si dimostri|Si calcoli|Si provi|Si trovi|Trovare|Find|Prove|Show that|Compute|Calculate|Dimostrare|Verificare)\b`)
**Output**: rimuove `#+ `. Se la riga inizia con `N. ` (numero + punto), converte in `**N.** resto`. Altrimenti testo plain.
- [ ] **Step 1: Aggiungi costante regex a livello di modulo** (dopo le costanti esistenti, prima di `_SYMBOL_PUA_MAP`)
Trova la riga `_VERSE_NUM_RE = re.compile(` (circa riga 160) e aggiungi **dopo**:
```python
_MATH_SYMBOLS_RE = re.compile(
r"[=+∈∀∃≤≥∞∑∫∂→↔⊂⊃∩∪αβγδεζηθικλμνξοπρστυφχψωΑΒΓΔΕΖΗΘΙΚΛΜΝΞΟΠΡΣΤΥΦΧΨΩ]"
)
_EXERCISE_TRIGGER_RE = re.compile(
r"\b(Si dimostri|Si calcoli|Si provi|Si trovi|Trovare|Find|Prove|Show that"
r"|Compute|Calculate|Dimostrare|Verificare)\b",
re.IGNORECASE,
)
_MATH_HDR_RE = re.compile(r"^(#{2,3})\s+(.+)$")
_NUMBERED_PREFIX_RE = re.compile(r"^(\d+(?:\.\d+)*[.)])\s+(.+)$", re.DOTALL)
```
- [ ] **Step 2: Aggiungi la funzione `_t_math_header_demotion`** (prima dell'orchestratore `apply_transforms`)
Trova la riga `# ─── Orchestratore` e aggiungi **prima**:
```python
def _t_math_header_demotion(text: str) -> tuple[str, int]:
lines = text.split("\n")
result, count = [], 0
for line in lines:
m = _MATH_HDR_RE.match(line)
if not m:
result.append(line)
continue
body = m.group(2)
if len(body) <= 100:
result.append(line)
continue
has_math = len(_MATH_SYMBOLS_RE.findall(body)) >= 3
has_exercise = bool(_EXERCISE_TRIGGER_RE.search(body))
if not (has_math or has_exercise):
result.append(line)
continue
nm = _NUMBERED_PREFIX_RE.match(body)
if nm:
result.append(f"**{nm.group(1)}** {nm.group(2)}")
else:
result.append(body)
count += 1
return "\n".join(result), count
```
- [ ] **Step 3: Registra la trasformazione in `_transforms`**
Nell'orchestratore `apply_transforms`, trova la riga:
```python
("n_garbage_headers_rimossi", _t_remove_garbage_headers),
```
e aggiungi **dopo**:
```python
("n_formula_headers_demotati", _t_math_header_demotion),
```
- [ ] **Step 4: Aggiungi la stat key al print in `runner.py`**
Trova in `conversione/_pipeline/runner.py` il blocco di print delle statistiche (dopo `apply_transforms`) e aggiungi:
```python
print(f" Formula-hdr demotati: {t['n_formula_headers_demotati']}")
```
- [ ] **Step 5: Verifica su caso sintetico**
```bash
.venv/bin/python -c "
from conversione._pipeline.transforms import apply_transforms
# Caso 1: header esercizio lungo → deve essere demotato
testo = '### 3. Si dimostri la formula per le equazioni di secondo grado ax^2 + bx + c = 0 e si analizzi il segno del discriminante b^2 - 4ac per tutti i valori reali.'
pulito, stats = apply_transforms(testo)
assert '###' not in pulito, f'Header non demotato: {pulito!r}'
print('Caso 1 OK:', pulito[:80])
# Caso 2: header titolo corto → NON deve essere demotato
testo2 = '### Teorema di Cauchy'
pulito2, _ = apply_transforms(testo2)
assert '###' in pulito2, f'Header legittimo demotato: {pulito2!r}'
print('Caso 2 OK:', pulito2)
# Caso 3: header con molti simboli math + lungo → demotato
testo3 = '### Sia f: R→R tale che ∀x∈R si abbia f(x) = ∑_{n=0}^{∞} aₙxⁿ con ∫f dx = g(x) + C per ogni x∈[a,b].'
pulito3, stats3 = apply_transforms(testo3)
print('Caso 3:', '###' not in pulito3, stats3.get('n_formula_headers_demotati'))
print('Stats:', stats.get('n_formula_headers_demotati'))
"
```
Atteso: Caso 1 e 3 demotati, Caso 2 intatto.
- [ ] **Step 6: Commit**
```bash
git add conversione/_pipeline/transforms.py conversione/_pipeline/runner.py
git commit -m "feat(transforms): aggiungi _t_math_header_demotion per header esercizi e formule"
```
---
## Task 4: `report.py` — metrica `formula_headers_residui`
**Files:**
- Modify: `conversione/_pipeline/report.py`
- [ ] **Step 1: Aggiungi funzione di scan formula-header e integrala nel report**
Nella funzione `build_report()`, dopo la definizione di `_scan()` (circa riga 53), aggiungi:
```python
def _scan_formula_headers(max_n: int = 10) -> list[dict]:
_math_sym = re.compile(
r"[=+∈∀∃≤≥∞∑∫∂→↔⊂⊃∩∪αβγδεζηθικλμνξοπρστυφχψωΑΒΓΔΕΖΗΘΙΚΛΜΝΞΟΠΡΣΤΥΦΧΨΩ]"
)
_ex_trigger = re.compile(
r"\b(Si dimostri|Si calcoli|Si provi|Si trovi|Trovare|Find|Prove|Show that"
r"|Compute|Calculate|Dimostrare|Verificare)\b",
re.IGNORECASE,
)
hits = []
for i, line in enumerate(text_lines):
m = re.match(r"^(#{2,3})\s+(.+)$", line)
if not m:
continue
body = m.group(2)
if len(body) <= 100:
continue
has_math = len(_math_sym.findall(body)) >= 3
has_ex = bool(_ex_trigger.search(body))
if has_math or has_ex:
hits.append({"riga": i + 1, "testo": line.strip()[:120]})
if len(hits) >= max_n:
break
return hits
```
- [ ] **Step 2: Aggiungi la metrica ai `residui`**
Trova nel dict `residui` la riga:
```python
"pua_markers": _scan(r'[-]'),
```
e aggiungi **dopo**:
```python
"formula_headers": _scan_formula_headers(),
```
Poi nel dict principale `report["residui"]`, trova la riga:
```python
"pua_markers_esempi": residui["pua_markers"],
```
e aggiungi **dopo**:
```python
"formula_headers": len(residui["formula_headers"]),
"formula_headers_esempi": residui["formula_headers"],
```
- [ ] **Step 3: Verifica**
```bash
.venv/bin/python -c "
import json
from pathlib import Path
from conversione._pipeline.report import build_report
from conversione._pipeline.transforms import apply_transforms
testo = open('conversione/analisi1/raw.md').read()
clean, t = apply_transforms(testo)
from conversione._pipeline.structure import analyze
tmp = Path('/tmp/test_report')
tmp.mkdir(exist_ok=True)
(tmp / 'clean.md').write_text(clean)
profile = analyze(tmp / 'clean.md')
rp = build_report('test', tmp, clean, t, profile, 5.0)
r = json.loads(rp.read_text())
print('formula_headers residui:', r['residui']['formula_headers'])
print('formula_headers esempi:', len(r['residui']['formula_headers_esempi']))
"
```
Atteso: count numerico (può essere 0 se la demozione ha funzionato bene), nessun errore.
- [ ] **Step 4: Commit**
```bash
git add conversione/_pipeline/report.py
git commit -m "feat(report): aggiungi metrica formula_headers_residui"
```
---
## Task 5: `validator.py` — penalità formula headers
**Files:**
- Modify: `conversione/_pipeline/validator.py`
- [ ] **Step 1: Aggiungi la penalità in `_score()`**
Trova in `_score()` la riga:
```python
_pen("pua_markers", 2, 20, "caratteri PUA font Symbol")
```
e aggiungi **dopo**:
```python
_pen("formula_headers", 3, 15, "formula/esercizio come header")
```
- [ ] **Step 2: Aggiungi colonna `fhdr` nell'output tabellare di `validate()`**
Trova in `validate()` la riga che costruisce `header`:
```python
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"
)
```
Sostituiscila con:
```python
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}{'fhdr':>5}"
f"{'med':>6}"
f" {'voto':>4} grade"
)
```
Trova il `print(...)` dentro il loop `for r in rows:` e aggiungi `fhdr`:
```python
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"{res.get('formula_headers', 0):>5}"
f"{dist.get('mediana', 0):>6}"
f" {s:>4} {_grade(s)}"
)
```
Aggiorna anche la riga finale `print("\nColonne: ...")`:
```python
print(
"\nColonne: bare=header vuoti corte=sez<150ch lunghe=sez>1500ch "
"btk=backtick br=<br>inline enc=simboli encoding fhdr=formula-header med=mediana chars\n"
)
```
- [ ] **Step 3: Verifica**
```bash
.venv/bin/python -c "
from conversione._pipeline.validator import _score
r = {'structure': {'livello_struttura': 3}, 'anomalie': {}, 'residui': {'formula_headers': 5}}
score, detail = _score(r)
print(score, detail)
assert any('formula' in d for d in detail), 'Penalità formula non applicata'
print('OK')
"
```
Atteso: penalità `formula/esercizio come header ×5 15` nel detail.
- [ ] **Step 4: Commit**
```bash
git add conversione/_pipeline/validator.py
git commit -m "feat(validator): aggiungi penalità formula_headers, colonna fhdr nel report"
```
---
## Task 6: Test di integrazione su analisi1
- [ ] **Step 1: Riesegui la pipeline su analisi1**
```bash
.venv/bin/python conversione/ --stem analisi1 --force 2>&1
```
Atteso: completamento senza errori, print `Formula-hdr demotati: N` visibile.
- [ ] **Step 2: Valida e confronta con il report precedente**
```bash
.venv/bin/python conversione/ validate analisi1 --detail
```
Confronta con il vecchio voto del `report.json` originale. Il voto deve essere ≥ al precedente.
- [ ] **Step 3: Verifica riduzione PUA bracket**
```bash
python3 -c "
import json
r = json.load(open('conversione/analisi1/report.json'))
pua = r['residui']['pua_markers']
fhdr = r['residui'].get('formula_headers', 'N/A')
print(f'PUA residui: {pua} (era 10+ prima)')
print(f'Formula headers residui: {fhdr}')
"
```
Atteso: `pua_markers` ridotto rispetto al run precedente (era 10 nel report originale).
- [ ] **Step 4: Commit finale se tutto OK**
```bash
git add conversione/analisi1/
git commit -m "chore: rigenera output analisi1 con pipeline ottimizzata"
```
@@ -1,80 +0,0 @@
# Pipeline ottimizzazione — Design Spec
*2026-04-30*
## Obiettivo
Eliminare la necessità di revisione manuale del `clean.md` per tutti i tipi di PDF (accademici/matematici, giuridici, tecnici) ottimizzando i parametri di opendataloader-pdf e aggiungendo trasformazioni mirate.
## Scope
Nessun hybrid backend. Solo Java + trasformazioni Python.
---
## 1. `converter.py` — Parametri adattivi
### 1.1 Rilevamento PDF taggato
Funzione `_is_tagged_pdf(pdf_path) -> bool` usando PyMuPDF (`fitz`):
```python
doc = fitz.open(str(pdf_path))
tagged = "StructTreeRoot" in doc.pdf_catalog()
doc.close()
```
### 1.2 Nuovi parametri fissi (tutti i PDF)
- `table_method="cluster"` — tabelle senza bordi visibili
- `content_safety_off=["tiny", "hidden-ocg"]` — evita filtraggio di footnote e layer OCG
### 1.3 Parametro condizionale
- `use_struct_tree=tagged` — attivo solo se il PDF è taggato
Una sola conversione Java, zero overhead per PDF non taggati.
---
## 2. `transforms.py` — Due aggiunte
### 2.1 PUA bracket TeX (U+F8EBF8F8)
Aggiunge al `_SYMBOL_PUA_MAP` i glifoni bracket di Computer Modern font che appaiono come PUA:
`U+F8EB, U+F8EC, U+F8ED, U+F8EE, U+F8EF, U+F8F0, U+F8F1, U+F8F2, U+F8F3, U+F8F4, U+F8F5, U+F8F6, U+F8F7, U+F8F8, U+F8F9, U+F8FA, U+F8FB, U+F8FC, U+F8FD, U+F8FE`
→ tutti mappati a `""` (pezzi di parentesi non ricostruibili come singolo glifo)
Il `_SYMBOL_PUA_RE` si aggiorna automaticamente essendo costruito dalla mappa.
### 2.2 Nuova trasformazione `_t_math_header_demotion`
Demota a testo semplice gli header `##`/`###` che sono in realtà enunciati di esercizi o formule lunghe.
**Criteri di demozione** (tutti devono valere):
- Livello `##` o `###`
- Lunghezza testo > 100 caratteri
- Almeno uno tra:
- ≥ 3 simboli matematici (`=`, `+`, `∈`, `∀`, `∃`, `≤`, `≥`, `∞`, lettere greche Unicode, `lim`, `sup`, `inf`, `∑`, `∫`)
- Matcha pattern traccia esercizio: `(Si dimostri|Si calcoli|Si provi|Si trovi|Trovare|Find|Prove|Show|Compute|Calculate)\b`
**Output**: rimuove `#+ ` iniziale. Se numerata (`N. testo`), converte in `**N.** testo`. Altrimenti testo plain.
**Posizione in `_transforms`**: gruppo "Rifinitura", dopo `_t_garbage_headers`.
**Stat key**: `n_formula_headers_demotati`
---
## 3. `report.py` — Nuova metrica residua
`build_report()` aggiunge contatore `formula_headers_residui`:
- Conta header `##`/`###` nel `clean.md` finale che superano ancora i criteri math (sopra)
- Mostra fino a 3 esempi in `formula_headers_esempi`
---
## 4. `validator.py` — Nuova penalità
| Problema | Penalità | Cap |
|----------|----------|-----|
| Formula/esercizio come header residuo | 3/cad | 15 |
---
## File modificati
1. `conversione/_pipeline/converter.py``_is_tagged_pdf()` + nuovi parametri
2. `conversione/_pipeline/transforms.py` — PUA map + `_t_math_header_demotion`
3. `conversione/_pipeline/report.py``formula_headers_residui`
4. `conversione/_pipeline/validator.py` — nuova penalità