Upload prompt_to.py

prompt_to.py

@@ -0,0 +1,500 @@
+import os
+import sqlite3
+import json
+import pandas as pd
+import re
+from agno.agent import Agent
+from agno.models.groq import Groq
+import streamlit as st
+from dotenv import load_dotenv
+import graphviz
+from clean_database import clean_database
+
+# Charger les variables d'environnement depuis le fichier .env
+load_dotenv()
+
+# Fonctions de normalisation de la base de données
+def detecter_format(fichier):
+    extension = os.path.splitext(fichier)[1].lower()
+    if extension in [".db", ".sqlite"]:
+        return "sqlite"
+    elif extension == ".json":
+        return "json"
+    elif extension == ".csv":
+        return "csv"
+    elif extension in [".xls", ".xlsx"]:
+        return "excel"
+    else:
+        return "inconnu"
+
+def json_to_sqlite(fichier_json):
+    with open(fichier_json, 'r', encoding='utf-8') as file:
+        data = json.load(file)
+    db_name = 'temp_json.db'
+    conn = sqlite3.connect(db_name)
+    cursor = conn.cursor()
+    table_name = "data"
+    columns = ', '.join(data[0].keys())
+    placeholders = ', '.join('?' * len(data[0]))
+    cursor.execute(f"CREATE TABLE IF NOT EXISTS {table_name} ({columns})")
+    for row in data:
+        cursor.execute(f"INSERT INTO {table_name} ({columns}) VALUES ({placeholders})", tuple(row.values()))
+    conn.commit()
+    conn.close()
+    return db_name
+
+def csv_to_sqlite(fichier_csv):
+    df = pd.read_csv(fichier_csv, encoding='utf-8')
+    db_name = 'temp_csv.db'
+    conn = sqlite3.connect(db_name)
+    df.to_sql('data', conn, if_exists='replace', index=False)
+    conn.close()
+    return db_name
+
+def excel_to_sqlite(fichier_excel):
+    df = pd.read_excel(fichier_excel, encoding='utf-8')
+    db_name = 'temp_excel.db'
+    conn = sqlite3.connect(db_name)
+    df.to_sql('data', conn, if_exists='replace', index=False)
+    conn.close()
+    return db_name
+
+def preparer_bdd(input_path):
+    format_detecte = detecter_format(input_path)
+    if format_detecte == "sqlite":
+        return input_path
+    elif format_detecte == "json":
+        return json_to_sqlite(input_path)
+    elif format_detecte == "csv":
+        return csv_to_sqlite(input_path)
+    elif format_detecte == "excel":
+        return excel_to_sqlite(input_path)
+    else:
+        st.error("Format non supporté. Utilisez un fichier .db, .json, .csv, .xls ou .xlsx")
+        return None
+
+def extraire_bdd(db_path):
+    with sqlite3.connect(db_path) as conn:
+        cursor = conn.cursor()
+        tables = [table[0] for table in cursor.execute(
+            "SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%';"
+        ).fetchall()]
+        if not tables:
+            st.warning("Aucune table trouvée.")
+            return ""
+        output_bdd = ""
+        for table in tables:
+            output_bdd += f"\n📌 Table `{table}` :\n"
+            cursor.execute(f"PRAGMA table_info({table});")
+            columns = cursor.fetchall()
+            col_names = [col[1] for col in columns]
+            col_types = [col[2] for col in columns]
+            output_bdd += f"📝 Colonnes: {', '.join(f'{name} ({ctype})' for name, ctype in zip(col_names, col_types))}\n"
+            cursor.execute(f"SELECT * FROM {table}")
+            rows = cursor.fetchall()
+            if rows:
+                output_bdd += "\n".join(f" -> {row}" for row in rows) + "\n"
+            else:
+                output_bdd += "⚠️ Aucune donnée trouvée.\n"
+        return output_bdd
+
+# Création de l'agent Groq pour la vérification de la normalisation
+normalization_checker = Agent(
+    model=Groq(id="deepseek-r1-distill-llama-70b", temperature=0.0, top_p=0),
+    description="Vérifie si une base de données est normalisée.",
+    instructions=[
+        "Analyse la structure de la base de données fournie en entrée.",
+        "Détermine si elle respecte les formes normales (1NF, 2NF, 3NF, BCNF).",
+        "Migre les données de l'ancienne base de données vers la nouvelle base de données normalisée sans générer de code SQL.",
+        "Assurez-vous que toutes les données sont correctement transférées et que les relations sont maintenues.",
+        "Si la base n'est pas normalisée, propose une version améliorée sans générer de schéma SQL."
+    ],
+    markdown=True
+)
+
+def analyser_bdd(output_bdd: str):
+    prompt = f"""
+    Voici la structure et les données de la base SQLite :
+    {output_bdd}
+    Tu es un expert en base de données exécute l'algorithme suivant pour normaliser et migrer la base de données. Affiche **uniquement** le résultat final sans explication détaillée et Ne pas ajouter de nouvelles colonnes non mentionnées dans la base de données d'origine.
+    ---
+    ### **Algorithme de normalisation et migration**
+    Début
+       Initialiser une variable `resultat_final` pour accumuler les résultats.
+       # Analyser la structure actuelle de la base de données
+       Pour chaque table `T` dans la base de données faire
+          Détecter la clé primaire existante ou en attribuer une si nécessaire.
+          # Appliquer la 1NF : Atomisation des attributs
+          Si `T` contient des attributs non atomiques alors
+             Transformer les attributs en valeurs atomiques.
+             Assigner les clés primaires et ajouter les clés étrangères si nécessaire.
+          Fin Si
+          # Appliquer la 2NF : Éliminer les dépendances fonctionnelles partielles
+          Si `T` contient une clé primaire composite et des attributs qui dépendent d’une partie seulement de la clé alors
+             Décomposer `T` en nouvelles tables en respectant la 2NF.
+             Assigner les clés primaires et ajouter les clés étrangères si nécessaire.
+          Fin Si
+          # Appliquer la 3NF : Éliminer les dépendances transitives
+          Si `T` contient des dépendances transitives (un attribut dépend d’un autre attribut non clé) alors
+             Décomposer `T` en nouvelles tables pour isoler ces dépendances.
+             Assigner les clés primaires et ajouter les clés étrangères si nécessaire.
+          Fin Si
+          # Appliquer la BCNF : Suppression des dépendances anormales
+          Si `T` contient une dépendance fonctionnelle où un attribut non clé détermine une clé candidate alors
+             Décomposer `T` en nouvelles relations respectant la BCNF.
+             Assigner les clés primaires et ajouter les clés étrangères si nécessaire.
+          Fin Si
+          # Migrer les données
+          Pour chaque nouvelle table normalisée faire
+             Identifier les données à migrer depuis l'ancienne base de données.
+             Transformer ou réorganiser les données selon le schéma normalisé proposé.
+             Insérer les données transformées dans la nouvelle table normalisée.
+          Fin Pour
+          # Le résultat doit absolument respecter cette structure.
+          Ajouter "📝 Nouvelles tables proposées :" à `resultat_final`.
+          Pour chaque nouvelle table `N` créée faire
+             Ajouter "📌 Table `nom_nouvelle_table` :" à `resultat_final`.
+             Ajouter "📝 Colonnes: colonne1 (type1), colonne2 (type2), ..." à `resultat_final`.
+             Ajouter "🔑 Clé primaire: `colonne_PK`" si définie.
+             Ajouter "🔗 Clé étrangère: `colonne_FK` → `table_referencée` (`colonne_referencée`)" si applicable.
+             Ajouter "📋 Données :" à `resultat_final`.
+             Pour chaque enregistrement migré dans la table faire
+                Ajouter " - `valeur1`, `valeur2`, ..." à `resultat_final`.
+             Fin Pour
+          Fin Pour
+       # Vérification finale avant affichage
+       Afficher `resultat_final`
+       Afficher "✅ Normalisation complète et migration réussie."
+    Fin
+    ---
+    """
+    response = normalization_checker.run(prompt)
+    resultat = response.content.strip()
+    resultat_sans_think = re.sub(r"<think>.*?</think>", "", resultat, flags=re.DOTALL).strip()
+    return resultat_sans_think
+
+# Création de l'agent 2 (vérifie la proposition de normalisation)
+normalization_validator = Agent(
+    model=Groq(id="qwen-qwq-32b", temperature=0.0, top_p=0),
+    description="Vérifie si la base de données normalisée proposée est correcte.",
+    instructions=[
+        "Analyse la normalisation proposée et vérifie si elle respecte les formes normales.",
+        "Compare la proposition avec la base de données générée pour s'assurer de leur correspondance.",
+        "Donne une proposition améliorée si nécessaire."
+    ],
+    markdown=True
+)
+
+def verifier_normalisation(proposition_normalisee: str, output_bdd: str):
+    prompt = f"""
+    Voici la base de données générée après application de la normalisation :
+    Vérifie si la normalisation proposée correspond bien à {output_bdd}.
+    Voici une proposition de base de données normalisée :
+    {proposition_normalisee}
+    - Tu es un expert en base de données exécute l'algorithme suivant pour vérifier si la base obtenue correspond bien à la normalisation attendue. Affiche **uniquement** le résultat final sans explication détaillée.
+    ---
+    ### **Algorithme de vérification et correction des formes normales**
+    Début
+       Initialiser une variable `corrections_appliquees` = False
+       Pour chaque table dans la base de données faire
+          Si tous les attributs sont atomiques alors
+             printf("📌 Vérification de la table `nom_table`")
+             printf("✅ La table `nom_table` est en 1NF")
+             Si la table ne contient pas de dépendances fonctionnelles partielles alors
+                printf("✅ La table `nom_table` est en 2NF")
+                Si la table ne contient pas de dépendances transitives alors
+                   printf("✅ La table `nom_table` est en 3NF")
+                   Si chaque dépendance fonctionnelle est basée sur une clé candidate alors
+                      printf("✅ La table `nom_table` est en BCNF")
+                   Sinon
+                      printf("❌ La table `nom_table` ne respecte pas la BCNF → Correction appliquée.")
+                      Appliquer la correction en décomposant la table en relations BCNF.
+                      corrections_appliquees = True
+                   Fin Si
+                Sinon
+                   printf("❌ La table `nom_table` ne respecte pas la 3NF → Correction appliquée.")
+                   Appliquer la correction en éliminant les dépendances transitives.
+                   corrections_appliquees = True
+                Fin Si
+             Sinon
+                printf("❌ La table `nom_table` ne respecte pas la 2NF → Correction appliquée.")
+                Appliquer la correction en éliminant les dépendances partielles.
+                corrections_appliquees = True
+             Fin Si
+          Sinon
+             printf("❌ La table `nom_table` ne respecte pas la 1NF → Correction appliquée.")
+             Appliquer la correction en atomisant les attributs.
+             corrections_appliquees = True
+          Fin Si
+       Fin Pour
+       Si corrections_appliquees == True alors
+          printf("⚠️ Des corrections ont été appliquées durant la vérification.")
+       Fin Si
+       printf("🔍 Normalisation terminée.")
+    Fin
+    ---
+    """
+    response = normalization_validator.run(prompt)
+    resultat = response.content.strip()
+    resultat_sans_think = re.sub(r"<think>.*?</think>", "", resultat, flags=re.DOTALL).strip()
+    return resultat_sans_think
+
+# Création de l'agent 3 (crée la nouvelle bdd normalisée)
+sql_generator = Agent(
+    model=Groq(id="whisper-large-v3", temperature=0.0, top_p=0),
+    description="Crée la nouvelle bdd normalisée.",
+    instructions=[
+        "Analyse le résultat de normalisation et génère les requêtes SQL pour créer la nouvelle bdd."
+    ],
+    markdown=True
+)
+
+def génération_des_requêtes(proposition_normalisee: str):
+    prompt = f"""
+    Voici une proposition de base de données normalisée :
+    {proposition_normalisee}
+    Génère uniquement les requêtes SQL pour créer et insérer des données dans la nouvelle base de données normalisée (crée les nouvelle tables et insérée tous les donnez et n'ajoutes rien).
+    """
+    response = sql_generator.run(prompt)
+    resultat = response.content.strip()
+    resultat_sans_think = re.sub(r"<think>.*?</think>", "", resultat, flags=re.DOTALL).strip()
+    return resultat_sans_think
+
+def nettoyer_requetes_sql(texte):
+    # Expression régulière pour capturer les requêtes CREATE TABLE et INSERT INTO
+    pattern = r"(CREATE TABLE.*?;|INSERT INTO.*?;)"
+    requetes_sql = re.findall(pattern, texte, re.DOTALL | re.IGNORECASE)
+    return "\n\n".join(req.strip() for req in requetes_sql)
+
+# --- Fonctions utilitaires ---
+def create_sqlite_db_from_file(sql_file, db_name):
+    if os.path.exists(db_name):
+        os.remove(db_name)
+
+    conn = sqlite3.connect(db_name)
+    cursor = conn.cursor()
+
+    with open(sql_file, 'r') as file:
+        sql_script = file.read()
+
+    cursor.executescript(sql_script)
+    conn.commit()
+    conn.close()
+
+def get_database_structure(db_name):
+    conn = sqlite3.connect(db_name)
+    cursor = conn.cursor()
+
+    # Récupérer les tables
+    cursor.execute("SELECT name FROM sqlite_master WHERE type='table';")
+    tables = cursor.fetchall()
+
+    structure = ""
+    for table in tables:
+        table_name = table[0]
+        structure += f"Table: {table_name}\n"
+
+        # Récupérer les colonnes
+        cursor.execute(f"PRAGMA table_info({table_name});")
+        columns = cursor.fetchall()
+        for column in columns:
+            structure += f"  Colonne: {column[1]} (Type: {column[2]})\n"
+
+    conn.close()
+    return structure
+
+def get_sample_data(db_name, limit=3):
+    conn = sqlite3.connect(db_name)
+    cursor = conn.cursor()
+    sample_text = ""
+    cursor.execute("SELECT name FROM sqlite_master WHERE type='table';")
+    tables = cursor.fetchall()
+    for table in tables:
+        table_name = table[0]
+        sample_text += f"\nExtrait de la table {table_name} :\n"
+        df = pd.read_sql_query(f"SELECT * FROM {table_name} LIMIT {limit}", conn)
+        sample_text += df.to_string(index=False)
+        sample_text += "\n"
+    conn.close()
+    return sample_text
+
+def transform_prompt_to_sql(prompt, db_structure):
+    instructions = f"""
+    Voici la structure de la base de données :
+    {db_structure}
+    Voici la requête de l'utilisateur :
+    {prompt}
+
+    Transforme ce prompt en une requête SQL en utilisant la structure de la base de données fournie.
+    """
+    response = sql_transformer.run(instructions)
+    return response.content.strip()
+
+def clean_sql_query(response):
+    # Utiliser une expression régulière pour extraire uniquement la requête SQL
+    match = re.search(r'```sql\s*(.*?)\s*```', response, re.DOTALL)
+    if match:
+        return match.group(1).strip()
+    else:
+        raise ValueError("Impossible d'extraire la requête SQL de la réponse.")
+
+def execute_sql_query(db_name, sql_query):
+    conn = sqlite3.connect(db_name)
+    cursor = conn.cursor()
+    cursor.execute(sql_query)
+    results = cursor.fetchall()
+    conn.close()
+    return results
+
+def visualize_database_schema_from_sql(sql_file):
+    with open(sql_file, 'r') as file:
+        sql_script = file.read()
+
+    # Extraction des tables et colonnes
+    tables = {}
+    pks = {}
+    fks = []
+
+    create_table_pattern = re.compile(r'CREATE TABLE\s+(\w+)\s*\((.*?)\);', re.DOTALL | re.IGNORECASE)
+    fk_pattern = re.compile(r'FOREIGN KEY\s*\((\w+)\)\s*REFERENCES\s+(\w+)\s*\((\w+)\)', re.IGNORECASE)
+    pk_inline_pattern = re.compile(r'^(\w+)\s+\w+.*PRIMARY KEY', re.IGNORECASE)
+    pk_constraint_pattern = re.compile(r'PRIMARY KEY\s*\((.*?)\)', re.IGNORECASE)
+
+
+    for match in create_table_pattern.finditer(sql_script):
+        table_name = match.group(1)
+        body = match.group(2)
+        lines = [line.strip() for line in body.split(',') if line.strip()]
+        columns = []
+        primary_keys = []
+
+        for line in lines:
+            if line.upper().startswith("FOREIGN KEY"):
+                fk_match = fk_pattern.search(line)
+                if fk_match:
+                    source_col = fk_match.group(1)
+                    ref_table = fk_match.group(2)
+                    ref_col = fk_match.group(3)
+                    fks.append((table_name, source_col, ref_table, ref_col))
+            elif line.upper().startswith("PRIMARY KEY"):
+                pk_match = pk_constraint_pattern.search(line)
+                if pk_match:
+                    pk_cols = [pk.strip() for pk in pk_match.group(1).split(',')]
+                    primary_keys.extend(pk_cols)
+            else:
+                parts = line.split()
+                if len(parts) == 0:
+                    continue  # ligne vide ou espace
+                col_name = parts[0].strip("()")  # on enlève les parenthèses résiduelles
+                if col_name.isidentifier():  # vérifie que c’est un nom valide
+                    if col_name not in columns:  # Vérifie si la colonne n'est pas déjà présente
+                        columns.append(col_name)
+                    inline_pk = pk_inline_pattern.match(line)
+                    if inline_pk:
+                        if inline_pk.group(1) not in primary_keys:  # Vérifie si la clé primaire n'est pas déjà présente
+                            primary_keys.append(inline_pk.group(1))
+
+        tables[table_name] = columns
+        pks[table_name] = primary_keys
+
+
+    # Construction du graphe avec style MERISE
+    dot = graphviz.Digraph(format='png')
+    dot.attr('node', shape='plaintext')
+
+    for table, columns in tables.items():
+        rows = []
+        for col in columns:
+            if col in pks.get(table, []):
+                rows.append(f'<TR><TD ALIGN="LEFT"><U>{col}</U></TD></TR>')
+            else:
+                rows.append(f'<TR><TD ALIGN="LEFT">{col}</TD></TR>')
+
+        label = f"""<<TABLE BORDER="1" CELLBORDER="0" CELLSPACING="0">
+            <TR><TD BGCOLOR="lightgray"><B>{table}</B></TD></TR>
+            {''.join(rows)}
+        </TABLE>>"""
+
+        dot.node(table, label=label)
+
+    for src_table, src_col, tgt_table, tgt_col in fks:
+        dot.edge(src_table, tgt_table, label=f"{src_col} → {tgt_col}")
+
+    st.graphviz_chart(dot)
+
+# --- Config Agent ---
+sql_transformer = Agent(
+    model=Groq(id="llama3-70b-8192", temperature=0.0, top_p=0),
+    description="Transforme une requête en langage naturel en requête SQL.",
+    instructions=[
+        "Analyse la requête fournie en entrée.",
+        "Transforme la requête en langage naturel en une requête SQL.",
+    ],
+    markdown=True
+)
+
+# --- Interface Streamlit ---
+st.set_page_config(page_title="Assistant SQL avec LLM", layout="wide")
+st.title("🧠📊 Assistant SQL intelligent (avec LLM)")
+
+db_name = "database.db"
+sql_file = "output.sql"
+
+# Champ de saisie pour le fichier d'entrée
+input_file = st.file_uploader("Choisissez un fichier (.db, .json, .csv, .xls ou .xlsx)", type=["db", "sqlite", "json", "csv", "xls", "xlsx"])
+
+if input_file:
+    # Enregistrer le fichier téléchargé temporairement
+    temp_file_path = f"temp_{input_file.name}"
+    with open(temp_file_path, "wb") as f:
+        f.write(input_file.getbuffer())
+
+    prepared_db_path = preparer_bdd(temp_file_path)
+    if prepared_db_path:
+        st.success(f"Base de données prête : {prepared_db_path}")
+        clean_database(prepared_db_path)
+        st.success("Base de données nettoyée.")
+        output_bdd = extraire_bdd(prepared_db_path)
+        if output_bdd:
+            proposition_normalisee = output_bdd
+            while True:
+                resultat = analyser_bdd(proposition_normalisee)
+                verification = verifier_normalisation(resultat, output_bdd)
+                if "⚠️ Des corrections ont été appliquées durant la vérification." in verification:
+                    proposition_normalisee = verification
+                else:
+                    st.text("\n🔍 Résultat final de l'analyse :\n" + resultat)
+                    st.text("\n✅ Vérification de la normalisation :\n" + verification)
+                    break
+
+            # Générer les requêtes SQL pour la nouvelle base de données normalisée
+            sql_content = génération_des_requêtes(resultat)
+            requetes_nettoyees = nettoyer_requetes_sql(sql_content)
+
+            # Écrire les requêtes SQL dans le fichier output.sql
+            with open(sql_file, "w", encoding='utf-8') as fichier_sql:
+                fichier_sql.write(requetes_nettoyees)
+
+            st.success(f"Fichier SQL généré: {sql_file}")
+            create_sqlite_db_from_file(sql_file, db_name)
+
+            # Visualiser le schéma de la base de données
+            st.subheader("Schéma des relations entre les tables")
+            visualize_database_schema_from_sql(sql_file)
+
+# Champ de saisie pour les requêtes en langage naturel
+prompt = st.text_input("💬 Pose ta question :", placeholder="Ex: Liste les patients atteints de diabète")
+
+if prompt:
+    st.write("🛠️ Génération SQL en cours…")
+    structure = get_database_structure(db_name)
+    response = transform_prompt_to_sql(prompt, structure)
+    sql_query = clean_sql_query(response)
+    st.code(sql_query, language="sql")
+
+    try:
+        df = execute_sql_query(db_name, sql_query)
+        st.success("✅ Requête exécutée avec succès !")
+        st.dataframe(df)
+    except Exception as e:
+        st.error(f"❌ Erreur lors de l'exécution de la requête : {e}")