import gradio as gr import os import json import shutil from pathlib import Path import google.generativeai as genai from huggingface_hub import HfApi, create_repo import tempfile import uuid from PIL import Image import io import hashlib from typing import List, Dict, Set import pdf2image import fitz # PyMuPDF from dotenv import load_dotenv # Charger les variables d'environnement load_dotenv() # Récupérer les clés API depuis les variables d'environnement DEFAULT_GEMINI_API_KEY = os.getenv("GEMINI_API_KEY", "") DEFAULT_HF_TOKEN = os.getenv("HF_TOKEN", "") def generate_unique_id() -> str: """Génère un ID unique de 50 caractères""" # Générer deux UUID4 et les combiner uuid1 = str(uuid.uuid4()) uuid2 = str(uuid.uuid4()) # Prendre les 25 premiers caractères de chaque UUID return f"{uuid1[:25]}{uuid2[:25]}" def check_duplicates(entries: List[Dict]) -> List[Dict]: """Vérifie et supprime les doublons dans les entrées du dataset""" seen_questions = set() seen_images = set() unique_entries = [] for entry in entries: # Créer une clé unique basée sur la question et l'image question_key = entry['query'].lower().strip() image_key = entry['image'] # Vérifier si la question est similaire à une question existante is_duplicate = False for seen_question in seen_questions: # Calculer la similarité entre les questions if question_key in seen_question or seen_question in question_key: is_duplicate = True break # Vérifier si l'image a déjà été utilisée trop de fois image_count = sum(1 for e in unique_entries if e['image'] == image_key) if not is_duplicate and image_count < 5: # Limite à 5 questions par image seen_questions.add(question_key) seen_images.add(image_key) unique_entries.append(entry) return unique_entries def process_files(api_key: str, hf_token: str, files: List[str], dataset_name: str, progress=gr.Progress()) -> str: """ Traite les fichiers (images ou PDFs) et crée le dataset """ try: print(f"🔍 Début du traitement avec {len(files)} fichiers") # Dossier temporaire pour toutes les images avant mélange with tempfile.TemporaryDirectory() as temp_images_dir: print("📂 Création du dossier temporaire pour les images") # Compteur pour numéroter les images image_counter = 1 # Liste pour stocker tous les chemins d'images all_images = [] # Traiter d'abord tous les PDFs print("📄 Traitement des PDFs...") for file in files: if file.name.lower().endswith('.pdf'): print(f"📄 Conversion du PDF: {file.name}") try: # Ouvrir le PDF avec PyMuPDF pdf_document = fitz.open(file.name) # Convertir chaque page en image for page_num in range(len(pdf_document)): page = pdf_document[page_num] pix = page.get_pixmap() image_path = os.path.join(temp_images_dir, f"{image_counter}.png") pix.save(image_path) all_images.append(image_path) print(f" 📄 Page {page_num + 1} convertie en {image_counter}.png") image_counter += 1 pdf_document.close() except Exception as e: print(f"❌ Erreur lors de la conversion du PDF {file.name}: {str(e)}") continue # Traiter ensuite toutes les images print("🖼️ Traitement des images...") for file in files: file_lower = file.name.lower() if file_lower.endswith(('.png', '.jpg', '.jpeg')): try: # Copier et renommer l'image new_path = os.path.join(temp_images_dir, f"{image_counter}.png") # Convertir en PNG si nécessaire if file_lower.endswith(('.jpg', '.jpeg')): img = Image.open(file.name) img.save(new_path, 'PNG') else: shutil.copy2(file.name, new_path) all_images.append(new_path) print(f"🖼️ Image {file.name} copiée en {image_counter}.png") image_counter += 1 except Exception as e: print(f"❌ Erreur lors du traitement de l'image {file.name}: {str(e)}") continue if not all_images: return "❌ Erreur: Aucune image valide trouvée. Veuillez fournir des fichiers PDF ou des images (PNG, JPG, JPEG)." # Mélanger toutes les images print(f"🔄 Mélange des {len(all_images)} images...") import random random.shuffle(all_images) # Créer un nouveau dossier pour les images mélangées et renumérotées with tempfile.TemporaryDirectory() as final_images_dir: final_images = [] for i, image_path in enumerate(all_images, 1): new_path = os.path.join(final_images_dir, f"{i}.png") shutil.copy2(image_path, new_path) final_images.append(new_path) print(f"📝 Image {os.path.basename(image_path)} renumérotée en {i}.png") print(f"✅ Total des images à traiter: {len(final_images)}") # Continuer avec le traitement des images return process_images(api_key, hf_token, final_images, dataset_name, progress) except Exception as e: return f"❌ Erreur lors du traitement des fichiers: {str(e)}" def process_images(api_key, hf_token, images, dataset_name, progress=gr.Progress()): """ Traite les images et crée le dataset """ try: print(f"🔍 Début du traitement avec {len(images)} images") print(f"🔑 API Key: {api_key[:5]}...") print(f"🔑 HF Token: {hf_token[:5]}...") print(f"📁 Dataset name: {dataset_name}") if not api_key or not hf_token: print("❌ Erreur: API Key ou HF Token manquant") return "❌ Erreur: Veuillez entrer votre clé API Google Gemini et votre token Hugging Face." # Configuration de l'API Gemini print("⚙️ Configuration de l'API Gemini...") genai.configure(api_key=api_key) model = genai.GenerativeModel('gemini-1.5-flash') # Créer d'abord le repository sur Hugging Face try: print("📦 Création du repository sur Hugging Face...") api = HfApi(token=hf_token) create_repo(dataset_name, repo_type="dataset", token=hf_token, exist_ok=True) print("✅ Repository créé avec succès") except Exception as e: print(f"❌ Erreur lors de la création du repository: {str(e)}") return f"❌ Erreur lors de la création du repository sur Hugging Face: {str(e)}" # Création d'un dossier temporaire pour le dataset with tempfile.TemporaryDirectory() as temp_dir: print(f"📂 Création du dossier temporaire: {temp_dir}") # Extraire le nom du dataset sans le nom d'utilisateur repo_name = dataset_name.split('/')[-1] dataset_path = Path(temp_dir) / repo_name dataset_path.mkdir() # Création des dossiers pour les splits splits = ['train', 'validation', 'test'] for split in splits: (dataset_path / split / 'images').mkdir(parents=True) print(f"📁 Création du dossier {split}") # Mélanger les images aléatoirement import random random.shuffle(images) total_images = len(images) progress(0, desc="Démarrage du traitement...") # Créer un dossier temporaire pour les images renommées with tempfile.TemporaryDirectory() as renamed_images_dir: renamed_images = [] print("🔄 Renommage des images...") # Renommer et copier toutes les images d'abord for i, image in enumerate(images, 1): new_image_path = Path(renamed_images_dir) / f"{i}.png" shutil.copy2(image, new_image_path) renamed_images.append(str(new_image_path)) print(f"📝 Image {image} renommée en {i}.png") # Traitement des images renommées for i, image in enumerate(renamed_images, 1): print(f"\n🖼️ Traitement de l'image {i}/{total_images}") progress(i / total_images, desc=f"Traitement de l'image {i}/{total_images}") # Déterminer le split (80% train, 10% validation, 10% test) if i <= len(images) * 0.8: split = 'train' elif i <= len(images) * 0.9: split = 'validation' else: split = 'test' print(f"📂 Split: {split}") # Copier l'image image_path = dataset_path / split / 'images' / f"{i}.png" print(f"📄 Copie de l'image vers: {image_path}") shutil.copy2(image, image_path) # Générer les questions/réponses avec Gemini all_qa_pairs = [] # Une seule tentative par image with open(image, 'rb') as img_file: img_data = img_file.read() # Générer un nombre aléatoire de questions à poser nb_questions = random.randint(1, 5) print(f"❓ Génération de {nb_questions} questions...") prompt = f"""Tu es un expert en analyse financière et en création de datasets. Examine attentivement ce document financier et génère exactement {nb_questions} questions/réponses en français, quelle que soit la langue du document. ÉTAPE 1 - IDENTIFICATION DE LA LANGUE DU DOCUMENT : - Analyse le texte dans l'image - Identifie la langue principale (fr, en, de, etc.) - Cette information servira uniquement de métadonnée IMPORTANT : Toutes les questions et réponses DOIVENT être en français, même si le document est dans une autre langue ! Format de réponse requis (JSON) : [ {{ "query": "Question financière en français", "answer": "Réponse en français", "langue_document": "code ISO de la langue source (fr, en, de, etc.)", "is_negative": false }} ] Instructions pour la création du dataset : 1. QUESTIONS FINANCIÈRES : - Analyse des montants, ratios et variations - Stratégies et objectifs financiers - Risques et opportunités - Dates et échéances importantes 2. QUESTIONS NÉGATIVES : - Au moins 1 question sur {nb_questions} où l'information n'est PAS dans le document - Pour ces questions, mettre "is_negative": true - La réponse DOIT commencer par "Cette information ne figure pas dans le document" 3. RÈGLES STRICTES : - Questions et réponses TOUJOURS en français - Questions précises et non ambiguës - Pas de répétitions - Format JSON strict - Au moins 1 question négative par image La réponse doit être uniquement le JSON, sans texte supplémentaire.""" response = model.generate_content([ prompt, {"mime_type": "image/png", "data": img_data} ]) try: # Nettoyer la réponse pour ne garder que le JSON response_text = response.text.strip() if response_text.startswith("```json"): response_text = response_text[7:] if response_text.endswith("```"): response_text = response_text[:-3] response_text = response_text.strip() # Extraire et formater les Q/R qa_pairs = json.loads(response_text) # Vérifier que c'est une liste if not isinstance(qa_pairs, list): raise ValueError("La réponse n'est pas une liste") # Vérifier qu'il y a au moins une question négative has_negative = False for qa in qa_pairs: if qa.get("is_negative", False): if not qa["answer"].startswith("Cette information ne figure pas dans le document"): qa["answer"] = "Cette information ne figure pas dans le document. " + qa["answer"] has_negative = True if not has_negative: print("⚠️ Attention: Aucune question négative générée, on réessaie...") continue # Vérifier que chaque élément a les bons champs for qa in qa_pairs: if not all(key in qa for key in ["query", "answer", "langue_document", "is_negative"]): raise ValueError("Un élément ne contient pas tous les champs requis") # Vérifier que la langue est fr if qa["langue_document"] != "fr": qa["langue_document"] = "fr" # Générer un ID unique qa["id"] = generate_unique_id() # Ajouter le chemin de l'image avec le nouveau nom qa["image"] = f"images/{i}.png" qa["file_name"] = f"images/{i}.png" all_qa_pairs.extend(qa_pairs) except json.JSONDecodeError as e: print(f"Erreur JSON: {str(e)}") continue except ValueError as e: print(f"Erreur de format: {str(e)}") continue # Vérifier et supprimer les doublons unique_qa_pairs = check_duplicates(all_qa_pairs) # Créer les entrées pour le JSONL for qa in unique_qa_pairs: entry = { "id": qa["id"], "image": qa["image"], "query": qa["query"], "answer": qa["answer"], "langue_document": qa["langue_document"], "file_name": qa["file_name"], "is_negative": qa["is_negative"] } # Ajouter au fichier JSONL correspondant jsonl_path = dataset_path / split / "metadata.jsonl" with open(jsonl_path, 'a', encoding='utf-8') as f: f.write(json.dumps(entry, ensure_ascii=False) + '\n') # Créer le fichier README.md (model card) with open(dataset_path / "README.md", 'w', encoding='utf-8') as f: f.write(f"""--- license: apache-2.0 task_categories: - document-question-answering - visual-question-answering language: - fr tags: - finance - vlm - document-ai - question-answering pretty_name: {dataset_name.split('/')[-1]} size_categories: - n<1K --- # {dataset_name.split('/')[-1]} ## Description Ce dataset a été créé pour l'entraînement de modèles Vision-Langage (VLM) spécialisés dans l'analyse de documents financiers. Il a été généré automatiquement en utilisant l'API Google Gemini pour analyser des documents financiers et produire des questions/réponses pertinentes en français. ## Objectif L'objectif de ce dataset est de permettre l'entraînement de mini-modèles VLM spécialisés dans les tâches financières, en leur permettant d'atteindre des performances proches des grands modèles comme GPT-4V ou Gemini, mais avec une empreinte plus légère et une spécialisation métier. ## Caractéristiques - **Langue** : Questions et réponses en français - **Domaine** : Finance et analyse de documents financiers - **Format** : Images (PNG) + métadonnées (JSONL) - **Types de questions** : - Analyse quantitative (montants, ratios, variations) - Analyse qualitative (stratégies, risques, opportunités) - Questions négatives (informations non présentes) - **Structure** : - Train (80%) - Validation (10%) - Test (10%) ## Métadonnées Chaque entrée du dataset contient : - Un ID unique - Le chemin de l'image - Une question en français - La réponse correspondante - La langue source du document - Un indicateur de question négative ## Génération Ce dataset a été généré automatiquement en utilisant : 1. L'API Google Gemini pour l'analyse des documents 2. Un prompt spécialisé pour la génération de questions/réponses financières 3. Un système de validation pour assurer la qualité et la cohérence ## Utilisation Ce dataset est particulièrement adapté pour : - L'entraînement de mini-modèles VLM spécialisés en finance - Le fine-tuning de modèles existants pour des tâches financières - L'évaluation de modèles sur des tâches de compréhension de documents financiers ## Citation Si vous utilisez ce dataset, veuillez citer : ```bibtex @misc{{dataset-{dataset_name.split('/')[-1]}, author = {{Martial ROBERGE}}, title = {{{dataset_name.split('/')[-1]}}}, year = {{2024}}, publisher = {{Hugging Face}}, organization = {{Lexia France}}, contact = {{martial@lexiapro.fr}} }} ``` ## Création Dataset créé par Martial ROBERGE (Lexia France) en utilisant [Mini-VLM Dataset Builder](https://huggingface.co/spaces/Marsouuu/french-visual-dataset-builder-v1). ## Licence Ce dataset est distribué sous licence Apache 2.0. """) # Créer le fichier LICENSE with open(dataset_path / "LICENSE", 'w') as f: f.write(""" Apache License Version 2.0, January 2004 http://www.apache.org/licenses/ Copyright 2024 Martial ROBERGE - Lexia France Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License. """) progress(0.9, desc="Upload du dataset sur Hugging Face...") try: # Uploader le dataset api.upload_folder( folder_path=str(dataset_path), repo_id=dataset_name, repo_type="dataset" ) progress(1.0, desc="Terminé !") return f"✅ Dataset créé avec succès !\n\nAccédez à votre dataset : https://huggingface.co/datasets/{dataset_name}" except Exception as e: return f"❌ Erreur lors de l'upload du dataset sur Hugging Face: {str(e)}" except Exception as e: return f"❌ Erreur: {str(e)}" # Interface Gradio with gr.Blocks() as demo: gr.Markdown(""" # 🎯 Mini-VLM Dataset Builder ## Créateur de datasets financiers en français pour mini-modèles VLM Cette application permet de créer des datasets de questions/réponses en français à partir de documents financiers (en français ou autres langues) pour entraîner des modèles Vision-Langage (VLM) légers et performants. ### 🔄 Processus : 1. Upload de documents (PDF, images) 2. Analyse automatique par Gemini 3. Génération de Q/R en français 4. Création du dataset sur Hugging Face ### ⚠️ Prérequis : - [Clé API Gemini](https://makersuite.google.com/app/apikey) - [Token Hugging Face](https://huggingface.co/settings/tokens) """) with gr.Row(): with gr.Column(scale=1): api_key = gr.Textbox( label="🔑 Clé API Google Gemini", type="password", placeholder="Entrez votre clé API Gemini", value="" ) hf_token = gr.Textbox( label="🔑 Token Hugging Face", type="password", placeholder="Entrez votre token Hugging Face", value="" ) dataset_name = gr.Textbox( label="📁 Nom du dataset", placeholder="votre-username/nom-du-dataset", info="Format : username/nom-du-dataset (ex: marsouuu/finance-dataset-fr)" ) with gr.Column(scale=1): files = gr.File( label="📄 Documents financiers (PDF, PNG, JPG, JPEG)", file_count="multiple", height=200 ) gr.Markdown(""" ### 📊 Caractéristiques : - Questions et réponses en français - 1 à 5 Q/R par document - Questions négatives incluses - Split train/val/test automatique """) submit_btn = gr.Button("🚀 Créer le dataset", variant="primary", scale=2) output = gr.Textbox( label="📝 Résultat", lines=3, interactive=False ) submit_btn.click( fn=process_files, inputs=[api_key, hf_token, files, dataset_name], outputs=output ) if __name__ == "__main__": demo.launch() else: # Pour Hugging Face Spaces app = demo