rag-from-scratch/conversione/_pipeline/extract.py

"""Estrazione PDF: verifica dipendenze, validazione, metadati, conversione → raw Markdown."""
import re
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}"


# ─── Metadati PDF ────────────────────────────────────────────────────────────

def extract_metadata(pdf_path: Path) -> dict:
    """
    Estrae title, author, year e page count dal PDF tramite fitz.
    Restituisce un dict con chiavi sempre presenti (stringa vuota se assenti).
    """
    try:
        import fitz
        doc  = fitz.open(str(pdf_path))
        meta = doc.metadata
        pages = len(doc)
        doc.close()

        def _clean(s: str) -> str:
            return s.strip() if s else ""

        year = ""
        creation = meta.get("creationDate", "")
        m = re.match(r"D:(\d{4})", creation)
        if m:
            year = m.group(1)

        return {
            "source": pdf_path.name,
            "title":  _clean(meta.get("title",  "")),
            "author": _clean(meta.get("author", "")),
            "year":   year,
            "pages":  pages,
        }
    except Exception:
        return {"source": pdf_path.name, "title": "", "author": "", "year": "", "pages": 0}


# ─── 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)
      markdown_page_separator       → inserisce separatore + marker pagina tra pagine
      replace_invalid_chars         → sostituisce caratteri non validi con spazio
    """
    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,
        markdown_page_separator="\n\n---\n<!-- page: %page-number% -->\n\n",
        replace_invalid_chars=" ",
        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