rag-from-scratch/docs/superpowers/plans/2026-04-30-pipeline-ottimizzazione.md

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

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

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

.venv/bin/python -c "from conversione._pipeline.converter import convert_pdf, _is_tagged_pdf; print('OK')"

Atteso: OK

• Step 4: Commit

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+F8EB–U+F8FE)

Files:

• Modify: conversione/_pipeline/transforms.py (sezione _SYMBOL_PUA_MAP, righe ~28–127)

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 }):

    "": "",  # 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

.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

.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

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+03B1–U+03C9 e U+0391–U+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:

_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:

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:

        ("n_garbage_headers_rimossi",      _t_remove_garbage_headers),

e aggiungi dopo:

        ("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:

    print(f"     Formula-hdr demotati:  {t['n_formula_headers_demotati']}")

• Step 5: Verifica su caso sintetico

.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

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:

    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:

        "pua_markers":      _scan(r'[-]'),

e aggiungi dopo:

        "formula_headers":  _scan_formula_headers(),

Poi nel dict principale report["residui"], trova la riga:

            "pua_markers_esempi":         residui["pua_markers"],

e aggiungi dopo:

            "formula_headers":            len(residui["formula_headers"]),
            "formula_headers_esempi":     residui["formula_headers"],

• Step 3: Verifica

.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

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:

    _pen("pua_markers",      2, 20, "caratteri PUA font Symbol")

e aggiungi dopo:

    _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:

    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:

    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:

        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: ..."):

    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

.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

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

.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

.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

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

git add conversione/analisi1/
git commit -m "chore: rigenera output analisi1 con pipeline ottimizzata"