davide/rag-from-scratch
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(conversione): 7 nuovi transform pipeline, refactor validate — media 92→99/100

Browse Source 
- dot-leader continui, strip "- " in allcaps, backtick orfani LaTeX
- TOC list removal, extract_article_headers, extract_math_environments, merge_title_headers
- validate.py: interfaccia semplificata, rimosso codice morto
This commit is contained in:
Davide Grilli
2026-04-17 07:47:56 +02:00
parent bcf2e688aa
commit 265ac92b6c
2 changed files with 271 additions and 173 deletions
Download Patch File Download Diff File Expand all files Collapse all files
conversione/pipeline.py
+197 -9
View File
@@ -169,7 +169,9 @@ def _is_allcaps_line(line: str) -> bool:
def _allcaps_to_header(raw_line: str) -> str:
text = raw_line.strip().rstrip(".").rstrip("?").strip()
# Rimuovi eventuale prefisso di lista "- " o "* " prima di creare l'header
text = re.sub(r"^[-*+]\s+", "", raw_line.strip())
text = text.rstrip(".").rstrip("?").strip()
_ORD_IT_PAT = "|".join(_ORDINALS_IT.keys())
m = re.match(rf"^CAPITOLO ({_ORD_IT_PAT})\. (.+)", text)
@@ -192,6 +194,152 @@ def _allcaps_to_header(raw_line: str) -> str:
return f"## {_sentence_case(text)}"
def _extract_math_environments(text: str) -> tuple[str, int]:
"""
Converte paragrafi che iniziano con ambienti matematici in header ###.
'Teorema 1.6.3 (principio di induzione) Sia A ⊆ N...'
→ '### Teorema 1.6.3 (principio di induzione)\n\nSia A ⊆ N...'
Riconosce: Definizione, Teorema, Lemma, Proposizione, Corollario,
Osservazione, Nota, Esempio (solo con numero di sezione).
Non tocca paragrafi che già iniziano con un header Markdown.
Deve girare PRIMA del merge paragrafi (step 5) per sfruttare i blocchi intatti.
"""
_ENVS = (
r"Definizione|Teorema|Lemma|Proposizione|"
r"Corollario|Osservazione|Nota|Esempio"
)
count = 0
blocks = text.split("\n\n")
result = []
for block in blocks:
stripped = block.strip()
if not stripped or stripped.startswith("#"):
result.append(block)
continue
m = re.match(
rf"^({_ENVS})\s+((?:\d+\.?){{1,4}})\s*(.*)",
stripped,
re.DOTALL,
)
if not m:
result.append(block)
continue
env = m.group(1)
num = m.group(2).rstrip(".")
rest = m.group(3).strip()
# Titolo opzionale tra parentesi: "(principio di induzione)"
title_m = re.match(r"^(\([^)]{2,60}\))\s+(.*)", rest, re.DOTALL)
if title_m:
header = f"### {env} {num} {title_m.group(1)}"
body = title_m.group(2).strip()
else:
header = f"### {env} {num}."
body = rest
result.append(f"{header}\n\n{body}" if body else header)
count += 1
return "\n\n".join(result), count
def _merge_title_headers(text: str) -> tuple[str, int]:
"""
Fonde header numerici isolati con il sottotitolo breve che li segue.
'### N.\n\nSottotitolo (riga singola ≤ 80 char, senza punto finale)'
→ '### N. Sottotitolo'
Caso tipico: parti di un'opera (es. Nietzsche) dove il numero di sezione
e il titolo della sezione sono in blocchi Markdown separati.
Non tocca header con titolo già inline né header seguiti da testo lungo.
"""
count = 0
blocks = re.split(r"\n{2,}", text)
result = []
i = 0
while i < len(blocks):
block = blocks[i]
stripped = block.strip()
if (
re.match(r"^#{2,3} \d+\.\s*$", stripped)
and i + 1 < len(blocks)
):
nxt = blocks[i + 1].strip()
# Sottotitolo valido: riga singola, ≤ 80 char, non header, non numerazione pura
if (
nxt
and "\n" not in nxt
and len(nxt) <= 80
and not nxt.startswith("#")
and not re.match(r"^\d+[\.\)]\s", nxt)
):
result.append(stripped.rstrip() + " " + nxt)
count += 1
i += 2
continue
result.append(block)
i += 1
return re.sub(r"\n{3,}", "\n\n", "\n\n".join(result)), count
def _extract_article_headers(text: str) -> tuple[str, int]:
"""
Converte voci di articolo dal formato lista Markdown al formato header ###.
'- Art. N[suffix]. Titolo. Corpo testo...' → '### Art. N[suffix]. Titolo.\n\nCorpo testo...'
'- Art. N[suffix]. (…) (1)' → '### Art. N[suffix].\n\n(…) (1)'
Gestisce suffissi come: Art. 4-bis., Art. 14-ter., Art. 1-quinquies.
Il titolo è la prima frase con iniziale maiuscola che termina con '.' prima di
ulteriore testo (es. "Leggi. La formazione..." → titolo "Leggi", corpo "La formazione...").
Se il testo non ha titolo separabile, tutto diventa il corpo.
"""
count = 0
def _repl(m: re.Match) -> str:
nonlocal count
num = m.group(1)
rest = m.group(2).strip()
# Titolo: frase con iniziale maiuscola, max 75 char, termina con '.',
# seguita da almeno un'altra frase (minimo 5 char) che inizia con maiuscola
# o con '(' / cifra (note a piè o continuazione corpo).
title_m = re.match(
r"^([A-ZÀÈÉÌÍÒÓÙÚ].{1,74}?)\.\s+([A-ZÀÈÉÌÍÒÓÙÚ\(\d].{4,})",
rest,
)
if title_m:
count += 1
return (
f"### Art. {num}. {title_m.group(1)}.\n\n"
f"{title_m.group(2).strip()}"
)
# Nessun titolo separabile: tutto è corpo
if rest:
count += 1
return f"### Art. {num}.\n\n{rest}"
# Articolo senza testo inline (es. "- Art. 5. (…) (1)" già estratto sopra,
# oppure articolo vuoto nella lista)
count += 1
return f"### Art. {num}."
text = re.sub(
r"^-\s+Art\.\s+([\d]+[a-z\-]*)\.\s*(.*)",
_repl,
text,
flags=re.MULTILINE,
)
return text, count
def apply_transforms(text: str) -> tuple[str, dict]:
"""
Applica le trasformazioni strutturali al Markdown grezzo.
@@ -203,6 +351,9 @@ def apply_transforms(text: str) -> tuple[str, dict]:
"n_accenti_corretti": 0,
"n_dotleader_rimossi": 0,
"n_header_concat_fixati": 0,
"n_articoli_estratti": 0,
"n_ambienti_matematici": 0,
"n_titoli_uniti": 0,
"n_header_allcaps": 0,
"n_sezioni_numerate": 0,
"n_paragrafi_uniti": 0,
@@ -224,13 +375,23 @@ def apply_transforms(text: str) -> tuple[str, dict]:
text = re.sub(r"([eEaAuUiIoO])`", lambda m: _ACCENT_MAP[m.group(1)], text)
stats["n_accenti_corretti"] = n_bt_before - text.count("`")
# Backtick orfani: artefatti LaTeX rimasti dopo la correzione vocale
# (es. "propriet`" da "proprietà", "continuit`" da "continuità").
# In testi PDF non esistono backtick legittimi → rimozione sicura.
n_bt_orfani = text.count("`")
if n_bt_orfani:
text = re.sub(r"`", "", text)
stats["n_accenti_corretti"] += n_bt_orfani
# 0b_pre. Rimuovi righe con dot-leader (voci di indice/sommario)
# Esempi: "- 1.1 Alfabeto greco . . . . . . 1", "3.4 Continuità . . . . 205"
# Pattern: almeno 3 occorrenze di ". " consecutive nella riga
# Cattura sia ". . . ." (spazi) sia "......." (punti continui, tipici dei TOC PDF)
_DOTLEADER_RE = r"^[^\n]*(?:(?:\. ){3,}|\.{4,})[^\n]*$"
stats["n_dotleader_rimossi"] = len(
re.findall(r"^[^\n]*(?:\. ){3,}[^\n]*$", text, re.MULTILINE)
re.findall(_DOTLEADER_RE, text, re.MULTILINE)
)
text = re.sub(r"^[^\n]*(?:\. ){3,}[^\n]*$", "", text, flags=re.MULTILINE)
text = re.sub(_DOTLEADER_RE, "", text, flags=re.MULTILINE)
# 0b_pre2. Rimuovi righe che sono solo numerali romani (indicatori di pagina TOC)
# Esempi: "i", "ii", "iii", "iv", "v" su riga isolata (footer pagine indice LaTeX)
@@ -306,6 +467,12 @@ def apply_transforms(text: str) -> tuple[str, dict]:
flags=re.MULTILINE,
)
# 0e. Converti voci articolo "- Art. N. Titolo. Corpo" → "### Art. N. Titolo.\n\nCorpo"
# Eseguito dopo la promozione h4+ → h3 (0d) per non duplicare Art. già header.
# Eseguito prima del merge paragrafi (5): il boundary ### previene la fusione.
text, n_art = _extract_article_headers(text)
stats["n_articoli_estratti"] = n_art
# 1. Rimuovi **bold** negli header esistenti: ## **Titolo** → ## Titolo
text = re.sub(
r"^(#{1,6})\s+\*\*(.+?)\*\*\s*$",
@@ -324,18 +491,26 @@ def apply_transforms(text: str) -> tuple[str, dict]:
text = re.sub(r"^(#{1,6}) (.+)$", _norm_allcaps_header, text, flags=re.MULTILINE)
# 2. Rimuovi righe TOC: header "# Indice", "# Contents", ecc.
# Rimuove la riga stessa; le voci subordinate (dot-leader) sono già rimosse da 0b_pre.
# L'header rimasto senza corpo viene poi eliminato dal transform 9.
# + le voci lista numeriche che seguono (TOC senza dot-leader, es. Nietzsche):
# "- 1. Dei pregiudizi dei filosofi" → rimossa se viene subito dopo un header TOC.
# Le voci con dot-leader sono già rimosse da 0b_pre.
# Gli header rimasti senza corpo vengono poi eliminati dal transform 9.
lines = text.split("\n")
new_lines = []
_in_toc = False
for line in lines:
# Stripping del prefisso markdown (##, #, ecc.) prima del confronto keyword
bare = re.sub(r"^#+\s*", "", line.strip())
bare = re.sub(r"^#+\s*", "", line.strip())
first_word = bare.split(".")[0].strip().lower()
if first_word in _TOC_KEYWORDS:
stats["toc_rimosso"] = True
else:
new_lines.append(line)
_in_toc = True
continue
if _in_toc:
# Salta righe vuote e voci lista numeriche (- N. Titolo / - N Titolo)
if re.match(r"^\s*$", line) or re.match(r"^\s*[-*+]\s+\d", line):
continue
_in_toc = False
new_lines.append(line)
text = "\n".join(new_lines)
# 3. Converti righe ALL-CAPS standalone → ## header
@@ -419,6 +594,11 @@ def apply_transforms(text: str) -> tuple[str, dict]:
flags=re.MULTILINE,
)
# 4d. Converti ambienti matematici (Teorema/Definizione/...) → ### header
# Eseguito prima del merge paragrafi (5) per sfruttare i blocchi intatti.
text, n_math = _extract_math_environments(text)
stats["n_ambienti_matematici"] = n_math
# 5. Unisci paragrafi spezzati da salti pagina PDF
_SENTENCE_END = set(".?!»)\"'")
blocks = text.split("\n\n")
@@ -470,6 +650,11 @@ def apply_transforms(text: str) -> tuple[str, dict]:
cleaned.append(block)
text = re.sub(r"\n{3,}", "\n\n", "\n\n".join(cleaned))
# 9b. Fondi header numerici isolati con il sottotitolo breve successivo
# "### N.\n\nSottotitolo" → "### N. Sottotitolo" (es. parti Nietzsche)
text, n_titoli = _merge_title_headers(text)
stats["n_titoli_uniti"] = n_titoli
return text, stats
@@ -734,6 +919,9 @@ def run(stem: str, project_root: Path, force: bool) -> bool:
print(f" Accenti corretti: {t_stats['n_accenti_corretti']}")
print(f" Dot-leader rimossi: {t_stats['n_dotleader_rimossi']}")
print(f" Header concat fixati: {t_stats['n_header_concat_fixati']}")
print(f" Articoli → ###: {t_stats['n_articoli_estratti']}")
print(f" Ambienti matematici: {t_stats['n_ambienti_matematici']}")
print(f" Titoli header uniti: {t_stats['n_titoli_uniti']}")
print(f" TOC rimosso: {'sì' if t_stats['toc_rimosso'] else 'no'}")
print(f" ALL-CAPS → ##: {t_stats['n_header_allcaps']}")
print(f" Sezioni → ###: {t_stats['n_sezioni_numerate']}")
conversione/validate.py
+74 -164
View File
@@ -1,12 +1,10 @@
#!/usr/bin/env python3
"""
conversione/validate.py — Validazione batch di tutti gli stem convertiti
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 per misurare la bontà del
Markdown prodotto.
e assegna un voto (0-100) a ogni documento.
Voto:
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
@@ -16,57 +14,54 @@ Voto:
Uso:
python conversione/validate.py # tutti gli stem
python conversione/validate.py analisi1 # stem specifico
python conversione/validate.py --stem analisi1
python conversione/validate.py --analisi1 # compatibilità
python conversione/validate.py a b c # stem multipli
"""
import json
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) -> int:
"""
Calcola un punteggio 0-100 sulla qualità del Markdown prodotto.
Penalità:
- struttura assente o piatta → -40 / -15
- backtick residui nel testo → -2 per occorrenza (max -30)
- URL / watermark residui → -5 per occorrenza (max -15)
- immagini residue → -5 per occorrenza (max -10)
- dot-leader residui → -5 per occorrenza (max -10)
- header senza titolo (bare) → -3 per occorrenza (max -15)
- troppe sezioni > 1500 chars → -5 / -10 (in % sul totale h3)
struttura assente / piatta → −40 / −15
backtick residui → −2/cad (max −30)
URL / watermark → −5/cad (max −15)
immagini residue → −5/cad (max −10)
dot-leader residui → −5/cad (max −10)
bare headers → −3/cad (max −15)
sezioni >1500ch >35/60% → −5 / −10
"""
score = 100
score = 100
structure = r.get("structure", {})
anomalie = r.get("anomalie", {})
residui = r.get("residui", {})
anomalie = r.get("anomalie", {})
residui = r.get("residui", {})
livello = structure.get("livello_struttura", 0)
n_h3 = max(structure.get("n_h3", 0), 1)
livello = structure.get("livello_struttura", 0)
n_h3 = max(structure.get("n_h3", 0), 1)
# Struttura
if livello == 0:
score -= 40
elif livello == 1:
score -= 15
# Residui nel testo
score -= min(30, residui.get("backtick", 0) * 2)
score -= min(15, residui.get("url", 0) * 5)
score -= min(10, residui.get("immagini", 0) * 5)
score -= min(10, residui.get("dotleader", 0) * 5)
# Anomalie strutturali
score -= min(15, anomalie.get("bare_headers", 0) * 3)
# Sezioni troppo lunghe (in % sul totale delle sezioni ###)
long_ratio = anomalie.get("long_sections", 0) / n_h3
if long_ratio > 0.6:
if long_ratio > 0.60:
score -= 10
elif long_ratio > 0.35:
score -= 5
@@ -75,82 +70,7 @@ def _score(r: dict) -> int:
def _grade(score: int) -> str:
if score >= 90: return "A"
if score >= 75: return "B"
if score >= 60: return "C"
if score >= 40: return "D"
return "F"
# ─── CLI ─────────────────────────────────────────────────────────────────────
def _normalize_target(token: str) -> str:
"""
Normalizza un target CLI in stem:
- analisi1
- --analisi1 (compatibilità)
- conversione/analisi1/report.json
- analisi1.pdf / analisi1.md / report.json
"""
raw = token.strip()
if not raw:
return raw
# Compatibilità con invocazione tipo: --analisi1
if raw.startswith("--") and len(raw) > 2:
raw = raw[2:]
p = Path(raw)
# Path diretto al report
if p.name == "report.json" and p.parent.name:
return p.parent.name
name = p.name
if name.endswith((".pdf", ".md", ".json")):
name = Path(name).stem
return name
def _parse_cli_args(argv: list[str]) -> list[str]:
parser = argparse.ArgumentParser(
description="Valida i report Markdown prodotti in conversione/<stem>/report.json"
)
parser.add_argument(
"targets",
nargs="*",
help="Stem, file o path da validare (es: analisi1 oppure conversione/analisi1/report.json)",
)
parser.add_argument(
"-s",
"--stem",
action="append",
default=[],
help="Stem specifico (ripetibile, es: --stem analisi1 --stem nietzsche)",
)
args, unknown = parser.parse_known_args(argv)
targets = [*args.targets, *args.stem]
# Compatibilità: `python validate.py --analisi1`
for tok in unknown:
if tok.startswith("--") and len(tok) > 2:
targets.append(tok[2:])
else:
parser.error(f"Argomento non riconosciuto: {tok}")
stems = []
seen = set()
for t in targets:
stem = _normalize_target(t)
if not stem or stem in seen:
continue
seen.add(stem)
stems.append(stem)
return stems
return next(g for threshold, g in _GRADES if score >= threshold)
# ─── Validazione ─────────────────────────────────────────────────────────────
@@ -158,100 +78,90 @@ def _parse_cli_args(argv: list[str]) -> list[str]:
def validate(stems: list[str], project_root: Path) -> None:
conv_dir = project_root / "conversione"
if stems:
paths = [conv_dir / s / "report.json" for s in stems]
else:
paths = sorted(conv_dir.glob("*/report.json"))
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 = []
for path in paths:
if not path.exists():
rows.append({"stem": path.parent.name, "_missing": True})
continue
r = json.loads(path.read_text(encoding="utf-8"))
rows.append(r)
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_stem = max(len(r.get("stem", "stem")) for r in rows) + 2
col = max(len(r.get("stem", "stem")) for r in rows) + 2
header = (
f"{'stem':<{col_stem}}"
f"{'stem':<{col}}"
f"{'h2':>4}{'h3':>5} "
f"{'strategia':<20}"
f"{'bare':>5}{'corte':>6}{'lunghe':>7}"
f"{'backtick':>9}{'dotlead':>8}{'url':>4}"
f" {'voto':>4} {'grade'}"
f" {'voto':>4} grade"
)
sep = "─" * len(header)
print()
print(header)
print(sep)
print(f"\n{header}\n{sep}")
scores = []
scored_docs = []
# ── Righe ─────────────────────────────────────────────────────────────
for r in rows:
if r.get("_missing"):
print(f"{r['stem']:<{col_stem}} (report.json non trovato)")
print(f"{r['stem']:<{col}} (report.json non trovato)")
continue
stem = r.get("stem", "?")
structure = r.get("structure", {})
anomalie = r.get("anomalie", {})
residui = r.get("residui", {})
h2 = structure.get("n_h2", 0)
h3 = structure.get("n_h3", 0)
strat = structure.get("strategia_chunking", "?")
bare = anomalie.get("bare_headers", 0)
corte = anomalie.get("short_sections", 0)
lunghe = anomalie.get("long_sections", 0)
backtick = residui.get("backtick", 0)
dotlead = residui.get("dotleader", 0)
url = residui.get("url", 0)
s = _score(r)
g = _grade(s)
st = r.get("structure", {})
an = r.get("anomalie", {})
res = r.get("residui", {})
s = _score(r)
scores.append(s)
scored_docs.append((stem, s, g))
print(
f"{stem:<{col_stem}}"
f"{h2:>4}{h3:>5} "
f"{strat:<20}"
f"{bare:>5}{corte:>6}{lunghe:>7}"
f"{backtick:>9}{dotlead:>8}{url:>4}"
f" {s:>4} {g}"
f"{r['stem']:<{col}}"
f"{st.get('n_h2', 0):>4}"
f"{st.get('n_h3', 0):>5} "
f"{st.get('strategia_chunking','?'):<20}"
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):>9}"
f"{res.get('dotleader', 0):>8}"
f"{res.get('url', 0):>4}"
f" {s:>4} {_grade(s)}"
)
# ── Riepilogo ─────────────────────────────────────────────────────────
print(sep)
if scores:
media = sum(scores) / len(scores)
grade_media = _grade(int(media))
print(f"Documenti: {len(scores)} "
f"Voto medio: {media:.0f}/100 {grade_media} "
f"(A≥90 B≥75 C≥60 D≥40 F<40)")
if len(scored_docs) == 1:
stem, score, grade = scored_docs[0]
print(f"Voto finale Markdown ({stem}): {score}/100 {grade}")
else:
voti = ", ".join(
f"{stem}={score}/100 {grade}"
for stem, score, grade in scored_docs
)
print(f"Voti Markdown: {voti}")
print()
print("Penalità: struttura assente −40, backtick residui −2/cad, "
"bare headers −3/cad, sezioni >1500ch >35% −5")
print()
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(
"\nPenalità: struttura assente −40, backtick −2/cad, "
"bare headers −3/cad, sezioni >1500ch >35% −5\n"
)
# ─── Entry point ─────────────────────────────────────────────────────────────
if __name__ == "__main__":
project_root = Path(__file__).parent.parent
stems = _parse_cli_args(sys.argv[1:])
validate(stems, project_root)
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.",
)
args = parser.parse_args()
validate(args.stems, Path(__file__).parent.parent)