feat(pdf-to-md): sostituisci report.md con report.json + validate.py

pipeline.py produce conversione/<stem>/report.json invece di
structure_profile.json + report.md. Il JSON contiene tutto:
trasformazioni, struttura, distribuzione lunghezze sezioni,
anomalie (bare_headers, short/long sections) e residui con esempi.

Fix: bare_headers flagga solo header senza corpo < 30 chars;
header numerati con corpo lungo (aforismi) non sono anomalie.

Nuovo validate.py legge tutti i report.json e stampa tabella
di stato per ogni stem ( / ⚠️ / ) con soglie configurabili.

README aggiornato con sezione validazione batch e struttura report.json.
This commit is contained in:
2026-04-16 15:53:46 +02:00
parent b8d3c459c8
commit e06b65272b
3 changed files with 338 additions and 17 deletions
+128 -4
View File
@@ -30,6 +30,7 @@ import re
import subprocess
import sys
import tempfile
from datetime import datetime
from pathlib import Path
@@ -563,14 +564,136 @@ def analyze(md_path: Path) -> dict:
}
# ─── Report di conversione ───────────────────────────────────────────────────
def build_report(
stem: str,
out_dir: Path,
clean_text: str,
t_stats: dict,
profile: dict,
reduction: float,
) -> Path:
"""
Genera conversione/<stem>/report.json con tutte le metriche di qualità:
statistiche trasformazioni, struttura, distribuzione lunghezze, anomalie
e problemi residui. Leggibile da validate.py per la validazione batch.
"""
text_lines = clean_text.split("\n")
# ── Raccolta sezioni ### con corpo ────────────────────────────────────
sections: list[tuple[str, str]] = []
cur_hdr: str | None = None
cur_body: list[str] = []
for line in text_lines:
if re.match(r"^### ", line):
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()))
lengths = [len(body) for _, body in sections]
# ── Distribuzione lunghezze ───────────────────────────────────────────
def _pct(data: list[int], p: float) -> int:
if not data:
return 0
s = sorted(data)
return s[max(0, min(len(s) - 1, int(len(s) * p)))]
distribution = {
"min": min(lengths) if lengths else 0,
"p25": _pct(lengths, 0.25),
"mediana": _pct(lengths, 0.50),
"p75": _pct(lengths, 0.75),
"max": max(lengths) if lengths else 0,
}
# ── Anomalie ──────────────────────────────────────────────────────────
# Header solo-numero senza corpo sostanziale: anomalia solo se il corpo
# è vuoto o < 30 chars. Un body lungo è una sezione numerata legittima
# (es. aforismi numerati dove il numero è l'identificatore della sezione).
bare_hdrs = [
{"header": hdr, "corpo_inizio": body[:120].replace("\n", " ")}
for hdr, body in sections
if re.match(r"^### \d+\.\s*$", hdr) and len(body.strip()) < 30
]
short_secs = [
{"header": hdr, "chars": length, "testo": body[:80].replace("\n", " ")}
for (hdr, body), length in zip(sections, lengths)
if 0 < length < 150
]
long_secs = [
{"header": hdr, "chars": length}
for (hdr, _), length in zip(sections, lengths)
if length > 1500
]
# ── Problemi residui (max 10 esempi ciascuno) ─────────────────────────
def _scan(pattern: str, max_n: int = 10) -> list[dict]:
hits = []
for i, line in enumerate(text_lines):
if re.search(pattern, line) and not re.match(r"^#+ ", line):
hits.append({"riga": i + 1, "testo": line.strip()[:120]})
if len(hits) >= max_n:
break
return hits
residui = {
"backtick": _scan(r"`"),
"dotleader": _scan(r"(?:\. ){3,}"),
"url": _scan(r"^(https?://|www\.)\S+"),
"immagini": _scan(r"!\[[^\]]*\]\([^)]*\)"),
}
# ── Composizione report ───────────────────────────────────────────────
report = {
"stem": stem,
"timestamp": datetime.now().strftime("%Y-%m-%d %H:%M"),
"transforms": {
**t_stats,
"riduzione_pct": round(reduction),
},
"structure": profile,
"distribution": distribution,
"anomalie": {
"bare_headers": len(bare_hdrs),
"short_sections": len(short_secs),
"long_sections": len(long_secs),
"bare_headers_list": bare_hdrs,
"short_sections_list": short_secs,
"long_sections_list": long_secs,
},
"residui": {
"backtick": len(residui["backtick"]),
"dotleader": len(residui["dotleader"]),
"url": len(residui["url"]),
"immagini": len(residui["immagini"]),
"backtick_esempi": residui["backtick"],
"dotleader_esempi": residui["dotleader"],
"url_esempi": residui["url"],
"immagini_esempi": residui["immagini"],
},
}
report_path = out_dir / "report.json"
report_path.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
return report_path
# ─── Pipeline principale ──────────────────────────────────────────────────────
def run(stem: str, project_root: Path, force: bool) -> bool:
pdf_path = project_root / "sources" / f"{stem}.pdf"
out_dir = project_root / "conversion" / stem
out_dir = project_root / "conversione" / stem
raw_out = out_dir / "raw.md"
clean_out = out_dir / "clean.md"
profile_out = out_dir / "structure_profile.json"
print(f"\n{'' * 52}")
print(f" {stem}")
@@ -623,7 +746,6 @@ def run(stem: str, project_root: Path, force: bool) -> bool:
raw_out.write_text(raw_text, encoding="utf-8")
clean_out.write_text(clean_text, encoding="utf-8")
profile = analyze(clean_out)
profile_out.write_text(json.dumps(profile, ensure_ascii=False, indent=2), encoding="utf-8")
_LIVELLO_DESC = {3: "ricca (h3)", 2: "parziale (h2)", 1: "paragrafi", 0: "testo piatto"}
print(f" ✅ Struttura: livello {profile['livello_struttura']}{_LIVELLO_DESC[profile['livello_struttura']]}")
@@ -634,10 +756,12 @@ 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_stats, profile, reduction)
print(f"\n Output:")
print(f" conversione/{stem}/raw.md (immutabile)")
print(f" conversione/{stem}/clean.md")
print(f" conversione/{stem}/structure_profile.json")
print(f" conversione/{stem}/report.json")
print(f"\n clean.md pronto per la suddivisione in chunk.")
return True