Create app.py

app.py

@@ -0,0 +1,190 @@
+import os
+import json
+import yaml
+import zipfile
+import time
+from pathlib import Path
+from typing import List, Dict
+import pandas as pd
+import streamlit as st
+from docling.document_converter import DocumentConverter, PdfFormatOption
+from docling.datamodel.base_models import InputFormat
+from docling.datamodel.document import ConversionStatus
+from docling.datamodel.pipeline_options import (
+    PdfPipelineOptions,
+    AcceleratorOptions,
+    AcceleratorDevice,
+    TableStructureOptions,
+    TableFormerMode,
+    EasyOcrOptions,
+    TesseractCliOcrOptions,
+    TesseractOcrOptions,
+    RapidOcrOptions,
+    OcrMacOptions,
+)
+from loguru import logger
+
+# Répertoires de sortie
+OUTPUT_DIR = Path("output")
+OUTPUT_DIR.mkdir(exist_ok=True)
+
+FIGURES_DIR = OUTPUT_DIR / "figures"
+FIGURES_DIR.mkdir(exist_ok=True)
+
+TABLES_DIR = OUTPUT_DIR / "tables"
+TABLES_DIR.mkdir(exist_ok=True)
+
+# Fonction de création de DocumentConverter
+def create_document_converter(
+    use_ocr: bool,
+    export_figures: bool,
+    export_tables: bool,
+    accelerator: str,
+    ocr_engine: str,
+    table_mode: str,
+    ocr_languages: List[str],
+) -> DocumentConverter:
+    accelerator_options = AcceleratorOptions(
+        num_threads=8,
+        device=AcceleratorDevice[accelerator.upper()],
+    )
+
+    table_structure_options = TableStructureOptions(
+        mode=table_mode,
+        do_cell_matching=True,
+    )
+
+    if ocr_engine == "easyocr":
+        ocr_options = EasyOcrOptions(lang=ocr_languages)
+    elif ocr_engine == "tesseract_cli":
+        ocr_options = TesseractCliOcrOptions(lang=ocr_languages)
+    elif ocr_engine == "tesserocr":
+        ocr_options = TesseractOcrOptions(lang=ocr_languages)
+    elif ocr_engine == "rapidocr":
+        ocr_options = RapidOcrOptions(lang=ocr_languages)
+    elif ocr_engine == "ocrmac":
+        ocr_options = OcrMacOptions(lang=ocr_languages)
+    else:
+        raise ValueError(f"Moteur OCR non pris en charge : {ocr_engine}")
+
+    pipeline_options = PdfPipelineOptions(
+        do_ocr=use_ocr,
+        generate_page_images=True,
+        generate_picture_images=export_figures,
+        generate_table_images=export_tables,
+        accelerator_options=accelerator_options,
+        table_structure_options=table_structure_options,
+        ocr_options=ocr_options,
+    )
+
+    return DocumentConverter(
+        allowed_formats=[
+            InputFormat.PDF,
+            InputFormat.DOCX,
+            InputFormat.PPTX,
+            InputFormat.HTML,
+            InputFormat.IMAGE,
+        ],
+        format_options={
+            InputFormat.PDF: PdfFormatOption(pipeline_options=pipeline_options),
+        },
+    )
+
+# Fonction pour créer un fichier ZIP
+def create_zip_file(output_dir: Path, zip_name: str) -> str:
+    zip_path = output_dir / zip_name
+    with zipfile.ZipFile(zip_path, "w") as zipf:
+        for file_path in output_dir.rglob("*"):
+            if file_path.is_file():
+                zipf.write(file_path, arcname=file_path.relative_to(output_dir))
+    return str(zip_path)
+
+# Fonction pour calculer les métriques (exemple : temps d'inférence, nombre de tables, figures, etc.)
+def compute_metrics(conversion_results, start_time):
+    metrics = {
+        "total_documents": len(conversion_results),
+        "successful_conversions": sum(1 for res in conversion_results if res.status == ConversionStatus.SUCCESS),
+        "failed_conversions": sum(1 for res in conversion_results if res.status != ConversionStatus.SUCCESS),
+        "total_time": time.time() - start_time,
+        "tables_extracted": sum(len(res.document.tables) for res in conversion_results if res.status == ConversionStatus.SUCCESS),
+        "figures_extracted": sum(len(res.document.pictures) for res in conversion_results if res.status == ConversionStatus.SUCCESS),
+    }
+    return metrics
+
+# Interface Streamlit
+st.set_page_config(page_title="Docling Dynamic Processor", layout="wide")
+st.title("Docling Dynamic Processor - Analyse et Extraction de Documents")
+
+# Formulaire de configuration
+st.sidebar.header("Options de configuration")
+use_ocr = st.sidebar.checkbox("Activer l'OCR", value=True)
+export_figures = st.sidebar.checkbox("Exporter les figures", value=True)
+export_tables = st.sidebar.checkbox("Exporter les tableaux", value=True)
+accelerator = st.sidebar.selectbox("Accélérateur", ["cpu", "cuda", "mps"], index=0)
+ocr_engine = st.sidebar.selectbox("Moteur OCR", ["easyocr", "tesseract_cli", "tesserocr", "rapidocr", "ocrmac"])
+table_mode = st.sidebar.selectbox("Mode Table", ["accurate", "fast"], index=0)
+ocr_languages = st.sidebar.text_input("Langues OCR (ex: eng, fra)", value="eng").split(",")
+
+# Zone de téléchargement
+uploaded_files = st.file_uploader("Téléchargez vos fichiers (PDF, DOCX, etc.)", type=["pdf", "docx", "pptx"], accept_multiple_files=True)
+
+if st.button("Lancer le traitement") and uploaded_files:
+    # Sauvegarder les fichiers téléchargés
+    input_paths = []
+    for uploaded_file in uploaded_files:
+        file_path = OUTPUT_DIR / uploaded_file.name
+        with open(file_path, "wb") as f:
+            f.write(uploaded_file.read())
+        input_paths.append(file_path)
+
+    # Démarrer le traitement
+    start_time = time.time()
+    converter = create_document_converter(
+        use_ocr,
+        export_figures,
+        export_tables,
+        accelerator,
+        ocr_engine,
+        table_mode,
+        ocr_languages,
+    )
+    conv_results = list(converter.convert_all(input_paths, raises_on_error=False))
+
+    # Traiter les fichiers et collecter les résultats
+    exported_files = {"figures": [], "tables": [], "exports": []}
+    for conv_res in conv_results:
+        if conv_res.status == ConversionStatus.SUCCESS:
+            doc_filename = conv_res.input.file.stem
+
+            # Export des tableaux
+            for table_ix, table in enumerate(conv_res.document.tables):
+                csv_file = OUTPUT_DIR / f"{doc_filename}-table-{table_ix+1}.csv"
+                table.export_to_dataframe().to_csv(csv_file, index=False)
+                exported_files["tables"].append(str(csv_file))
+
+            # Export des figures (sous forme d'images)
+            for fig_ix, figure in enumerate(conv_res.document.pictures):
+                fig_file = FIGURES_DIR / f"{doc_filename}-figure-{fig_ix+1}.png"
+                figure.image.save(fig_file)
+                exported_files["figures"].append(str(fig_file))
+
+    # Générer un ZIP contenant tous les fichiers
+    zip_file = create_zip_file(OUTPUT_DIR, "exported_results.zip")
+
+    # Calcul des métriques
+    metrics = compute_metrics(conv_results, start_time)
+
+    # Afficher les résultats
+    st.success("Traitement terminé!")
+    st.metric("Documents traités avec succès", metrics["successful_conversions"])
+    st.metric("Échecs", metrics["failed_conversions"])
+    st.metric("Temps total (s)", f"{metrics['total_time']:.2f}")
+    st.metric("Total des tableaux extraits", metrics["tables_extracted"])
+    st.metric("Total des figures extraites", metrics["figures_extracted"])
+
+    st.download_button(
+        label="Télécharger tous les résultats (ZIP)",
+        data=open(zip_file, "rb").read(),
+        file_name="exported_results.zip",
+        mime="application/zip",
+    )